fix(mcp): a failed handshake must not strand the child process

An MCP server whose `initialize` answer is an error — a broken or version-
mismatched connector — starts fine and then never exits. `McpServer::start`
returned `Err` correctly, but the `Child` lives in the read-loop task rather
than in the returned value, so `kill_on_drop` followed a task nothing ever
drops. Every retry therefore left a live process holding three pipes and a
pidfd, and the supervisor's retry ceiling is deliberately not permanent.

The end state was not a dead connector but a dead instance: the process hit
its 1024-descriptor limit, `accept()` began failing with EMFILE, and incoming
connections queued on a socket nobody could accept from — while the process,
the port and every other connector still looked healthy. Observed in
production at ~5h from the first bad handshake to unreachable, with 229
orphaned interpreters.

`stop_server`/`stop_all` had the same hole from the other side: they document
the dropped handle as killing the process, but the task holds its end of
stdin, so the child stayed blocked on a read that would never return.

Both close with one seam. `McpServer` now owns a oneshot sender whose receiver
the read-loop selects on; nothing ever sends, so the drop is the message. That
covers a deliberate stop, the last `Arc` going away, and a `?` in `start()`
unwinding past the local binding before it was ever returned — including the
caller's `timeout`, which drops the same future. The loop then kills and, as
importantly, reaps: an unreaped child trades the orphan for a zombie holding
the same pipes.

Both leaks are covered by tests that fail without the fix.

Also raise LimitNOFILE to 65536: the installers write it, and update.sh heals
an existing unit additively, leaving an admin's own value alone. The leak is
the bug, but 1024 for a process sharing descriptors between the listener,
every user's SQLite handles and three pipes per connector is thin regardless.

crates/mcp-client/src/server.rs

