Timeout MCP requests, reap stdio child, detect unexpected EOF

alexylon · alexylon · commit 05d60ab2680a · 2026-04-23T00:51:58.000+03:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -18,6 +18,8 @@ All notable changes to Sofos are documented in this file.
 
 ### Fixed
 
+- **MCP requests are now bounded at 120 seconds.** A frozen stdio MCP server used to wedge every subsequent MCP call because `BufRead::read_line` blocks indefinitely and the stdout mutex serialises requests. The blocking I/O now runs on a dedicated worker thread wrapped in `tokio::time::timeout`; on expiry the child is killed so queued callers fail fast with EPIPE rather than piling up on a mutex that will never release. HTTP MCP clients gain an explicit `reqwest` timeout for the same reason — previously they inherited the library default, which is unlimited.
+- **MCP child processes are now reaped on drop.** `std::process::Child::drop` does not wait, so every MCP server previously lingered as a zombie until sofos itself exited. `StdioClient` now has an explicit `Drop` that kills and waits the subprocess.
 - **"Goodbye!" no longer prints on the same line as the status row on exit.** The TUI left the cursor parked at the end of the status line; when the session summary short-circuited on zero usage, `print_goodbye` then printed `Goodbye!` flush against `… thinking: 5120 tok`. The teardown now emits an escape-newline on the no-summary path so the message always starts fresh.
 - **Safe-mode tool list now matches reality.** `SAFE_MODE_MESSAGE` told the model it had `list_directory, read_file and web_search`, but `get_read_only_tools()` actually exposes `list_directory, read_file, glob_files, web_fetch, web_search` (+ `search_code` when ripgrep is present). The model was being given a false toolset description on safe-mode entry.
 - **Corrupted session file on save no longer wipes the in-memory conversation.** `save_session` re-read the prior file to preserve `created_at`; if the file was unparseable (hand-edited, partial prior write, schema drift) the JSON error bubbled up and the current turn's messages were dropped on the floor. The stamp now falls back to `now` on any read/parse failure — losing a date is cheaper than losing the session. Regression test added (`save_session_survives_corrupted_prior_file`).
diff --git a/src/mcp/client.rs b/src/mcp/client.rs
@@ -7,6 +7,15 @@ use std::io::{BufRead, BufReader, Write};
 use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+/// Ceiling on a single MCP request (stdio read + HTTP round-trip). A
+/// misbehaving MCP server used to freeze every subsequent MCP call
+/// because `BufRead::read_line` blocks indefinitely and the stdout
+/// mutex serialises all requests. Two minutes is generous enough for
+/// slow remote search backends (PubMed / ClinicalTrials.gov) while
+/// still bounding a frozen server's blast radius to a single turn.
+const MCP_REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
 
 fn create_init_request() -> InitializeRequest {
     InitializeRequest {
@@ -95,14 +104,122 @@ impl McpClient {
     }
 }
 
+/// Take the stdin and (buffered) stdout out of a freshly-spawned child.
+/// Returning `Err` here is essentially unreachable because
+/// `Command::spawn` with `Stdio::piped()` on both ends always populates
+/// `Child::stdin` / `Child::stdout`, but we still handle it — leaking
+/// a live `Child` out to `Drop` would fail to reap it (the std type
+/// doesn't wait).
+fn take_child_pipes(process: &mut Child, server_name: &str) -> Result<(ChildStdin, ChildStdout)> {
+    let stdin = process.stdin.take().ok_or_else(|| {
+        SofosError::McpError(format!(
+            "Failed to get stdin for MCP server '{}'",
+            server_name
+        ))
+    })?;
+    let stdout = process.stdout.take().ok_or_else(|| {
+        SofosError::McpError(format!(
+            "Failed to get stdout for MCP server '{}'",
+            server_name
+        ))
+    })?;
+    Ok((stdin, stdout))
+}
+
+/// Write a single stdio message to the server (JSON-RPC request *or*
+/// notification). Shared by `send_request` and `send_notification` so
+/// the lock/write/flush sequence lives in one place.
+fn stdio_write_blocking(
+    server_name: &str,
+    stdin: &Arc<Mutex<ChildStdin>>,
+    payload: &str,
+) -> Result<()> {
+    let mut stdin_guard = stdin
+        .lock()
+        .map_err(|e| SofosError::McpError(format!("Failed to lock stdin: {}", e)))?;
+    writeln!(stdin_guard, "{}", payload).map_err(|e| {
+        SofosError::McpError(format!(
+            "Failed to write to MCP server '{}': {}",
+            server_name, e
+        ))
+    })?;
+    stdin_guard.flush().map_err(|e| {
+        SofosError::McpError(format!(
+            "Failed to flush stdin for MCP server '{}': {}",
+            server_name, e
+        ))
+    })?;
+    Ok(())
+}
+
+/// Run one stdio MCP request on a worker thread. Owns no `self` so it
+/// can be freely moved into `tokio::task::spawn_blocking`. Returns the
+/// parsed `JsonRpcResponse` envelope (not the `result` payload) so the
+/// caller can distinguish a server-reported error from a transport
+/// failure.
+fn stdio_request_blocking(
+    server_name: &str,
+    stdin: &Arc<Mutex<ChildStdin>>,
+    stdout: &Arc<Mutex<BufReader<ChildStdout>>>,
+    request_json: &str,
+) -> Result<JsonRpcResponse> {
+    stdio_write_blocking(server_name, stdin, request_json)?;
+
+    let mut stdout_guard = stdout
+        .lock()
+        .map_err(|e| SofosError::McpError(format!("Failed to lock stdout: {}", e)))?;
+
+    let mut response_line = String::new();
+    let bytes_read = stdout_guard.read_line(&mut response_line).map_err(|e| {
+        SofosError::McpError(format!(
+            "Failed to read from MCP server '{}': {}",
+            server_name, e
+        ))
+    })?;
+    // Zero bytes from `read_line` means the server closed stdout
+    // cleanly — typically a crash or exit between requests.
+    // Surface that plainly so the user isn't chasing a bogus
+    // "parse error" message for what's really a dead server.
+    if bytes_read == 0 {
+        return Err(SofosError::McpError(format!(
+            "MCP server '{}' closed stdout unexpectedly (server crashed or exited?)",
+            server_name
+        )));
+    }
+
+    serde_json::from_str(&response_line).map_err(|e| {
+        SofosError::McpError(format!(
+            "Failed to parse response from MCP server '{}': {}",
+            server_name, e
+        ))
+    })
+}
+
 pub struct StdioClient {
     server_name: String,
-    _process: Arc<Mutex<Child>>,
+    process: Arc<Mutex<Child>>,
     stdin: Arc<Mutex<ChildStdin>>,
     stdout: Arc<Mutex<BufReader<ChildStdout>>>,
     next_id: Arc<AtomicU64>,
 }
 
+impl Drop for StdioClient {
+    fn drop(&mut self) {
+        // `Child::drop` does NOT wait on the subprocess, so without
+        // this the MCP server lingers as a zombie until sofos itself
+        // exits. A detached reap task spawned by `kill_child_detached`
+        // may also be running when Drop fires — the mutex serialises
+        // them, and the kill/wait pair is idempotent: a second `kill`
+        // after the child already exited returns `InvalidInput`, and
+        // a second `wait` after the child was already reaped returns
+        // a harmless error. Both are discarded.
+        if let Ok(mut child) = self.process.lock() {
+            let _ = child.kill();
+            let _ = child.wait();
+        }
+    }
+}
+
 impl StdioClient {
     pub async fn new(server_name: String, config: McpServerConfig) -> Result<Self> {
         let command = config
@@ -129,23 +246,23 @@ impl StdioClient {
             ))
         })?;
 
