@sigitcodereview / sigit
Code Commits Issues Pulls Cloud Agent CI/CD
main
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 235 lines 7.64 KB
Raw
1 //! Regression: a tool's child process must never share the agent's stdin.
2 //!
3 //! In ACP mode stdin is the JSON-RPC pipe from the editor. `run_command` spawns
4 //! through the platform shell, and a child that inherits that pipe can read the
5 //! client's next request out of it. The request is then gone: the agent never
6 //! sees it, the editor waits for a reply that cannot come, and the session is
7 //! wedged for good — the symptom filed as "acp: model hangs".
8
9 use std::collections::VecDeque;
10 use std::io::{BufRead, BufReader, Read, Write};
11 use std::net::TcpListener;
12 use std::process::{Child, ChildStdin, Command, Stdio};
13 use std::sync::Mutex;
14 use std::sync::mpsc::{Receiver, channel};
15 use std::time::{Duration, Instant};
16
17 use serde_json::{Value, json};
18
19 const TIMEOUT: Duration = Duration::from_secs(30);
20
21 fn sse_body(chunks: &[Value]) -> String {
22 let mut body = String::new();
23 for chunk in chunks {
24 body.push_str(&format!("data: {chunk}\n\n"));
25 }
26 body.push_str("data: [DONE]\n\n");
27 body
28 }
29
30 fn sse_tool_call(id: &str, name: &str, arguments: &str) -> String {
31 sse_body(&[json!({
32 "choices": [{"delta": {"tool_calls": [{
33 "index": 0,
34 "id": id,
35 "function": {"name": name, "arguments": arguments},
36 }]}}]
37 })])
38 }
39
40 fn sse_text(text: &str) -> String {
41 sse_body(&[json!({"choices": [{"delta": {"content": text}}]})])
42 }
43
44 struct FakeEndpoint {
45 port: u16,
46 }
47
48 fn start_fake_endpoint(responses: Vec<String>) -> FakeEndpoint {
49 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
50 let port = listener.local_addr().unwrap().port();
51 let queue = Mutex::new(VecDeque::from(responses));
52
53 std::thread::spawn(move || {
54 for stream in listener.incoming() {
55 let Ok(mut stream) = stream else { continue };
56 let mut reader = BufReader::new(match stream.try_clone() {
57 Ok(clone) => clone,
58 Err(_) => continue,
59 });
60 let mut content_length = 0usize;
61 loop {
62 let mut line = String::new();
63 if reader.read_line(&mut line).unwrap_or(0) == 0 {
64 break;
65 }
66 let line = line.trim();
67 if line.is_empty() {
68 break;
69 }
70 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
71 content_length = length.trim().parse().unwrap_or(0);
72 }
73 }
74 let mut body = vec![0u8; content_length];
75 if reader.read_exact(&mut body).is_err() {
76 continue;
77 }
78 let payload = queue
79 .lock()
80 .unwrap()
81 .pop_front()
82 .unwrap_or_else(|| "data: [DONE]\n\n".to_string());
83 let response = format!(
84 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\
85 content-length: {}\r\nconnection: close\r\n\r\n{}",
86 payload.len(),
87 payload
88 );
89 let _ = stream.write_all(response.as_bytes());
90 let _ = stream.flush();
91 }
92 });
93
94 FakeEndpoint { port }
95 }
96
97 struct AgentUnderTest {
98 child: Child,
99 stdin: ChildStdin,
100 incoming: Receiver<Value>,
101 next_id: u64,
102 }
103
104 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
105 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
106 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
107 .env("OPENAI_API_KEY", "test-key")
108 .env("SIGIT_MODEL", "scripted-model")
109 .env("SIGIT_CONFIG_DIR", config_dir)
110 .env("SIGIT_MCP", "off")
111 .env("SIGIT_PERMISSIONS", "allow")
112 .env_remove("SIGIT_LOCAL_INFERENCE")
113 .stdin(Stdio::piped())
114 .stdout(Stdio::piped())
115 .stderr(Stdio::null())
116 .spawn()
117 .expect("spawn sigit in ACP mode");
118
119 let stdout = child.stdout.take().unwrap();
120 let (message_tx, incoming) = channel();
121 std::thread::spawn(move || {
122 for line in BufReader::new(stdout).lines() {
123 let Ok(line) = line else { break };
124 if let Ok(message) = serde_json::from_str::<Value>(&line)
125 && message_tx.send(message).is_err()
126 {
127 break;
128 }
129 }
130 });
131
132 let stdin = child.stdin.take().unwrap();
133 AgentUnderTest {
134 child,
135 stdin,
136 incoming,
137 next_id: 0,
138 }
139 }
140
141 impl AgentUnderTest {
142 fn request(&mut self, method: &str, params: Value) -> u64 {
143 self.next_id += 1;
144 let id = self.next_id;
145 let mut line =
146 json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
147 line.push('\n');
148 self.stdin.write_all(line.as_bytes()).expect("write stdin");
149 self.stdin.flush().expect("flush stdin");
150 id
151 }
152
153 fn wait_for_response(&mut self, id: u64) -> Value {
154 let deadline = Instant::now() + TIMEOUT;
155 loop {
156 let remaining = deadline.saturating_duration_since(Instant::now());
157 let Ok(message) = self.incoming.recv_timeout(remaining) else {
158 panic!("timed out waiting for the response to request {id}");
159 };
160 if message["id"] == id && message.get("method").is_none() {
161 assert!(
162 message.get("error").is_none(),
163 "request {id} failed: {message}"
164 );
165 return message;
166 }
167 }
168 }
169 }
170
171 impl Drop for AgentUnderTest {
172 fn drop(&mut self) {
173 let _ = self.child.kill();
174 let _ = self.child.wait();
175 }
176 }
177
178 #[test]
179 fn a_command_that_reads_stdin_cannot_eat_the_clients_next_request() {
180 let scratch = std::env::temp_dir().join(format!("sigit_acp_stdin_{}", std::process::id()));
181 let config_dir = scratch.join("config");
182 let work = scratch.join("work");
183 std::fs::create_dir_all(&config_dir).unwrap();
184 std::fs::create_dir_all(&work).unwrap();
185 let stolen = work.join("stolen.txt");
186
187 // A command that drains stdin to a file. With stdin inherited it reads the
188 // JSON-RPC pipe; with stdin closed it gets EOF and exits immediately.
189 let command = format!("cat > {}", stolen.display());
190 let endpoint = start_fake_endpoint(vec![
191 sse_tool_call(
192 "call_1",
193 "run_command",
194 &json!({"command": command, "cwd": work.to_str().unwrap()}).to_string(),
195 ),
196 sse_text("done"),
197 sse_text("still here"),
198 ]);
199
200 let mut agent = spawn_agent(endpoint.port, &config_dir);
201
202 let id = agent.request(
203 "initialize",
204 json!({"protocolVersion": 1, "clientCapabilities": {}}),
205 );
206 agent.wait_for_response(id);
207
208 let id = agent.request("session/new", json!({"cwd": work, "mcpServers": []}));
209 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
210 .as_str()
211 .expect("session id")
212 .to_string();
213
214 let first = agent.request(
215 "session/prompt",
216 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": "commit it"}]}),
217 );
218 agent.wait_for_response(first);
219
220 // The client's next request must reach the agent.
221 let second = agent.request(
222 "session/prompt",
223 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": "and again"}]}),
224 );
225 agent.wait_for_response(second);
226
227 let swallowed = std::fs::read_to_string(&stolen).unwrap_or_default();
228 assert!(
229 swallowed.is_empty(),
230 "the tool's child read {} bytes off the agent's stdin: {swallowed}",
231 swallowed.len()
232 );
233
234 std::fs::remove_dir_all(&scratch).ok();
235 }