@sigitcodereview / sigit
Code Commits Issues Pulls Cloud Agent CI/CD
feature/acp-v1-conformance
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 277 lines 9.12 KB
Raw
1 //! Regression: an ACP tool call carries the programmatic tool name.
2 //!
3 //! A `tool_call` has always carried a human title; `name` is what lets a
4 //! client tell `run_command` from `read_file` without parsing it. The protocol
5 //! puts the name on the first report of a call and forbids it changing
6 //! afterwards, so it goes on `ToolCall` and never on a later
7 //! `ToolCallUpdate`. `session/load` replays the call, so the replay needs it
8 //! too (issue #196).
9 //!
10 //! This drives the real binary against a scripted OpenAI-compatible endpoint.
11
12 use std::collections::VecDeque;
13 use std::io::{BufRead, BufReader, Read, Write};
14 use std::net::TcpListener;
15 use std::process::{Child, ChildStdin, Command, Stdio};
16 use std::sync::Mutex;
17 use std::sync::mpsc::{Receiver, channel};
18 use std::time::{Duration, Instant};
19
20 use serde_json::{Value, json};
21
22 const TIMEOUT: Duration = Duration::from_secs(30);
23
24 fn sse_body(chunks: &[Value]) -> String {
25 let mut body = String::new();
26 for chunk in chunks {
27 body.push_str(&format!("data: {chunk}\n\n"));
28 }
29 body.push_str("data: [DONE]\n\n");
30 body
31 }
32
33 fn sse_tool_call(id: &str, name: &str, arguments: &str) -> String {
34 sse_body(&[json!({
35 "choices": [{"delta": {"tool_calls": [{
36 "index": 0,
37 "id": id,
38 "function": {"name": name, "arguments": arguments},
39 }]}}]
40 })])
41 }
42
43 fn sse_text(text: &str) -> String {
44 sse_body(&[json!({"choices": [{"delta": {"content": text}}]})])
45 }
46
47 struct FakeEndpoint {
48 port: u16,
49 }
50
51 fn start_fake_endpoint(responses: Vec<String>) -> FakeEndpoint {
52 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
53 let port = listener.local_addr().unwrap().port();
54 let queue = Mutex::new(VecDeque::from(responses));
55
56 std::thread::spawn(move || {
57 for stream in listener.incoming() {
58 let Ok(mut stream) = stream else { continue };
59 let mut reader = BufReader::new(match stream.try_clone() {
60 Ok(clone) => clone,
61 Err(_) => continue,
62 });
63 let mut content_length = 0usize;
64 loop {
65 let mut line = String::new();
66 if reader.read_line(&mut line).unwrap_or(0) == 0 {
67 break;
68 }
69 let line = line.trim();
70 if line.is_empty() {
71 break;
72 }
73 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
74 content_length = length.trim().parse().unwrap_or(0);
75 }
76 }
77 let mut body = vec![0u8; content_length];
78 if reader.read_exact(&mut body).is_err() {
79 continue;
80 }
81 let payload = queue
82 .lock()
83 .unwrap()
84 .pop_front()
85 .unwrap_or_else(|| "data: [DONE]\n\n".to_string());
86 let response = format!(
87 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\
88 content-length: {}\r\nconnection: close\r\n\r\n{}",
89 payload.len(),
90 payload
91 );
92 let _ = stream.write_all(response.as_bytes());
93 let _ = stream.flush();
94 }
95 });
96
97 FakeEndpoint { port }
98 }
99
100 struct AgentUnderTest {
101 child: Child,
102 stdin: ChildStdin,
103 incoming: Receiver<Value>,
104 next_id: u64,
105 }
106
107 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
108 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
109 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
110 .env("OPENAI_API_KEY", "test-key")
111 .env("SIGIT_MODEL", "scripted-model")
112 .env("SIGIT_CONFIG_DIR", config_dir)
113 .env("SIGIT_MCP", "off")
114 .env("SIGIT_PERMISSIONS", "allow")
115 .env_remove("SIGIT_LOCAL_INFERENCE")
116 .stdin(Stdio::piped())
117 .stdout(Stdio::piped())
118 .stderr(Stdio::null())
119 .spawn()
120 .expect("spawn sigit in ACP mode");
121
122 let stdout = child.stdout.take().unwrap();
123 let (message_tx, incoming) = channel();
124 std::thread::spawn(move || {
125 for line in BufReader::new(stdout).lines() {
126 let Ok(line) = line else { break };
127 if let Ok(message) = serde_json::from_str::<Value>(&line)
128 && message_tx.send(message).is_err()
129 {
130 break;
131 }
132 }
133 });
134
135 let stdin = child.stdin.take().unwrap();
136 AgentUnderTest {
137 child,
138 stdin,
139 incoming,
140 next_id: 0,
141 }
142 }
143
144 impl AgentUnderTest {
145 fn request(&mut self, method: &str, params: Value) -> u64 {
146 self.next_id += 1;
147 let id = self.next_id;
148 let mut line =
149 json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
150 line.push('\n');
151 self.stdin.write_all(line.as_bytes()).expect("write stdin");
152 self.stdin.flush().expect("flush stdin");
153 id
154 }
155
156 /// The response to `id`, plus every `session/update` payload that arrived
157 /// before it — which for a prompt and for `session/load` is the whole
158 /// stream the client would render.
159 fn wait_for_response_with_updates(&mut self, id: u64) -> (Value, Vec<Value>) {
160 let mut updates = Vec::new();
161 let deadline = Instant::now() + TIMEOUT;
162 loop {
163 let remaining = deadline.saturating_duration_since(Instant::now());
164 let Ok(message) = self.incoming.recv_timeout(remaining) else {
165 panic!("timed out waiting for the response to request {id}");
166 };
167 if message["method"] == "session/update" {
168 updates.push(message["params"]["update"].clone());
169 }
170 if message["id"] == id && message.get("method").is_none() {
171 assert!(
172 message.get("error").is_none(),
173 "request {id} failed: {message}"
174 );
175 return (message, updates);
176 }
177 }
178 }
179 }
180
181 impl Drop for AgentUnderTest {
182 fn drop(&mut self) {
183 let _ = self.child.kill();
184 let _ = self.child.wait();
185 }
186 }
187
188 fn tool_call(updates: &[Value], id: &str) -> Value {
189 updates
190 .iter()
191 .find(|update| update["sessionUpdate"] == "tool_call" && update["toolCallId"] == id)
192 .unwrap_or_else(|| panic!("no tool_call for {id} in {updates:#?}"))
193 .clone()
194 }
195
196 #[test]
197 fn a_tool_call_carries_its_name_live_and_on_replay() {
198 let scratch = std::env::temp_dir().join(format!("sigit_acp_tool_name_{}", std::process::id()));
199 let config_dir = scratch.join("config");
200 let work = scratch.join("work");
201 let _ = std::fs::remove_dir_all(&scratch);
202 std::fs::create_dir_all(&config_dir).unwrap();
203 std::fs::create_dir_all(&work).unwrap();
204 let notes = work.join("notes.txt");
205 std::fs::write(&notes, "hello from the notes\n").unwrap();
206
207 let endpoint = start_fake_endpoint(vec![
208 sse_tool_call("call_1", "read_file", &json!({"path": notes}).to_string()),
209 sse_text("read it"),
210 ]);
211
212 let mut agent = spawn_agent(endpoint.port, &config_dir);
213
214 let id = agent.request(
215 "initialize",
216 json!({"protocolVersion": 1, "clientCapabilities": {}}),
217 );
218 agent.wait_for_response_with_updates(id);
219
220 let id = agent.request("session/new", json!({"cwd": work, "mcpServers": []}));
221 let (response, updates) = agent.wait_for_response_with_updates(id);
222 assert!(
223 !updates
224 .iter()
225 .any(|update| update["sessionUpdate"] == "tool_call"),
226 "session/new reports no tool call: {updates:#?}"
227 );
228 let session_id = response["result"]["sessionId"]
229 .as_str()
230 .expect("session id")
231 .to_string();
232
233 let id = agent.request(
234 "session/prompt",
235 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": "what do the notes say?"}]}),
236 );
237 let (_, live) = agent.wait_for_response_with_updates(id);
238
239 let first = tool_call(&live, "call_1");
240 assert_eq!(
241 first["name"], "read_file",
242 "the first report of a call must carry its tool name: {first}"
243 );
244
245 let later: Vec<&Value> = live
246 .iter()
247 .filter(|update| {
248 update["sessionUpdate"] == "tool_call_update" && update["toolCallId"] == "call_1"
249 })
250 .collect();
251 assert!(
252 !later.is_empty(),
253 "a finished call should have been updated at least once: {live:#?}"
254 );
255 for update in later {
256 assert!(
257 update.get("name").is_none(),
258 "the name belongs to the first report only: {update}"
259 );
260 }
261
262 // A thread reopened from the session store replays the call, so the replay
263 // has to carry the name too.
264 let id = agent.request(
265 "session/load",
266 json!({"sessionId": session_id, "cwd": work, "mcpServers": []}),
267 );
268 let (_, replay) = agent.wait_for_response_with_updates(id);
269 let replayed = tool_call(&replay, "call_1");
270 assert_eq!(
271 replayed["name"], "read_file",
272 "a replayed tool call must carry its tool name: {replayed}"
273 );
274
275 drop(agent);
276 let _ = std::fs::remove_dir_all(&scratch);
277 }