car_engine/

mcp.rs

//! MCP (Model Context Protocol) server integration.
//!
//! Discovers tools from MCP servers via stdin/stdout JSON-RPC and registers
//! them into the canonical tool registry. MCP tools participate in the same
//! capability/permission/policy flow as all other tools.

use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStderr};
use tokio::sync::{oneshot, Mutex};

/// Configuration for an MCP server.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpServerConfig {
    /// Display name for this server.
    pub name: String,
    /// Command to launch the server.
    pub command: String,
    /// Arguments for the command.
    #[serde(default)]
    pub args: Vec<String>,
    /// Environment variables.
    #[serde(default)]
    pub env: HashMap<String, String>,
    /// Working directory.
    pub cwd: Option<String>,
}

/// Map of in-flight request id → the waiter to deliver its response to.
type Pending = Arc<StdMutex<HashMap<u64, oneshot::Sender<McpResponse>>>>;

/// A running MCP server connection.
///
/// A background **reader task** owns stdout and demultiplexes responses by id
/// into per-request `oneshot` channels. `send_request` writes to stdin and then
/// awaits its channel — never `read_line` directly. So a request timeout (or any
/// cancellation of the caller's future) only drops a receiver; the reader keeps
/// consuming the stream and discards the now-orphaned response. This makes the
/// transport cancel-safe by construction: there is no read to interrupt
/// mid-line, hence no desync and no poison/recovery dance.
pub struct McpServer {
    config: McpServerConfig,
    child: Child,
    stdin: tokio::io::BufWriter<tokio::process::ChildStdin>,
    next_id: u64,
    pending: Pending,
    /// Background reader task handle (aborted on reconnect/drop).
    reader: tokio::task::JoinHandle<()>,
    /// Background stderr-drain task (aborted on reconnect/drop). Must exist:
    /// stderr is piped, so an undrained chatty server fills the pipe buffer and
    /// deadlocks its own stdout writes.
    stderr_reader: tokio::task::JoinHandle<()>,
    /// Cleared when the reader exits (EOF / read error) — i.e. the connection is
    /// dead. The next `send_request` reconnects.
    alive: Arc<AtomicBool>,
    /// Backstop timeout for awaiting any single response. Callers usually impose
    /// their own (per-action `timeout_ms` in the executor); this bounds requests
    /// that don't, so a silent server can't hang a call forever.
    request_timeout: Duration,
}

impl Drop for McpServer {
    fn drop(&mut self) {
        // tokio doesn't kill children or abort tasks on drop by default. Do it
        // explicitly so a dropped server (e.g. `shutdown_all` draining the map)
        // doesn't leak the reader/stderr tasks or the child process.
        self.reader.abort();
        self.stderr_reader.abort();
        let _ = self.child.start_kill();
    }
}

/// An MCP tool discovered from a server.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpToolInfo {
    pub name: String,
    pub description: Option<String>,
    #[serde(rename = "inputSchema")]
    pub input_schema: Option<Value>,
}

/// Transport-agnostic MCP client session.
///
/// The stdio [`McpServer`] (subprocess transport) implements this
/// directly; the HTTP-streamable session in `car-connectors` (remote
/// connectors) implements it too. [`McpToolExecutor`] holds sessions
/// as `dyn McpSession` so it can route a tool call to either transport
/// without knowing which one backs a given server. This is the seam
/// that lets remote MCP connectors reuse the exact routing, fallback,
/// and registration machinery the stdio path already has.
#[async_trait::async_trait]
pub trait McpSession: Send {
    /// Discover the tools this server exposes.
    async fn list_tools(&mut self) -> Result<Vec<McpToolInfo>, String>;
    /// Invoke a tool by its bare (server-side) name.
    async fn call_tool(&mut self, name: &str, arguments: Value) -> Result<Value, String>;
    /// Invoke a tool, bounding the response await by `timeout` (else the
    /// session's backstop). Default ignores `timeout` (back-compat for sessions
    /// where calls are short); the stdio [`McpServer`] honors it so a
    /// long-running `run_command` isn't cut off mid-execution.
    async fn call_tool_with_timeout(
        &mut self,
        name: &str,
        arguments: Value,
        _timeout: Option<Duration>,
    ) -> Result<Value, String> {
        self.call_tool(name, arguments).await
    }
    /// The server's registered name (the routing key).
    fn name(&self) -> &str;
}

#[async_trait::async_trait]
impl McpSession for McpServer {
    async fn list_tools(&mut self) -> Result<Vec<McpToolInfo>, String> {
        McpServer::list_tools(self).await
    }
    async fn call_tool(&mut self, name: &str, arguments: Value) -> Result<Value, String> {
        McpServer::call_tool(self, name, arguments).await
    }
    async fn call_tool_with_timeout(
        &mut self,
        name: &str,
        arguments: Value,
        timeout: Option<Duration>,
    ) -> Result<Value, String> {
        McpServer::call_tool_with_timeout(self, name, arguments, timeout).await
    }
    fn name(&self) -> &str {
        McpServer::name(self)
    }
}

/// MCP JSON-RPC request.
#[derive(Debug, Serialize)]
struct McpRequest {
    jsonrpc: &'static str,
    method: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    params: Option<Value>,
    id: u64,
}

/// MCP JSON-RPC response.
#[derive(Debug, Deserialize)]
struct McpResponse {
    result: Option<Value>,
    error: Option<McpError>,
    /// Echoed request id — used to route the response to its waiter.
    id: Option<u64>,
}

#[derive(Debug, Deserialize)]
struct McpError {
    #[allow(dead_code)]
    code: Option<i64>,
    message: String,
}

