soothe_client/appkit/
turn_boundary.rs

1//! DaemonSession turn-end contract for the pool TurnRunner path.
2
3use serde_json::Value;
4
5use crate::stream_terminal::{is_turn_end_custom_data, is_turn_progress_chunk, STREAM_END};
6use crate::turn_boundary::{
7    frame_turn_id, is_idle_terminal_allowed, is_turn_terminal_allowed, parse_turn_generation,
8    turn_ids_match,
9};
10
11/// Completion event for turn-scoped `soothe.stream.end`.
12pub const TURN_END_STREAM_END: &str = STREAM_END;
13/// Completion event for gated `status=idle`.
14pub const TURN_END_IDLE: &str = "status.idle";
15/// Completion event for `status=stopped` after running.
16pub const TURN_END_STOPPED: &str = "status.stopped";
17
18/// Per-turn progress flags (DaemonSession parity; not shared across chats).
19#[derive(Debug, Default, Clone)]
20pub struct TurnLifecycleGate {
21    /// Saw `status=running` for this turn.
22    pub saw_running: bool,
23    /// Saw non-intake turn progress (`messages` / step customs).
24    pub saw_turn_progress: bool,
25    /// Bound turn_id from status=running.
26    pub expected_turn_id: Option<String>,
27    /// Cancellation notice seen.
28    pub cancellation_seen: bool,
29}
30
31impl TurnLifecycleGate {
32    /// Update gate flags from one decoded inbound message (status or event frame).
33    pub fn observe(&mut self, msg: &Value) {
34        if msg.get("type").and_then(|v| v.as_str()) == Some("status") {
35            if let Some(state) = msg.get("state").and_then(|v| v.as_str()) {
36                self.observe_status(state, frame_turn_id(Some(msg)));
37            }
38            return;
39        }
40        if msg.get("type").and_then(|v| v.as_str()) == Some("event") || msg.get("mode").is_some() {
41            let mode = msg.get("mode").and_then(|v| v.as_str()).unwrap_or("");
42            let data = msg.get("data").cloned().unwrap_or(Value::Null);
43            self.observe_event(mode, &data);
44        }
45    }
46
47    /// Observe a status frame.
48    pub fn observe_status(&mut self, state: &str, turn_id: Option<String>) {
49        if state.eq_ignore_ascii_case("running") {
50            self.saw_running = true;
51            if let Some(status_turn) = turn_id {
52                let new_gen = parse_turn_generation(Some(&status_turn));
53                let old_gen = parse_turn_generation(self.expected_turn_id.as_deref());
54                if self.expected_turn_id.is_none()
55                    || (new_gen.is_some()
56                        && (old_gen.is_none() || new_gen.unwrap() >= old_gen.unwrap()))
57                {
58                    if self
59                        .expected_turn_id
60                        .as_ref()
61                        .is_some_and(|e| e != &status_turn)
62                    {
63                        self.saw_turn_progress = false;
64                    }
65                    self.expected_turn_id = Some(status_turn);
66                }
67            }
68        }
69    }
70
71    /// Observe an event frame.
72    pub fn observe_event(&mut self, mode: &str, data: &Value) {
73        if is_turn_progress_chunk(mode, data) {
74            self.saw_turn_progress = true;
75        }
76    }
77
78    /// Whether turn-scoped `stream.end` may end the turn.
79    pub fn allow_stream_end(&self, frame_turn: Option<&str>) -> bool {
80        is_turn_terminal_allowed(
81            self.expected_turn_id.as_deref(),
82            frame_turn,
83            self.saw_running,
84            self.saw_turn_progress,
85        )
86    }
87
88    /// Whether `status=idle` may soft-complete the turn.
89    pub fn allow_idle_complete(&self, frame_turn: Option<&str>) -> bool {
90        is_idle_terminal_allowed(
91            self.expected_turn_id.as_deref(),
92            frame_turn,
93            self.saw_running,
94            self.saw_turn_progress,
95            self.cancellation_seen,
96        )
97    }
98}
99
100/// Applies DaemonSession end rules to pool frames.
101#[derive(Debug, Default)]
102pub struct TurnBoundary {
103    /// Progress gate for this turn.
104    pub gate: TurnLifecycleGate,
105    /// True after a terminal boundary was observed.
106    pub ended: bool,
107    /// Wire completion reason when `ended`.
108    pub reason: String,
109}
110
111impl TurnBoundary {
112    /// Feed a status frame (turn_id omitted — prefer [`Self::feed_status_turn`]).
113    pub fn feed_status(&mut self, state: &str) -> Option<&'static str> {
114        self.feed_status_turn(state, None)
115    }
116
117    /// Feed a status frame with wire `turn_id`.
118    pub fn feed_status_turn(&mut self, state: &str, turn_id: Option<&str>) -> Option<&'static str> {
119        if self.ended {
120            return static_reason(&self.reason);
121        }
122        self.gate
123            .observe_status(state, turn_id.map(|s| s.to_string()));
124        if state.eq_ignore_ascii_case("stopped") && self.gate.saw_running {
125            if self.gate.expected_turn_id.is_some()
126                && !turn_ids_match(self.gate.expected_turn_id.as_deref(), turn_id)
127            {
128                return None;
129            }
130            return Some(self.mark(TURN_END_STOPPED));
131        }
132        if state.eq_ignore_ascii_case("idle") && self.gate.allow_idle_complete(turn_id) {
133            return Some(self.mark(TURN_END_IDLE));
134        }
135        None
136    }
137
138    /// Feed an event frame (outer turn_id omitted).
139    pub fn feed_event(&mut self, mode: &str, data: &Value) -> Option<&'static str> {
140        self.feed_event_turn(mode, data, None)
141    }
142
143    /// Feed an event frame with outer-frame `turn_id`.
144    pub fn feed_event_turn(
145        &mut self,
146        mode: &str,
147        data: &Value,
148        frame_turn: Option<&str>,
149    ) -> Option<&'static str> {
150        if self.ended {
151            return static_reason(&self.reason);
152        }
153        self.gate.observe_event(mode, data);
154        let data_turn = frame_turn_id(Some(data));
155        let tid = data_turn.as_deref().or(frame_turn);
156        if mode == "custom" && is_turn_end_custom_data(data) && self.gate.allow_stream_end(tid) {
157            return Some(self.mark(TURN_END_STREAM_END));
158        }
159        None
160    }
161
162    fn mark(&mut self, reason: &'static str) -> &'static str {
163        self.ended = true;
164        self.reason = reason.to_string();
165        reason
166    }
167}
168
169fn static_reason(reason: &str) -> Option<&'static str> {
170    match reason {
171        TURN_END_STREAM_END => Some(TURN_END_STREAM_END),
172        TURN_END_IDLE => Some(TURN_END_IDLE),
173        TURN_END_STOPPED => Some(TURN_END_STOPPED),
174        _ => None,
175    }
176}
177
178/// True for TurnBoundary completion_event values (not phase deliverables).
179pub fn is_daemon_turn_end_event(completion_event: &str) -> bool {
180    matches!(
181        completion_event.trim(),
182        TURN_END_STREAM_END | TURN_END_IDLE | TURN_END_STOPPED
183    )
184}
soothe_client/appkit/turn_boundary.rs

soothe_client/appkit/
turn_boundary.rs
