@sigitcodereview / sigit
Code Commits Issues Pulls Cloud Agent CI/CD
feature/cargo-workspace
claude/github-issue-40-0au5ua claude/linked-issue-acp-tui-u4vvpa claude/minor-release-cut-vkugow claude/sigit-acp-local-chat-cx6380 claude/sigit-cloud-agent-expansion-reox0a claude/tool-permission-system claude/zen-feynman-0u78dk development feature/acp-cancel-request feature/acp-session-list-cursor feature/acp-v1-conformance feature/agent-tools-multiedit-glob-todos-remember feature/cargo-workspace feature/hooks feature/load-local-model-explicitly feature/sigit-agent-profile feature/sigit-code-cloud feature/xcode-mcp-server fix/acp-replay-loaded-session-history fix/acp-replay-reasoning-and-attachments fix/dsml-doubled-bar-markers main release/v1.5.10 release/v1.5.6 release/v1.6.1
rs 490 lines 17.7 KB
Raw
1 //! Two ACP sessions in one agent process, on an HTTP backend.
2 //!
3 //! An editor runs a single sigit process for every thread it has open. A turn
4 //! spends most of its time waiting on the endpoint, and nothing in that wait
5 //! needs the process's working directory or workspace roots, so a second
6 //! thread should be able to open, run a tool, and finish while the first is
7 //! still streaming. Each thread also has to find its own working directory
8 //! again when its next tool call runs.
9
10 use std::io::{BufRead, BufReader, Read, Write};
11 use std::net::{TcpListener, TcpStream};
12 use std::process::{Child, ChildStdin, Command, Stdio};
13 use std::sync::atomic::{AtomicBool, Ordering};
14 use std::sync::mpsc::{Receiver, channel};
15 use std::sync::{Arc, Mutex};
16 use std::time::{Duration, Instant};
17
18 use serde_json::{Value, json};
19
20 const TIMEOUT: Duration = Duration::from_secs(30);
21
22 /// The prompt text that makes the endpoint hold its reply open.
23 const HOLD: &str = "HOLD-THIS-TURN";
24
25 /// The prompt text that makes the model ask for a command instead of a read,
26 /// which is what the permission gate stops on.
27 const RUN: &str = "RUN-A-COMMAND";
28
29 fn sse(events: &[Value]) -> String {
30 events
31 .iter()
32 .map(|event| format!("data: {event}\n\n"))
33 .collect()
34 }
35
36 fn text_delta(text: &str) -> Value {
37 json!({"choices": [{"delta": {"content": text}}]})
38 }
39
40 fn finish(reason: &str) -> Value {
41 json!({"choices": [{"delta": {}, "finish_reason": reason}]})
42 }
43
44 fn read_marker_call() -> Value {
45 json!({
46 "choices": [{"delta": {"tool_calls": [{
47 "index": 0,
48 "id": "call_marker",
49 "function": {"name": "read_file", "arguments": "{\"path\":\"marker.txt\"}"},
50 }]}}]
51 })
52 }
53
54 /// Writes `ran.txt` into the working directory. Plain redirection, so the
55 /// same line works under `sh` and `cmd`.
56 fn run_command_call() -> Value {
57 json!({
58 "choices": [{"delta": {"tool_calls": [{
59 "index": 0,
60 "id": "call_run",
61 "function": {"name": "run_command", "arguments": "{\"command\":\"echo ran > ran.txt\"}"},
62 }]}}]
63 })
64 }
65
66 /// An endpoint that answers each connection on its own thread.
67 ///
68 /// A request that already carries a tool result gets a closing line of text.
69 /// Otherwise the model "asks" to read `marker.txt`, a relative path, so the
70 /// result shows which directory the agent resolved it against. A first request
71 /// mentioning [`HOLD`] streams one fragment and then waits for `release`
72 /// before asking for the file. One mentioning [`RUN`] asks for a command.
73 struct Endpoint {
74 port: u16,
75 requests: Arc<Mutex<Vec<Value>>>,
76 release: Arc<AtomicBool>,
77 }
78
79 fn read_request(stream: &TcpStream) -> Option<Value> {
80 let mut reader = BufReader::new(stream.try_clone().ok()?);
81 let mut content_length = 0usize;
82 loop {
83 let mut line = String::new();
84 if reader.read_line(&mut line).ok()? == 0 {
85 return None;
86 }
87 let line = line.trim();
88 if line.is_empty() {
89 break;
90 }
91 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
92 content_length = length.trim().parse().unwrap_or(0);
93 }
94 }
95 let mut body = vec![0u8; content_length];
96 reader.read_exact(&mut body).ok()?;
97 serde_json::from_slice(&body).ok()
98 }
99
100 fn mentions(request: &Value, needle: &str) -> bool {
101 request["messages"]
102 .as_array()
103 .is_some_and(|messages| messages.iter().any(|m| m.to_string().contains(needle)))
104 }
105
106 fn has_tool_result(request: &Value) -> bool {
107 request["messages"]
108 .as_array()
109 .is_some_and(|messages| messages.iter().any(|m| m["role"] == "tool"))
110 }
111
112 fn start_endpoint() -> Endpoint {
113 let listener = TcpListener::bind("127.0.0.1:0").expect("bind endpoint");
114 let port = listener.local_addr().unwrap().port();
115 let requests = Arc::new(Mutex::new(Vec::new()));
116 let release = Arc::new(AtomicBool::new(false));
117
118 let (seen, gate) = (Arc::clone(&requests), Arc::clone(&release));
119 std::thread::spawn(move || {
120 for stream in listener.incoming() {
121 let Ok(mut stream) = stream else { continue };
122 let (seen, gate) = (Arc::clone(&seen), Arc::clone(&gate));
123 std::thread::spawn(move || {
124 let Some(request) = read_request(&stream) else {
125 return;
126 };
127 seen.lock().unwrap().push(request.clone());
128
129 // No content-length: the body ends when the connection closes,
130 // which lets a fragment go out before the rest is decided.
131 let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\
132 connection: close\r\n\r\n";
133 if stream.write_all(head.as_bytes()).is_err() {
134 return;
135 }
136
137 let body = if has_tool_result(&request) {
138 sse(&[text_delta("done"), finish("stop")])
139 } else {
140 if mentions(&request, HOLD) {
141 let first = sse(&[text_delta("holding")]);
142 if stream.write_all(first.as_bytes()).is_err() || stream.flush().is_err() {
143 return;
144 }
145 let deadline = Instant::now() + TIMEOUT;
146 while !gate.load(Ordering::Acquire) && Instant::now() < deadline {
147 std::thread::sleep(Duration::from_millis(20));
148 }
149 }
150 let call = if mentions(&request, RUN) {
151 run_command_call()
152 } else {
153 read_marker_call()
154 };
155 sse(&[call, finish("tool_calls")])
156 };
157 let _ = stream.write_all(body.as_bytes());
158 let _ = stream.write_all(b"data: [DONE]\n\n");
159 let _ = stream.flush();
160 });
161 }
162 });
163
164 Endpoint {
165 port,
166 requests,
167 release,
168 }
169 }
170
171 struct AgentUnderTest {
172 child: Child,
173 stdin: ChildStdin,
174 incoming: Receiver<Value>,
175 next_id: u64,
176 /// Messages read while waiting for something else, oldest first.
177 seen: Vec<Value>,
178 }
179
180 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
181 spawn_agent_with_permissions(port, config_dir, "allow")
182 }
183
184 fn spawn_agent_with_permissions(
185 port: u16,
186 config_dir: &std::path::Path,
187 permissions: &str,
188 ) -> AgentUnderTest {
189 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
190 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
191 .env("OPENAI_API_KEY", "test-key")
192 .env("SIGIT_MODEL", "scripted-model")
193 .env("SIGIT_CONFIG_DIR", config_dir)
194 .env("SIGIT_MCP", "off")
195 .env("SIGIT_PERMISSIONS", permissions)
196 .env_remove("SIGIT_LOCAL_INFERENCE")
197 .env_remove("SIGIT_MAX_TOOL_ROUNDS")
198 .stdin(Stdio::piped())
199 .stdout(Stdio::piped())
200 .stderr(Stdio::null())
201 .spawn()
202 .expect("spawn sigit in ACP mode");
203
204 let stdout = child.stdout.take().unwrap();
205 let (message_tx, incoming) = channel();
206 std::thread::spawn(move || {
207 for line in BufReader::new(stdout).lines() {
208 let Ok(line) = line else { break };
209 if let Ok(message) = serde_json::from_str::<Value>(&line)
210 && message_tx.send(message).is_err()
211 {
212 break;
213 }
214 }
215 });
216
217 let stdin = child.stdin.take().unwrap();
218 AgentUnderTest {
219 child,
220 stdin,
221 incoming,
222 next_id: 0,
223 seen: Vec::new(),
224 }
225 }
226
227 impl AgentUnderTest {
228 fn request(&mut self, method: &str, params: Value) -> u64 {
229 self.next_id += 1;
230 let id = self.next_id;
231 let mut line =
232 json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
233 line.push('\n');
234 self.stdin.write_all(line.as_bytes()).expect("write stdin");
235 self.stdin.flush().expect("flush stdin");
236 id
237 }
238
239 /// Read messages until one satisfies `wanted`, keeping every message read
240 /// on the way in `seen`. Panics with `what` when none arrives in time.
241 fn wait_for(&mut self, what: &str, wanted: impl Fn(&Value) -> bool) -> Value {
242 let deadline = Instant::now() + TIMEOUT;
243 loop {
244 let remaining = deadline.saturating_duration_since(Instant::now());
245 let Ok(message) = self.incoming.recv_timeout(remaining) else {
246 panic!("timed out waiting for {what}");
247 };
248 self.seen.push(message.clone());
249 if wanted(&message) {
250 return message;
251 }
252 }
253 }
254
255 fn wait_for_response(&mut self, id: u64) -> Value {
256 let message = self.wait_for(&format!("the response to request {id}"), |message| {
257 message["id"] == id && message.get("method").is_none()
258 });
259 assert!(
260 message.get("error").is_none(),
261 "request {id} failed: {message}"
262 );
263 message
264 }
265
266 /// Answer a request the agent sent, such as a permission request.
267 fn respond(&mut self, id: &Value, result: Value) {
268 let mut line = json!({"jsonrpc": "2.0", "id": id, "result": result}).to_string();
269 line.push('\n');
270 self.stdin.write_all(line.as_bytes()).expect("write stdin");
271 self.stdin.flush().expect("flush stdin");
272 }
273
274 fn close_session(&mut self, session_id: &str) -> u64 {
275 self.request("session/close", json!({"sessionId": session_id}))
276 }
277
278 fn responded(&self, id: u64) -> bool {
279 self.seen
280 .iter()
281 .any(|message| message["id"] == id && message.get("method").is_none())
282 }
283
284 fn new_session(&mut self, cwd: &std::path::Path) -> String {
285 let id = self.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
286 self.wait_for_response(id)["result"]["sessionId"]
287 .as_str()
288 .expect("session id")
289 .to_string()
290 }
291
292 fn send_prompt(&mut self, session_id: &str, text: &str) -> u64 {
293 self.request(
294 "session/prompt",
295 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": text}]}),
296 )
297 }
298 }
299
300 impl Drop for AgentUnderTest {
301 fn drop(&mut self) {
302 let _ = self.child.kill();
303 let _ = self.child.wait();
304 }
305 }
306
307 /// The tool result a follow-up request carries.
308 fn tool_result(request: &Value) -> String {
309 request["messages"]
310 .as_array()
311 .into_iter()
312 .flatten()
313 .filter(|message| message["role"] == "tool")
314 .map(|message| message["content"].to_string())
315 .collect()
316 }
317
318 #[test]
319 fn a_second_session_opens_runs_and_finishes_while_the_first_is_streaming() {
320 let dir = std::env::temp_dir().join(format!("sigit_acp_concurrent_{}", std::process::id()));
321 let (work_a, work_b) = (dir.join("work-a"), dir.join("work-b"));
322 for (work, marker) in [(&work_a, "marker-from-a"), (&work_b, "marker-from-b")] {
323 std::fs::create_dir_all(work).unwrap();
324 std::fs::write(work.join("marker.txt"), marker).unwrap();
325 }
326 std::fs::create_dir_all(dir.join("config")).unwrap();
327
328 let endpoint = start_endpoint();
329 let mut agent = spawn_agent(endpoint.port, &dir.join("config"));
330 let id = agent.request(
331 "initialize",
332 json!({"protocolVersion": 1, "clientCapabilities": {}}),
333 );
334 agent.wait_for_response(id);
335
336 // Thread A starts a turn and the endpoint holds it open mid-stream.
337 let session_a = agent.new_session(&work_a);
338 let prompt_a = agent.send_prompt(&session_a, HOLD);
339 agent.wait_for("thread A's first streamed fragment", |message| {
340 let update = &message["params"]["update"];
341 message["params"]["sessionId"] == session_a.as_str()
342 && update["sessionUpdate"] == "agent_message_chunk"
343 && update["content"]["text"]
344 .as_str()
345 .is_some_and(|text| text.contains("holding"))
346 });
347
348 // Thread B opens and runs a whole turn, tool call included, meanwhile.
349 let session_b = agent.new_session(&work_b);
350 let prompt_b = agent.send_prompt(&session_b, "read the marker");
351 let response_b = agent.wait_for_response(prompt_b);
352 assert_eq!(response_b["result"]["stopReason"], "end_turn");
353 assert!(
354 !agent.responded(prompt_a),
355 "thread A finished before it was released, so it was never mid-turn"
356 );
357
358 // Thread A picks up where it was, in its own directory.
359 endpoint.release.store(true, Ordering::Release);
360 let response_a = agent.wait_for_response(prompt_a);
361 assert_eq!(response_a["result"]["stopReason"], "end_turn");
362
363 let requests = endpoint.requests.lock().unwrap().clone();
364 let follow_ups: Vec<&Value> = requests.iter().filter(|r| has_tool_result(r)).collect();
365 assert_eq!(
366 follow_ups.len(),
367 2,
368 "one tool round per thread: {requests:?}"
369 );
370 for request in follow_ups {
371 let result = tool_result(request);
372 if mentions(request, HOLD) {
373 assert!(result.contains("marker-from-a"), "thread A read {result}");
374 } else {
375 assert!(result.contains("marker-from-b"), "thread B read {result}");
376 }
377 }
378 // Neither thread's conversation reached the other's requests.
379 for request in requests.iter().filter(|r| !mentions(r, HOLD)) {
380 assert!(!mentions(request, "holding"), "{request}");
381 }
382
383 std::fs::remove_dir_all(&dir).ok();
384 }
385
386 fn initialized_agent(port: u16, config_dir: &std::path::Path, permissions: &str) -> AgentUnderTest {
387 std::fs::create_dir_all(config_dir).unwrap();
388 let mut agent = spawn_agent_with_permissions(port, config_dir, permissions);
389 let id = agent.request(
390 "initialize",
391 json!({"protocolVersion": 1, "clientCapabilities": {}}),
392 );
393 agent.wait_for_response(id);
394 agent
395 }
396
397 #[test]
398 fn closing_a_session_mid_stream_ends_its_turn_and_runs_no_tool() {
399 let dir = std::env::temp_dir().join(format!("sigit_acp_close_stream_{}", std::process::id()));
400 let (work_a, work_b) = (dir.join("work-a"), dir.join("work-b"));
401 for (work, marker) in [(&work_a, "marker-from-a"), (&work_b, "marker-from-b")] {
402 std::fs::create_dir_all(work).unwrap();
403 std::fs::write(work.join("marker.txt"), marker).unwrap();
404 }
405
406 let endpoint = start_endpoint();
407 let mut agent = initialized_agent(endpoint.port, &dir.join("config"), "allow");
408
409 // Thread B exists first, so thread A is the installed one when it closes.
410 let session_b = agent.new_session(&work_b);
411 let session_a = agent.new_session(&work_a);
412 let prompt_a = agent.send_prompt(&session_a, HOLD);
413 agent.wait_for("thread A's first streamed fragment", |message| {
414 message["params"]["sessionId"] == session_a.as_str()
415 && message["params"]["update"]["sessionUpdate"] == "agent_message_chunk"
416 });
417
418 // The editor closes the thread while its turn is waiting on the endpoint.
419 let close_a = agent.close_session(&session_a);
420 let response_a = agent.wait_for_response(prompt_a);
421 assert_eq!(response_a["result"]["stopReason"], "cancelled");
422 agent.wait_for_response(close_a);
423
424 // The endpoint now asks for the file. Nobody is left to read it.
425 endpoint.release.store(true, Ordering::Release);
426
427 // The id is gone, and a request for it gets an answer instead of waiting
428 // behind anything the closed turn left held.
429 let late = agent.send_prompt(&session_a, "anyone there");
430 let refused = agent.wait_for("the answer to a prompt on the closed thread", |message| {
431 message["id"] == late && message.get("method").is_none()
432 });
433 assert!(refused.get("error").is_some(), "{refused}");
434
435 // Thread B still runs its tool in its own directory.
436 let prompt_b = agent.send_prompt(&session_b, "read the marker");
437 let response_b = agent.wait_for_response(prompt_b);
438 assert_eq!(response_b["result"]["stopReason"], "end_turn");
439
440 let requests = endpoint.requests.lock().unwrap().clone();
441 let follow_ups: Vec<&Value> = requests.iter().filter(|r| has_tool_result(r)).collect();
442 assert_eq!(
443 follow_ups.len(),
444 1,
445 "only thread B ran a tool: {requests:?}"
446 );
447 let result = tool_result(follow_ups[0]);
448 assert!(result.contains("marker-from-b"), "thread B read {result}");
449
450 std::fs::remove_dir_all(&dir).ok();
451 }
452
453 #[test]
454 fn closing_a_session_at_a_permission_prompt_ends_its_turn_without_an_answer() {
455 let dir = std::env::temp_dir().join(format!("sigit_acp_close_ask_{}", std::process::id()));
456 let work = dir.join("work");
457 std::fs::create_dir_all(&work).unwrap();
458
459 let endpoint = start_endpoint();
460 let mut agent = initialized_agent(endpoint.port, &dir.join("config"), "ask");
461
462 let session = agent.new_session(&work);
463 let prompt = agent.send_prompt(&session, RUN);
464 let permission = agent.wait_for("the permission request", |message| {
465 message["method"] == "session/request_permission"
466 });
467
468 // The client closes the thread and never answers the request. The turn
469 // must not keep the close waiting on an answer that is not coming.
470 let close = agent.close_session(&session);
471 let response = agent.wait_for_response(prompt);
472 assert_eq!(response["result"]["stopReason"], "cancelled");
473 agent.wait_for_response(close);
474
475 // An approval that arrives after the close runs nothing.
476 agent.respond(
477 &permission["id"],
478 json!({"outcome": {"outcome": "selected", "optionId": "allow_once"}}),
479 );
480 let late = agent.send_prompt(&session, "anyone there");
481 agent.wait_for("the answer to a prompt on the closed thread", |message| {
482 message["id"] == late && message.get("method").is_none()
483 });
484 assert!(
485 !work.join("ran.txt").exists(),
486 "the command ran after the close"
487 );
488
489 std::fs::remove_dir_all(&dir).ok();
490 }