Add stdio transport to the MCP client

index f0a0de8..7490f38 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -103,16 +103,24 @@ feeds results back. Neither the loop nor ACP/TUI surfaces depend on a concrete b calls `skill` with a name) loads the full `SKILL.md` body. The `skill` tool is appended in the `*_as_specs`/`build_tool_specs` layer (not in `all_tools()`) so its description can be dynamic, and only when at least one skill exists. -- **`src/mcp.rs`** — [Model Context Protocol](https://modelcontextprotocol.io) *client*. Connects to - MCP servers over the **Streamable HTTP** transport (one JSON-RPC POST endpoint; replies are - `application/json` or SSE), runs the `initialize`/`tools/list` handshake, and forwards `tools/call`. - Discovery is best-effort at startup (`mcp::init`, called from both branches of `main()`) and cached - in a process-global so the synchronous spec builders (`mcp::tool_specs`) and the async dispatch - (`mcp::call_tool`) can both read it. Tools are namespaced `mcp__<server>__<tool>`, appended in the - `*_as_specs`/`build_tool_specs` layer and routed in `tools::execute_tool` via `mcp::is_mcp_tool`. The - official server (`<cloud>/mcp`, default `https://sigit.si/api/v1/mcp`) is baked in and authed with the - cloud session token; extra servers live in `mcp.toml` (global `$SIGIT_CONFIG_DIR/mcp.toml` and - project-local `.sigit/mcp.toml`). stdio transport is not supported. +- **`src/mcp.rs`** — [Model Context Protocol](https://modelcontextprotocol.io) *client*. Two + transports: **Streamable HTTP** (one JSON-RPC POST endpoint, `url` in `mcp.toml`; replies are + `application/json` or SSE) and **stdio** (`command` + optional `args`/`[server.env]` in + `mcp.toml`; sigit spawns the server and speaks newline-delimited JSON-RPC over its + stdin/stdout, stderr inherited into sigit's log). `url` and `command` are mutually exclusive — + both or neither is a config error, logged and skipped. Both transports run the same + `initialize`/`tools/list` handshake and forward `tools/call`. Discovery is best-effort at + startup (`mcp::init`, called from both branches of `main()`) and cached in a process-global so + the synchronous spec builders (`mcp::tool_specs`) and the async dispatch (`mcp::call_tool`) can + both read it; `/reload` does *not* re-run it, so config changes need a restart. stdio children + live for the process; a dead child fails calls with an in-band error string (no auto-restart). + Tools are namespaced `mcp__<server>__<tool>`, appended in the `*_as_specs`/`build_tool_specs` + layer and routed in `tools::execute_tool` via `mcp::is_mcp_tool`. The official server + (`<cloud>/mcp`, default `https://sigit.si/api/v1/mcp`) is baked in (always HTTP) and authed + with the cloud session token; extra servers live in `mcp.toml` (global + `$SIGIT_CONFIG_DIR/mcp.toml` and project-local `.sigit/mcp.toml`). The stdio path is covered by + `tests/mcp_stdio.rs`, driven by the test-only `src/bin/mcp_stdio_stub.rs` helper binary + (excluded from the published crate via `exclude` in `Cargo.toml`). - **`src/permissions.rs`** — tool permission policy. Every tool call passes through `decision_for` before executing: read-only tools always run; mutating tools (and all `mcp__*`/unknown tools) are governed by, in order: per-session plan mode (`/plan` — deny all

index 1c28a0a..4141386 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,11 @@ readme = "README.md" keywords = ["sigit", "cli", "ai", "coding-agent", "llm"] categories = ["command-line-utilities"] authors = ["Seto Elkahfi <seto@ondeinference.com>"] +# `src/bin/` holds test-only helper binaries (the MCP stdio stub used by the +# integration tests). Excluding it keeps them out of the published crate, so +# `cargo install sigit` installs exactly one binary. The integration test that +# spawns the stub goes with it, or the packaged crate's tests wouldn't compile. +exclude = ["src/bin/", "tests/mcp_stdio.rs"] [[bin]] name = "sigit" @@ -25,7 +30,7 @@ agent-client-protocol = { version = "1.0", features = ["unstable_session_fork", onde = "1.1.2" # Async runtime -tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "io-std", "io-util", "sync", "time"] } +tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "io-std", "io-util", "sync", "time", "process"] } tokio-util = { version = "0.7", features = ["compat"] } futures = "0.3"

new file mode 100644 index 0000000..d3ea985 --- /dev/null +++ b/src/bin/mcp_stdio_stub.rs @@ -0,0 +1,109 @@ +//! Test-only MCP stdio server stub, used by `tests/mcp_stdio.rs`. +//! +//! Speaks the MCP stdio transport: newline-delimited JSON-RPC 2.0, one message +//! per line on stdin/stdout. It answers `initialize`, `tools/list` (a single +//! `echo` tool) and `tools/call` (echoes `text` back, prefixed with the +//! `STUB_PREFIX` env var so tests can verify env propagation). +//! +//! Flags that script failure modes: +//! - `--fail`: exit(1) immediately, before reading anything (a server whose +//! process dies at spawn). +//! - `--exit-after-list`: exit(0) right after answering `tools/list` (a server +//! that dies after discovery, exercising the dead-child call path). +//! +//! After the `initialize` response it also emits a server-initiated +//! notification the client must log and ignore. +//! +//! This binary is excluded from the published crate (see `exclude` in +//! `Cargo.toml`); it exists only for the integration tests. + +use std::io::{BufRead, Write}; + +use serde_json::{Value, json}; + +fn main() { + let args: Vec<String> = std::env::args().skip(1).collect(); + if args.iter().any(|a| a == "--fail") { + std::process::exit(1); + } + let exit_after_list = args.iter().any(|a| a == "--exit-after-list"); + + let stdin = std::io::stdin(); + let stdout = std::io::stdout(); + + for line in stdin.lock().lines() { + let Ok(line) = line else { break }; + if line.trim().is_empty() { + continue; + } + let Ok(message) = serde_json::from_str::<Value>(&line) else { + continue; + }; + let id = message.get("id").cloned(); + let method = message + .get("method") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + + let result = match method.as_str() { + "initialize" => Some(json!({ + "protocolVersion": "2025-06-18", + "capabilities": { "tools": {} }, + "serverInfo": { "name": "mcp-stdio-stub", "version": "0.0.0" } + })), + "tools/list" => Some(json!({ + "tools": [{ + "name": "echo", + "description": "Echo the text back.", + "inputSchema": { + "type": "object", + "properties": { "text": { "type": "string" } }, + "required": ["text"] + } + }] + })), + "tools/call" => { + let text = message + .pointer("/params/arguments/text") + .and_then(Value::as_str) + .unwrap_or_default(); + let prefix = std::env::var("STUB_PREFIX").unwrap_or_default(); + Some(json!({ + "content": [{ "type": "text", "text": format!("{prefix}{text}") }] + })) + } + // Notifications (`notifications/initialized`) and anything else + // without a scripted answer fall through. + _ => None, + }; + + let mut out = stdout.lock(); + match (id, result) { + (Some(id), Some(result)) => { + let response = json!({ "jsonrpc": "2.0", "id": id, "result": result }); + writeln!(out, "{response}").ok(); + } + (Some(id), None) => { + let response = json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": -32601, "message": "method not found" } + }); + writeln!(out, "{response}").ok(); + } + // A notification: nothing to answer. + (None, _) => {} + } + if method == "initialize" { + // Server-initiated traffic the client must ignore. + let notification = json!({ "jsonrpc": "2.0", "method": "notifications/stub" }); + writeln!(out, "{notification}").ok(); + } + out.flush().ok(); + + if exit_after_list && method == "tools/list" { + std::process::exit(0); + } + } +}

index ca6f7c3..e4e3e9b 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -6,11 +6,20 @@ //! When the model calls an MCP tool, the call is forwarded to the owning server //! and the result fed back into the agent loop. //! -//! Transport: the modern **Streamable HTTP** transport — a single HTTP endpoint -//! the client POSTs JSON-RPC 2.0 messages to. The server answers either with a -//! single `application/json` body or a `text/event-stream` (SSE) stream that -//! carries the JSON-RPC response. Both are handled here. stdio transport is not -//! supported (siGit Code never spawns child processes for inference). +//! Transports: +//! +//! - **Streamable HTTP** — a single HTTP endpoint the client POSTs JSON-RPC 2.0 +//! messages to. The server answers either with a single `application/json` +//! body or a `text/event-stream` (SSE) stream that carries the JSON-RPC +//! response. Both are handled here. Configured with `url` in `mcp.toml`. +//! - **stdio** — siGit spawns the server as a child process and exchanges +//! newline-delimited JSON-RPC messages over its stdin/stdout (the server's +//! stderr flows into siGit's own log stream). Configured with `command` +//! (plus optional `args` and `[server.env]`) in `mcp.toml`. This is how most +//! published MCP servers (filesystem, Playwright, GitHub, ...) are run. +//! +//! `url` and `command` are mutually exclusive; an entry with both, or neither, +//! is a config error that is logged and skipped. //! //! ## The official server //! @@ -30,6 +39,13 @@ //! stored in a process-global so the synchronous tool-spec builders //! ([`tool_specs`]) and the async dispatch ([`call_tool`]) can both read it. //! +//! stdio children live for the sigit process. When a child dies (EOF or an I/O +//! error on its pipes) the server is marked dead and later calls return an +//! in-band error string the model can react to; there is no automatic restart. +//! `/reload` does *not* re-run discovery ([`init`] is once-per-process), so a +//! changed `mcp.toml` or a dead server needs a sigit restart. At process exit +//! children see EOF on their stdin and exit on their own. +//! //! Tools are namespaced `mcp__<server>__<tool>` so they never collide with //! built-in tools or with each other across servers. This mirrors the //! convention used by other MCP-aware agents. @@ -39,15 +55,18 @@ //! are unused, so the dead-code lint is suppressed there only. #![cfg_attr(not(unix), allow(dead_code))] -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::path::PathBuf; -use std::sync::OnceLock; +use std::sync::Mutex as StdMutex; use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::{Arc, OnceLock}; use std::time::Duration; use serde::Deserialize; use serde_json::{Value, json}; -use tokio::sync::Mutex; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::process::{Child, ChildStdin, ChildStdout, Command}; +use tokio::sync::{Mutex, oneshot}; use crate::backend::ToolSpec; @@ -91,6 +110,25 @@ struct McpTool { struct ServerConn { /// Sanitized server name used in tool namespacing and the `/mcp` listing. name: String, + /// Display endpoint for the `/mcp` listing: the URL for HTTP servers, the + /// command line for stdio servers. + endpoint: String, + /// The live transport. `None` when a stdio server failed to even spawn. + transport: Option<Transport>, + /// Tools discovered at startup. Empty when the server failed to connect. + tools: Vec<McpTool>, + /// Connection error, if the handshake failed. Surfaced by `/mcp`. + error: Option<String>, +} + +/// How a connected server is reached. +enum Transport { + Http(HttpConn), + Stdio(StdioConn), +} + +/// Streamable HTTP connection state. +struct HttpConn { /// Streamable HTTP endpoint (the single POST URL). url: String, /// Extra headers sent on every request (e.g. `Authorization`). @@ -98,10 +136,59 @@ struct ServerConn { /// Session id handed back by the server on `initialize`, echoed on every /// later request via the `Mcp-Session-Id` header. session_id: Mutex<Option<String>>, - /// Tools discovered at startup. Empty when the server failed to connect. - tools: Vec<McpTool>, - /// Connection error, if the handshake failed. Surfaced by `/mcp`. - error: Option<String>, +} + +/// stdio connection state: a child process speaking newline-delimited JSON-RPC +/// over its stdin/stdout. +struct StdioConn { + /// The child's stdin. The mutex serializes writes so concurrent requests + /// can't interleave bytes on the pipe; `None` once the pipe broke. + writer: Mutex<Option<ChildStdin>>, + /// State shared with the background reader task that owns the child's + /// stdout. + shared: Arc<StdioShared>, + /// JSON-RPC id source. Ids are per-connection so the reader task can route + /// each response to the request that carries its id. + next_id: AtomicI64, +} + +/// State shared between a [`StdioConn`] and its background reader task. +struct StdioShared { + /// Server name, for log lines. + name: String, + /// In-flight requests awaiting a response, keyed by JSON-RPC id. Dropping + /// a sender (when the connection dies) wakes the waiter with an error. + pending: StdMutex<HashMap<i64, oneshot::Sender<Value>>>, + /// Why the connection is unusable, once it is (EOF, I/O error, kill). + dead: StdMutex<Option<String>>, + /// The child handle, kept so a dead/failed connection can kill and reap + /// the process. Taken on death. + child: StdMutex<Option<Child>>, +} + +impl StdioShared { + fn dead_reason(&self) -> Option<String> { + self.dead.lock().unwrap().clone() + } + + /// Mark the connection unusable: record the reason (first one wins), fail + /// every in-flight request, and kill + reap the child, best effort. + fn mark_dead(&self, reason: &str) { + { + let mut dead = self.dead.lock().unwrap(); + if dead.is_none() { + *dead = Some(reason.to_string()); + } + } + // Dropping the senders wakes every waiter with a recv error. + self.pending.lock().unwrap().clear(); + if let Some(mut child) = self.child.lock().unwrap().take() { + let _ = child.start_kill(); + tokio::spawn(async move { + let _ = child.wait().await; + }); + } + } } /// The process-global MCP state: a shared HTTP client plus every configured @@ -125,19 +212,70 @@ fn official_url() -> String { ) } -/// A server entry as written in `mcp.toml`. +/// A server entry as written in `mcp.toml`. Exactly one of `url` (Streamable +/// HTTP) or `command` (stdio) selects the transport. #[derive(Debug, Deserialize)] struct ServerEntry { name: String, - url: String, + /// Streamable HTTP endpoint. Mutually exclusive with `command`. + #[serde(default)] + url: Option<String>, + /// stdio server executable. Mutually exclusive with `url`. + #[serde(default)] + command: Option<String>, + /// Arguments for `command`. + #[serde(default)] + args: Vec<String>, + /// Extra environment variables for `command`, added on top of the + /// inherited environment. + #[serde(default)] + env: BTreeMap<String, String>, /// Set `enabled = false` to keep an entry in the file but skip connecting. #[serde(default)] enabled: Option<bool>, - /// Static headers, e.g. `Authorization = "Bearer ..."`. + /// Static headers, e.g. `Authorization = "Bearer ..."`. HTTP only. #[serde(default)] headers: BTreeMap<String, String>, } +impl ServerEntry { + /// Resolve the entry's transport. `url` and `command` are mutually + /// exclusive and exactly one is required; anything else is a config error. + fn transport_def(&self) -> Result<TransportDef, String> { + let url = self.url.as_deref().map(str::trim).filter(|v| !v.is_empty()); + let command = self + .command + .as_deref() + .map(str::trim) + .filter(|v| !v.is_empty()); + match (url, command) { + (Some(_), Some(_)) => { + Err("has both `url` and `command`; a server uses exactly one transport".to_string()) + } + (None, None) => { + Err("needs either `url` (Streamable HTTP) or `command` (stdio)".to_string()) + } + (Some(url), None) => Ok(TransportDef::Http { + url: url.to_string(), + headers: self + .headers + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + }), + (None, Some(command)) => Ok(TransportDef::Stdio { + command: command.to_string(), + args: self.args.clone(), + env: self + .env + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + }), + } + } +} + /// The `mcp.toml` schema. #[derive(Debug, Default, Deserialize)] struct McpFile { @@ -149,12 +287,43 @@ struct McpFile { server: Vec<ServerEntry>, } +/// How to reach a configured server, before connecting. +#[derive(Debug, Clone)] +enum TransportDef { + Http { + url: String, + headers: Vec<(String, String)>, + }, + Stdio { + command: String, + args: Vec<String>, + env: Vec<(String, String)>, + }, +} + +impl TransportDef { + /// Human-readable endpoint for logs and the `/mcp` listing: the URL for + /// HTTP, the command line for stdio. + fn endpoint(&self) -> String { + match self { + TransportDef::Http { url, .. } => url.clone(), + TransportDef::Stdio { command, args, .. } => { + let mut line = command.clone(); + for arg in args { + line.push(' '); + line.push_str(arg); + } + line + } + } + } +} + /// A resolved server definition, before connecting. #[derive(Debug, Clone)] struct ServerDef { name: String, - url: String, - headers: Vec<(String, String)>, + transport: TransportDef, } /// Config files to read, in priority order (later wins on a name clash): @@ -220,22 +389,21 @@ fn load_configs() -> Vec<ServerDef> { continue; } let name = sanitize(&entry.name); - if name.is_empty() || entry.url.trim().is_empty() { - log::warn!( - "mcp: skipping server with empty name/url in {}", - path.display() - ); + if name.is_empty() { + log::warn!("mcp: skipping server with empty name in {}", path.display()); continue; } - let headers = entry.headers.into_iter().collect(); - upsert( - &mut defs, - ServerDef { - name, - url: entry.url.trim().to_string(), - headers, - }, - ); + let transport = match entry.transport_def() { + Ok(transport) => transport, + Err(error) => { + log::warn!( + "mcp: skipping server '{name}' in {}: {error}", + path.display() + ); + continue; + } + }; + upsert(&mut defs, ServerDef { name, transport }); } } @@ -258,8 +426,10 @@ fn load_configs() -> Vec<ServerDef> { } defs.push(ServerDef { name: "sigit".to_string(), - url: official_url(), - headers, + transport: TransportDef::Http { + url: official_url(), + headers, + }, }); } @@ -341,11 +511,33 @@ pub async fn init() { /// Run the handshake against one server and collect its tools. Always returns a /// `ServerConn`; failures land in its `error` field rather than propagating. async fn connect(http: &reqwest::Client, def: ServerDef) -> ServerConn { + let endpoint = def.transport.endpoint(); + let transport = match &def.transport { + TransportDef::Http { url, headers } => Transport::Http(HttpConn { + url: url.clone(), + headers: headers.clone(), + session_id: Mutex::new(None), + }), + TransportDef::Stdio { command, args, env } => { + match spawn_stdio(&def.name, command, args, env) { + Ok(conn) => Transport::Stdio(conn), + Err(error) => { + return ServerConn { + name: def.name, + endpoint, + transport: None, + tools: Vec::new(), + error: Some(error), + }; + } + } + } + }; + let mut conn = ServerConn { - name: def.name.clone(), - url: def.url.clone(), - headers: def.headers.clone(), - session_id: Mutex::new(None), + name: def.name, + endpoint, + transport: Some(transport), tools: Vec::new(), error: None, }; @@ -364,31 +556,33 @@ async fn connect(http: &reqwest::Client, def: ServerDef) -> ServerConn { Err(_) => conn.error = Some(format!("timed out after {}s", HANDSHAKE_TIMEOUT.as_secs())), } + // A stdio child that failed its handshake is useless — kill it rather than + // leave it running for the rest of the process. + if let Some(error) = conn.error.clone() + && let Some(Transport::Stdio(stdio)) = &conn.transport + { + stdio.shared.mark_dead(&error); + } + conn } -/// The `initialize` request: negotiate protocol version and capture the session -/// id from the response headers (handled inside [`post_rpc`]). +/// The `initialize` request: negotiate protocol version and (on HTTP) capture +/// the session id from the response headers (handled inside [`post_rpc`]). async fn initialize(http: &reqwest::Client, conn: &ServerConn) -> Result<(), String> { - let body = json!({ - "jsonrpc": "2.0", - "id": 0, - "method": "initialize", - "params": { - "protocolVersion": PROTOCOL_VERSION, - "capabilities": {}, - "clientInfo": { "name": "sigit", "version": env!("CARGO_PKG_VERSION") } - } + let params = json!({ + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": { "name": "sigit", "version": env!("CARGO_PKG_VERSION") } }); - post_rpc(http, conn, &body, HANDSHAKE_TIMEOUT).await?; + rpc_request(http, conn, "initialize", params, HANDSHAKE_TIMEOUT).await?; Ok(()) } /// The `notifications/initialized` notification. Servers expect it before -/// fielding requests; it carries no id and yields a 202 with no body. +/// fielding requests; it carries no id and no response. async fn notify_initialized(http: &reqwest::Client, conn: &ServerConn) -> Result<(), String> { - let body = json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }); - post_notification(http, conn, &body, HANDSHAKE_TIMEOUT).await + rpc_notify(http, conn, "notifications/initialized", HANDSHAKE_TIMEOUT).await } /// `tools/list`, following `nextCursor` pagination, mapped into [`McpTool`]s. @@ -401,8 +595,7 @@ async fn list_tools(http: &reqwest::Client, conn: &ServerConn) -> Result<Vec<Mcp Some(c) => json!({ "cursor": c }), None => json!({}), }; - let body = json!({ "jsonrpc": "2.0", "id": 0, "method": "tools/list", "params": params }); - let result = post_rpc(http, conn, &body, HANDSHAKE_TIMEOUT).await?; + let result = rpc_request(http, conn, "tools/list", params, HANDSHAKE_TIMEOUT).await?; for tool in result .get("tools") @@ -521,32 +714,40 @@ pub async fn call_tool(full_name: &str, arguments: &str) -> String { } impl Mcp { - /// Send a `tools/call` and render the result into text. Retries once after a - /// re-`initialize` if the session was dropped (HTTP 404), which is how - /// Streamable HTTP signals an expired session. + /// Send a `tools/call` and render the result into text. On HTTP, retries + /// once after a re-`initialize` if the session was dropped (HTTP 404), + /// which is how Streamable HTTP signals an expired session. async fn call( &self, server: &ServerConn, remote_name: &str, args: Value, ) -> Result<String, String> { - let body = json!({ - "jsonrpc": "2.0", - "id": 0, - "method": "tools/call", - "params": { "name": remote_name, "arguments": args } - }); - - let result = match post_rpc(&self.http, server, &body, CALL_TIMEOUT).await { - Ok(result) => result, - Err(error) if error.contains("returned 404") => { - // Session expired — drop it, re-handshake, and retry once. - *server.session_id.lock().await = None; - initialize(&self.http, server).await?; - notify_initialized(&self.http, server).await?; - post_rpc(&self.http, server, &body, CALL_TIMEOUT).await? + let params = json!({ "name": remote_name, "arguments": args }); + let result = match &server.transport { + None => return Err(format!("server '{}' is not connected", server.name)), + Some(Transport::Stdio(stdio)) => { + stdio.request("tools/call", params, CALL_TIMEOUT).await? + } + Some(Transport::Http(http_conn)) => { + let body = json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "tools/call", + "params": params + }); + match post_rpc(&self.http, &server.name, http_conn, &body, CALL_TIMEOUT).await { + Ok(result) => result, + Err(error) if error.contains("returned 404") => { + // Session expired — drop it, re-handshake, and retry once. + *http_conn.session_id.lock().await = None; + initialize(&self.http, server).await?; + notify_initialized(&self.http, server).await?; + post_rpc(&self.http, &server.name, http_conn, &body, CALL_TIMEOUT).await? + } + Err(error) => return Err(error), + } } - Err(error) => return Err(error), }; Ok(render_tool_result(&result)) @@ -611,6 +812,242 @@ fn truncate(text: String) -> String { format!("{kept}\n\n[output truncated to {RESULT_CHAR_LIMIT} characters]") } +// ── Transport-generic JSON-RPC dispatch ───────────────────────────────────── + +/// Send a JSON-RPC request over whichever transport the server uses and return +/// its `result`. +async fn rpc_request( + http: &reqwest::Client, + conn: &ServerConn, + method: &str, + params: Value, + timeout: Duration, +) -> Result<Value, String> { + match &conn.transport { + None => Err(format!("server '{}' is not connected", conn.name)), + Some(Transport::Http(http_conn)) => { + let body = json!({ "jsonrpc": "2.0", "id": 0, "method": method, "params": params }); + post_rpc(http, &conn.name, http_conn, &body, timeout).await + } + Some(Transport::Stdio(stdio)) => stdio.request(method, params, timeout).await, + } +} + +/// Send a JSON-RPC notification (no id, no response expected). +async fn rpc_notify( + http: &reqwest::Client, + conn: &ServerConn, + method: &str, + timeout: Duration, +) -> Result<(), String> { + match &conn.transport { + None => Err(format!("server '{}' is not connected", conn.name)), + Some(Transport::Http(http_conn)) => { + let body = json!({ "jsonrpc": "2.0", "method": method }); + post_notification(http, &conn.name, http_conn, &body, timeout).await + } + Some(Transport::Stdio(stdio)) => stdio.notify(method).await, + } +} + +// ── stdio JSON-RPC plumbing ───────────────────────────────────────────────── + +/// Spawn a stdio MCP server and start its background reader task. The child's +/// stderr is inherited so it lands in sigit's own log stream; the given env +/// vars are added on top of the inherited environment. +fn spawn_stdio( + name: &str, + command: &str, + args: &[String], + env: &[(String, String)], +) -> Result<StdioConn, String> { + let mut cmd = Command::new(command); + cmd.args(args) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::inherit()) + .kill_on_drop(true); + for (key, value) in env { + cmd.env(key, value); + } + let mut child = cmd + .spawn() + .map_err(|error| format!("failed to spawn `{command}`: {error}"))?; + let stdin = child + .stdin + .take() + .ok_or_else(|| "child stdin was not captured".to_string())?; + let stdout = child + .stdout + .take() + .ok_or_else(|| "child stdout was not captured".to_string())?; + + let shared = Arc::new(StdioShared { + name: name.to_string(), + pending: StdMutex::new(HashMap::new()), + dead: StdMutex::new(None), + child: StdMutex::new(Some(child)), + }); + tokio::spawn(stdio_reader(BufReader::new(stdout), Arc::clone(&shared))); + + Ok(StdioConn { + writer: Mutex::new(Some(stdin)), + shared, + next_id: AtomicI64::new(1), + }) +} + +/// Background task owning a stdio child's stdout: parses one JSON-RPC message +/// per line and routes each response to the pending request that carries its +/// id. Server-initiated requests and notifications (anything with a `method`) +/// are logged and ignored — siGit doesn't support server→client calls. On EOF +/// or a read error the connection is marked dead, which fails every in-flight +/// request and reaps the child. +async fn stdio_reader(mut stdout: BufReader<ChildStdout>, shared: Arc<StdioShared>) { + let mut line = String::new(); + loop { + line.clear(); + match stdout.read_line(&mut line).await { + Ok(0) => { + shared.mark_dead("server closed its stdout (process exited)"); + return; + } + Ok(_) => {} + Err(error) => { + shared.mark_dead(&format!("read error: {error}")); + return; + } + } + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let message: Value = match serde_json::from_str(trimmed) { + Ok(message) => message, + Err(error) => { + log::warn!("mcp: '{}' sent a non-JSON line: {error}", shared.name); + continue; + } + }; + if let Some(method) = message.get("method").and_then(Value::as_str) { + log::debug!( + "mcp: ignoring server-initiated '{method}' from '{}'", + shared.name + ); + continue; + } + let Some(id) = message.get("id").and_then(Value::as_i64) else { + log::warn!( + "mcp: '{}' sent a response without a usable id; ignoring", + shared.name + ); + continue; + }; + let waiter = shared.pending.lock().unwrap().remove(&id); + match waiter { + Some(sender) => { + let _ = sender.send(message); + } + None => log::debug!( + "mcp: '{}' answered unknown/expired request id {id}; ignoring", + shared.name + ), + } + } +} + +impl StdioConn { + /// Send a JSON-RPC request and await its response, correlated by id. Fails + /// fast (in-band, never panicking) when the child has died. + async fn request( + &self, + method: &str, + params: Value, + timeout: Duration, + ) -> Result<Value, String> { + let name = &self.shared.name; + if let Some(reason) = self.shared.dead_reason() { + return Err(format!("stdio server '{name}' is not running: {reason}")); + } + + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let body = json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }); + let (sender, receiver) = oneshot::channel(); + self.shared.pending.lock().unwrap().insert(id, sender); + + if let Err(error) = self.write_line(&body).await { + self.shared.pending.lock().unwrap().remove(&id); + self.shared.mark_dead(&error); + return Err(format!("stdio server '{name}': {error}")); + } + + let message = match tokio::time::timeout(timeout, receiver).await { + Ok(Ok(message)) => message, + // Our sender was dropped: the connection died mid-request. + Ok(Err(_)) => { + let reason = self + .shared + .dead_reason() + .unwrap_or_else(|| "connection closed".to_string()); + return Err(format!("stdio server '{name}' is not running: {reason}")); + } + Err(_) => { + self.shared.pending.lock().unwrap().remove(&id); + return Err(format!( + "request to stdio server '{name}' timed out after {}s", + timeout.as_secs() + )); + } + }; + + if let Some(error) = message.get("error") { + let code = error.get("code").and_then(Value::as_i64).unwrap_or(0); + let msg = error + .get("message") + .and_then(Value::as_str) + .unwrap_or("unknown error"); + return Err(format!("'{name}' JSON-RPC error {code}: {msg}")); + } + message + .get("result") + .cloned() + .ok_or_else(|| format!("response from '{name}' had no result")) + } + + /// Send a JSON-RPC notification (no id, no response). + async fn notify(&self, method: &str) -> Result<(), String> { + let body = json!({ "jsonrpc": "2.0", "method": method }); + if let Err(error) = self.write_line(&body).await { + self.shared.mark_dead(&error); + return Err(format!("stdio server '{}': {error}", self.shared.name)); + } + Ok(()) + } + + /// Write one newline-delimited JSON-RPC message. The writer mutex keeps + /// concurrent requests from interleaving bytes on the pipe. + async fn write_line(&self, body: &Value) -> Result<(), String> { + let mut guard = self.writer.lock().await; + let Some(writer) = guard.as_mut() else { + return Err("stdin already closed".to_string()); + }; + let mut line = body.to_string(); + line.push('\n'); + let result = async { + writer.write_all(line.as_bytes()).await?; + writer.flush().await + } + .await; + if let Err(error) = result { + // A broken pipe is unrecoverable; drop the writer so later calls + // fail fast. + *guard = None; + return Err(format!("write failed: {error}")); + } + Ok(()) + } +} + // ── Streamable HTTP JSON-RPC plumbing ─────────────────────────────────────── /// POST a JSON-RPC request and return its `result`. Handles both an @@ -618,7 +1055,8 @@ fn truncate(text: String) -> String { /// session id from the response headers, and maps a JSON-RPC `error` to `Err`. async fn post_rpc( http: &reqwest::Client, - conn: &ServerConn, + name: &str, + conn: &HttpConn, body: &Value, timeout: Duration, ) -> Result<Value, String> { @@ -659,8 +1097,7 @@ async fn post_rpc( let detail = response.text().await.unwrap_or_default(); let detail: String = detail.chars().take(500).collect(); return Err(format!( - "server '{}' returned {}: {detail}", - conn.name, + "server '{name}' returned {}: {detail}", status.as_u16() )); } @@ -668,14 +1105,14 @@ async fn post_rpc( let text = response .text() .await - .map_err(|error| format!("reading response from '{}': {error}", conn.name))?; + .map_err(|error| format!("reading response from '{name}': {error}"))?; let message = if content_type.contains("text/event-stream") { parse_sse_response(&text) - .ok_or_else(|| format!("no JSON-RPC message in SSE reply from '{}'", conn.name))? + .ok_or_else(|| format!("no JSON-RPC message in SSE reply from '{name}'"))? } else { serde_json::from_str::<Value>(&text) - .map_err(|error| format!("parsing response from '{}': {error}", conn.name))? + .map_err(|error| format!("parsing response from '{name}': {error}"))? }; if let Some(error) = message.get("error") { @@ -684,20 +1121,21 @@ async fn post_rpc( .get("message") .and_then(Value::as_str) .unwrap_or("unknown error"); - return Err(format!("'{}' JSON-RPC error {code}: {msg}", conn.name)); + return Err(format!("'{name}' JSON-RPC error {code}: {msg}")); } message .get("result") .cloned() - .ok_or_else(|| format!("response from '{}' had no result", conn.name)) + .ok_or_else(|| format!("response from '{name}' had no result")) } /// POST a JSON-RPC notification (no id, no response expected). A non-success /// status is an error; an empty 202 body is the normal case. async fn post_notification( http: &reqwest::Client, - conn: &ServerConn, + name: &str, + conn: &HttpConn, body: &Value, timeout: Duration, ) -> Result<(), String> { @@ -708,8 +1146,7 @@ async fn post_notification( .map_err(|error| format!("notification to {} failed: {error}", conn.url))?; if !response.status().is_success() { return Err(format!( - "server '{}' rejected notification: {}", - conn.name, + "server '{name}' rejected notification: {}", response.status().as_u16() )); } @@ -721,7 +1158,7 @@ async fn post_notification( /// session id once we have one. async fn build_request( http: &reqwest::Client, - conn: &ServerConn, + conn: &HttpConn, body: &Value, timeout: Duration, ) -> reqwest::RequestBuilder { @@ -779,7 +1216,8 @@ fn parse_sse_response(body: &str) -> Option<Value> { // ── Status reporting (`/mcp`) ──────────────────────────────────────────────── /// Human-readable summary of configured MCP servers and their tools, for the -/// `/mcp` slash command. +/// `/mcp` slash command. Shows the URL for HTTP servers and the command line +/// for stdio servers. pub fn status_summary() -> String { let Some(mcp) = MCP.get() else { return "MCP is not initialized.".to_string(); @@ -799,13 +1237,13 @@ pub fn status_summary() -> String { match &server.error { Some(error) => lines.push(format!( "- {} ({}) — unavailable: {error}", - server.name, server.url + server.name, server.endpoint )), None => { lines.push(format!( "- {} ({}) — {} tool(s)", server.name, - server.url, + server.endpoint, server.tools.len() )); for tool in &server.tools { @@ -867,23 +1305,117 @@ mod tests { ); } + #[test] + fn parses_stdio_server_with_args_and_env() { + let toml = r#" + [[server]] + name = "fs" + command = "npx" + args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"] + + [server.env] + LOG_LEVEL = "debug" + TOKEN = "abc" + "#; + let parsed: McpFile = toml::from_str(toml).unwrap(); + assert_eq!(parsed.server.len(), 1); + let entry = &parsed.server[0]; + assert_eq!(entry.command.as_deref(), Some("npx")); + assert_eq!(entry.args.len(), 3); + assert_eq!( + entry.env.get("LOG_LEVEL").map(String::as_str), + Some("debug") + ); + assert_eq!(entry.env.get("TOKEN").map(String::as_str), Some("abc")); + + let def = entry.transport_def().expect("valid stdio entry"); + match def { + TransportDef::Stdio { command, args, env } => { + assert_eq!(command, "npx"); + assert_eq!(args[0], "-y"); + assert_eq!(env.len(), 2); + } + TransportDef::Http { .. } => panic!("expected a stdio transport"), + } + } + + #[test] + fn entry_with_url_and_command_is_a_config_error() { + let toml = r#" + [[server]] + name = "confused" + url = "https://example.com/mcp" + command = "npx" + "#; + let parsed: McpFile = toml::from_str(toml).unwrap(); + let error = parsed.server[0].transport_def().unwrap_err(); + assert!(error.contains("both"), "unexpected error: {error}"); + } + + #[test] + fn entry_with_neither_url_nor_command_is_a_config_error() { + let toml = r#" + [[server]] + name = "empty" + "#; + let parsed: McpFile = toml::from_str(toml).unwrap(); + let error = parsed.server[0].transport_def().unwrap_err(); + assert!(error.contains("needs"), "unexpected error: {error}"); + } + + #[test] + fn blank_url_or_command_counts_as_absent() { + let toml = r#" + [[server]] + name = "blank" + url = " " + command = "server-bin" + "#; + let parsed: McpFile = toml::from_str(toml).unwrap(); + // A blank url is treated as absent, so this resolves to stdio. + match parsed.server[0].transport_def().expect("stdio") { + TransportDef::Stdio { command, .. } => assert_eq!(command, "server-bin"), + TransportDef::Http { .. } => panic!("expected stdio"), + } + } + + #[test] + fn endpoint_renders_url_or_command_line() { + let http = TransportDef::Http { + url: "https://example.com/mcp".into(), + headers: vec![], + }; + assert_eq!(http.endpoint(), "https://example.com/mcp"); + + let stdio = TransportDef::Stdio { + command: "npx".into(), + args: vec!["-y".into(), "server-fs".into()], + env: vec![], + }; + assert_eq!(stdio.endpoint(), "npx -y server-fs"); + } + #[test] fn upsert_replaces_same_name() { let mut defs = vec![ServerDef { name: "a".into(), - url: "u1".into(), - headers: vec![], + transport: TransportDef::Http { + url: "u1".into(), + headers: vec![], + }, }]; upsert( &mut defs, ServerDef { name: "a".into(), - url: "u2".into(), - headers: vec![], + transport: TransportDef::Http { + url: "u2".into(), + headers: vec![], + }, }, ); assert_eq!(defs.len(), 1); - assert_eq!(defs[0].url, "u2"); + assert_eq!(defs[0].transport.endpoint(), "u2"); } #[test]

new file mode 100644 index 0000000..de42a9b --- /dev/null +++ b/tests/mcp_stdio.rs @@ -0,0 +1,371 @@ +//! End-to-end stdio MCP transport test against the real binary. +//! +//! Spawns `sigit` in ACP mode wired to a scripted OpenAI-compatible SSE +//! endpoint (the same harness as `acp_permissions.rs`) and to a temp +//! `SIGIT_CONFIG_DIR` whose `mcp.toml` configures three stdio servers, all +//! backed by the `mcp_stdio_stub` helper binary: +//! +//! - `stub` — healthy; exposes one `echo` tool and prefixes replies with the +//! `STUB_PREFIX` env var from `[server.env]`, proving env propagation. +//! - `dying` — completes the discovery handshake, then exits. Its tool is +//! offered to the model, but calling it must fail with an in-band error. +//! - `deadone` — exits(1) at spawn; discovery must record it unavailable and +//! offer no tools for it. +//! +//! The scripted model calls `mcp__stub__echo`, then `mcp__dying__echo`, then +//! finishes. The test asserts the tool round-trip, the dead-child error, and +//! the `/mcp` listing (command lines shown, dead server flagged). + +use std::collections::VecDeque; +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::TcpListener; +use std::process::{Child, ChildStdin, Command, Stdio}; +use std::sync::mpsc::{Receiver, channel}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use serde_json::{Value, json}; + +const TIMEOUT: Duration = Duration::from_secs(60); + +// ── Scripted OpenAI-compatible endpoint ───────────────────────────────────── + +fn sse_body(events: &[Value]) -> String { + let mut body = String::new(); + for event in events { + body.push_str("data: "); + body.push_str(&event.to_string()); + body.push_str("\n\n"); + } + body.push_str("data: [DONE]\n\n"); + body +} + +fn sse_tool_call(id: &str, name: &str, arguments: &str) -> String { + sse_body(&[json!({ + "choices": [{"delta": {"tool_calls": [{ + "index": 0, + "id": id, + "function": {"name": name, "arguments": arguments}, + }]}}] + })]) +} + +fn sse_text(text: &str) -> String { + sse_body(&[json!({"choices": [{"delta": {"content": text}}]})]) +} + +/// Serves one scripted SSE response per request and records each request body. +struct FakeEndpoint { + port: u16, + requests: Arc<Mutex<Vec<Value>>>, +} + +fn start_fake_endpoint(responses: Vec<String>) -> FakeEndpoint { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint"); + let port = listener.local_addr().unwrap().port(); + let requests: Arc<Mutex<Vec<Value>>> = Arc::default(); + let recorded = Arc::clone(&requests); + let queue = Mutex::new(VecDeque::from(responses)); + + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let mut reader = BufReader::new(match stream.try_clone() { + Ok(clone) => clone, + Err(_) => continue, + }); + let mut content_length = 0usize; + loop { + let mut line = String::new(); + if reader.read_line(&mut line).unwrap_or(0) == 0 { + break; + } + let line = line.trim(); + if line.is_empty() { + break; + } + if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") { + content_length = length.trim().parse().unwrap_or(0); + } + } + let mut body = vec![0u8; content_length]; + if reader.read_exact(&mut body).is_err() { + continue; + } + if let Ok(request) = serde_json::from_slice::<Value>(&body) { + recorded.lock().unwrap().push(request); + } + let payload = queue + .lock() + .unwrap() + .pop_front() + .unwrap_or_else(|| sse_text("out of scripted responses")); + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\ + content-length: {}\r\nconnection: close\r\n\r\n{}", + payload.len(), + payload + ); + let _ = stream.write_all(response.as_bytes()); + } + }); + + FakeEndpoint { port, requests } +} + +// ── ACP client over the binary's stdio ────────────────────────────────────── + +struct AgentUnderTest { + child: Child, + stdin: ChildStdin, + incoming: Receiver<Value>, + next_id: u64, +} + +fn spawn_agent(port: u16, config_dir: &std::path::Path, cwd: &std::path::Path) -> AgentUnderTest { + let mut child = Command::new(env!("CARGO_BIN_EXE_sigit")) + .current_dir(cwd) + .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}")) + .env("OPENAI_API_KEY", "test-key") + .env("SIGIT_MODEL", "scripted-model") + .env("SIGIT_CONFIG_DIR", config_dir) + // MCP stays ON (that's what we test), but the baked-in official + // server must not phone home from CI. + .env("SIGIT_MCP_OFFICIAL", "off") + .env_remove("SIGIT_MCP") + // MCP tools are mutating and would otherwise wait at the permission + // gate; permissions have their own test. + .env("SIGIT_PERMISSIONS", "allow") + .env_remove("SIGIT_LOCAL_INFERENCE") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn sigit in ACP mode"); + + let stdout = child.stdout.take().unwrap(); + let (message_tx, incoming) = channel(); + std::thread::spawn(move || { + for line in BufReader::new(stdout).lines() { + let Ok(line) = line else { break }; + if let Ok(message) = serde_json::from_str::<Value>(&line) + && message_tx.send(message).is_err() + { + break; + } + } + }); + + let stdin = child.stdin.take().unwrap(); + AgentUnderTest { + child, + stdin, + incoming, + next_id: 0, + } +} + +impl AgentUnderTest { + fn send(&mut self, message: Value) { + let mut line = message.to_string(); + line.push('\n'); + self.stdin + .write_all(line.as_bytes()) + .expect("write to agent stdin"); + self.stdin.flush().expect("flush agent stdin"); + } + + fn request(&mut self, method: &str, params: Value) -> u64 { + self.next_id += 1; + let id = self.next_id; + self.send(json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params})); + id + } + + /// Wait for the response to our request `id`, collecting the raw JSON of + /// every `session/update` notification that arrives before it. + fn wait_for_response_collecting_updates(&mut self, id: u64) -> (Value, String) { + let deadline = Instant::now() + TIMEOUT; + let mut updates = String::new(); + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + match self.incoming.recv_timeout(remaining) { + Ok(message) if message["id"] == id && message.get("method").is_none() => { + assert!( + message.get("error").is_none(), + "request {id} failed: {message}" + ); + return (message, updates); + } + Ok(message) => { + if message["method"] == "session/update" { + updates.push_str(&message["params"].to_string()); + updates.push('\n'); + } + } + Err(_) => panic!("timed out waiting for response to request {id}"), + } + } + } + + fn wait_for_response(&mut self, id: u64) -> Value { + self.wait_for_response_collecting_updates(id).0 + } +} + +impl Drop for AgentUnderTest { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +// ── The round-trip ────────────────────────────────────────────────────────── + +#[test] +fn stdio_mcp_discovery_call_and_dead_child() { + let stub = env!("CARGO_BIN_EXE_mcp_stdio_stub"); + + let endpoint = start_fake_endpoint(vec![ + sse_tool_call("call_1", "mcp__stub__echo", r#"{"text":"hello"}"#), + sse_tool_call("call_2", "mcp__dying__echo", r#"{"text":"gone"}"#), + sse_text("done"), + ]); + + let scratch = std::env::temp_dir().join(format!("sigit_mcp_stdio_{}", std::process::id())); + let config_dir = scratch.join("config"); + let cwd = scratch.join("cwd"); + std::fs::create_dir_all(&config_dir).unwrap(); + std::fs::create_dir_all(&cwd).unwrap(); + + // TOML literal strings (single quotes) keep Windows backslashes intact. + let mcp_toml = format!( + r#"official = false + +[[server]] +name = "stub" +command = '{stub}' + +[server.env] +STUB_PREFIX = "pfx:" + +[[server]] +name = "dying" +command = '{stub}' +args = ["--exit-after-list"] + +[[server]] +name = "deadone" +command = '{stub}' +args = ["--fail"] +"# + ); + std::fs::write(config_dir.join("mcp.toml"), mcp_toml).unwrap(); + + let mut agent = spawn_agent(endpoint.port, &config_dir, &cwd); + + let id = agent.request( + "initialize", + json!({"protocolVersion": 1, "clientCapabilities": {}}), + ); + agent.wait_for_response(id); + + let id = agent.request("session/new", json!({"cwd": cwd, "mcpServers": []})); + let session_id = agent.wait_for_response(id)["result"]["sessionId"] + .as_str() + .expect("session id") + .to_string(); + + // ── /mcp: listing shows command lines and flags the dead server ───── + let prompt_id = agent.request( + "session/prompt", + json!({ + "sessionId": session_id, + "prompt": [{"type": "text", "text": "/mcp"}], + }), + ); + let (_, listing) = agent.wait_for_response_collecting_updates(prompt_id); + // Raw JSON of the update notifications; escape the path the way JSON does + // so Windows backslashes compare correctly. + let stub_json = serde_json::to_string(stub).unwrap(); + let stub_escaped = stub_json.trim_matches('"'); + assert!( + listing.contains("mcp__stub__echo"), + "/mcp must list the healthy server's tool, got: {listing}" + ); + assert!( + listing.contains(stub_escaped), + "/mcp must show the stdio server's command line, got: {listing}" + ); + assert!( + listing.contains("--exit-after-list"), + "/mcp must include the args in the command line, got: {listing}" + ); + assert!( + listing.contains("unavailable"), + "/mcp must flag the server that died at spawn, got: {listing}" + ); + + // ── One prompt: echo round-trip, then the dead-child call ─────────── + let prompt_id = agent.request( + "session/prompt", + json!({ + "sessionId": session_id, + "prompt": [{"type": "text", "text": "use the stub tools"}], + }), + ); + let response = agent.wait_for_response(prompt_id); + assert_eq!(response["result"]["stopReason"], "end_turn"); + + // ── What the endpoint saw ──────────────────────────────────────────── + let requests = endpoint.requests.lock().unwrap(); + // Slash commands never reach the model, so all three completions belong + // to the tool-calling prompt. + assert_eq!(requests.len(), 3, "expected exactly three completions"); + + // The offered tool specs must include both live servers' echo tools and + // nothing from the server that failed discovery. + let tools = requests[0]["tools"].to_string(); + assert!( + tools.contains("mcp__stub__echo"), + "stub tool missing from specs: {tools}" + ); + assert!( + tools.contains("mcp__dying__echo"), + "dying server's tool missing from specs: {tools}" + ); + assert!( + !tools.contains("mcp__deadone__"), + "a server that failed discovery must contribute no tools: {tools}" + ); + + // The echo call's result must round-trip, carrying the [server.env] + // prefix (proving env vars reached the child). + let messages = requests[1]["messages"].as_array().expect("messages"); + let result = messages + .iter() + .find(|message| message["role"] == "tool" && message["tool_call_id"] == "call_1") + .expect("tool result for the echo call"); + assert_eq!( + result["content"].as_str().unwrap_or_default(), + "pfx:hello", + "echo result should carry the env-var prefix" + ); + + // The call to the server that died after discovery must come back as an + // in-band error string, not hang or crash the agent. + let messages = requests[2]["messages"].as_array().expect("messages"); + let result = messages + .iter() + .find(|message| message["role"] == "tool" && message["tool_call_id"] == "call_2") + .expect("tool result for the dead server's call"); + let content = result["content"].as_str().unwrap_or_default(); + assert!( + content.contains("Error") && content.contains("stdio server 'dying'"), + "dead-child call must fail in-band, got: {content}" + ); + + drop(agent); + let _ = std::fs::remove_dir_all(&scratch); +}