car_server_core/host.rs
1//! WS-coupled host pieces that stay in `car-server-core` after the #418 lift.
2//!
3//! `HostState`, the [`EventSubscriber`] trait, and the channel/approval types
4//! moved to `car-server-types`; they are re-exported here so `crate::host::*`
5//! keeps resolving across the dispatcher with no call-site churn.
6//!
7//! What stays: [`RunTraceSubscriber`] — it owns a `WsChannel` write sink and
8//! needs a bounded write + failure signal (to exit its drain task on a wedged
9//! socket) that the simple `EventSubscriber::send_text` contract doesn't model.
10
11pub use car_server_types::host::*;
12
13use crate::session::WsChannel;
14use futures::SinkExt;
15use std::sync::Arc;
16use std::time::Duration;
17use tokio_tungstenite::tungstenite::Message;
18
19/// Capacity of each run-trace subscriber's bounded channel (agent run
20/// tracing, U4). One slot per buffered `runs.trace.event` between the
21/// producer (the recorder / lifecycle path) and the per-subscriber drain
22/// task. Sized generously: a healthy CarHost drains far faster than turns
23/// are produced, so the buffer only fills if a socket genuinely wedges —
24/// at which point `try_send` drops the event rather than blocking the
25/// producer (invariant #2). 256 absorbs a burst of fast turns without
26/// dropping under normal load.
27pub const RUN_TRACE_CHANNEL_CAP: usize = 256;
28
29/// One live `runs.trace.event` subscriber — the producer side of the
30/// bounded channel whose drain task writes frames to the subscriber's
31/// WebSocket (agent run tracing, U4).
32///
33/// Keyed in [`crate::session::ServerState::run_subscribers`] by
34/// `(run_id, host_client_id)` so two CarHost windows on the same run are
35/// independent streams (the explicit fanout the single-subscriber-
36/// per-method notification registry can't provide), and so disconnect
37/// cleanup can drop exactly this connection's subscriptions.
38///
39/// The producer holds ONLY the `tx` and calls [`RunTraceSubscriber::push`]
40/// — a non-blocking `try_send`. It never touches the WS socket, so a slow
41/// CarHost can never stall the recorder, the `runs` lock, or any other
42/// in-flight RPC (invariant #2). The dedicated drain task owns the socket
43/// write.
44pub struct RunTraceSubscriber {
45 /// The connection that subscribed — its WS `client_id`.
46 pub host_client_id: String,
47 /// Non-blocking producer handle into the drain task's channel.
48 tx: tokio::sync::mpsc::Sender<car_proto::RunTraceEvent>,
49}
50
51impl RunTraceSubscriber {
52 /// Spawn a drain task bound to `channel` and return the producer-side
53 /// subscriber handle. The drain task serializes each
54 /// `runs.trace.event` to a JSON-RPC notification frame and writes it
55 /// to the subscriber's WS; it exits when the `tx` is dropped (the
56 /// subscriber is removed on unsubscribe / disconnect) — at which point
57 /// the channel closes and `recv()` returns `None`.
58 pub fn spawn(host_client_id: String, channel: Arc<WsChannel>) -> Self {
59 let (tx, mut rx) =
60 tokio::sync::mpsc::channel::<car_proto::RunTraceEvent>(RUN_TRACE_CHANNEL_CAP);
61 tokio::spawn(async move {
62 while let Some(event) = rx.recv().await {
63 let Ok(json) = serde_json::to_string(&serde_json::json!({
64 "jsonrpc": "2.0",
65 "method": "runs.trace.event",
66 "params": event,
67 })) else {
68 continue;
69 };
70 // A wedged socket should not hang the drain task forever —
71 // bound the write so a dead connection's task exits and
72 // frees the channel rather than parking on a full TCP
73 // buffer. On any write error the connection is gone; stop
74 // draining (the subscriber is reaped on disconnect).
75 let mut guard = channel.write.lock().await;
76 let send = tokio::time::timeout(
77 Duration::from_secs(10),
78 guard.send(Message::Text(json.into())),
79 )
80 .await;
81 drop(guard);
82 match send {
83 Ok(Ok(())) => {}
84 // Timed out or errored — the socket is unusable. Exit
85 // the drain loop; the producer's `try_send` will start
86 // dropping (bounded) and the subscriber is cleaned up
87 // on disconnect.
88 _ => break,
89 }
90 }
91 });
92 Self { host_client_id, tx }
93 }
94
95 /// Non-blocking push of one event onto the drain channel. Returns
96 /// `false` when the channel is full (a wedged subscriber) — the event
97 /// is dropped rather than blocking the producer (invariant #2). The
98 /// producer treats the drop as best-effort: the client detects the
99 /// resulting cursor gap and re-subscribes to backfill (R8).
100 pub fn push(&self, event: car_proto::RunTraceEvent) -> bool {
101 self.tx.try_send(event).is_ok()
102 }
103
104 #[cfg(test)]
105 pub fn is_closed(&self) -> bool {
106 self.tx.is_closed()
107 }
108}
109
110#[cfg(test)]
111mod tests {
112 use super::*;
113 use std::collections::HashMap;
114 use tokio::sync::Mutex;
115
116 /// Build a `WsChannel` whose writes land in a `futures::mpsc`
117 /// receiver the test can drain to observe the drain task's frames.
118 /// `futures::channel::mpsc::UnboundedSender` already implements
119 /// `Sink<Message>`; we map its `SendError` into the tungstenite error
120 /// the `WsSink` alias expects.
121 fn channel_with_capture() -> (
122 Arc<WsChannel>,
123 futures::channel::mpsc::UnboundedReceiver<Message>,
124 ) {
125 use futures::sink::SinkExt;
126 let (tx, rx) = futures::channel::mpsc::unbounded::<Message>();
127 let sink: crate::session::WsSink =
128 Box::pin(tx.sink_map_err(|_| tokio_tungstenite::tungstenite::Error::ConnectionClosed));
129 let channel = Arc::new(WsChannel {
130 write: Mutex::new(sink),
131 pending: Mutex::new(HashMap::new()),
132 active_actions: Mutex::new(HashMap::new()),
133 next_id: std::sync::atomic::AtomicU64::new(0),
134 });
135 (channel, rx)
136 }
137
138 #[tokio::test]
139 async fn run_trace_subscriber_drains_event_to_socket() {
140 let (channel, mut rx) = channel_with_capture();
141 let sub = RunTraceSubscriber::spawn("host-1".to_string(), channel);
142 let event = car_proto::RunTraceEvent {
143 run_id: "run-1".to_string(),
144 agent_id: "agent-a".to_string(),
145 record: car_proto::RunRecord::Started(car_proto::RunStarted {
146 run_id: "run-1".to_string(),
147 client_id: None,
148 agent_id: "agent-a".to_string(),
149 intent: "go".to_string(),
150 outcome_description: None,
151 started_at: chrono::Utc::now(),
152 }),
153 cursor: 0,
154 status: car_proto::RunLiveStatus::InProgress,
155 };
156 assert!(sub.push(event), "push onto a fresh channel succeeds");
157
158 // The drain task serializes the event to a runs.trace.event frame.
159 use futures::StreamExt;
160 let frame = tokio::time::timeout(Duration::from_secs(2), rx.next())
161 .await
162 .expect("drain task wrote within the deadline")
163 .expect("a frame");
164 let text = match frame {
165 Message::Text(t) => t.to_string(),
166 other => panic!("expected a text frame, got {other:?}"),
167 };
168 let json: serde_json::Value = serde_json::from_str(&text).unwrap();
169 assert_eq!(json["method"], "runs.trace.event");
170 assert_eq!(json["params"]["run_id"], "run-1");
171 assert_eq!(json["params"]["status"], "in_progress");
172 }
173
174 #[tokio::test]
175 async fn dropping_subscriber_ends_its_drain_task() {
176 let (channel, _rx) = channel_with_capture();
177 let sub = RunTraceSubscriber::spawn("host-1".to_string(), channel);
178 assert!(!sub.is_closed());
179 // Dropping the subscriber drops its sender; the drain task's
180 // recv() returns None and the task exits. We can only assert the
181 // sender side here (the task is detached), but is_closed flips
182 // once the receiver is gone — which happens when the task ends.
183 drop(sub);
184 // No panic / hang is the assertion; the spawned task tears down.
185 }
186}