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 use std::collections::VecDeque;
18 use std::io::{BufRead, BufReader, Read, Write};
19 use std::net::TcpListener;
20 use std::process::{Child, ChildStdin, Command, Stdio};
21 use std::sync::mpsc::{Receiver, channel};
22 use std::sync::{Arc, Mutex};
23 use std::time::{Duration, Instant};
24
25 use serde_json::{Value, json};
26
27 const TIMEOUT: Duration = Duration::from_secs(60);
28
29 // ── Scripted OpenAI-compatible endpoint ─────────────────────────────────────
30
31 fn sse_body(events: &[Value]) -> String {
32 let mut body = String::new();
33 for event in events {
34 body.push_str("data: ");
35 body.push_str(&event.to_string());
36 body.push_str("\n\n");
37 }
38 body.push_str("data: [DONE]\n\n");
39 body
40 }
41
42 fn sse_tool_call(id: &str, name: &str, arguments: &str) -> String {
43 sse_body(&[json!({
44 "choices": [{"delta": {"tool_calls": [{
45 "index": 0,
46 "id": id,
47 "function": {"name": name, "arguments": arguments},
48 }]}}]
49 })])
50 }
51
52 fn sse_text(text: &str) -> String {
53 sse_body(&[json!({"choices": [{"delta": {"content": text}}]})])
54 }
55
56 /// Serves one scripted SSE response per request and records each request body.
57 struct FakeEndpoint {
58 port: u16,
59 requests: Arc<Mutex<Vec<Value>>>,
60 }
61
62 fn start_fake_endpoint(responses: Vec<String>) -> FakeEndpoint {
63 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
64 let port = listener.local_addr().unwrap().port();
65 let requests: Arc<Mutex<Vec<Value>>> = Arc::default();
66 let recorded = Arc::clone(&requests);
67 let queue = Mutex::new(VecDeque::from(responses));
68
69 std::thread::spawn(move || {
70 // `connection: close` below means one request per connection, so the
71 // serial accept loop matches the agent's serial completion requests.
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) -> AgentUnderTest {
127 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
128 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
129 .env("OPENAI_API_KEY", "test-key")
130 .env("SIGIT_MODEL", "scripted-model")
131 .env("SIGIT_CONFIG_DIR", config_dir)
132 .env("SIGIT_MCP", "off")
133 // A fresh config dir means the default permission mode, `ask` — make
134 // sure the environment can't turn the gate off underneath the test.
135 .env_remove("SIGIT_PERMISSIONS")
136 .env_remove("SIGIT_LOCAL_INFERENCE")
137 .stdin(Stdio::piped())
138 .stdout(Stdio::piped())
139 .stderr(Stdio::null())
140 .spawn()
141 .expect("spawn sigit in ACP mode");
142
143 let stdout = child.stdout.take().unwrap();
144 let (message_tx, incoming) = channel();
145 std::thread::spawn(move || {
146 for line in BufReader::new(stdout).lines() {
147 let Ok(line) = line else { break };
148 if let Ok(message) = serde_json::from_str::<Value>(&line)
149 && message_tx.send(message).is_err()
150 {
151 break;
152 }
153 }
154 });
155
156 let stdin = child.stdin.take().unwrap();
157 AgentUnderTest {
158 child,
159 stdin,
160 incoming,
161 next_id: 0,
162 }
163 }
164
165 impl AgentUnderTest {
166 fn send(&mut self, message: Value) {
167 let mut line = message.to_string();
168 line.push('\n');
169 self.stdin
170 .write_all(line.as_bytes())
171 .expect("write to agent stdin");
172 self.stdin.flush().expect("flush agent stdin");
173 }
174
175 fn request(&mut self, method: &str, params: Value) -> u64 {
176 self.next_id += 1;
177 let id = self.next_id;
178 self.send(json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}));
179 id
180 }
181
182 fn respond(&mut self, id: Value, result: Value) {
183 self.send(json!({"jsonrpc": "2.0", "id": id, "result": result}));
184 }
185
186 /// Skip notifications and unrelated traffic until `matches` is satisfied.
187 fn wait_for(&mut self, what: &str, matches: impl Fn(&Value) -> bool) -> Value {
188 let deadline = Instant::now() + TIMEOUT;
189 loop {
190 let remaining = deadline.saturating_duration_since(Instant::now());
191 match self.incoming.recv_timeout(remaining) {
192 Ok(message) if matches(&message) => return message,
193 Ok(_) => continue,
194 Err(_) => panic!("timed out waiting for {what}"),
195 }
196 }
197 }
198
199 /// The response to one of *our* requests (has our id, no `method`).
200 fn wait_for_response(&mut self, id: u64) -> Value {
201 let response = self.wait_for(&format!("response to request {id}"), |message| {
202 message["id"] == id && message.get("method").is_none()
203 });
204 assert!(
205 response.get("error").is_none(),
206 "request {id} failed: {response}"
207 );
208 response
209 }
210
211 /// A request *from* the agent (has a `method` and its own id).
212 fn wait_for_agent_request(&mut self, method: &str) -> Value {
213 self.wait_for(&format!("agent request {method}"), |message| {
214 message["method"] == method && message.get("id").is_some()
215 })
216 }
217 }
218
219 impl Drop for AgentUnderTest {
220 fn drop(&mut self) {
221 let _ = self.child.kill();
222 let _ = self.child.wait();
223 }
224 }
225
226 // ── The round-trip ──────────────────────────────────────────────────────────
227
228 #[test]
229 fn permission_round_trip_cancel_then_allow() {
230 let endpoint = start_fake_endpoint(vec![
231 sse_tool_call("call_1", "run_command", r#"{"command":"echo sigit-first"}"#),
232 sse_tool_call(
233 "call_2",
234 "run_command",
235 r#"{"command":"echo sigit-approved"}"#,
236 ),
237 sse_text("done"),
238 ]);
239
240 let scratch = std::env::temp_dir().join(format!("sigit_acp_perm_{}", std::process::id()));
241 let config_dir = scratch.join("config");
242 let cwd = scratch.join("cwd");
243 std::fs::create_dir_all(&config_dir).unwrap();
244 std::fs::create_dir_all(&cwd).unwrap();
245
246 let mut agent = spawn_agent(endpoint.port, &config_dir);
247
248 let id = agent.request(
249 "initialize",
250 json!({"protocolVersion": 1, "clientCapabilities": {}}),
251 );
252 agent.wait_for_response(id);
253
254 let id = agent.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
255 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
256 .as_str()
257 .expect("session id")
258 .to_string();
259
260 // ── Turn 1: cancel at the permission gate ───────────────────────────
261 let prompt_id = agent.request(
262 "session/prompt",
263 json!({
264 "sessionId": session_id,
265 "prompt": [{"type": "text", "text": "run the first command"}],
266 }),
267 );
268
269 let permission = agent.wait_for_agent_request("session/request_permission");
270 let params = &permission["params"];
271 assert_eq!(params["sessionId"], session_id.as_str());
272 let title = params["toolCall"]["title"].as_str().expect("title");
273 assert!(
274 title.contains("run_command") && title.contains("echo sigit-first"),
275 "the dialog must show the tool and its arguments, got: {title}"
276 );
277 assert_eq!(
278 params["toolCall"]["rawInput"]["command"], "echo sigit-first",
279 "full arguments must travel as rawInput"
280 );
281 let option_ids: Vec<&str> = params["options"]
282 .as_array()
283 .expect("options")
284 .iter()
285 .map(|option| option["optionId"].as_str().unwrap_or_default())
286 .collect();
287 assert_eq!(option_ids, ["allow_once", "allow_session", "reject_once"]);
288
289 agent.respond(
290 permission["id"].clone(),
291 json!({"outcome": {"outcome": "cancelled"}}),
292 );
293
294 let response = agent.wait_for_response(prompt_id);
295 assert_eq!(response["result"]["stopReason"], "cancelled");
296
297 // ── Turn 2: history must be repaired; then approve once ─────────────
298 let prompt_id = agent.request(
299 "session/prompt",
300 json!({
301 "sessionId": session_id,
302 "prompt": [{"type": "text", "text": "run the second command"}],
303 }),
304 );
305
306 let permission = agent.wait_for_agent_request("session/request_permission");
307 agent.respond(
308 permission["id"].clone(),
309 json!({"outcome": {"outcome": "selected", "optionId": "allow_once"}}),
310 );
311
312 let response = agent.wait_for_response(prompt_id);
313 assert_eq!(response["result"]["stopReason"], "end_turn");
314
315 // ── What the endpoint saw ────────────────────────────────────────────
316 let requests = endpoint.requests.lock().unwrap();
317 assert_eq!(requests.len(), 3, "expected exactly three completions");
318
319 // Request 2 replays the full history: the cancelled round's tool call
320 // must be answered by a `role: "tool"` message, not left dangling.
321 let messages = requests[1]["messages"].as_array().expect("messages");
322 let call_position = messages
323 .iter()
324 .position(|message| message["tool_calls"][0]["id"] == "call_1")
325 .expect("cancelled turn's assistant tool call in replayed history");
326 let repair = &messages[call_position + 1];
327 assert_eq!(repair["role"], "tool", "dangling tool call not closed out");
328 assert_eq!(repair["tool_call_id"], "call_1");
329 assert!(
330 repair["content"]
331 .as_str()
332 .unwrap_or_default()
333 .contains("cancelled"),
334 "repair message should say the turn was cancelled: {repair}"
335 );
336
337 // Request 3 carries the approved call's real output.
338 let messages = requests[2]["messages"].as_array().expect("messages");
339 let result = messages
340 .iter()
341 .find(|message| message["role"] == "tool" && message["tool_call_id"] == "call_2")
342 .expect("tool result for the approved call");
343 assert!(
344 result["content"]
345 .as_str()
346 .unwrap_or_default()
347 .contains("sigit-approved"),
348 "the approved command's output should reach the endpoint: {result}"
349 );
350
351 drop(agent);
352 let _ = std::fs::remove_dir_all(&scratch);
353 }