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};
6
7/// Completion event for turn-scoped `soothe.stream.end`.
8pub const TURN_END_STREAM_END: &str = STREAM_END;
9/// Completion event for gated `status=idle`.
10pub const TURN_END_IDLE: &str = "status.idle";
11/// Completion event for `status=stopped` after running.
12pub const TURN_END_STOPPED: &str = "status.stopped";
13
14/// Per-turn progress flags (DaemonSession parity; not shared across chats).
15#[derive(Debug, Default, Clone)]
16pub struct TurnLifecycleGate {
17    /// Saw `status=running` for this turn.
18    pub saw_running: bool,
19    /// Saw any event payload after filtering.
20    pub saw_stream_payload: bool,
21    /// Saw non-intake turn progress (`messages` / step customs).
22    pub saw_turn_progress: bool,
23}
24
25impl TurnLifecycleGate {
26    /// Update gate flags from one decoded inbound message (status or event frame).
27    pub fn observe(&mut self, msg: &Value) {
28        if msg.get("type").and_then(|v| v.as_str()) == Some("status") {
29            if let Some(state) = msg.get("state").and_then(|v| v.as_str()) {
30                self.observe_status(state);
31            }
32            return;
33        }
34        if msg.get("type").and_then(|v| v.as_str()) == Some("event") || msg.get("mode").is_some() {
35            let mode = msg.get("mode").and_then(|v| v.as_str()).unwrap_or("");
36            let data = msg.get("data").cloned().unwrap_or(Value::Null);
37            self.observe_event(mode, &data);
38        }
39    }
40
41    /// Observe a status frame.
42    pub fn observe_status(&mut self, state: &str) {
43        if state.eq_ignore_ascii_case("running") {
44            self.saw_running = true;
45        }
46    }
47
48    /// Observe an event frame.
49    pub fn observe_event(&mut self, mode: &str, data: &Value) {
50        self.saw_stream_payload = true;
51        if is_turn_progress_chunk(mode, data) {
52            self.saw_turn_progress = true;
53        }
54    }
55
56    /// Whether turn-scoped `stream.end` may end the turn.
57    pub fn allow_stream_end(&self) -> bool {
58        self.saw_running && self.saw_turn_progress
59    }
60
61    /// Whether `status=idle` may soft-complete the turn.
62    pub fn allow_idle_complete(&self) -> bool {
63        self.saw_running && self.saw_stream_payload
64    }
65}
66
67/// Applies DaemonSession end rules to pool frames.
68#[derive(Debug, Default)]
69pub struct TurnBoundary {
70    /// Progress gate for this turn.
71    pub gate: TurnLifecycleGate,
72    /// True after a terminal boundary was observed.
73    pub ended: bool,
74    /// Wire completion reason when `ended`.
75    pub reason: String,
76}
77
78impl TurnBoundary {
79    /// Feed a status frame. Returns Some(reason) when the turn ends.
80    pub fn feed_status(&mut self, state: &str) -> Option<&'static str> {
81        if self.ended {
82            return static_reason(&self.reason);
83        }
84        self.gate.observe_status(state);
85        if state.eq_ignore_ascii_case("stopped") && self.gate.saw_running {
86            return Some(self.mark(TURN_END_STOPPED));
87        }
88        if state.eq_ignore_ascii_case("idle") && self.gate.allow_idle_complete() {
89            return Some(self.mark(TURN_END_IDLE));
90        }
91        None
92    }
93
94    /// Feed an event frame. Returns Some(reason) when the turn ends.
95    pub fn feed_event(&mut self, mode: &str, data: &Value) -> Option<&'static str> {
96        if self.ended {
97            return static_reason(&self.reason);
98        }
99        self.gate.observe_event(mode, data);
100        if mode == "custom" && is_turn_end_custom_data(data) && self.gate.allow_stream_end() {
101            return Some(self.mark(TURN_END_STREAM_END));
102        }
103        None
104    }
105
106    fn mark(&mut self, reason: &'static str) -> &'static str {
107        self.ended = true;
108        self.reason = reason.to_string();
109        reason
110    }
111}
112
113fn static_reason(reason: &str) -> Option<&'static str> {
114    match reason {
115        TURN_END_STREAM_END => Some(TURN_END_STREAM_END),
116        TURN_END_IDLE => Some(TURN_END_IDLE),
117        TURN_END_STOPPED => Some(TURN_END_STOPPED),
118        _ => None,
119    }
120}
121
122/// True for TurnBoundary completion_event values (not phase deliverables).
123pub fn is_daemon_turn_end_event(completion_event: &str) -> bool {
124    matches!(
125        completion_event.trim(),
126        TURN_END_STREAM_END | TURN_END_IDLE | TURN_END_STOPPED
127    )
128}
129
130#[cfg(test)]
131mod tests {
132    use super::*;
133    use serde_json::json;
134
135    #[test]
136    fn ignores_pre_running_idle() {
137        let mut b = TurnBoundary::default();
138        assert!(b.feed_status("idle").is_none());
139        b.feed_status("running");
140        b.feed_event(
141            "messages",
142            &json!([{"type":"AIMessageChunk","content":"x"}]),
143        );
144        assert_eq!(b.feed_status("idle"), Some(TURN_END_IDLE));
145    }
146
147    #[test]
148    fn stream_end_requires_running_and_progress() {
149        let mut b = TurnBoundary::default();
150        let end = json!({"type": STREAM_END, "scope": "turn"});
151        assert!(b.feed_event("custom", &end).is_none());
152        b.feed_status("running");
153        assert!(b.feed_event("custom", &end).is_none());
154        b.feed_event("messages", &json!({}));
155        assert_eq!(b.feed_event("custom", &end), Some(TURN_END_STREAM_END));
156    }
157
158    #[test]
159    fn stopped_after_running() {
160        let mut b = TurnBoundary::default();
161        assert!(b.feed_status("stopped").is_none());
162        b.feed_status("running");
163        assert_eq!(b.feed_status("stopped"), Some(TURN_END_STOPPED));
164    }
165
166    #[test]
167    fn daemon_turn_end_event_helper() {
168        assert!(is_daemon_turn_end_event(TURN_END_STREAM_END));
169        assert!(!is_daemon_turn_end_event(
170            "soothe.protocol.message.goal_completion"
171        ));
172    }
173}
soothe_client/appkit/turn_boundary.rs

soothe_client/appkit/
turn_boundary.rs