@@ -225,6 +225,31 @@ pub struct McpServer {
     /// manager can tell "this handle is dead" from "this call failed". Shared with
     /// that task, which is the only writer.
     alive: Arc<std::sync::atomic::AtomicBool>,
+    /// Ties the child process's life to this handle's.
+    ///
+    /// The read-loop owns the `Child` — it needs `wait()` for the exit status — so
+    /// `Command::kill_on_drop` follows *that task*, which nothing ever drops, rather
+    /// than this value. On its own that leaves two ways to strand a live child:
+    /// `stop_server`/`stop_all` drop a handle whose process then keeps running (the
+    /// task still holds its end of stdin, so the child blocks on a read that never
+    /// returns), and — the one that took an instance down — a `start()` that fails
+    /// *after* the spawn never produces a handle at all, so there is nothing to drop.
+    ///
+    /// That second case is the expensive one, because the natural failure is a server
+    /// which starts fine and answers `initialize` wrong: it never exits by itself, so
+    /// the supervisor mints one orphan (three pipes and a pidfd) per retry, and the
+    /// retry ceiling is deliberately not permanent. The end state is not a dead
+    /// connector but a dead *app* — the process hits its file-descriptor limit,
+    /// `accept()` begins failing with `EMFILE`, and connections pile up on a socket
+    /// nobody can accept from.
+    ///
+    /// Holding the sender here closes both: the read-loop selects on the matching
+    /// receiver, which resolves as soon as this field is dropped — whether that is a
+    /// deliberate stop, the last `Arc` going away, or a `?` in `start()` unwinding
+    /// past the local `server` binding before it was ever returned. The caller's
+    /// `timeout` is covered by the same mechanism, since dropping the `start()`
+    /// future drops that binding too.
+    _kill_on_drop: oneshot::Sender<()>,
 }
 
 impl McpServer {
@@ -328,6 +353,11 @@ impl McpServer {
             Arc::new(Mutex::new(HashMap::new()));
         let pending_elicitations = Arc::new(AtomicUsize::new(0));
 
+        // Created before the read-loop task, which is what holds the child and so is
+        // the only thing that can kill it. The sender goes into `server` below — see
+        // `McpServer::_kill_on_drop` for what that buys.
+        let (kill_tx, mut kill_rx) = oneshot::channel::<()>();
+
         let alive                    = Arc::new(std::sync::atomic::AtomicBool::new(true));
         let alive_bg                 = Arc::clone(&alive);
         let pending_bg               = pending.clone();
@@ -340,8 +370,26 @@ impl McpServer {
         tokio::spawn(async move {
             let mut child = child;
             let mut lines = BufReader::new(stdout).lines();
+            // Set when the handle went away, so the epitaph below tells a deliberate
+            // teardown apart from a server that died on its own.
+            let mut killed_by_client = false;
             loop {
-                match lines.next_line().await {
+                let next = tokio::select! {
+                    // A chatty server must not be able to starve the kill signal.
+                    biased;
+                    // Resolves when the `McpServer` holding the sender is dropped.
+                    // Nothing ever sends, so the value is always `Err(RecvError)` —
+                    // the drop *is* the message.
+                    _ = &mut kill_rx => {
+                        // `start_kill` only signals; the `wait()` below is what
+                        // reaps the child and releases its pipes.
+                        let _ = child.start_kill();
+                        killed_by_client = true;
+                        break;
+                    }
+                    line = lines.next_line() => line,
+                };
+                match next {
                     Ok(Some(line)) if !line.trim().is_empty() => {
                         if let Ok(msg) = serde_json::from_str::<Value>(&line) {
                             let has_method = msg.get("method").is_some();
@@ -380,12 +428,20 @@ impl McpServer {
                     _ => break,
                 }
             }
-            let exit_info = match child.wait().await {
-                Ok(status) if !status.success() => format!(
-                    "process exited with {}",
-                    status.code().map(|c| c.to_string()).unwrap_or_else(|| "signal".into())
-                ),
-                _ => "process exited unexpectedly".into(),
+            // Always reap, including after `start_kill`, which only signals: skipping
+            // this would trade the orphan for a zombie, and a zombie still holds the
+            // pipes that made the original leak fatal.
+            let status = child.wait().await;
+            let exit_info = if killed_by_client {
+                "stopped by the client".to_string()
+            } else {
+                match status {
+                    Ok(status) if !status.success() => format!(
+                        "process exited with {}",
+                        status.code().map(|c| c.to_string()).unwrap_or_else(|| "signal".into())
+                    ),
+                    _ => "process exited unexpectedly".into(),
+                }
             };
             // Publish the death *before* failing the pending calls: a caller woken
             // by the error below must find `is_alive() == false`, or it would
@@ -412,6 +468,10 @@ impl McpServer {
             pending_elicitations,
             server_capabilities: json!({}),
             alive,
+            // From here the child's life follows this binding: every `?` below drops
+            // it on the way out, which is what kills a server that started but never
+            // finished its handshake.
+            _kill_on_drop: kill_tx,
         };
 
         let init = server.request("initialize", json!({
@@ -465,6 +525,8 @@ impl McpServer {
             }
         }
 
+        // `..server` moves the kill sender into the returned value, so the child now
+        // outlives the handshake and dies with the handle instead.
         Ok(McpServer { tools, server_capabilities, ..server })
     }
 

crates/mcp-client/tests/startup_failure.rs

@@ -0,0 +1,203 @@
+//! A failed startup handshake must not leave the child process behind.
+//!
+//! Reproduces the failure that took a production instance down: a connector whose
+//! server starts fine, answers `initialize` with `-32601 Method not found`, and then
+//! never exits. `McpServer::start` returns `Err`, but the `Child` lives in the
+//! read-loop task rather than in the returned value — so before `KillOnStartFailure`
+//! nothing reaped it, and the supervisor's retry loop minted one orphan (three pipes
+//! and a pidfd) per attempt until the process hit its file-descriptor limit and
+//! stopped accepting connections altogether.
+//!
+//! The assertion is deliberately about the *process*, not about the error: the error
+//! was always correct, and it is the corpse that mattered. Skipped if `python3` is
+//! absent.
+
+#![cfg(unix)]
+
+use std::io::Write;
+use std::process::Command;
+use std::time::{Duration, Instant};
+
+use mcp_client::config::{McpServerConfig, McpTransport};
+use mcp_client::server::McpServer;
+
+/// Answers the handshake wrong and then hangs forever, ignoring stdin. The hanging
+/// is the point: a broken server that *exits* cleans up after itself and leaks
+/// nothing, so a test against one would pass with or without the fix.
+const WEDGED_SERVER: &str = r#"
+import sys, json, os, time
+
+with open(sys.argv[1], "w") as f:
+    f.write(str(os.getpid()))
+    f.flush()
+
+raw = sys.stdin.readline()
+msg = json.loads(raw)
+sys.stdout.write(json.dumps({
+    "jsonrpc": "2.0", "id": msg.get("id"),
+    "error": {"code": -32601, "message": "Method not found: initialize"}}) + "\n")
+sys.stdout.flush()
+
+while True:
+    time.sleep(3600)
+"#;
+
+/// Completes the handshake, then hangs forever the way a real idle connector does —
+/// blocked on a stdin the client holds open. Nothing about this server is broken; it
+/// is the *handle* being dropped that must end it.
+const HEALTHY_SERVER: &str = r#"
+import sys, json, os
+
+with open(sys.argv[1], "w") as f:
+    f.write(str(os.getpid()))
+    f.flush()
+
+def send(obj):
+    sys.stdout.write(json.dumps(obj) + "\n")
+    sys.stdout.flush()
+
+while True:
+    raw = sys.stdin.readline()
+    if not raw:
+        break
+    raw = raw.strip()
+    if not raw:
+        continue
+    msg = json.loads(raw)
+    mid, method = msg.get("id"), msg.get("method")
+    if method == "initialize":
+        send({"jsonrpc": "2.0", "id": mid, "result": {
+            "protocolVersion": "2025-11-25", "capabilities": {},
+            "serverInfo": {"name": "healthy", "version": "0"}}})
+    elif method == "tools/list":
+        send({"jsonrpc": "2.0", "id": mid, "result": {"tools": []}})
+    elif mid is not None:
+        send({"jsonrpc": "2.0", "id": mid, "result": {}})
+"#;
+
+fn python3_available() -> bool {
+    Command::new("python3").arg("--version").output().is_ok()
+}
+
+/// True while `pid` still names a process — including a zombie, which is what makes
+/// this an assertion about reaping and not merely about killing.
+fn alive(pid: &str) -> bool {
+    Command::new("kill")
+        .args(["-0", pid])
+        .output()
+        .map(|o| o.status.success())
+        .unwrap_or(false)
+}
+
+/// Writes `script` to a temp file and returns a config that runs it, plus the path
+/// the server will record its own pid at.
+fn fake_server(name: &str, script: &str) -> (McpServerConfig, std::path::PathBuf, std::path::PathBuf) {
+    let stamp       = format!("{}_{}", std::process::id(), name);
+    let script_path = std::env::temp_dir().join(format!("skald_{stamp}.py"));
+    let pid_path    = std::env::temp_dir().join(format!("skald_{stamp}.pid"));
+    std::fs::File::create(&script_path)
+        .unwrap()
+        .write_all(script.as_bytes())
+        .unwrap();
+
+    let cfg = McpServerConfig {
+        name:      name.to_string(),
+        transport: McpTransport::Stdio,
+        command:   Some("python3".to_string()),
+        args:      Some(vec![
+            script_path.to_string_lossy().to_string(),
+            pid_path.to_string_lossy().to_string(),
+        ]),
+        env:       None,
+        url:       None,
+        api_key:   None,
+        launch_in: None,
+    };
+    (cfg, script_path, pid_path)
+}
+
+fn recorded_pid(pid_path: &std::path::Path) -> String {
+    let pid = std::fs::read_to_string(pid_path)
+        .expect("the fake server should have recorded its pid");
+    let pid = pid.trim().to_string();
+    assert!(!pid.is_empty(), "empty pid file");
+    pid
+}
+
+/// Waits for `pid` to disappear, then reports whether it leaked. The kill is
+/// asynchronous — a dropped sender wakes the read-loop, which kills and then reaps —
+/// so this samples rather than checking once.
+async fn leaked(pid: &str) -> bool {
+    let deadline = Instant::now() + Duration::from_secs(10);
+    while alive(pid) && Instant::now() < deadline {
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    let leaked = alive(pid);
+    if leaked {
+        // Don't let a failing test leave behind the orphan it just detected.
+        let _ = Command::new("kill").args(["-9", pid]).output();
+    }
+    leaked
+}
+
+#[tokio::test]
+async fn failed_handshake_kills_and_reaps_the_child() {
+    if !python3_available() {
+        eprintln!("python3 not found — skipping startup-failure integration test");
+        return;
+    }
+
+    let (cfg, script_path, pid_path) = fake_server("wedged", WEDGED_SERVER);
+
+    // `McpServer` is not `Debug`, so unwrap the Result by hand rather than
+    // `expect_err`.
+    let err = match McpServer::start(&cfg, None, None, None).await {
+        Ok(_)  => panic!("a server that rejects `initialize` must not start"),
+        Err(e) => e,
+    };
+    assert!(
+        err.to_string().contains("protocol error"),
+        "unexpected error: {err}"
+    );
+
+    let pid    = recorded_pid(&pid_path);
+    let leaked = leaked(&pid).await;
+
+    let _ = std::fs::remove_file(&script_path);
+    let _ = std::fs::remove_file(&pid_path);
+
+    assert!(
+        !leaked,
+        "child {pid} survived a failed handshake — this is the file-descriptor leak"
+    );
+}
+
+/// The other half of the same defect: `stop_server`/`stop_all` drop the handle and
+/// document that as killing the process, but the child lives in the read-loop task,
+/// which holds its end of stdin — so before this fix the server simply stayed
+/// blocked on a read that would never return.
+#[tokio::test]
+async fn dropping_the_handle_kills_and_reaps_the_child() {
+    if !python3_available() {
+        eprintln!("python3 not found — skipping handle-drop integration test");
+        return;
+    }
+
+    let (cfg, script_path, pid_path) = fake_server("healthy", HEALTHY_SERVER);
+
+    let server = McpServer::start(&cfg, None, None, None)
+        .await
+        .unwrap_or_else(|e| panic!("the healthy fake server should start: {e}"));
+
+    let pid = recorded_pid(&pid_path);
+    assert!(alive(&pid), "the server should be running while the handle is held");
+
+    drop(server);
+
+    let leaked = leaked(&pid).await;
+
+    let _ = std::fs::remove_file(&script_path);
+    let _ = std::fs::remove_file(&pid_path);
+
+    assert!(!leaked, "child {pid} outlived the handle that owned it");
+}