1 //! End-to-end stdio MCP transport test against the real binary.
2 //!
3 //! Spawns `sigit` in ACP mode wired to a scripted OpenAI-compatible SSE
4 //! endpoint (the same harness as `acp_permissions.rs`) and to a temp
5 //! `SIGIT_CONFIG_DIR` whose `mcp.toml` configures three stdio servers, all
6 //! backed by the `mcp_stdio_stub` helper binary:
7 //!
8 //! - `stub` — healthy; exposes one `echo` tool and prefixes replies with the
9 //! `STUB_PREFIX` env var from `[server.env]`, proving env propagation.
10 //! - `dying` — completes the discovery handshake, then exits. Its tool is
11 //! offered to the model, but calling it must fail with an in-band error.
12 //! - `deadone` — exits(1) at spawn; discovery must record it unavailable and
13 //! offer no tools for it.
14 //!
15 //! The scripted model calls `mcp__stub__echo`, then `mcp__dying__echo`, then
16 //! finishes. The test asserts the tool round-trip, the dead-child error, and
17 //! the `/mcp` listing (command lines shown, dead server flagged).
18
19 use std::collections::VecDeque;
20 use std::io::{BufRead, BufReader, Read, Write};
21 use std::net::TcpListener;
22 use std::process::{Child, ChildStdin, Command, Stdio};
23 use std::sync::mpsc::{Receiver, channel};
24 use std::sync::{Arc, Mutex};
25 use std::time::{Duration, Instant};
26
27 use serde_json::{Value, json};
28
29 const TIMEOUT: Duration = Duration::from_secs(60);
30
31 // ── Scripted OpenAI-compatible endpoint ─────────────────────────────────────
32
33 fn sse_body(events: &[Value]) -> String {
34 let mut body = String::new();
35 for event in events {
36 body.push_str("data: ");
37 body.push_str(&event.to_string());
38 body.push_str("\n\n");
39 }
40 body.push_str("data: [DONE]\n\n");
41 body
42 }
43
44 fn sse_tool_call(id: &str, name: &str, arguments: &str) -> String {
45 sse_body(&[json!({
46 "choices": [{"delta": {"tool_calls": [{
47 "index": 0,
48 "id": id,
49 "function": {"name": name, "arguments": arguments},
50 }]}}]
51 })])
52 }
53
54 fn sse_text(text: &str) -> String {
55 sse_body(&[json!({"choices": [{"delta": {"content": text}}]})])
56 }
57
58 /// Serves one scripted SSE response per request and records each request body.
59 struct FakeEndpoint {
60 port: u16,
61 requests: Arc<Mutex<Vec<Value>>>,
62 }
63
64 fn start_fake_endpoint(responses: Vec<String>) -> FakeEndpoint {
65 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
66 let port = listener.local_addr().unwrap().port();
67 let requests: Arc<Mutex<Vec<Value>>> = Arc::default();
68 let recorded = Arc::clone(&requests);
69 let queue = Mutex::new(VecDeque::from(responses));
70
71 std::thread::spawn(move || {
72 for stream in listener.incoming() {
73 let Ok(mut stream) = stream else { continue };
74 let mut reader = BufReader::new(match stream.try_clone() {
75 Ok(clone) => clone,
76 Err(_) => continue,
77 });
78 let mut content_length = 0usize;
79 loop {
80 let mut line = String::new();
81 if reader.read_line(&mut line).unwrap_or(0) == 0 {
82 break;
83 }
84 let line = line.trim();
85 if line.is_empty() {
86 break;
87 }
88 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
89 content_length = length.trim().parse().unwrap_or(0);
90 }
91 }
92 let mut body = vec![0u8; content_length];
93 if reader.read_exact(&mut body).is_err() {
94 continue;
95 }
96 if let Ok(request) = serde_json::from_slice::<Value>(&body) {
97 recorded.lock().unwrap().push(request);
98 }
99 let payload = queue
100 .lock()
101 .unwrap()
102 .pop_front()
103 .unwrap_or_else(|| sse_text("out of scripted responses"));
104 let response = format!(
105 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\
106 content-length: {}\r\nconnection: close\r\n\r\n{}",
107 payload.len(),
108 payload
109 );
110 let _ = stream.write_all(response.as_bytes());
111 }
112 });
113
114 FakeEndpoint { port, requests }
115 }
116
117 // ── ACP client over the binary's stdio ──────────────────────────────────────
118
119 struct AgentUnderTest {
120 child: Child,
121 stdin: ChildStdin,
122 incoming: Receiver<Value>,
123 next_id: u64,
124 }
125
126 fn spawn_agent(port: u16, config_dir: &std::path::Path, cwd: &std::path::Path) -> AgentUnderTest {
127 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
128 .current_dir(cwd)
129 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
130 .env("OPENAI_API_KEY", "test-key")
131 .env("SIGIT_MODEL", "scripted-model")
132 .env("SIGIT_CONFIG_DIR", config_dir)
133 // MCP stays ON (that's what we test), but the baked-in official
134 // server must not phone home from CI.
135 .env("SIGIT_MCP_OFFICIAL", "off")
136 .env_remove("SIGIT_MCP")
137 // MCP tools are mutating and would otherwise wait at the permission
138 // gate; permissions have their own test.
139 .env("SIGIT_PERMISSIONS", "allow")
140 .env_remove("SIGIT_LOCAL_INFERENCE")
141 .stdin(Stdio::piped())
142 .stdout(Stdio::piped())
143 .stderr(Stdio::null())
144 .spawn()
145 .expect("spawn sigit in ACP mode");
146
147 let stdout = child.stdout.take().unwrap();
148 let (message_tx, incoming) = channel();
149 std::thread::spawn(move || {
150 for line in BufReader::new(stdout).lines() {
151 let Ok(line) = line else { break };
152 if let Ok(message) = serde_json::from_str::<Value>(&line)
153 && message_tx.send(message).is_err()
154 {
155 break;
156 }
157 }
158 });
159
160 let stdin = child.stdin.take().unwrap();
161 AgentUnderTest {
162 child,
163 stdin,
164 incoming,
165 next_id: 0,
166 }
167 }
168
169 impl AgentUnderTest {
170 fn send(&mut self, message: Value) {
171 let mut line = message.to_string();
172 line.push('\n');
173 self.stdin
174 .write_all(line.as_bytes())
175 .expect("write to agent stdin");
176 self.stdin.flush().expect("flush agent stdin");
177 }
178
179 fn request(&mut self, method: &str, params: Value) -> u64 {
180 self.next_id += 1;
181 let id = self.next_id;
182 self.send(json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}));
183 id
184 }
185
186 /// Wait for the response to our request `id`, collecting the raw JSON of
187 /// every `session/update` notification that arrives before it.
188 fn wait_for_response_collecting_updates(&mut self, id: u64) -> (Value, String) {
189 let deadline = Instant::now() + TIMEOUT;
190 let mut updates = String::new();
191 loop {
192 let remaining = deadline.saturating_duration_since(Instant::now());
193 match self.incoming.recv_timeout(remaining) {
194 Ok(message) if message["id"] == id && message.get("method").is_none() => {
195 assert!(
196 message.get("error").is_none(),
197 "request {id} failed: {message}"
198 );
199 return (message, updates);
200 }
201 Ok(message) => {
202 if message["method"] == "session/update" {
203 updates.push_str(&message["params"].to_string());
204 updates.push('\n');
205 }
206 }
207 Err(_) => panic!("timed out waiting for response to request {id}"),
208 }
209 }
210 }
211
212 fn wait_for_response(&mut self, id: u64) -> Value {
213 self.wait_for_response_collecting_updates(id).0
214 }
215 }
216
217 impl Drop for AgentUnderTest {
218 fn drop(&mut self) {
219 let _ = self.child.kill();
220 let _ = self.child.wait();
221 }
222 }
223
224 // ── The round-trip ──────────────────────────────────────────────────────────
225
226 #[test]
227 fn stdio_mcp_discovery_call_and_dead_child() {
228 let stub = env!("CARGO_BIN_EXE_mcp_stdio_stub");
229
230 let endpoint = start_fake_endpoint(vec![
231 sse_tool_call("call_1", "mcp__stub__echo", r#"{"text":"hello"}"#),
232 sse_tool_call("call_2", "mcp__dying__echo", r#"{"text":"gone"}"#),
233 sse_text("done"),
234 ]);
235
236 let scratch = std::env::temp_dir().join(format!("sigit_mcp_stdio_{}", std::process::id()));
237 let config_dir = scratch.join("config");
238 let cwd = scratch.join("cwd");
239 std::fs::create_dir_all(&config_dir).unwrap();
240 std::fs::create_dir_all(&cwd).unwrap();
241
242 // TOML literal strings (single quotes) keep Windows backslashes intact.
243 let mcp_toml = format!(
244 r#"official = false
245
246 [[server]]
247 name = "stub"
248 command = '{stub}'
249
250 [server.env]
251 STUB_PREFIX = "pfx:"
252
253 [[server]]
254 name = "dying"
255 command = '{stub}'
256 args = ["--exit-after-list"]
257
258 [[server]]
259 name = "deadone"
260 command = '{stub}'
261 args = ["--fail"]
262 "#
263 );
264 std::fs::write(config_dir.join("mcp.toml"), mcp_toml).unwrap();
265
266 let mut agent = spawn_agent(endpoint.port, &config_dir, &cwd);
267
268 let id = agent.request(
269 "initialize",
270 json!({"protocolVersion": 1, "clientCapabilities": {}}),
271 );
272 agent.wait_for_response(id);
273
274 let id = agent.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
275 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
276 .as_str()
277 .expect("session id")
278 .to_string();
279
280 // ── /mcp: listing shows command lines and flags the dead server ─────
281 let prompt_id = agent.request(
282 "session/prompt",
283 json!({
284 "sessionId": session_id,
285 "prompt": [{"type": "text", "text": "/mcp"}],
286 }),
287 );
288 let (_, listing) = agent.wait_for_response_collecting_updates(prompt_id);
289 // Raw JSON of the update notifications; escape the path the way JSON does
290 // so Windows backslashes compare correctly.
291 let stub_json = serde_json::to_string(stub).unwrap();
292 let stub_escaped = stub_json.trim_matches('"');
293 assert!(
294 listing.contains("mcp__stub__echo"),
295 "/mcp must list the healthy server's tool, got: {listing}"
296 );
297 assert!(
298 listing.contains(stub_escaped),
299 "/mcp must show the stdio server's command line, got: {listing}"
300 );
301 assert!(
302 listing.contains("--exit-after-list"),
303 "/mcp must include the args in the command line, got: {listing}"
304 );
305 assert!(
306 listing.contains("unavailable"),
307 "/mcp must flag the server that died at spawn, got: {listing}"
308 );
309
310 // ── One prompt: echo round-trip, then the dead-child call ───────────
311 let prompt_id = agent.request(
312 "session/prompt",
313 json!({
314 "sessionId": session_id,
315 "prompt": [{"type": "text", "text": "use the stub tools"}],
316 }),
317 );
318 let response = agent.wait_for_response(prompt_id);
319 assert_eq!(response["result"]["stopReason"], "end_turn");
320
321 // ── What the endpoint saw ────────────────────────────────────────────
322 let requests = endpoint.requests.lock().unwrap();
323 // Slash commands never reach the model, so all three completions belong
324 // to the tool-calling prompt.
325 assert_eq!(requests.len(), 3, "expected exactly three completions");
326
327 // The offered tool specs must include both live servers' echo tools and
328 // nothing from the server that failed discovery.
329 let tools = requests[0]["tools"].to_string();
330 assert!(
331 tools.contains("mcp__stub__echo"),
332 "stub tool missing from specs: {tools}"
333 );
334 assert!(
335 tools.contains("mcp__dying__echo"),
336 "dying server's tool missing from specs: {tools}"
337 );
338 assert!(
339 !tools.contains("mcp__deadone__"),
340 "a server that failed discovery must contribute no tools: {tools}"
341 );
342
343 // The echo call's result must round-trip, carrying the [server.env]
344 // prefix (proving env vars reached the child).
345 let messages = requests[1]["messages"].as_array().expect("messages");
346 let result = messages
347 .iter()
348 .find(|message| message["role"] == "tool" && message["tool_call_id"] == "call_1")
349 .expect("tool result for the echo call");
350 assert_eq!(
351 result["content"].as_str().unwrap_or_default(),
352 "pfx:hello",
353 "echo result should carry the env-var prefix"
354 );
355
356 // The call to the server that died after discovery must come back as an
357 // in-band error string, not hang or crash the agent.
358 let messages = requests[2]["messages"].as_array().expect("messages");
359 let result = messages
360 .iter()
361 .find(|message| message["role"] == "tool" && message["tool_call_id"] == "call_2")
362 .expect("tool result for the dead server's call");
363 let content = result["content"].as_str().unwrap_or_default();
364 assert!(
365 content.contains("Error") && content.contains("stdio server 'dying'"),
366 "dead-child call must fail in-band, got: {content}"
367 );
368
369 drop(agent);
370 let _ = std::fs::remove_dir_all(&scratch);
371 }