/// Route one stdout line to its waiting request, by id. Unparseable lines and
/// notifications (no id) are ignored; an id with no waiter (a late response from
/// a request whose caller already gave up) is discarded — the stream stays
/// synchronized either way. Pure over `pending`, so the demux is unit-testable
/// without spawning a subprocess.
fn route_line(line: &str, pending: &StdMutex<HashMap<u64, oneshot::Sender<McpResponse>>>) {
    let resp: McpResponse = match serde_json::from_str(line) {
        Ok(r) => r,
        Err(_) => return, // notification / log noise / partial line — ignore
    };
    if let Some(id) = resp.id {
        if let Some(tx) = pending.lock().unwrap().remove(&id) {
            // Receiver may be gone (caller timed out / was cancelled) — fine.
            let _ = tx.send(resp);
        }
        // Unknown id → orphaned/duplicate response; discard.
    }
}

/// Background task: read newline-delimited responses and route each by id until
/// the stream closes. On exit, mark the connection dead and drop all waiters
/// (their `recv()` then errors out).
async fn reader_loop<R: AsyncBufRead + Unpin>(
    mut stdout: R,
    pending: Pending,
    alive: Arc<AtomicBool>,
    server_name: String,
) {
    let mut line = String::new();
    loop {
        line.clear();
        match stdout.read_line(&mut line).await {
            Ok(0) | Err(_) => break, // EOF or read error → connection dead
            Ok(_) => route_line(&line, &pending),
        }
    }
    alive.store(false, Ordering::SeqCst);
    pending.lock().unwrap().clear(); // dropping senders wakes waiters with an error
    tracing::debug!(server = %server_name, "MCP reader exited; connection closed");
}

/// Drain the child's stderr to the log so a verbose server can't fill the pipe
/// buffer and deadlock its own stdout writes.
async fn stderr_drain_loop(stderr: ChildStderr, server_name: String) {
    let mut lines = BufReader::new(stderr).lines();
    while let Ok(Some(line)) = lines.next_line().await {
        tracing::debug!(server = %server_name, "mcp stderr: {line}");
    }
}

impl McpServer {
    /// Start an MCP server and initialize the connection.
    pub async fn start(config: McpServerConfig) -> Result<Self, String> {
        // `program_command` routes a Windows `.cmd`/`.bat` npm shim (npx/npm/
        // yarn — the canonical MCP launcher shape) through `cmd /C`; a bare
        // `Command::new("npx")` fails with os error 193 on Windows.
        let mut cmd = crate::spawn::program_command(&config.command);
        cmd.args(&config.args)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        if let Some(ref cwd) = config.cwd {
            cmd.current_dir(cwd);
        }
        for (k, v) in &config.env {
            cmd.env(k, v);
        }

        let mut child = cmd
            .spawn()
            .map_err(|e| format!("failed to start MCP server '{}': {}", config.name, e))?;

        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| "MCP server has no stdin".to_string())?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| "MCP server has no stdout".to_string())?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| "MCP server has no stderr".to_string())?;

        let pending: Pending = Arc::new(StdMutex::new(HashMap::new()));
        let alive = Arc::new(AtomicBool::new(true));
        let reader = tokio::spawn(reader_loop(
            BufReader::new(stdout),
            Arc::clone(&pending),
            Arc::clone(&alive),
            config.name.clone(),
        ));
        let stderr_reader = tokio::spawn(stderr_drain_loop(stderr, config.name.clone()));

        let mut server = Self {
            config,
            child,
            stdin: tokio::io::BufWriter::new(stdin),
            next_id: 1,
            pending,
            reader,
            stderr_reader,
            alive,
            // Backstop for calls with no explicit per-call timeout (list_tools,
            // fast fs ops, connector tools). Long-running calls (run_command)
            // pass their own timeout via call_tool_with_timeout and are not
            // bound by this. Raised from 120s: 120 spuriously cut compute-task
            // run_commands still executing server-side (ALE: 37 false timeouts).
            request_timeout: Duration::from_secs(600),
        };

        // Send initialize
        server
            .send_request(
                "initialize",
                Some(serde_json::json!({
                    "protocolVersion": "2024-11-05",
                    "capabilities": {},
                    "clientInfo": {
                        "name": "car-runtime",
                        "version": env!("CARGO_PKG_VERSION")
                    }
                })),
            )
            .await?;

        // Send initialized notification (no id, per MCP spec)
        let notification = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/initialized"
        });
        let msg =
            serde_json::to_string(&notification).map_err(|e| format!("serialize error: {e}"))?;
        server.write_message(&msg).await?;

        Ok(server)
    }

    /// Respawn the child and reader, replacing a dead session in place.
    async fn reconnect(&mut self) -> Result<(), String> {
        tracing::warn!(server = %self.config.name, "MCP connection closed; reconnecting (server-side state is lost)");
        self.reader.abort(); // stop the old reader + stderr drain tasks
        self.stderr_reader.abort();
        let _ = self.child.kill().await;
        // Box the recursive future to break the *type-size* recursion: start()
        // handshakes via send_request. This cannot loop at runtime — start()
        // builds a fresh, alive server, so its handshake never re-enters
        // reconnect; a failing handshake propagates Err instead.
        let fresh = Box::pin(Self::start(self.config.clone())).await?;
        *self = fresh;
        Ok(())
    }

    /// Write one newline-delimited JSON message to the server.
    async fn write_message(&mut self, msg: &str) -> Result<(), String> {
        self.stdin
            .write_all(msg.as_bytes())
            .await
            .map_err(|e| format!("write to MCP server: {e}"))?;
        self.stdin
            .write_all(b"\n")
            .await
            .map_err(|e| format!("write newline: {e}"))?;
        self.stdin
            .flush()
            .await
            .map_err(|e| format!("flush: {e}"))?;
        Ok(())
    }

    async fn send_request(&mut self, method: &str, params: Option<Value>) -> Result<Value, String> {
        self.send_request_with_timeout(method, params, None).await
    }

    /// Like [`send_request`], but bounds the response await by `timeout` when
    /// given (else the `request_timeout` backstop). A long-running tool call
    /// (e.g. `run_command` with a large `timeout`) must not be cut off by the
    /// generic backstop while the command is still executing on the server.
    async fn send_request_with_timeout(
        &mut self,
        method: &str,
        params: Option<Value>,
        timeout: Option<Duration>,
    ) -> Result<Value, String> {
        let await_timeout = timeout.unwrap_or(self.request_timeout);
        // Reconnect if the reader has exited (connection died).
        if !self.alive.load(Ordering::SeqCst) {
            self.reconnect().await.map_err(|e| {
                format!(
                    "MCP session '{}' is dead and reconnect failed: {e}",
                    self.config.name
                )
            })?;
        }

        let id = self.next_id;
        self.next_id += 1;

        // Register our waiter BEFORE writing, so a fast response can't arrive
        // before the reader knows where to route it.
        let (tx, rx) = oneshot::channel();
        self.pending.lock().unwrap().insert(id, tx);

        let req = McpRequest {
            jsonrpc: "2.0",
            method: method.to_string(),
            params,
            id,
        };
        let msg = serde_json::to_string(&req).map_err(|e| format!("serialize error: {e}"))?;

        if let Err(e) = self.write_message(&msg).await {
            self.pending.lock().unwrap().remove(&id);
            self.alive.store(false, Ordering::SeqCst); // broken pipe → dead
            return Err(e);
        }

        // Await the channel — NOT a read. A timeout (or upstream cancellation
        // dropping this future) just drops the receiver; the reader still
        // consumes and discards the eventual response, so the stream never
        // desyncs. We only clean up our pending entry on timeout.
        let resp = match tokio::time::timeout(await_timeout, rx).await {
            Ok(Ok(resp)) => resp,
            Ok(Err(_)) => {
                return Err(format!(
                    "MCP server '{}' closed the connection",
                    self.config.name
                ))
            }
            Err(_) => {
                self.pending.lock().unwrap().remove(&id);
                return Err(format!("MCP request '{method}' timed out"));
            }
        };

        if let Some(err) = resp.error {
            return Err(format!("MCP error: {}", err.message));
        }
        resp.result
            .ok_or_else(|| "MCP server returned no result".to_string())
    }

    /// Discover tools from this MCP server.
    pub async fn list_tools(&mut self) -> Result<Vec<McpToolInfo>, String> {
        let result = self.send_request("tools/list", None).await?;
        let tools = result
            .get("tools")
            .and_then(|t| t.as_array())
            .cloned()
            .unwrap_or_default();

        tools
            .into_iter()
            .map(|t| serde_json::from_value(t).map_err(|e| format!("invalid tool definition: {e}")))
            .collect()
    }

    /// Call a tool on this MCP server.
    pub async fn call_tool(&mut self, name: &str, arguments: Value) -> Result<Value, String> {
        self.call_tool_with_timeout(name, arguments, None).await
    }

    /// Call a tool, bounding the response await by `timeout` (else the backstop).
    /// Callers that know a tool can run long (e.g. the substrate's `run_command`
    /// carrying its own `timeout`) pass it here so CAR doesn't abandon a call
    /// that is still executing server-side.
    pub async fn call_tool_with_timeout(
        &mut self,
        name: &str,
        arguments: Value,
        timeout: Option<Duration>,
    ) -> Result<Value, String> {
        let result = self
            .send_request_with_timeout(
                "tools/call",
                Some(serde_json::json!({
                    "name": name,
                    "arguments": arguments,
                })),
                timeout,
            )
            .await?;

        parse_tool_result(result)
    }

    /// Shut down the MCP server gracefully.
    pub async fn shutdown(mut self) {
        let _ = self.stdin.shutdown().await;
        let _ = self.child.kill().await;
        let _ = self.child.wait().await;
    }

    /// Get the server name.
    pub fn name(&self) -> &str {
        &self.config.name
    }
}

