1 //! End-to-end ACP permission round-trip against the real binary.
2 //!
3 //! Spawns `sigit` in ACP mode (stdin piped, so not a TTY) wired to a scripted
4 //! OpenAI-compatible SSE endpoint via the `OPENAI_BASE_URL` override, then
5 //! drives newline-delimited JSON-RPC over stdio. The scripted model calls
6 //! `run_command` — a mutating tool — so the agent must send
7 //! `session/request_permission` mid-turn (the exact path the spawned-handler /
8 //! `turn_lock` design exists for). The test answers it twice:
9 //!
10 //! 1. `cancelled` — the prompt must stop with `stopReason: "cancelled"`, and
11 //! the *next* request to the endpoint must show the abandoned round closed
12 //! out with `role: "tool"` results, or a strict OpenAI-compatible endpoint
13 //! would reject the whole session.
14 //! 2. `selected: allow_once` — the tool must actually execute and its output
15 //! travel back to the endpoint as a tool result.
16 //!
17 //! A second test covers live progress: a slow command's `in_progress` tool call
18 //! must reach the client while the command is still running, which is what the
19 //! editor's spinner is driven from. A third test covers what the client
20 //! renders: text from two tool rounds must reach it as separate paragraphs,
21 //! since ACP clients concatenate consecutive agent-message chunks into one
22 //! block.
23
24 use std::collections::VecDeque;
25 use std::io::{BufRead, BufReader, Read, Write};
26 use std::net::TcpListener;
27 use std::process::{Child, ChildStdin, Command, Stdio};
28 use std::sync::mpsc::{Receiver, channel};
29 use std::sync::{Arc, Mutex};
30 use std::time::{Duration, Instant};
31
32 use serde_json::{Value, json};
33
34 const TIMEOUT: Duration = Duration::from_secs(60);
35
36 // ── Scripted OpenAI-compatible endpoint ─────────────────────────────────────
37
38 fn sse_body(events: &[Value]) -> String {
39 let mut body = String::new();
40 for event in events {
41 body.push_str("data: ");
42 body.push_str(&event.to_string());
43 body.push_str("\n\n");
44 }
45 body.push_str("data: [DONE]\n\n");
46 body
47 }
48
49 fn sse_tool_call(id: &str, name: &str, arguments: &str) -> String {
50 sse_body(&[json!({
51 "choices": [{"delta": {"tool_calls": [{
52 "index": 0,
53 "id": id,
54 "function": {"name": name, "arguments": arguments},
55 }]}}]
56 })])
57 }
58
59 fn sse_text(text: &str) -> String {
60 sse_body(&[json!({"choices": [{"delta": {"content": text}}]})])
61 }
62
63 /// One round that says something and then asks for a tool — the shape that used
64 /// to leave the next round's text glued to this sentence.
65 fn sse_text_then_tool_call(text: &str, id: &str, name: &str, arguments: &str) -> String {
66 sse_body(&[
67 json!({"choices": [{"delta": {"content": text}}]}),
68 json!({
69 "choices": [{"delta": {"tool_calls": [{
70 "index": 0,
71 "id": id,
72 "function": {"name": name, "arguments": arguments},
73 }]}}]
74 }),
75 ])
76 }
77
78 /// Serves one scripted SSE response per request and records each request body.
79 struct FakeEndpoint {
80 port: u16,
81 requests: Arc<Mutex<Vec<Value>>>,
82 }
83
84 fn start_fake_endpoint(responses: Vec<String>) -> FakeEndpoint {
85 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
86 let port = listener.local_addr().unwrap().port();
87 let requests: Arc<Mutex<Vec<Value>>> = Arc::default();
88 let recorded = Arc::clone(&requests);
89 let queue = Mutex::new(VecDeque::from(responses));
90
91 std::thread::spawn(move || {
92 // `connection: close` below means one request per connection, so the
93 // serial accept loop matches the agent's serial completion requests.
94 for stream in listener.incoming() {
95 let Ok(mut stream) = stream else { continue };
96 let mut reader = BufReader::new(match stream.try_clone() {
97 Ok(clone) => clone,
98 Err(_) => continue,
99 });
100 let mut content_length = 0usize;
101 loop {
102 let mut line = String::new();
103 if reader.read_line(&mut line).unwrap_or(0) == 0 {
104 break;
105 }
106 let line = line.trim();
107 if line.is_empty() {
108 break;
109 }
110 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
111 content_length = length.trim().parse().unwrap_or(0);
112 }
113 }
114 let mut body = vec![0u8; content_length];
115 if reader.read_exact(&mut body).is_err() {
116 continue;
117 }
118 if let Ok(request) = serde_json::from_slice::<Value>(&body) {
119 recorded.lock().unwrap().push(request);
120 }
121 let payload = queue
122 .lock()
123 .unwrap()
124 .pop_front()
125 .unwrap_or_else(|| sse_text("out of scripted responses"));
126 let response = format!(
127 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\
128 content-length: {}\r\nconnection: close\r\n\r\n{}",
129 payload.len(),
130 payload
131 );
132 let _ = stream.write_all(response.as_bytes());
133 }
134 });
135
136 FakeEndpoint { port, requests }
137 }
138
139 // ── ACP client over the binary's stdio ──────────────────────────────────────
140
141 struct AgentUnderTest {
142 child: Child,
143 stdin: ChildStdin,
144 incoming: Receiver<Value>,
145 next_id: u64,
146 }
147
148 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
149 spawn_agent_with_permissions(port, config_dir, None)
150 }
151
152 /// `permissions` sets `SIGIT_PERMISSIONS`; `None` leaves the default `ask` mode
153 /// in place, which is what the permission round-trip needs.
154 fn spawn_agent_with_permissions(
155 port: u16,
156 config_dir: &std::path::Path,
157 permissions: Option<&str>,
158 ) -> AgentUnderTest {
159 let mut command = Command::new(env!("CARGO_BIN_EXE_sigit"));
160 command
161 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
162 .env("OPENAI_API_KEY", "test-key")
163 .env("SIGIT_MODEL", "scripted-model")
164 .env("SIGIT_CONFIG_DIR", config_dir)
165 .env("SIGIT_MCP", "off")
166 .env_remove("SIGIT_LOCAL_INFERENCE")
167 .stdin(Stdio::piped())
168 .stdout(Stdio::piped())
169 .stderr(Stdio::null());
170 match permissions {
171 // A fresh config dir means the default permission mode, `ask` — make
172 // sure the environment can't turn the gate off underneath the test.
173 None => command.env_remove("SIGIT_PERMISSIONS"),
174 Some(mode) => command.env("SIGIT_PERMISSIONS", mode),
175 };
176 let mut child = command.spawn().expect("spawn sigit in ACP mode");
177
178 let stdout = child.stdout.take().unwrap();
179 let (message_tx, incoming) = channel();
180 std::thread::spawn(move || {
181 for line in BufReader::new(stdout).lines() {
182 let Ok(line) = line else { break };
183 if let Ok(message) = serde_json::from_str::<Value>(&line)
184 && message_tx.send(message).is_err()
185 {
186 break;
187 }
188 }
189 });
190
191 let stdin = child.stdin.take().unwrap();
192 AgentUnderTest {
193 child,
194 stdin,
195 incoming,
196 next_id: 0,
197 }
198 }
199
200 impl AgentUnderTest {
201 fn send(&mut self, message: Value) {
202 let mut line = message.to_string();
203 line.push('\n');
204 self.stdin
205 .write_all(line.as_bytes())
206 .expect("write to agent stdin");
207 self.stdin.flush().expect("flush agent stdin");
208 }
209
210 fn request(&mut self, method: &str, params: Value) -> u64 {
211 self.next_id += 1;
212 let id = self.next_id;
213 self.send(json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}));
214 id
215 }
216
217 fn respond(&mut self, id: Value, result: Value) {
218 self.send(json!({"jsonrpc": "2.0", "id": id, "result": result}));
219 }
220
221 /// Skip notifications and unrelated traffic until `matches` is satisfied.
222 fn wait_for(&mut self, what: &str, matches: impl Fn(&Value) -> bool) -> Value {
223 let deadline = Instant::now() + TIMEOUT;
224 loop {
225 let remaining = deadline.saturating_duration_since(Instant::now());
226 match self.incoming.recv_timeout(remaining) {
227 Ok(message) if matches(&message) => return message,
228 Ok(_) => continue,
229 Err(_) => panic!("timed out waiting for {what}"),
230 }
231 }
232 }
233
234 /// The response to one of *our* requests (has our id, no `method`).
235 fn wait_for_response(&mut self, id: u64) -> Value {
236 let response = self.wait_for(&format!("response to request {id}"), |message| {
237 message["id"] == id && message.get("method").is_none()
238 });
239 assert!(
240 response.get("error").is_none(),
241 "request {id} failed: {response}"
242 );
243 response
244 }
245
246 /// The first `session/update` whose payload satisfies `matches`.
247 ///
248 /// Only used by tests that run with the permission gate off, so a
249 /// `session/request_permission` means the run is misconfigured — say that
250 /// rather than waiting out the timeout on a request nobody will answer.
251 fn wait_for_update(&mut self, what: &str, matches: impl Fn(&Value) -> bool) -> Value {
252 let message = self.wait_for(what, |message| {
253 assert!(
254 message["method"] != "session/request_permission",
255 "unexpected permission request while waiting for {what}: {message}"
256 );
257 message["method"] == "session/update" && matches(&message["params"]["update"])
258 });
259 message["params"]["update"].clone()
260 }
261
262 /// The response to `session/prompt`, paired with every `agent_message_chunk`
263 /// seen on the way, concatenated exactly as a client renders them.
264 fn wait_for_prompt(&mut self, id: u64) -> (Value, String) {
265 let mut rendered = String::new();
266 let deadline = Instant::now() + TIMEOUT;
267 loop {
268 let remaining = deadline.saturating_duration_since(Instant::now());
269 let Ok(message) = self.incoming.recv_timeout(remaining) else {
270 panic!("timed out waiting for the prompt response");
271 };
272 let update = &message["params"]["update"];
273 if message["method"] == "session/update"
274 && update["sessionUpdate"] == "agent_message_chunk"
275 && let Some(chunk) = update["content"]["text"].as_str()
276 {
277 rendered.push_str(chunk);
278 }
279 if message["id"] == id && message.get("method").is_none() {
280 assert!(message.get("error").is_none(), "prompt failed: {message}");
281 return (message, rendered);
282 }
283 }
284 }
285
286 /// A request *from* the agent (has a `method` and its own id).
287 fn wait_for_agent_request(&mut self, method: &str) -> Value {
288 self.wait_for(&format!("agent request {method}"), |message| {
289 message["method"] == method && message.get("id").is_some()
290 })
291 }
292 }
293
294 impl Drop for AgentUnderTest {
295 fn drop(&mut self) {
296 let _ = self.child.kill();
297 let _ = self.child.wait();
298 }
299 }
300
301 // ── The round-trip ──────────────────────────────────────────────────────────
302
303 #[test]
304 fn permission_round_trip_cancel_then_allow() {
305 let endpoint = start_fake_endpoint(vec![
306 sse_tool_call("call_1", "run_command", r#"{"command":"echo sigit-first"}"#),
307 sse_tool_call(
308 "call_2",
309 "run_command",
310 r#"{"command":"echo sigit-approved"}"#,
311 ),
312 sse_text("done"),
313 ]);
314
315 let scratch = std::env::temp_dir().join(format!("sigit_acp_perm_{}", std::process::id()));
316 let config_dir = scratch.join("config");
317 let cwd = scratch.join("cwd");
318 std::fs::create_dir_all(&config_dir).unwrap();
319 std::fs::create_dir_all(&cwd).unwrap();
320
321 let mut agent = spawn_agent(endpoint.port, &config_dir);
322
323 let id = agent.request(
324 "initialize",
325 json!({"protocolVersion": 1, "clientCapabilities": {}}),
326 );
327 agent.wait_for_response(id);
328
329 let id = agent.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
330 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
331 .as_str()
332 .expect("session id")
333 .to_string();
334
335 // ── Turn 1: cancel at the permission gate ───────────────────────────
336 let prompt_id = agent.request(
337 "session/prompt",
338 json!({
339 "sessionId": session_id,
340 "prompt": [{"type": "text", "text": "run the first command"}],
341 }),
342 );
343
344 let permission = agent.wait_for_agent_request("session/request_permission");
345 let params = &permission["params"];
346 assert_eq!(params["sessionId"], session_id.as_str());
347 let title = params["toolCall"]["title"].as_str().expect("title");
348 assert!(
349 title.contains("run_command") && title.contains("echo sigit-first"),
350 "the dialog must show the tool and its arguments, got: {title}"
351 );
352 assert_eq!(
353 params["toolCall"]["rawInput"]["command"], "echo sigit-first",
354 "full arguments must travel as rawInput"
355 );
356 let option_ids: Vec<&str> = params["options"]
357 .as_array()
358 .expect("options")
359 .iter()
360 .map(|option| option["optionId"].as_str().unwrap_or_default())
361 .collect();
362 assert_eq!(option_ids, ["allow_once", "allow_session", "reject_once"]);
363
364 agent.respond(
365 permission["id"].clone(),
366 json!({"outcome": {"outcome": "cancelled"}}),
367 );
368
369 let response = agent.wait_for_response(prompt_id);
370 assert_eq!(response["result"]["stopReason"], "cancelled");
371
372 // ── Turn 2: history must be repaired; then approve once ─────────────
373 let prompt_id = agent.request(
374 "session/prompt",
375 json!({
376 "sessionId": session_id,
377 "prompt": [{"type": "text", "text": "run the second command"}],
378 }),
379 );
380
381 let permission = agent.wait_for_agent_request("session/request_permission");
382 agent.respond(
383 permission["id"].clone(),
384 json!({"outcome": {"outcome": "selected", "optionId": "allow_once"}}),
385 );
386
387 let response = agent.wait_for_response(prompt_id);
388 assert_eq!(response["result"]["stopReason"], "end_turn");
389
390 // ── What the endpoint saw ────────────────────────────────────────────
391 let requests = endpoint.requests.lock().unwrap();
392 assert_eq!(requests.len(), 3, "expected exactly three completions");
393
394 // The client explicitly requests automatic tool selection. Some
395 // OpenAI-compatible gateways do not honour OpenAI's implicit `auto`
396 // default, which otherwise lets the model describe a tool action in prose
397 // and end the turn without a tool call.
398 assert_eq!(requests[0]["tool_choice"], "auto");
399
400 // Request 2 replays the full history: the cancelled round's tool call
401 // must be answered by a `role: "tool"` message, not left dangling.
402 let messages = requests[1]["messages"].as_array().expect("messages");
403 let call_position = messages
404 .iter()
405 .position(|message| message["tool_calls"][0]["id"] == "call_1")
406 .expect("cancelled turn's assistant tool call in replayed history");
407 let repair = &messages[call_position + 1];
408 assert_eq!(repair["role"], "tool", "dangling tool call not closed out");
409 assert_eq!(repair["tool_call_id"], "call_1");
410 assert!(
411 repair["content"]
412 .as_str()
413 .unwrap_or_default()
414 .contains("cancelled"),
415 "repair message should say the turn was cancelled: {repair}"
416 );
417
418 // Request 3 carries the approved call's real output.
419 let messages = requests[2]["messages"].as_array().expect("messages");
420 let result = messages
421 .iter()
422 .find(|message| message["role"] == "tool" && message["tool_call_id"] == "call_2")
423 .expect("tool result for the approved call");
424 assert!(
425 result["content"]
426 .as_str()
427 .unwrap_or_default()
428 .contains("sigit-approved"),
429 "the approved command's output should reach the endpoint: {result}"
430 );
431
432 drop(agent);
433 let _ = std::fs::remove_dir_all(&scratch);
434 }
435
436 #[test]
437 fn a_slow_commands_progress_reaches_the_client_while_it_runs() {
438 let scratch = std::env::temp_dir().join(format!("sigit_acp_progress_{}", std::process::id()));
439 let config_dir = scratch.join("config");
440 let cwd = scratch.join("cwd");
441 std::fs::create_dir_all(&config_dir).unwrap();
442 std::fs::create_dir_all(&cwd).unwrap();
443
444 // The command has to run for a few seconds through `sh -c` / `cmd /C`.
445 // `sleep` is neither a cmd builtin nor on a Windows runner's PATH (Git's
446 // usr\bin, which carries the MSYS coreutils, is deliberately left off it),
447 // and the Windows job hung on it rather than failing fast. `ping` ships in
448 // System32, waits a second between echoes, and never reads stdin.
449 //
450 // Four seconds against the one-second assertion below is deliberate
451 // padding. A busy runner only ever widens the gap being measured — it
452 // cannot make the command finish sooner — so the one way a correct agent
453 // could fail here is this thread being descheduled for seconds between
454 // reading the two updates off the channel. Three seconds of slack covers
455 // that without weakening the assertion, which is separating four seconds
456 // from the 164µs the blocking version produced.
457 #[cfg(unix)]
458 let command = "sleep 4";
459 #[cfg(windows)]
460 let command = "ping -n 5 127.0.0.1";
461
462 // `SIGIT_PERMISSIONS=allow` skips the gate, so the only thing between the
463 // two updates below is the command itself.
464 let endpoint = start_fake_endpoint(vec![
465 sse_tool_call(
466 "call_1",
467 "run_command",
468 &json!({"command": command, "cwd": cwd}).to_string(),
469 ),
470 sse_text("done"),
471 ]);
472
473 let mut agent = spawn_agent_with_permissions(endpoint.port, &config_dir, Some("allow"));
474
475 let id = agent.request(
476 "initialize",
477 json!({"protocolVersion": 1, "clientCapabilities": {}}),
478 );
479 agent.wait_for_response(id);
480
481 let id = agent.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
482 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
483 .as_str()
484 .expect("session id")
485 .to_string();
486
487 let prompt_id = agent.request(
488 "session/prompt",
489 json!({
490 "sessionId": session_id,
491 "prompt": [{"type": "text", "text": "run the slow command"}],
492 }),
493 );
494
495 let started = agent.wait_for_update("the tool call starting", |update| {
496 update["sessionUpdate"] == "tool_call" && update["toolCallId"] == "call_1"
497 });
498 let in_progress_at = Instant::now();
499 assert_eq!(
500 started["status"], "in_progress",
501 "a running tool call must be announced as in_progress: {started}"
502 );
503
504 let finished = agent.wait_for_update("the tool call finishing", |update| {
505 update["sessionUpdate"] == "tool_call_update"
506 && update["toolCallId"] == "call_1"
507 && update["status"] == "completed"
508 });
509
510 // The command runs between the two updates. If the turn blocked the
511 // connection's actors, both would only reach the client once it had
512 // already finished, and the editor would never draw a spinner.
513 assert!(
514 in_progress_at.elapsed() >= Duration::from_millis(1_000),
515 "in_progress arrived only {:?} before completed — the client had no \
516 window to show progress in. `{command}` said: {}",
517 in_progress_at.elapsed(),
518 finished["rawOutput"]
519 );
520
521 let response = agent.wait_for_response(prompt_id);
522 assert_eq!(response["result"]["stopReason"], "end_turn");
523
524 drop(agent);
525 let _ = std::fs::remove_dir_all(&scratch);
526 }
527
528 #[test]
529 fn text_from_two_tool_rounds_reaches_the_client_as_separate_paragraphs() {
530 let endpoint = start_fake_endpoint(vec![
531 // `list_directory` is read-only, so it runs without a permission gate
532 // and the two rounds follow each other straight away.
533 sse_text_then_tool_call(
534 "I'll search for this pattern.",
535 "call_1",
536 "list_directory",
537 r#"{"path":"."}"#,
538 ),
539 sse_text("Let me broaden the search:"),
540 ]);
541
542 let scratch = std::env::temp_dir().join(format!("sigit_acp_spacing_{}", std::process::id()));
543 let config_dir = scratch.join("config");
544 let cwd = scratch.join("cwd");
545 std::fs::create_dir_all(&config_dir).unwrap();
546 std::fs::create_dir_all(&cwd).unwrap();
547
548 let mut agent = spawn_agent(endpoint.port, &config_dir);
549
550 let id = agent.request(
551 "initialize",
552 json!({"protocolVersion": 1, "clientCapabilities": {}}),
553 );
554 agent.wait_for_response(id);
555
556 let id = agent.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
557 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
558 .as_str()
559 .expect("session id")
560 .to_string();
561
562 let prompt_id = agent.request(
563 "session/prompt",
564 json!({
565 "sessionId": session_id,
566 "prompt": [{"type": "text", "text": "find it"}],
567 }),
568 );
569 let (response, rendered) = agent.wait_for_prompt(prompt_id);
570 assert_eq!(response["result"]["stopReason"], "end_turn");
571
572 assert!(
573 rendered.contains("pattern.\n\nLet me broaden"),
574 "the second round must open a new paragraph, got: {rendered:?}"
575 );
576 assert!(
577 !rendered.contains("pattern.Let me"),
578 "rounds must not run together into one sentence, got: {rendered:?}"
579 );
580
581 drop(agent);
582 let _ = std::fs::remove_dir_all(&scratch);
583 }