@sigitcodereview / sigit
Code Commits Issues Pulls Cloud Agent CI/CD
fix/dsml-doubled-bar-markers
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 280 lines 9.47 KB
Raw
1 //! The JSON-RPC error codes the agent answers with.
2 //!
3 //! ACP uses the standard codes: `-32602` for a request whose parameters are
4 //! wrong (an auth method that was never offered, a relative `cwd`), `-32002`
5 //! for a session id the agent does not know. `-32000` is reserved for
6 //! "authentication required", so a sign-in that fails is `-32603`.
7
8 use std::collections::VecDeque;
9 use std::io::{BufRead, BufReader, Read, Write};
10 use std::net::TcpListener;
11 use std::process::{Child, ChildStdin, Command, Stdio};
12 use std::sync::mpsc::{Receiver, channel};
13 use std::sync::{Arc, Mutex};
14 use std::time::{Duration, Instant};
15
16 use serde_json::{Value, json};
17
18 const TIMEOUT: Duration = Duration::from_secs(60);
19
20 fn sse_body(events: &[Value]) -> String {
21 let mut body = String::new();
22 for event in events {
23 body.push_str("data: ");
24 body.push_str(&event.to_string());
25 body.push_str("\n\n");
26 }
27 body.push_str("data: [DONE]\n\n");
28 body
29 }
30
31 fn sse_text(text: &str) -> String {
32 sse_body(&[json!({"choices": [{"delta": {"content": text}}]})])
33 }
34
35 /// Serves one scripted SSE response per request and records each request body.
36 // Shared harness; this file never reads the recorded requests.
37 #[allow(dead_code)]
38 struct FakeEndpoint {
39 port: u16,
40 requests: Arc<Mutex<Vec<Value>>>,
41 }
42
43 fn start_fake_endpoint(responses: Vec<String>) -> FakeEndpoint {
44 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
45 let port = listener.local_addr().unwrap().port();
46 let requests: Arc<Mutex<Vec<Value>>> = Arc::default();
47 let recorded = Arc::clone(&requests);
48 let queue = Mutex::new(VecDeque::from(responses));
49
50 std::thread::spawn(move || {
51 for stream in listener.incoming() {
52 let Ok(mut stream) = stream else { continue };
53 let mut reader = BufReader::new(match stream.try_clone() {
54 Ok(clone) => clone,
55 Err(_) => continue,
56 });
57 let mut content_length = 0usize;
58 loop {
59 let mut line = String::new();
60 if reader.read_line(&mut line).unwrap_or(0) == 0 {
61 break;
62 }
63 let line = line.trim();
64 if line.is_empty() {
65 break;
66 }
67 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
68 content_length = length.trim().parse().unwrap_or(0);
69 }
70 }
71 let mut body = vec![0u8; content_length];
72 if reader.read_exact(&mut body).is_err() {
73 continue;
74 }
75 if let Ok(request) = serde_json::from_slice::<Value>(&body) {
76 recorded.lock().unwrap().push(request);
77 }
78 let payload = queue
79 .lock()
80 .unwrap()
81 .pop_front()
82 .unwrap_or_else(|| sse_text("out of scripted responses"));
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 }
91 });
92
93 FakeEndpoint { port, requests }
94 }
95
96 // ── ACP client over the binary's stdio ──────────────────────────────────────
97
98 struct AgentUnderTest {
99 child: Child,
100 stdin: ChildStdin,
101 incoming: Receiver<Value>,
102 next_id: u64,
103 }
104
105 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
106 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
107 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
108 .env("OPENAI_API_KEY", "test-key")
109 .env("SIGIT_MODEL", "scripted-model")
110 .env("SIGIT_CONFIG_DIR", config_dir)
111 .env("SIGIT_MCP", "off")
112 .env("SIGIT_PERMISSIONS", "allow")
113 .env_remove("SIGIT_LOCAL_INFERENCE")
114 .stdin(Stdio::piped())
115 .stdout(Stdio::piped())
116 .stderr(Stdio::null())
117 .spawn()
118 .expect("spawn sigit in ACP mode");
119
120 let stdout = child.stdout.take().unwrap();
121 let (message_tx, incoming) = channel();
122 std::thread::spawn(move || {
123 for line in BufReader::new(stdout).lines() {
124 let Ok(line) = line else { break };
125 if let Ok(message) = serde_json::from_str::<Value>(&line)
126 && message_tx.send(message).is_err()
127 {
128 break;
129 }
130 }
131 });
132
133 let stdin = child.stdin.take().unwrap();
134 AgentUnderTest {
135 child,
136 stdin,
137 incoming,
138 next_id: 0,
139 }
140 }
141
142 impl AgentUnderTest {
143 fn send(&mut self, message: Value) {
144 let mut line = message.to_string();
145 line.push('\n');
146 self.stdin
147 .write_all(line.as_bytes())
148 .expect("write to agent stdin");
149 self.stdin.flush().expect("flush agent stdin");
150 }
151
152 fn request(&mut self, method: &str, params: Value) -> u64 {
153 self.next_id += 1;
154 let id = self.next_id;
155 self.send(json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}));
156 id
157 }
158
159 /// Wait for the response to our request `id`, collecting the raw JSON of
160 /// every `session/update` notification that arrives before it.
161 fn wait_for_response_collecting_updates(&mut self, id: u64) -> (Value, String) {
162 let deadline = Instant::now() + TIMEOUT;
163 let mut updates = String::new();
164 loop {
165 let remaining = deadline.saturating_duration_since(Instant::now());
166 match self.incoming.recv_timeout(remaining) {
167 Ok(message) if message["id"] == id && message.get("method").is_none() => {
168 assert!(
169 message.get("error").is_none(),
170 "request {id} failed: {message}"
171 );
172 return (message, updates);
173 }
174 Ok(message) => {
175 if message["method"] == "session/update" {
176 updates.push_str(&message["params"].to_string());
177 updates.push('\n');
178 }
179 }
180 Err(_) => panic!("timed out waiting for response to request {id}"),
181 }
182 }
183 }
184
185 fn wait_for_response(&mut self, id: u64) -> Value {
186 self.wait_for_response_collecting_updates(id).0
187 }
188 }
189
190 impl Drop for AgentUnderTest {
191 fn drop(&mut self) {
192 let _ = self.child.kill();
193 let _ = self.child.wait();
194 }
195 }
196
197 impl AgentUnderTest {
198 /// The raw answer to request `id`, error or not.
199 fn answer(&mut self, id: u64) -> Value {
200 let deadline = Instant::now() + TIMEOUT;
201 loop {
202 let remaining = deadline.saturating_duration_since(Instant::now());
203 match self.incoming.recv_timeout(remaining) {
204 Ok(message) if message["id"] == id && message.get("method").is_none() => {
205 return message;
206 }
207 Ok(_) => {}
208 Err(_) => panic!("timed out waiting for the answer to request {id}"),
209 }
210 }
211 }
212 }
213
214 fn open() -> (AgentUnderTest, std::path::PathBuf, std::path::PathBuf) {
215 let endpoint = start_fake_endpoint(vec![]);
216 let scratch = std::env::temp_dir().join(format!(
217 "sigit_acp_codes_{}_{}",
218 std::process::id(),
219 endpoint.port
220 ));
221 let config_dir = scratch.join("config");
222 std::fs::create_dir_all(&config_dir).unwrap();
223 let mut agent = spawn_agent(endpoint.port, &config_dir);
224 let id = agent.request(
225 "initialize",
226 json!({"protocolVersion": 1, "clientCapabilities": {}}),
227 );
228 agent.wait_for_response(id);
229 (agent, scratch, config_dir)
230 }
231
232 #[test]
233 fn an_auth_method_that_was_never_offered_is_invalid_params() {
234 let (mut agent, scratch, _) = open();
235 let id = agent.request("authenticate", json!({"methodId": "not-a-method"}));
236 let answer = agent.answer(id);
237 assert_eq!(answer["error"]["code"], -32602, "{answer}");
238 drop(agent);
239 let _ = std::fs::remove_dir_all(scratch);
240 }
241
242 #[test]
243 fn a_session_the_agent_does_not_know_is_resource_not_found() {
244 let (mut agent, scratch, _) = open();
245 for (method, params) in [
246 (
247 "session/prompt",
248 json!({"sessionId": "nope", "prompt": [{"type": "text", "text": "hi"}]}),
249 ),
250 (
251 "session/set_mode",
252 json!({"sessionId": "nope", "modeId": "permission-mode-auto"}),
253 ),
254 (
255 "session/set_config_option",
256 json!({"sessionId": "nope", "configId": "sigit-permission-mode",
257 "value": "permission-mode-auto"}),
258 ),
259 ] {
260 let id = agent.request(method, params);
261 let answer = agent.answer(id);
262 assert_eq!(answer["error"]["code"], -32002, "{method}: {answer}");
263 }
264 drop(agent);
265 let _ = std::fs::remove_dir_all(scratch);
266 }
267
268 #[test]
269 fn a_relative_cwd_filter_on_session_list_is_invalid_params() {
270 let (mut agent, scratch, _) = open();
271 let id = agent.request("session/list", json!({"cwd": "relative/dir"}));
272 let answer = agent.answer(id);
273 assert_eq!(answer["error"]["code"], -32602, "{answer}");
274
275 let id = agent.request("session/list", json!({}));
276 let answer = agent.answer(id);
277 assert!(answer.get("error").is_none(), "{answer}");
278 drop(agent);
279 let _ = std::fs::remove_dir_all(scratch);
280 }