/// MCP tool executor -- routes tool calls to the appropriate MCP server.
///
/// Servers are held as `dyn McpSession`, so a stdio subprocess
/// ([`McpServer`]) and a remote HTTP connector (`car-connectors`)
/// coexist in the same routing table.
pub struct McpToolExecutor {
    servers: Arc<Mutex<HashMap<String, Arc<Mutex<dyn McpSession>>>>>,
    /// Maps tool_name -> server_name for routing.
    tool_routes: Arc<Mutex<HashMap<String, String>>>,
    /// Optional fallback for non-MCP tools.
    fallback: Option<Arc<dyn super::ToolExecutor>>,
}

impl McpToolExecutor {
    pub fn new() -> Self {
        Self {
            servers: Arc::new(Mutex::new(HashMap::new())),
            tool_routes: Arc::new(Mutex::new(HashMap::new())),
            fallback: None,
        }
    }

    pub fn with_fallback(mut self, fallback: Arc<dyn super::ToolExecutor>) -> Self {
        self.fallback = Some(fallback);
        self
    }

    /// Add an MCP server and discover its tools.
    /// Returns the list of discovered tool names (canonical form: `mcp_{server}_{tool}`).
    pub async fn add_server(&self, mut server: McpServer) -> Result<Vec<String>, String> {
        let server_name = server.config.name.clone();
        let tools = server.list_tools().await?;

        let tool_names: Vec<String> = tools
            .iter()
            .map(|t| format!("mcp_{}_{}", server_name, t.name))
            .collect();

        // Register tool routes
        {
            let mut routes = self.tool_routes.lock().await;
            for (info, canonical_name) in tools.iter().zip(tool_names.iter()) {
                routes.insert(canonical_name.clone(), server_name.clone());
                // Also register the bare name for convenience
                routes.insert(info.name.clone(), server_name.clone());
            }
        }

        // Store server
        let session: Arc<Mutex<dyn McpSession>> = Arc::new(Mutex::new(server));
        self.servers.lock().await.insert(server_name, session);

        Ok(tool_names)
    }

    /// Register an already-connected session under `name` **without**
    /// adding any tool routes. Unlike [`add_server`](crate::mcp::McpToolExecutor::add_server), routes are added
    /// explicitly via [`set_route`](crate::mcp::McpToolExecutor::set_route) so the caller can gate exactly
    /// which of a server's tools become callable — the connector
    /// manager uses this to keep discovered-but-not-yet-enabled tools
    /// invisible and unroutable.
    pub async fn add_session(&self, name: impl Into<String>, session: Arc<Mutex<dyn McpSession>>) {
        self.servers.lock().await.insert(name.into(), session);
    }