-        let stdin = process.stdin.take().ok_or_else(|| {
-            SofosError::McpError(format!(
-                "Failed to get stdin for MCP server '{}'",
-                server_name
-            ))
-        })?;
-
-        let stdout = process.stdout.take().ok_or_else(|| {
-            SofosError::McpError(format!(
-                "Failed to get stdout for MCP server '{}'",
-                server_name
-            ))
-        })?;
+        // Take the pipes through a helper that reaps the child on
+        // error, so a (practically-impossible but not-type-proven)
+        // `None` from `take()` doesn't leak a zombie into the OS
+        // process table. Once the child is wrapped in `Self`, the
+        // `Drop` impl takes over.
+        let (stdin, stdout) = match take_child_pipes(&mut process, &server_name) {
+            Ok(pair) => pair,
+            Err(e) => {
+                let _ = process.kill();
+                let _ = process.wait();
+                return Err(e);
+            }
+        };
 
         let client = Self {
             server_name: server_name.clone(),
-            _process: Arc::new(Mutex::new(process)),
+            process: Arc::new(Mutex::new(process)),
             stdin: Arc::new(Mutex::new(stdin)),
             stdout: Arc::new(Mutex::new(BufReader::new(stdout))),
             next_id: Arc::new(AtomicU64::new(1)),
@@ -156,6 +273,53 @@ impl StdioClient {
         Ok(client)
     }
 
+    /// Run a blocking closure with the shared MCP timeout ceiling. On
+    /// timeout the child is killed off-thread so the async caller
+    /// doesn't pause the executor waiting for the OS to reap it. Used
+    /// by both `send_request` and `send_notification` so they share
+    /// the same lock/panic/timeout error vocabulary.
+    async fn run_with_timeout<T, F>(&self, label: &str, blocking: F) -> Result<T>
+    where
+        F: FnOnce() -> Result<T> + Send + 'static,
+        T: Send + 'static,
+    {
+        let task = tokio::task::spawn_blocking(blocking);
+        match tokio::time::timeout(MCP_REQUEST_TIMEOUT, task).await {
+            Ok(Ok(Ok(value))) => Ok(value),
+            Ok(Ok(Err(e))) => Err(e),
+            Ok(Err(join_err)) => Err(SofosError::McpError(format!(
+                "MCP worker panicked for server '{}' during {}: {}",
+                self.server_name, label, join_err
+            ))),
+            Err(_) => {
+                self.kill_child_detached();
+                Err(SofosError::McpError(format!(
+                    "MCP server '{}' {} timed out after {}s",
+                    self.server_name,
+                    label,
+                    MCP_REQUEST_TIMEOUT.as_secs()
+                )))
+            }
+        }
+    }
+
+    /// Kill + reap the child on a blocking thread without waiting for
+    /// it to finish here. Used from async timeout handlers so a slow
+    /// `Child::wait` (milliseconds in practice, but the call is
+    /// synchronous) never pauses the tokio executor. Firing and
+    /// forgetting is safe because tokio drains blocking tasks on
+    /// runtime shutdown, and the `Drop` impl is idempotent against
+    /// an already-reaped child.
+    fn kill_child_detached(&self) {
+        let process = Arc::clone(&self.process);
+        tokio::task::spawn_blocking(move || {
+            if let Ok(mut child) = process.lock() {
+                let _ = child.kill();
+                let _ = child.wait();
+            }
+        });
+    }
+
     async fn initialize(&self) -> Result<()> {
         let response = self
             .send_request(
@@ -175,47 +339,16 @@ impl StdioClient {
     async fn send_request(&self, method: &str, params: Option<Value>) -> Result<Value> {
         let id = self.next_id.fetch_add(1, Ordering::SeqCst);
         let request = JsonRpcRequest::new(id, method.to_string(), params);
-
         let request_json = serde_json::to_string(&request)?;
 
-        {
-            let mut stdin = self
-                .stdin
-                .lock()
-                .map_err(|e| SofosError::McpError(format!("Failed to lock stdin: {}", e)))?;
-            writeln!(stdin, "{}", request_json).map_err(|e| {
-                SofosError::McpError(format!(
-                    "Failed to write to MCP server '{}': {}",
-                    self.server_name, e
-                ))
-            })?;
-            stdin.flush().map_err(|e| {
-                SofosError::McpError(format!(
-                    "Failed to flush stdin for MCP server '{}': {}",
-                    self.server_name, e
-                ))
-            })?;
-        }
-
-        let mut stdout = self
-            .stdout
-            .lock()
-            .map_err(|e| SofosError::McpError(format!("Failed to lock stdout: {}", e)))?;
-
-        let mut response_line = String::new();
-        stdout.read_line(&mut response_line).map_err(|e| {
-            SofosError::McpError(format!(
-                "Failed to read from MCP server '{}': {}",
-                self.server_name, e
-            ))
-        })?;
-
-        let response: JsonRpcResponse = serde_json::from_str(&response_line).map_err(|e| {
-            SofosError::McpError(format!(
-                "Failed to parse response from MCP server '{}': {}",
-                self.server_name, e
-            ))
-        })?;
+        let server_name = self.server_name.clone();
+        let stdin = Arc::clone(&self.stdin);
+        let stdout = Arc::clone(&self.stdout);
+        let response = self
+            .run_with_timeout("request", move || {
+                stdio_request_blocking(&server_name, &stdin, &stdout, &request_json)
+            })
+            .await?;
 
         if let Some(error) = response.error {
             return Err(SofosError::McpError(format!(
@@ -237,27 +370,17 @@ impl StdioClient {
             "jsonrpc": "2.0",
             "method": method,
         });
-
         let notification_json = serde_json::to_string(&notification)?;
 
-        let mut stdin = self
-            .stdin
-            .lock()
-            .map_err(|e| SofosError::McpError(format!("Failed to lock stdin: {}", e)))?;
-        writeln!(stdin, "{}", notification_json).map_err(|e| {
-            SofosError::McpError(format!(
-                "Failed to write notification to MCP server '{}': {}",
-                self.server_name, e
-            ))
-        })?;
-        stdin.flush().map_err(|e| {
-            SofosError::McpError(format!(
-                "Failed to flush stdin for MCP server '{}': {}",
-                self.server_name, e
-            ))
-        })?;
-
-        Ok(())
+        // Notifications are one-way (no response to read), but the
+        // write itself can still block forever if the child has
+        // wedged its read side. Same timeout path as `send_request`.
+        let server_name = self.server_name.clone();
+        let stdin = Arc::clone(&self.stdin);
+        self.run_with_timeout("notification", move || {
+            stdio_write_blocking(&server_name, &stdin, &notification_json)
+        })
+        .await
     }
 
     pub async fn list_tools(&self) -> Result<Vec<McpTool>> {
@@ -285,7 +408,14 @@ impl HttpClient {
 
         let headers = config.headers.unwrap_or_default();
 
-        let client = reqwest::Client::new();
+        // A bare `reqwest::Client::new()` uses no request timeout at
+        // all, so a slow remote MCP server could stall a turn forever.
+        // Set the shared MCP ceiling at client-construction time so
+        // every call-site inherits it without extra threading.
+        let client = reqwest::Client::builder()
+            .timeout(MCP_REQUEST_TIMEOUT)
+            .build()
+            .map_err(|e| SofosError::McpError(format!("Failed to build MCP HTTP client: {}", e)))?;
 
         let http_client = Self {
             server_name: server_name.clone(),
