fix(broker): serialize terminal worker writes

Author: miyaontherelay

crates/broker/src/runtime/api.rs

@@ -736,35 +736,22 @@ impl BrokerRuntime {
                 timeout_ms,
                 reply,
             } => {
-                let Some(handle) = workers.workers.get_mut(&name) else {
+                if !workers.workers.contains_key(&name) {
                     let _ = reply.send(Err(format!("unknown worker '{}'", name)));
                     return;
-                };
+                }
 
                 let model_command = format!("/model {}\n", model);
-                let result = async {
-                    handle
-                        .stdin
-                        .write_all(model_command.as_bytes())
-                        .await
-                        .with_context(|| {
-                            format!("failed writing model command to worker '{}'", name)
-                        })?;
-                    handle
-                        .stdin
-                        .flush()
-                        .await
-                        .with_context(|| format!("failed flushing worker '{}' stdin", name))?;
-                    if let Some(timeout_ms) = timeout_ms {
-                        tracing::info!(
-                            name = %name,
-                            timeout_ms,
-                            "HTTP API set_model timeout_ms is currently advisory only"
-                        );
-                    }
-                    Ok::<(), anyhow::Error>(())
+                let result = workers
+                    .send_raw_to_worker(&name, model_command.into_bytes())
+                    .await;
+                if let Some(timeout_ms) = timeout_ms {
+                    tracing::info!(
+                        name = %name,
+                        timeout_ms,
+                        "HTTP API set_model timeout_ms is currently advisory only"
+                    );
                 }
-                .await;
 
                 match result {
                     Ok(()) => {

crates/broker/src/runtime/fleet.rs

@@ -19,10 +19,6 @@ const TERMINAL_INPUT_MAX_BYTES: usize = 64 * 1024;
 const TERMINAL_INPUT_MAX_BASE64_BYTES: usize = TERMINAL_INPUT_MAX_BYTES * 4 / 3 + 4;
 const TERMINAL_SNAPSHOT_TIMEOUT: Duration = Duration::from_secs(10);
 const TERMINAL_INPUT_ACK_TIMEOUT: Duration = Duration::from_secs(5);
-// Terminal worker writes share the broker event loop with fleet control and
-// worker lifecycle events. A wedged PTY must fail only its own attach instead
-// of awaiting an unbounded pipe write in that loop.
-const TERMINAL_WORKER_WRITE_TIMEOUT: Duration = Duration::from_millis(250);
 const TERMINAL_INPUT_MAX_IN_FLIGHT_PER_SESSION: usize = 16;
 // Relaycast currently limits a node to 32 terminal sessions. Keep that many
 // slots free from high-volume frames so every affected session can still get a
@@ -42,6 +38,44 @@ pub(super) fn try_send_terminal(
         .is_ok()
 }
 
+pub(super) fn fail_terminal_session(
+    terminal_control_tx: &mpsc::Sender<TerminalControlCommand>,
+    terminal_sessions: &mut HashMap<String, TerminalSession>,
+    terminal_snapshot_requests: &mut HashMap<String, TerminalSnapshotRequest>,
+    terminal_input_requests: &mut HashMap<String, TerminalInputRequest>,
+    session_id: String,
+    code: &str,
+    message: String,
+) {
+    terminal_sessions.remove(&session_id);
+    terminal_snapshot_requests.retain(|_, pending| pending.session_id != session_id);
+    terminal_input_requests.retain(|_, pending| pending.session_id != session_id);
+
+    // Error is useful when the terminal lane has room, but it is non-final and
+    // deliberately gives way to the reserved close capacity. Queue its close
+    // directly instead of routing through `send_terminal`: the generic
+    // backpressure fallback would otherwise produce a second close with a
+    // different reason.
+    let _ = try_send_terminal(
+        terminal_control_tx,
+        TerminalToCloud::Error {
+            session_id: session_id.clone(),
+            code: code.into(),
+            message: message.clone(),
+        },
+    );
+    if !try_send_terminal(
+        terminal_control_tx,
+        TerminalToCloud::Closed {
+            session_id: session_id.clone(),
+            code: Some(code.into()),
+            message: Some(message),
+        },
+    ) {
+        tracing::warn!(target = "relay_broker::terminal", session_id = %session_id, "terminal close could not be queued after session failure");
+    }
+}
+
 #[derive(Debug, Clone)]
 pub(super) struct PendingVerifiedSpawn {
     pub(super) invocation_id: String,
@@ -153,44 +187,29 @@ impl BrokerRuntime {
                                 pending_output_bytes: 0,
                             },
                         );
-                        // Request the grid asynchronously. The response is routed
-                        // from worker_events to the terminal lane, so this never
-                        // stalls heartbeat/action processing behind a PTY snapshot.
+                        // Queue the grid request on the worker-owned stdin
+                        // writer. The event loop never awaits a PTY pipe write;
+                        // the writer serializes complete frames and reports a
+                        // later pipe failure as a terminal session failure.
                         let request_id = format!("terminal_snapshot_{}", Uuid::new_v4().simple());
-                        match tokio::time::timeout(
-                            TERMINAL_WORKER_WRITE_TIMEOUT,
-                            self.workers.send_to_worker(
-                                agent_name.as_str(),
-                                "snapshot_pty",
-                                Some(RequestId::new(request_id.clone())),
-                                json!({ "format": "ansi" }),
-                            ),
-                        )
-                        .await
-                        {
-                            Ok(Ok(())) => {
-                                self.terminal_snapshot_requests.insert(
-                                    request_id,
-                                    TerminalSnapshotRequest {
-                                        session_id,
-                                        deadline: Instant::now() + TERMINAL_SNAPSHOT_TIMEOUT,
-                                    },
-                                );
-                            }
-                            Ok(Err(error)) => {
-                                self.fail_terminal_session(
-                                    session_id,
-                                    "snapshot_failed",
-                                    error.to_string(),
-                                );
-                            }
-                            Err(_) => {
-                                self.fail_terminal_session(
-                                    session_id,
-                                    "snapshot_timeout",
-                                    "terminal snapshot write timed out".into(),
-                                );
-                            }
+                        self.terminal_snapshot_requests.insert(
+                            request_id.clone(),
+                            TerminalSnapshotRequest {
+                                session_id: session_id.clone(),
+                                deadline: Instant::now() + TERMINAL_SNAPSHOT_TIMEOUT,
+                            },
+                        );
+                        if let Err(error) = self.workers.try_send_to_worker(
+                            agent_name.as_str(),
+                            "snapshot_pty",
+                            Some(RequestId::new(request_id.clone())),
+                            json!({ "format": "ansi" }),
+                        ) {
+                            self.fail_terminal_session(
+                                session_id,
+                                "snapshot_failed",
+                                error.to_string(),
+                            );
                         }
                     }
                 }
@@ -268,34 +287,20 @@ impl BrokerRuntime {
                     return;
                 }
                 let request_id = format!("terminal_input_{}", Uuid::new_v4().simple());
-                match tokio::time::timeout(
-                    TERMINAL_WORKER_WRITE_TIMEOUT,
-                    self.workers.send_to_worker(
-                        session.agent.as_str(),
-                        "write_pty",
-                        Some(RequestId::new(request_id.clone())),
-                        json!({ "data": data }),
-                    ),
-                )
-                .await
-                {
-                    Ok(Ok(())) => {
-                        self.terminal_input_requests.insert(
-                            request_id,
-                            TerminalInputRequest {
-                                session_id,
-                                deadline: Instant::now() + TERMINAL_INPUT_ACK_TIMEOUT,
-                            },
-                        );
-                    }
-                    Ok(Err(error)) => {
-                        self.fail_terminal_session(session_id, "input_failed", error.to_string())
-                    }
-                    Err(_) => self.fail_terminal_session(
-                        session_id,
-                        "input_timeout",
-                        "terminal input write timed out".into(),
-                    ),
+                self.terminal_input_requests.insert(
+                    request_id.clone(),
+                    TerminalInputRequest {
+                        session_id: session_id.clone(),
+                        deadline: Instant::now() + TERMINAL_INPUT_ACK_TIMEOUT,
+                    },
+                );
+                if let Err(error) = self.workers.try_send_to_worker(
+                    session.agent.as_str(),
+                    "write_pty",
+                    Some(RequestId::new(request_id.clone())),
+                    json!({ "data": data }),
+                ) {
+                    self.fail_terminal_session(session_id, "input_failed", error.to_string());
                 }
             }
             TerminalControlEvent::Message(TerminalFromCloud::Resize {
@@ -327,26 +332,13 @@ impl BrokerRuntime {
                     });
                     return;
                 }
-                match tokio::time::timeout(
-                    TERMINAL_WORKER_WRITE_TIMEOUT,
-                    self.workers.send_to_worker(
-                        session.agent.as_str(),
-                        "resize_pty",
-                        None,
-                        json!({ "rows": rows, "cols": cols }),
-                    ),
-                )
-                .await
-                {
-                    Ok(Ok(())) => {}
-                    Ok(Err(error)) => {
-                        self.fail_terminal_session(session_id, "resize_failed", error.to_string())
-                    }
-                    Err(_) => self.fail_terminal_session(
-                        session_id,
-                        "resize_timeout",
-                        "terminal resize write timed out".into(),
-                    ),
+                if let Err(error) = self.workers.try_send_to_worker(
+                    session.agent.as_str(),
+                    "resize_pty",
+                    None,
+                    json!({ "rows": rows, "cols": cols }),
+                ) {
+                    self.fail_terminal_session(session_id, "resize_failed", error.to_string());
                 }
             }
             TerminalControlEvent::Message(TerminalFromCloud::Close { session_id }) => {
@@ -399,21 +391,15 @@ impl BrokerRuntime {
     }
 
     fn fail_terminal_session(&mut self, session_id: String, code: &str, message: String) {
-        self.terminal_sessions.remove(&session_id);
-        self.terminal_snapshot_requests
-            .retain(|_, pending| pending.session_id != session_id);
-        self.terminal_input_requests
-            .retain(|_, pending| pending.session_id != session_id);
-        self.send_terminal(TerminalToCloud::Error {
-            session_id: session_id.clone(),
-            code: code.into(),
-            message: message.clone(),
-        });
-        self.send_terminal(TerminalToCloud::Closed {
+        fail_terminal_session(
+            &self.terminal_control_tx,
+            &mut self.terminal_sessions,
+            &mut self.terminal_snapshot_requests,
+            &mut self.terminal_input_requests,
             session_id,
-            code: Some(code.into()),
-            message: Some(message),
-        });
+            code,
+            message,
+        );
     }
 
     pub(super) async fn handle_fleet_control_event(&mut self, event: FleetControlEvent) {
@@ -1701,6 +1687,47 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn terminal_failure_queues_one_close_with_the_original_reason_at_reserve() {
+        let (tx, mut rx) = mpsc::channel(TERMINAL_CLOSE_RESERVE + 1);
+        assert!(try_send_terminal(
+            &tx,
+            TerminalToCloud::Output {
+                session_id: "session-a".into(),
+                chunk: "x".into(),
+                offset: None,
+            },
+        ));
+        // This leaves exactly the reserved close capacity. The non-final Error
+        // must be rejected, but the single final close must still carry the
+        // actual failure rather than an output_backpressure fallback.
+        fail_terminal_session(
+            &tx,
+            &mut HashMap::new(),
+            &mut HashMap::new(),
+            &mut HashMap::new(),
+            "session-a".into(),
+            "snapshot_failed",
+            "worker command queue is full".into(),
+        );
+
+        assert!(matches!(
+            rx.try_recv(),
+            Ok(TerminalControlCommand::Send(TerminalToCloud::Output { .. }))
+        ));
+        assert!(matches!(
+            rx.try_recv(),
+            Ok(TerminalControlCommand::Send(TerminalToCloud::Closed {
+                session_id,
+                code: Some(code),
+                message: Some(message),
+            })) if session_id == "session-a"
+                && code == "snapshot_failed"
+                && message == "worker command queue is full"
+        ));
+        assert!(rx.try_recv().is_err(), "failure must emit only one close");
+    }
+
     #[test]
     fn classify_fleet_delivery_injects_message_classes_and_acks_receipts() {
         // Mirrors relaycast parse_inbound_kind message-class alias set: any of

crates/broker/src/runtime/tests.rs

@@ -17,7 +17,9 @@ use crate::protocol::{
     HeadlessHarnessDriver, MessageInjectionMode, NativeHarnessConfig, RelayDelivery,
     ResolvedHarnessConfig,
 };
-use crate::worker::{AgentWorkState, WorkerEvent, WorkerHandle, WorkerRegistry};
+use crate::worker::{
+    spawn_worker_writer, AgentWorkState, WorkerEvent, WorkerHandle, WorkerRegistry,
+};
 use crate::{
     broker::injection_format::format_injection,
     util::{
@@ -71,7 +73,7 @@ fn env_test_lock() -> &'static Mutex<()> {
 async fn make_worker_registry_with_worker(name: &str) -> WorkerRegistry {
     let (tx, _rx) = mpsc::channel::<WorkerEvent>(16);
     let mut registry = WorkerRegistry::new(
-        tx,
+        tx.clone(),
         Vec::new(),
         PathBuf::from("/tmp/agent-relay-broker-tests"),
         Instant::now(),
@@ -83,10 +85,13 @@ async fn make_worker_registry_with_worker(name: &str) -> WorkerRegistry {
         .spawn()
         .expect("test worker process should spawn");
     let stdin = child.stdin.take().expect("test worker stdin should exist");
+    let generation = Uuid::new_v4();
+    let (command_tx, command_rx) = mpsc::channel(128);
+    spawn_worker_writer(tx, WorkerName::from(name), generation, stdin, command_rx);
     registry.workers.insert(
         WorkerName::from(name),
         WorkerHandle {
-            generation: Uuid::new_v4(),
+            generation,
             spec: AgentSpec {
                 name: WorkerName::from(name),
                 runtime: AgentRuntime::Pty,
@@ -106,7 +111,7 @@ async fn make_worker_registry_with_worker(name: &str) -> WorkerRegistry {
             parent: None,
             workspace_id: Some(WorkspaceId::new("ws_demo")),
             child,
-            stdin,
+            command_tx,
             harness_pid: None,
             spawned_at: Instant::now(),
             // Ready, so the orphan sweep's readiness deadline never applies to