stdiod: deterministic child shutdown — terminal report precedes any respawn ack (T-74)

crates/stdiod/PROTOCOL.md

@@ -306,6 +306,30 @@ pump EOF, input pump write error, a frame addressed to an exited child). The
 reference client uses a one-shot latch.
 *Source: `proc.rs::ChildDiagnostics::take_terminal_error` (`reported.swap`).*
 
+**T-74** When a client stops a child and then reports a spawn result for the
+same `server_id` (a kill-and-respawn: env update, spec update, a changed
+desired spec, or a restart of an unresponsive child), the terminal
+`server_offline` for the old child MUST be sent before the new
+`server_spawn_result`. Receivers MAY therefore treat
+`server_spawn_result{ok:true}` as clearing any terminal error they hold for
+that `server_id`. See T-42.
+
+This is an ordering requirement, not a delivery guarantee. A client whose
+outbound channel cannot accept the report (full, or disconnected mid-shutdown)
+MAY drop it, in which case the backend sees no `server_offline` at all and
+falls back to its own staleness handling (T-50); what a client MUST NOT do is
+let the report arrive after the ack. The reference client bounds its own
+shutdown: it joins both frame pumps under one budget, since either may hold the
+one-shot report, and if it has to abort a pump that has taken the report it
+sends the report itself, non-blocking, before returning. T-43 still holds
+across that fallback: the latch that hands out the report and the flag that
+records a completed send are updated without an await between them, so an
+aborted pump has either sent the report or left it unsent, never both.
+*Source: `proc.rs::ChildServer::shutdown` (joins both pumps, with a fallback
+send after an abort); `daemon.rs::Supervisor::try_spawn`
+(every respawn path awaits `shutdown` first); `registry.py::_dispatch_inbound`
+(`ServerSpawnResult` arm clears `last_error_by_server`).*
+
 **T-44** A spawn failure MUST produce **both** `server_spawn_result{ok:false}`
 and `tunnel_error{code:"spawn_failed", server_id}`, in that order. The backend
 treats `spawn_failed` and `server_offline` as terminal for the current child:
@@ -361,7 +385,7 @@ carry on) rather than failing the frame.
 
 | Code | Direction | Meaning | Governed by |
 |------|-----------|---------|-------------|
-| `server_offline` | client → backend | The child for `server_id` is gone: its output pump hit EOF or errored, its input pump failed to write, or a frame arrived for a child already known dead. Terminal for that child, and emitted at most once per child lifetime | T-42, T-43, T-47 |
+| `server_offline` | client → backend | The child for `server_id` is gone: its output pump hit EOF or errored, its input pump failed to write, or a frame arrived for a child already known dead. Terminal for that child, and emitted at most once per child lifetime | T-42, T-43, T-47, T-74 |
 | `spawn_failed` | client → backend | The client tried to start `server_id` and could not (binary missing, exec refused, module URI unrecognised). Always accompanies a `server_spawn_result{ok:false}` and follows it | T-44 |
 | `server_unresponsive` | client → backend | The child for `server_id` stopped consuming its input: the per-child queue filled, so this request could not be delivered. The client then kills and respawns the child | T-51 |
 | `stdio_tunnel_disabled` | backend → client | Device-wide (`server_id` null): the org has the `stdio_tunnel_enabled` feature flag off. The backend closes with 1008 straight after | T-07, T-49 |

crates/stdiod/crates/edison-stdiod/src/child_diagnostics.rs

