soothe_client/
inbound_priority.rs

1//! Inbound frame drop priorities for pending-event overflow handling.
2//!
3//! Lower priority values are kept longer under pressure; higher values are drop
4//! candidates first. Matches Python and Go client behavior.
5
6use serde_json::{Map, Value};
7
8use crate::stream_terminal::{is_turn_end_custom_data, STREAM_END};
9
10/// Never drop: terminals, status, errors.
11pub const DROP_PRIORITY_CRITICAL: i32 = 0;
12/// Prefer keep: cognition / tool batches.
13pub const DROP_PRIORITY_HIGH: i32 = 1;
14/// Default drop candidate: streaming chunks.
15pub const DROP_PRIORITY_NORMAL: i32 = 2;
16
17/// Pending-event cap (Python inbound queue parity).
18pub const DEFAULT_INBOUND_MAX_SIZE: usize = 20_000;
19
20/// Returns drop priority for an inbound frame.
21///
22/// `None` maps to [`DROP_PRIORITY_CRITICAL`] (same as a nil map in Go).
23pub fn inbound_frame_drop_priority(event: Option<&Map<String, Value>>) -> i32 {
24    let Some(event) = event else {
25        return DROP_PRIORITY_CRITICAL;
26    };
27
28    let mut event_type = value_as_str(event.get("type"));
29
30    if event_type == "event_batch" || event_type == "tool_call_updates_batch" {
31        return DROP_PRIORITY_HIGH;
32    }
33
34    if event_type == "next" {
35        if let Some(payload) = event.get("payload").and_then(|v| v.as_object()) {
36            let inner_mode = value_as_str(payload.get("mode"));
37            let inner_data = payload.get("data");
38            if inner_mode == "messages" {
39                if messages_wire_terminal(inner_data) {
40                    return DROP_PRIORITY_CRITICAL;
41                }
42                if let Some(arr) = inner_data.and_then(|v| v.as_array()) {
43                    if let Some(first) = arr.first().and_then(|v| v.as_object()) {
44                        if value_as_str(first.get("phase")) == "goal_completion" {
45                            return DROP_PRIORITY_CRITICAL;
46                        }
47                    }
48                }
49            }
50            if value_as_str(payload.get("type")) == "complete" {
51                return DROP_PRIORITY_CRITICAL;
52            }
53            if let Some(inner) = inner_data.and_then(|v| v.as_object()) {
54                return inbound_frame_drop_priority(Some(inner));
55            }
56            event_type = value_as_str(payload.get("type"));
57        }
58    }
59
60    if event_type == "complete" || event_type == "error" || event_type == "connection_ack" {
61        return DROP_PRIORITY_CRITICAL;
62    }
63
64    if event_type == "status" {
65        let state = value_as_str(event.get("state"));
66        if matches!(state.as_str(), "idle" | "running" | "stopped" | "detached") {
67            return DROP_PRIORITY_CRITICAL;
68        }
69    }
70
71    if event_type == "event" {
72        let mode = value_as_str(event.get("mode"));
73        let data = event.get("data");
74        if mode == "custom" {
75            if let Some(data) = data {
76                if is_turn_end_custom_data(data) {
77                    return DROP_PRIORITY_CRITICAL;
78                }
79                if let Some(m) = data.as_object() {
80                    let custom_type = value_as_str(m.get("type"));
81                    if has_prefix(&custom_type, "soothe.cognition.") {
82                        return DROP_PRIORITY_HIGH;
83                    }
84                    if has_prefix(&custom_type, "soothe.error.") || custom_type == "stream_degraded"
85                    {
86                        return DROP_PRIORITY_CRITICAL;
87                    }
88                    if custom_type == "soothe.ux.stream_tool_wire.tool_call_updates_batch" {
89                        return DROP_PRIORITY_HIGH;
90                    }
91                }
92            }
93        }
94        if mode == "messages" {
95            if messages_wire_terminal(data) {
96                return DROP_PRIORITY_CRITICAL;
97            }
98            if let Some(arr) = data.and_then(|v| v.as_array()) {
99                if let Some(first) = arr.first().and_then(|v| v.as_object()) {
100                    if value_as_str(first.get("phase")) == "goal_completion" {
101                        return DROP_PRIORITY_CRITICAL;
102                    }
103                }
104            }
105        }
106    }
107
108    DROP_PRIORITY_NORMAL
109}
110
111fn messages_wire_terminal(data: Option<&Value>) -> bool {
112    let Some(arr) = data.and_then(|v| v.as_array()) else {
113        return false;
114    };
115    let Some(body) = arr.first().and_then(|v| v.as_object()) else {
116        return false;
117    };
118    let t = value_as_str(body.get("type"));
119    t == STREAM_END || contains_str(&t, "stream.end")
120}
121
122fn value_as_str(v: Option<&Value>) -> String {
123    v.and_then(|v| v.as_str()).unwrap_or("").to_string()
124}
125
126fn has_prefix(s: &str, prefix: &str) -> bool {
127    s.len() >= prefix.len() && s.starts_with(prefix)
128}
129
130fn contains_str(s: &str, sub: &str) -> bool {
131    sub.is_empty() || s.contains(sub)
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137    use serde_json::json;
138
139    #[test]
140    fn nil_event_is_critical() {
141        assert_eq!(inbound_frame_drop_priority(None), DROP_PRIORITY_CRITICAL);
142    }
143
144    #[test]
145    fn status_idle_is_critical() {
146        let event = json!({"type": "status", "state": "idle"})
147            .as_object()
148            .cloned()
149            .unwrap();
150        assert_eq!(
151            inbound_frame_drop_priority(Some(&event)),
152            DROP_PRIORITY_CRITICAL
153        );
154    }
155
156    #[test]
157    fn updates_event_is_normal() {
158        let event = json!({"type": "event", "mode": "updates", "data": {}})
159            .as_object()
160            .cloned()
161            .unwrap();
162        assert_eq!(
163            inbound_frame_drop_priority(Some(&event)),
164            DROP_PRIORITY_NORMAL
165        );
166    }
167
168    #[test]
169    fn event_batch_is_high() {
170        let event = json!({"type": "event_batch"}).as_object().cloned().unwrap();
171        assert_eq!(
172            inbound_frame_drop_priority(Some(&event)),
173            DROP_PRIORITY_HIGH
174        );
175    }
176
177    #[test]
178    fn cognition_custom_is_high() {
179        let event = json!({
180            "type": "event",
181            "mode": "custom",
182            "data": {"type": "soothe.cognition.plan.created"}
183        })
184        .as_object()
185        .cloned()
186        .unwrap();
187        assert_eq!(
188            inbound_frame_drop_priority(Some(&event)),
189            DROP_PRIORITY_HIGH
190        );
191    }
192}
soothe_client/inbound_priority.rs

soothe_client/
inbound_priority.rs