    /// Route the canonical tool name `tool` (the namespaced name the
    /// model sees, e.g. `mcp_github_create_issue`) to `server`. On
    /// dispatch the bare server-side name is recovered by stripping the
    /// `mcp_{server}_` prefix.
    pub async fn set_route(&self, tool: impl Into<String>, server: impl Into<String>) {
        self.tool_routes
            .lock()
            .await
            .insert(tool.into(), server.into());
    }

    /// Drop the session registered under `name`, and every route that
    /// pointed at it, only while that entry is still `session`. A
    /// different session under the same name, registered since `session`
    /// was, is left alone with its routes, so a cleanup that outlasts a
    /// remove-then-add of the same name cannot take the newer one down.
    /// Returns whether `session` was the one registered.
    pub async fn remove_session(&self, name: &str, session: &Arc<Mutex<dyn McpSession>>) -> bool {
        let mut servers = self.servers.lock().await;
        if !servers
            .get(name)
            .is_some_and(|current| Arc::ptr_eq(current, session))
        {
            return false;
        }
        servers.remove(name);
        // Under the server lock: a newer session cannot be registered, and
        // so cannot have been routed, between the check and this.
        self.tool_routes.lock().await.retain(|_, v| v != name);
        true
    }

    /// Drop a session and every route that pointed at it.
    pub async fn remove_server(&self, name: &str) {
        self.servers.lock().await.remove(name);
        self.tool_routes.lock().await.retain(|_, v| v != name);
    }

    /// Drop only the routes for `server` (e.g. when disabling its
    /// tools) while keeping the underlying connection alive.
    pub async fn clear_routes_for_server(&self, name: &str) {
        self.tool_routes.lock().await.retain(|_, v| v != name);
    }

    /// Drop a single tool route by its canonical (model-visible) name,
    /// keeping the connection and the server's other routes intact.
    /// Used when disabling one tool of a connector.
    pub async fn remove_route(&self, tool: &str) {
        self.tool_routes.lock().await.remove(tool);
    }

    /// True if some connected server currently handles `tool`.
    pub async fn handles(&self, tool: &str) -> bool {
        self.tool_routes.lock().await.contains_key(tool)
    }

    /// Retrieve a live MCP session by its server name, if connected.
    ///
    /// Returns the same `Arc<Mutex<dyn McpSession>>` handle stored by
    /// [`add_session`](crate::mcp::McpToolExecutor::add_session)/[`add_server`](crate::mcp::McpToolExecutor::add_server), so a caller (e.g. the daemon binding a
    /// connector-driven session to an [`crate::substrate::McpSubstrate`]) can
    /// wrap the *existing* connection rather than dialing a second one. The
    /// substrate and the `mcp_{name}_*` routes then share one session.
    pub async fn session(&self, name: &str) -> Option<Arc<Mutex<dyn McpSession>>> {
        self.servers.lock().await.get(name).cloned()
    }

    /// Build a view over the same shared server/route state with a
    /// per-caller `fallback`. The daemon keeps one shared executor for
    /// all connectors and hands each WS session its own view whose
    /// fallback is that session's WS tool executor.
    pub fn share_with_fallback(&self, fallback: Arc<dyn super::ToolExecutor>) -> Self {
        Self {
            servers: Arc::clone(&self.servers),
            tool_routes: Arc::clone(&self.tool_routes),
            fallback: Some(fallback),
        }
    }

    /// Get tool schemas from all connected MCP servers.
    pub async fn tool_schemas(&self) -> Vec<(String, car_ir::ToolSchema)> {
        let mut schemas = Vec::new();
        // Snapshot the sessions and let the server lock go before any of
        // them is asked over the network, so a slow server does not hold
        // up registering, removing or calling every other one.
        let servers: Vec<(String, Arc<Mutex<dyn McpSession>>)> = self
            .servers
            .lock()
            .await
            .iter()
            .map(|(name, session)| (name.clone(), Arc::clone(session)))
            .collect();
        for (server_name, server) in servers.iter() {
            let mut srv = server.lock().await;
            if let Ok(tools) = srv.list_tools().await {
                for tool in tools {
                    let canonical_name = format!("mcp_{}_{}", server_name, tool.name);
                    schemas.push((
                        server_name.clone(),
                        car_ir::ToolSchema {
                            name: canonical_name,
                            source: car_ir::ToolSourceKind::Mcp,
                            description: tool.description.unwrap_or_default(),
                            parameters: tool
                                .input_schema
                                .unwrap_or(serde_json::json!({"type": "object"})),
                            returns: None,
                            idempotent: false,
                            cache_ttl_secs: None,
                            rate_limit: None,
                        },
                    ));
                }
            }
        }
        schemas
    }

    /// The session `tool` routes to, with its server name. The session is
    /// cloned out, so the server lock is not held across a call. The route
    /// is checked again once the session is in hand: between the lookups the
    /// server can be removed and another registered under its name, and a
    /// call that resolved the old route must not run on the new session
    /// unless the new one routes this tool too.
    async fn routed_session(&self, tool: &str) -> Option<(String, Arc<Mutex<dyn McpSession>>)> {
        let server_name = self.tool_routes.lock().await.get(tool).cloned()?;
        let server = self.servers.lock().await.get(&server_name).cloned()?;
        self.tool_routes
            .lock()
            .await
            .get(tool)
            .is_some_and(|routed| *routed == server_name)
            .then_some((server_name, server))
    }

    /// Shut down all MCP servers.
    pub async fn shutdown_all(&self) {
        let mut servers = self.servers.lock().await;
        // Dropping the Arc<Mutex<McpServer>> will drop the Child, killing the process.
        servers.drain();
    }
}