@@ -36,6 +36,13 @@ pub(crate) struct ChildDiagnostics {
     /// as crashed next to its own live PID.
     observed_exit: Arc<AtomicBool>,
     reported: Arc<AtomicBool>,
+    /// Set once a terminal report has actually been queued on the outbound
+    /// channel. `reported` says a path took ownership of the report;
+    /// this says the wire has it. Nothing awaits between the send completing
+    /// and this store, so a cancelled pump can never leave the pair
+    /// disagreeing, which is what lets `shutdown` decide whether it has to
+    /// send the report itself.
+    pub(crate) report_sent: Arc<AtomicBool>,
     /// One-shot guard so a single death produces a single `crashed` write,
     /// whichever pump gets there first.
     crash_published: Arc<AtomicBool>,
@@ -87,6 +94,11 @@ impl ChildDiagnostics {
         self.observed_exit.load(Ordering::Acquire)
     }
 
+    /// Whether a terminal report has reached the outbound channel.
+    pub(crate) fn report_sent(&self) -> bool {
+        self.report_sent.load(Ordering::Acquire)
+    }
+
     /// Flip this child's `state.json` entry to `crashed`, once, and only for
     /// a process that was seen to exit.
     ///

crates/stdiod/crates/edison-stdiod/src/daemon_reconcile.rs

@@ -60,6 +60,19 @@ impl Supervisor {
 
     /// Spawn one raw desired server after applying current local values. Reports
     /// the spawn result and retains placeholders so later respawns re-enrich cleanly.
+    ///
+    /// Ordering contract (PROTOCOL.md T-74): every kill-and-respawn path here
+    /// (`apply_snapshot`, `apply_delta`, `apply_spec_update`,
+    /// `apply_env_update`, `restart_unresponsive`) MUST `await`
+    /// [`ChildServer::shutdown`] for the outgoing child *before* calling this,
+    /// and this method's `server_spawn_result` send must stay after the
+    /// spawn. That is what makes the old child's terminal `server_offline`
+    /// reach the outbound channel first: `shutdown` returns only once the
+    /// stdout pump's report has been queued, and the channel behind
+    /// [`OutgoingHandle`] preserves the order in which sends complete, so the
+    /// WS writer emits the two frames in that order. The backend relies on it
+    /// to treat a successful ack as clearing a stored terminal error
+    /// (`registry.py::_dispatch_inbound`).
     async fn try_spawn(&mut self, raw: DesiredServer) {
         let server_id = raw.server_id.clone();
         let sensitive_arg_values = self

crates/stdiod/crates/edison-stdiod/src/daemon_tests.rs

@@ -23,8 +23,16 @@ fn env_store_for(test: &str) -> EnvStore {
 }
 
 fn supervisor_with(test: &str, children: HashMap<String, ChildServer>) -> Supervisor {
+    supervisor_with_outgoing(test, children, OutgoingHandle::new())
+}
+
+fn supervisor_with_outgoing(
+    test: &str,
+    children: HashMap<String, ChildServer>,
+    outgoing: OutgoingHandle,
+) -> Supervisor {
     let mut supervisor = Supervisor::new(
-        OutgoingHandle::new(),
+        outgoing,
         StateWriter::new(State::default()),
         env_store_for(test),
     );
@@ -91,6 +99,50 @@ async fn snapshot_reports_an_exited_child_as_crashed() {
     assert!(matches!(entries[0].state, ServerStatus::Crashed));
 }
 
+/// PROTOCOL.md T-74: a kill-and-respawn MUST put the old child's terminal
+/// `server_offline` on the outbound channel before the replacement's
+/// `server_spawn_result`, so the backend can treat a successful ack as
+/// clearing the stored error. The outbound channel is the observation point
+/// the WS writer drains in order, so frame order here is wire order.
+///
+/// Both frames are read with `try_recv`: they must already be queued by the
+/// time the respawn path returns, not merely arrive eventually.
+#[tokio::test]
+async fn respawn_queues_terminal_offline_before_the_spawn_ack() {
+    let spec = desired("filesystem", "sleep 30");
+    let outgoing = OutgoingHandle::new();
+    let (wire_tx, mut wire_rx) = mpsc::channel(8);
+    outgoing.set(wire_tx);
+    let child = ChildServer::spawn(&spec, &spec, outgoing.clone(), Vec::new(), None).unwrap();
+    let mut supervisor = supervisor_with_outgoing(
+        "respawn-order",
+        HashMap::from([("filesystem".to_string(), child)]),
+        outgoing,
+    );
+
+    supervisor.restart_unresponsive("filesystem").await;
+
+    let TunnelFrame::TunnelError(error) = wire_rx
+        .try_recv()
+        .expect("the old child's terminal error should already be queued")
+    else {
+        panic!("first frame should be the old child's terminal tunnel_error");
+    };
+    assert_eq!(error.code, "server_offline");
+    assert_eq!(error.server_id.as_deref(), Some("filesystem"));
+
+    let TunnelFrame::ServerSpawnResult(result) = wire_rx
+        .try_recv()
+        .expect("the replacement's spawn ack should follow it")
+    else {
+        panic!("second frame should be the replacement's server_spawn_result");
+    };
+    assert!(result.ok, "the replacement should have spawned");
+    assert_eq!(result.server_id, "filesystem");
+
+    supervisor.shutdown_children().await;
+}
+
 /// A child whose stdin has broken is terminal for MCP - the backend gets its
 /// `server_offline` - but the process may still be running, and the snapshot
 /// must not call a live PID crashed. It stays `running` until the supervisor

crates/stdiod/crates/edison-stdiod/src/proc.rs

@@ -32,6 +32,16 @@ use crate::child_diagnostics::ChildDiagnostics;
 use crate::state::StateWriter;
 use crate::tunnel::OutgoingHandle;
 
+/// How long [`ChildServer::shutdown`] waits for the frame pumps to finish
+/// their terminal report before falling back to aborting them. Shared by both
+/// joins. See the comment at the join site for why this is a join rather than
+/// an abort.
+const PUMP_JOIN_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// How long [`ChildServer::shutdown`] waits for an aborted pump to actually
+/// stop before deciding whether it has to send the terminal report itself.
+const PUMP_ABORT_GRACE: Duration = Duration::from_millis(200);
+
 /// Build the base `Command` for a child MCP server.
 ///
 /// On Unix this is just `Command::new(program).args(args)` - the inherited PATH
@@ -352,6 +362,9 @@ pub struct ChildServer {
     /// behind an async lock.
     pub pid: Option<u32>,
     pub outbound_tx: mpsc::Sender<serde_json::Value>,
+    /// Kept so [`shutdown`](Self::shutdown) can emit the terminal report
+    /// itself when it has to abort the pump that would have sent it.
+    tunnel_outgoing: OutgoingHandle,
     pub stdin_pump: JoinHandle<()>,
     pub stdout_pump: JoinHandle<()>,
     stderr_pump: JoinHandle<()>,
@@ -427,7 +440,7 @@ impl ChildServer {
         let stdout_pump = tokio::spawn(stdout_pump(
             enriched.server_id.clone(),
             stdout,
-            tunnel_outgoing,
+            tunnel_outgoing.clone(),
             diagnostics.clone(),
             Some(child.clone()),
         ));
@@ -451,6 +464,7 @@ impl ChildServer {
             child,
             pid,
             outbound_tx,
+            tunnel_outgoing,
             stdin_pump,
             stdout_pump,
             stderr_pump,
@@ -485,34 +499,121 @@ impl ChildServer {
         self.diagnostics.has_observed_exit()
     }
 
-    /// Kill the child and abort the pumps.
+    /// Kill the child and wind the pumps down.
+    ///
+    /// Returns only once the child's terminal `server_offline` has been queued
+    /// on the outbound channel, so a caller that respawns the same `server_id`
+    /// cannot enqueue the new child's `server_spawn_result` ahead of it - see
+    /// PROTOCOL.md T-74 and `daemon.rs::Supervisor::try_spawn`. Both frame
+    /// pumps are joined under one budget, because either of them can be the
+    /// one holding the terminal report. If the budget runs out, the report is
+    /// sent from here instead so that aborting a wedged pump cannot swallow it
+    /// (the exact hang T-42 exists to prevent).
     pub async fn shutdown(self) {
-        let mut child = self.child.lock().await;
-        if let Some(pid) = child.id() {
-            #[cfg(unix)]
-            {
-                let _ = Command::new("kill")
-                    .args(["-KILL", "--", &format!("-{pid}")])
-                    .status()
-                    .await;
+        let Self {
+            server_id,
+            child,
+            outbound_tx,
+            mut stdin_pump,
+            mut stdout_pump,
+            stderr_pump,
+            tunnel_outgoing,
+            diagnostics,
+            ..
+        } = self;
+        let status = {
+            let mut guard = child.lock().await;
+            if let Some(pid) = guard.id() {
+                #[cfg(unix)]
+                {
+                    let _ = Command::new("kill")
+                        .args(["-KILL", "--", &format!("-{pid}")])
+                        .status()
+                        .await;
+                }
+                #[cfg(windows)]
+                {
+                    let _ = Command::new("taskkill")
+                        .args(["/PID", &pid.to_string(), "/T", "/F"])
+                        .status()
+                        .await;
+                }
             }
-            #[cfg(windows)]
+            let _ = guard.start_kill();
+            let status = guard.wait().await.ok();
+            // Released before joining the pumps: a terminal report reads the
+            // exit status through this same lock and would otherwise deadlock
+            // until the budget below runs out.
+            status
+        };
+        // Reaped just above, so the process is observably gone even if no pump
+        // ever managed to read an exit status for it.
+        diagnostics.mark_observed_exit();
+
+        // Closing the frame channel ends the stdin pump at its `recv`, so the
+        // common case is a clean exit rather than a cancellation.
+        drop(outbound_tx);
+
+        // Join, don't abort: a pump's last act is the terminal
+        // `server_offline` report, and aborting it mid-flight is what made the
+        // ordering against a subsequent respawn a coin flip. Both pumps are
+        // joined because either can own that report - stdout on EOF, stdin on
+        // a failed write - and a one-shot latch means the one that got there
+        // first is the only one that will ever send it. The dead child's pipes
+        // are already at EOF, so the work left is `report_terminal`: at most
+        // ~100ms waiting for stderr to drain, ~100ms polling for the exit
+        // status, then one channel send. The budget is shared by both joins
+        // and bounds the pathological case where the send blocks because the
+        // outbound channel is full and the WS writer is wedged.
+        let deadline = tokio::time::Instant::now() + PUMP_JOIN_TIMEOUT;
+        let mut report_was_cancelled = false;
+        for (which, pump) in [("stdout", &mut stdout_pump), ("stdin", &mut stdin_pump)] {
+            if tokio::time::timeout_at(deadline, &mut *pump).await.is_ok() {
+                continue;
+            }
+            warn!(
+                server_id = %server_id,
+                pump = which,
+                "pump did not finish within the shutdown budget; aborting it",
+            );
+            pump.abort();
+            // `abort` only requests cancellation, which lands at the task's
+            // next await point. Wait for it: if the pump were still runnable
+            // it could complete a send between the check below and the
+            // fallback, and the backend would see two terminal errors for one
+            // child. When cancellation does not land in the grace period we
+            // leave the report alone rather than risk that duplicate - the
+            // backend's staleness teardown fails the calls either way.
+            if tokio::time::timeout(PUMP_ABORT_GRACE, &mut *pump)
+                .await
+                .is_ok()
             {
-                let _ = Command::new("taskkill")
-                    .args(["/PID", &pid.to_string(), "/T", "/F"])
-                    .status()
-                    .await;
+                report_was_cancelled = true;
+            }
+        }
+        stderr_pump.abort();
+
+        // The pump that owned the report was cancelled before it queued one,
+        // so send it from here. `reported` is consumed by whichever path took
+        // the error and `report_sent` only flips once a send has completed, so
+        // between them they say exactly what still has to go out: nothing when
+        // a send got through, a fresh error when no path ever took one, and a
+        // reconstruction when a path took one and died before sending it.
+        // Non-blocking on purpose - shutdown must not block on a wedged
+        // channel, and a full channel means the session is being torn down
+        // anyway.
+        if report_was_cancelled && !diagnostics.report_sent() {
+            let error = diagnostics
+                .take_terminal_error(&server_id, status.as_ref())
+                .unwrap_or_else(|| diagnostics.terminal_error(&server_id, status.as_ref()));
+            if !tunnel_outgoing.try_send(TunnelFrame::TunnelError(error)) {
+                warn!(
+                    server_id = %server_id,
+                    "could not queue the terminal report during shutdown; the backend will \
+                     fail in-flight calls when the session goes stale",
+                );
             }
         }
-        let _ = child.start_kill();
-        let _ = child.wait().await;
-        // Reaped here, so the process is observably gone even if no pump
-        // ever managed to read an exit status for it.
-        self.diagnostics.mark_observed_exit();
-        drop(child);
-        self.stdin_pump.abort();
-        self.stdout_pump.abort();
-        self.stderr_pump.abort();
     }
 }
 
@@ -591,7 +692,13 @@ async fn report_terminal(
     let error = diagnostics.take_terminal_error(server_id, status.as_ref());
     diagnostics.publish_crashed(server_id).await;
     if let Some(error) = error {
-        tunnel_outgoing.send(TunnelFrame::TunnelError(error)).await;
+        if tunnel_outgoing.send(TunnelFrame::TunnelError(error)).await {
+            // No await between the send completing and this store, so a pump
+            // cancelled at any point either queued the frame and recorded it
+            // or did neither. `shutdown` relies on that to avoid both a lost
+            // report and a duplicate one.
+            diagnostics.report_sent.store(true, Ordering::Release);
+        }
     }
 }