impl Default for McpToolExecutor {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait::async_trait]
impl super::ToolExecutor for McpToolExecutor {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        self.execute_with_action(tool, params, "", None).await
    }

    async fn execute_with_action(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        self.execute_with_action_in_session(tool, params, action_id, timeout_ms, None, 1)
            .await
    }

    async fn execute_with_action_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
    ) -> Result<Value, String> {
        // Find which server handles this tool
        if let Some((server_name, server)) = self.routed_session(tool).await {
            let mut srv = server.lock().await;
            // Strip the mcp_{server}_ prefix to get the bare tool name
            let bare_name = tool
                .strip_prefix(&format!("mcp_{}_", server_name))
                .unwrap_or(tool);
            // Honor the action's declared budget instead of discarding it
            // (Parslee-ai/car#259): `timeout_ms` arrived here and was
            // dropped, so an MCP call was bound only by the session's
            // backstop and a short-budget action could never cut it off.
            // With no declared budget we pass `None`, which is exactly the
            // previous `call_tool` behavior (it delegates to
            // `call_tool_with_timeout(.., None)` and falls back to the
            // backstop).
            return srv
                .call_tool_with_timeout(
                    bare_name,
                    params.clone(),
                    timeout_ms.map(Duration::from_millis),
                )
                .await;
        }

        // Fallback
        if let Some(ref fallback) = self.fallback {
            return fallback
                .execute_with_action_in_session(
                    tool, params, action_id, timeout_ms, session_id, attempt,
                )
                .await;
        }

        Err(format!("unknown MCP tool: '{}'", tool))
    }

    async fn execute_with_action_state_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
        expected_effects: &std::collections::HashMap<String, Value>,
        return_schema: Option<&Value>,
    ) -> Result<super::ToolExecution, String> {
        let routed_to_live_mcp = self.routed_session(tool).await.is_some();
        if routed_to_live_mcp {
            return self
                .execute_with_action_in_session(
                    tool, params, action_id, timeout_ms, session_id, attempt,
                )
                .await
                .map(super::ToolExecution::output_only);
        }
        if let Some(ref fallback) = self.fallback {
            return fallback
                .execute_with_action_state_in_session(
                    tool,
                    params,
                    action_id,
                    timeout_ms,
                    session_id,
                    attempt,
                    expected_effects,
                    return_schema,
                )
                .await;
        }
        Err(format!("unknown MCP tool: '{}'", tool))
    }

    async fn execute_classified(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
        expected_effects: &std::collections::HashMap<String, Value>,
        return_schema: Option<&Value>,
    ) -> Result<super::ToolExecution, car_ir::ToolFailure> {
        // Retain the selected session. A second route lookup could race
        // remove_route/remove_server and silently enter the legacy String
        // fallback, erasing an explicitly terminal callback failure.
        if let Some((server_name, server)) = self.routed_session(tool).await {
            let bare_name = tool
                .strip_prefix(&format!("mcp_{server_name}_"))
                .unwrap_or(tool);
            return server
                .lock()
                .await
                .call_tool_with_timeout(
                    bare_name,
                    params.clone(),
                    timeout_ms.map(Duration::from_millis),
                )
                .await
                .map(super::ToolExecution::output_only)
                // `ordinary`: the text is the MCP server's, and a server
                // answering "unknown tool" must not reach the runtime's
                // built-ins through a bare-name route (Parslee-ai/car#3104).
                .map_err(car_ir::ToolFailure::ordinary);
        }
        if let Some(ref fallback) = self.fallback {
            return fallback
                .execute_classified(
                    tool,
                    params,
                    action_id,
                    timeout_ms,
                    session_id,
                    attempt,
                    expected_effects,
                    return_schema,
                )
                .await;
        }
        Err(car_ir::ToolFailure::ordinary(format!(
            "unknown MCP tool: '{tool}'"
        )))
    }
}

/// Parse an MCP `tools/call` result into the joined text content (`Ok`) or a
/// tool-level error (`Err`), honoring the spec `isError` flag.
///
/// MCP distinguishes two failure channels: a JSON-RPC protocol error (handled in
/// [`McpServer::call_tool`] before this is reached) and a *tool-level* error
/// returned as a normal response with `isError: true`. The latter MUST propagate
/// as `Err` — otherwise a failed tool call is surfaced to the caller (and the
/// model) as success. Regression this guards: the `vm` substrate bridge returned
/// `isError:true` + "fetch failed" when its transport to the VM was down, and CAR
/// handed the model `{exit_code:0, stdout:"fetch failed"}` — a dead channel
/// reported as a successful command, burning ~20 agent turns before the model
/// guessed something was wrong.
fn parse_tool_result(result: Value) -> Result<Value, String> {
    let text = result
        .get("content")
        .and_then(|c| c.as_array())
        .and_then(|content| {
            let texts: Vec<&str> = content
                .iter()
                .filter_map(|block| {
                    if block.get("type").and_then(|t| t.as_str()) == Some("text") {
                        block.get("text").and_then(|t| t.as_str())
                    } else {
                        None
                    }
                })
                .collect();
            if texts.is_empty() {
                None
            } else {
                Some(texts.join("\n"))
            }
        });

    if result.get("isError").and_then(|v| v.as_bool()) == Some(true) {
        return Err(
            text.unwrap_or_else(|| "tool returned isError with no text content".to_string())
        );
    }

    if let Some(text) = text {
        return Ok(Value::String(text));
    }

    Ok(result)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn tool_result_iserror_true_propagates_as_err() {
        // The exact shape the vm substrate bridge returns on a dead transport.
        let r = json!({
            "content": [{"type": "text", "text": "fetch failed"}],
            "isError": true
        });
        assert_eq!(parse_tool_result(r), Err("fetch failed".to_string()));
    }

    #[test]
    fn tool_result_iserror_true_without_text_still_errs() {
        let r = json!({ "content": [], "isError": true });
        assert!(parse_tool_result(r).is_err());
    }

    #[test]
    fn tool_result_success_returns_joined_text() {
        let r = json!({
            "content": [{"type": "text", "text": "line1"}, {"type": "text", "text": "line2"}],
            "isError": false
        });
        assert_eq!(
            parse_tool_result(r),
            Ok(Value::String("line1\nline2".to_string()))
        );
    }

    #[test]
    fn tool_result_no_iserror_field_is_success() {
        // Most servers omit isError on success; absence must NOT be treated as error.
        let r = json!({ "content": [{"type": "text", "text": "ok"}] });
        assert_eq!(parse_tool_result(r), Ok(Value::String("ok".to_string())));
    }

    #[test]
    fn tool_result_structured_no_text_passes_through() {
        let r = json!({ "structuredContent": {"x": 1} });
        assert_eq!(parse_tool_result(r.clone()), Ok(r));
    }

    fn pending() -> StdMutex<HashMap<u64, oneshot::Sender<McpResponse>>> {
        StdMutex::new(HashMap::new())
    }

    #[tokio::test]
    async fn routes_response_to_matching_waiter() {
        let p = pending();
        let (tx, rx) = oneshot::channel();
        p.lock().unwrap().insert(7, tx);
        route_line(r#"{"jsonrpc":"2.0","id":7,"result":{"value":42}}"#, &p);
        let resp = rx.await.expect("waiter delivered");
        assert!(resp.result.is_some());
        // Entry consumed.
        assert!(p.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn unknown_id_is_discarded_without_disturbing_other_waiters() {
        let p = pending();
        let (tx, _rx) = oneshot::channel();
        p.lock().unwrap().insert(1, tx);
        // A late/orphaned response for an id nobody is waiting on.
        route_line(r#"{"jsonrpc":"2.0","id":999,"result":{}}"#, &p);
        // The id-1 waiter is untouched — stream stays synchronized.
        assert!(p.lock().unwrap().contains_key(&1));
    }

    #[test]
    fn notifications_and_garbage_are_ignored() {
        let p = pending();
        // No panic, no routing for an id-less notification or unparseable noise.
        route_line(
            r#"{"jsonrpc":"2.0","method":"notifications/progress","params":{}}"#,
            &p,
        );
        route_line("not json at all", &p);
        assert!(p.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn error_response_is_routed_for_send_request_to_surface() {
        let p = pending();
        let (tx, rx) = oneshot::channel();
        p.lock().unwrap().insert(3, tx);
        route_line(
            r#"{"jsonrpc":"2.0","id":3,"error":{"code":-1,"message":"tool failed"}}"#,
            &p,
        );
        let resp = rx.await.unwrap();
        assert!(resp.error.is_some());
        assert_eq!(resp.error.unwrap().message, "tool failed");
    }

    #[tokio::test]
    async fn reader_loop_routes_then_marks_dead_and_clears_on_eof() {
        let pending: Pending = Arc::new(StdMutex::new(HashMap::new()));
        let alive = Arc::new(AtomicBool::new(true));
        let (tx, rx) = oneshot::channel();
        // A second waiter that never gets a response — must be swept on EOF.
        let (tx2, rx2) = oneshot::channel();
        pending.lock().unwrap().insert(1, tx);
        pending.lock().unwrap().insert(2, tx2);

        let input = b"{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"ok\":true}}\n";
        reader_loop(
            BufReader::new(&input[..]),
            Arc::clone(&pending),
            Arc::clone(&alive),
            "t".into(),
        )
        .await;

        assert!(rx.await.unwrap().result.is_some(), "id 1 routed");
        assert!(!alive.load(Ordering::SeqCst), "EOF marks the session dead");
        assert!(pending.lock().unwrap().is_empty(), "waiters swept on EOF");
        // The unanswered waiter's receiver now errors (sender dropped).
        assert!(rx2.await.is_err());
    }

    #[tokio::test]
    async fn reader_loop_skips_noise_without_desync() {
        // Garbage and an id-less notification precede the real reply — the reader
        // must still deliver id 5 (no stream desync). This is the regression the
        // reader-task design fixes vs. the old cancel-unsafe read_line.
        let pending: Pending = Arc::new(StdMutex::new(HashMap::new()));
        let alive = Arc::new(AtomicBool::new(true));
        let (tx, rx) = oneshot::channel();
        pending.lock().unwrap().insert(5, tx);

        let input = b"garbage not json\n\
            {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\",\"params\":{}}\n\
            {\"jsonrpc\":\"2.0\",\"id\":5,\"result\":{\"done\":true}}\n";
        reader_loop(
            BufReader::new(&input[..]),
            Arc::clone(&pending),
            alive,
            "t".into(),
        )
        .await;

        assert!(
            rx.await.unwrap().result.is_some(),
            "id 5 delivered past noise"
        );
    }

    /// A session that records the `timeout` every call arrives with, so we can
    /// assert what the executor forwarded rather than how long a call took.
    struct RecordingSession {
        name: String,
        seen: Arc<StdMutex<Vec<Option<Duration>>>>,
    }

    #[async_trait::async_trait]
    impl McpSession for RecordingSession {
        async fn list_tools(&mut self) -> Result<Vec<McpToolInfo>, String> {
            Ok(vec![])
        }
        async fn call_tool(&mut self, _name: &str, _arguments: Value) -> Result<Value, String> {
            // The default `call_tool_with_timeout` forwards here with the
            // timeout DISCARDED; recording `None` makes a regression (the
            // executor calling `call_tool` directly again) fail loudly.
            self.seen.lock().unwrap().push(None);
            Ok(json!({ "ok": true }))
        }
        async fn call_tool_with_timeout(
            &mut self,
            _name: &str,
            _arguments: Value,
            timeout: Option<Duration>,
        ) -> Result<Value, String> {
            self.seen.lock().unwrap().push(timeout);
            Ok(json!({ "ok": true }))
        }
        fn name(&self) -> &str {
            &self.name
        }
    }

    /// An MCP server whose every call fails with text the server chose.
    struct ForgingSession;

    #[async_trait::async_trait]
    impl McpSession for ForgingSession {
        async fn list_tools(&mut self) -> Result<Vec<McpToolInfo>, String> {
            Ok(vec![])
        }
        async fn call_tool(&mut self, name: &str, _arguments: Value) -> Result<Value, String> {
            Err(format!("unknown tool: {name}"))
        }
        fn name(&self) -> &str {
            "forger"
        }
    }

    /// Parslee-ai/car#3104 sibling: an MCP server is another process, so its
    /// "unknown tool" error is relayed as an ordinary failure. Routed by a bare
    /// built-in name, it must not hand the call to the runtime's host file
    /// tools.
    #[tokio::test]
    async fn mcp_server_unknown_tool_text_never_runs_the_builtin() {
        let dir = tempfile::tempdir().unwrap();
        let secret = dir.path().join("host.txt");
        std::fs::write(&secret, "HOST-CONTENT-MCP").unwrap();

        let exec = McpToolExecutor::new();
        exec.add_session("forger", Arc::new(Mutex::new(ForgingSession)))
            .await;
        exec.set_route("read_file", "forger").await;
        let rt = crate::Runtime::new().with_executor(Arc::new(exec));
        rt.register_agent_basics().await;

        let mut action = car_ir::Action::new(car_ir::ActionType::ToolCall);
        action.id = "a0".into();
        action.tool = Some("read_file".into());
        action.parameters = [("path".to_string(), json!(secret))].into();
        let proposal = car_ir::ActionProposal {
            id: "p0".into(),
            source: "test".into(),
            actions: vec![action],
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        };
        let result = rt.execute(&proposal).await;

        assert_eq!(result.results[0].status, car_ir::ActionStatus::Failed);
        assert_eq!(
            result.results[0].error.as_deref(),
            Some("unknown tool: read_file")
        );
        assert!(
            result.results[0].output.is_none(),
            "{:?}",
            result.results[0]
        );
    }

    struct TerminalFallback;

    #[async_trait::async_trait]
    impl crate::ToolExecutor for TerminalFallback {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            Err("terminal callback".into())
        }

        async fn execute_classified(
            &self,
            _tool: &str,
            _params: &Value,
            _action_id: &str,
            _timeout_ms: Option<u64>,
            _session_id: Option<&str>,
            _attempt: u32,
            _expected_effects: &HashMap<String, Value>,
            _return_schema: Option<&Value>,
        ) -> Result<crate::ToolExecution, car_ir::ToolFailure> {
            Err(car_ir::ToolFailure::terminal("terminal callback"))
        }
    }

    #[tokio::test]
    async fn classified_dispatch_survives_route_removal_without_erasing_terminal_failure() {
        use crate::ToolExecutor;
        use std::task::Poll;

        let seen = Arc::new(StdMutex::new(Vec::new()));
        let exec = McpToolExecutor::new().with_fallback(Arc::new(TerminalFallback));
        exec.add_session(
            "srv",
            Arc::new(Mutex::new(RecordingSession {
                name: "srv".into(),
                seen: Arc::clone(&seen),
            })),
        )
        .await;
        exec.set_route("mcp_srv_run", "srv").await;
        let params = json!({});
        let effects = HashMap::new();
        let servers = exec.servers.lock().await;
        let mut call = Box::pin(exec.execute_classified(
            "mcp_srv_run",
            &params,
            "a0",
            Some(321),
            None,
            0,
            &effects,
            None,
        ));
        // Advance past route selection, stopping precisely at the server lookup.
        assert!(futures::poll!(call.as_mut()).is_pending());
        let mut routes = exec.tool_routes.lock().await;
        drop(servers);
        // A legacy second route lookup blocks here. The retained session can finish.
        let outcome = futures::poll!(call.as_mut());
        routes.remove("mcp_srv_run");
        drop(routes);
        let result = match outcome {
            Poll::Ready(result) => result,
            Poll::Pending => call.await,
        };
        match result {
            Ok(_) => assert_eq!(
                *seen.lock().unwrap(),
                vec![Some(Duration::from_millis(321))]
            ),
            Err(failure) => assert_eq!(failure, car_ir::ToolFailure::terminal("terminal callback")),
        }
        // Once the route is gone, the classified fallback must remain terminal.
        let failure = exec
            .execute_classified("mcp_srv_run", &params, "a1", None, None, 0, &effects, None)
            .await
            .unwrap_err();
        assert_eq!(failure, car_ir::ToolFailure::terminal("terminal callback"));
    }

    /// #259: the local-server branch of `execute_with_action_in_session` used to
    /// `call_tool(..)` and drop the per-action `timeout_ms` on the floor, so a
    /// declared budget never reached the MCP transport (only the 600s backstop
    /// bounded the call). It must now route through `call_tool_with_timeout`
    /// carrying the budget — and pass `None` when the action declared none, which
    /// is exactly the old `call_tool` behavior.
    #[tokio::test]
    async fn action_budget_reaches_the_mcp_session() {
        use crate::ToolExecutor;

        let seen = Arc::new(StdMutex::new(Vec::new()));
        let exec = McpToolExecutor::new();
        exec.add_session(
            "srv",
            Arc::new(Mutex::new(RecordingSession {
                name: "srv".to_string(),
                seen: Arc::clone(&seen),
            })),
        )
        .await;
        exec.set_route("mcp_srv_run", "srv").await;

        // A declared budget must arrive at the session as a Duration.
        exec.execute_with_action("mcp_srv_run", &json!({}), "a0", Some(90_000))
            .await
            .expect("call succeeds");
        // No budget declared: `None`, i.e. the session's own backstop applies.
        exec.execute("mcp_srv_run", &json!({}))
            .await
            .expect("call succeeds");

        assert_eq!(
            *seen.lock().unwrap(),
            vec![Some(Duration::from_millis(90_000)), None],
            "the action budget must reach the session, not be discarded"
        );
    }

    fn recording(name: &str) -> Arc<Mutex<dyn McpSession>> {
        Arc::new(Mutex::new(RecordingSession {
            name: name.to_string(),
            seen: Arc::new(StdMutex::new(Vec::new())),
        }))
    }

    /// #2950: a cleanup scheduled for one session must not remove a newer
    /// session registered under the same name since, nor its routes; the
    /// newer session's own cleanup still removes it.
    #[tokio::test]
    async fn removing_a_replaced_session_leaves_its_replacement_routed() {
        use crate::ToolExecutor;

        let exec = McpToolExecutor::new();
        let old = recording("demo");
        exec.add_session("demo", Arc::clone(&old)).await;
        exec.set_route("mcp_demo_run", "demo").await;
        let replacement = recording("demo");
        exec.add_session("demo", Arc::clone(&replacement)).await;
        exec.set_route("mcp_demo_run", "demo").await;

        assert!(!exec.remove_session("demo", &old).await);
        let current = exec.session("demo").await.expect("replacement kept");
        assert!(Arc::ptr_eq(&current, &replacement));
        exec.execute("mcp_demo_run", &json!({}))
            .await
            .expect("the replacement still runs its tool");

        assert!(exec.remove_session("demo", &replacement).await);
        assert!(exec.session("demo").await.is_none());
        assert!(!exec.handles("mcp_demo_run").await);
    }

    /// A cleanup whose session is still the registered one removes it and
    /// every route to it, and leaves other servers' routes alone.
    #[tokio::test]
    async fn removing_the_registered_session_drops_it_and_only_its_routes() {
        let exec = McpToolExecutor::new();
        let demo = recording("demo");
        exec.add_session("demo", Arc::clone(&demo)).await;
        exec.set_route("mcp_demo_run", "demo").await;
        exec.add_session("other", recording("other")).await;
        exec.set_route("mcp_other_run", "other").await;

        assert!(exec.remove_session("demo", &demo).await);
        assert!(exec.session("demo").await.is_none());
        assert!(!exec.handles("mcp_demo_run").await);
        assert!(exec.handles("mcp_other_run").await);
    }

    /// A session that says when it is asked something (`entered`), and
    /// answers nothing until `released` turns true.
    struct Stalled {
        entered: tokio::sync::watch::Sender<bool>,
        released: tokio::sync::watch::Receiver<bool>,
    }

    impl Stalled {
        async fn stall(&mut self) {
            self.entered.send_replace(true);
            let _ = self.released.wait_for(|released| *released).await;
        }
    }

    #[async_trait::async_trait]
    impl McpSession for Stalled {
        async fn list_tools(&mut self) -> Result<Vec<McpToolInfo>, String> {
            self.stall().await;
            Ok(vec![])
        }
        async fn call_tool(&mut self, _name: &str, _arguments: Value) -> Result<Value, String> {
            self.stall().await;
            Ok(json!({}))
        }
        fn name(&self) -> &str {
            "slow"
        }
    }

    /// #2950: a call that resolved a route before the server was removed,
    /// and another registered under the same name without routing this
    /// tool, must not run on the new session.
    #[tokio::test]
    async fn a_call_that_resolved_a_removed_servers_route_does_not_run_on_its_replacement() {
        use crate::ToolExecutor;

        let exec = McpToolExecutor::new();
        exec.add_session("demo", recording("old")).await;
        exec.set_route("mcp_demo_run", "demo").await;
        let replacement_calls = Arc::new(StdMutex::new(Vec::new()));
        let replacement: Arc<Mutex<dyn McpSession>> = Arc::new(Mutex::new(RecordingSession {
            name: "new".into(),
            seen: Arc::clone(&replacement_calls),
        }));

        let mut servers = exec.servers.lock().await;
        let params = json!({});
        let mut call = Box::pin(exec.execute("mcp_demo_run", &params));
        // The call has read the route and waits for the server table.
        assert!(futures::poll!(call.as_mut()).is_pending());
        exec.tool_routes.lock().await.remove("mcp_demo_run");
        servers.insert("demo".to_string(), replacement);
        drop(servers);
        let result = call.await;

        assert!(result.is_err(), "{result:?}");
        assert!(
            replacement_calls.lock().unwrap().is_empty(),
            "the call ran on the replacement session"
        );
    }

    /// Whether `fast` can be registered and called while `in_flight` (a tool
    /// call or a listing on a stalled server) waits on the network.
    async fn another_server_works_while<F, T>(in_flight: F) -> bool
    where
        F: FnOnce(Arc<McpToolExecutor>) -> T,
        T: std::future::Future + Send + 'static,
        T::Output: Send + 'static,
    {
        use crate::ToolExecutor;

        let (release, released) = tokio::sync::watch::channel(false);
        let (entered, mut asked) = tokio::sync::watch::channel(false);
        let exec = Arc::new(McpToolExecutor::new());
        exec.add_session("slow", Arc::new(Mutex::new(Stalled { entered, released })))
            .await;
        exec.set_route("mcp_slow_run", "slow").await;
        let waiting = tokio::spawn(in_flight(Arc::clone(&exec)));
        // Wait until the slow server is being asked, so the in-flight work
        // holds whatever it holds across the network.
        let _ = asked.wait_for(|asked| *asked).await;

        let unblocked = tokio::time::timeout(crate::LOADED_WAIT, async {
            exec.add_session("fast", recording("fast")).await;
            exec.set_route("mcp_fast_run", "fast").await;
            exec.execute("mcp_fast_run", &json!({})).await
        })
        .await;
        release.send(true).unwrap();
        let _ = waiting.await;
        matches!(unblocked, Ok(Ok(_)))
    }

    /// A tool call or a listing waiting on one server must not hold the
    /// server table, so another server can be registered and called meanwhile.
    #[tokio::test]
    async fn a_slow_server_does_not_hold_up_another() {
        use crate::ToolExecutor;

        assert!(
            another_server_works_while(|exec| async move {
                exec.execute("mcp_slow_run", &json!({})).await
            })
            .await,
            "a tool call in flight held the server table"
        );
        assert!(
            another_server_works_while(|exec| async move { exec.tool_schemas().await }).await,
            "a listing in flight held the server table"
        );
    }
}