@sigitcodereview / sigit
Code Commits Issues Pulls Cloud Agent CI/CD
release/v1.5.10
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 356 lines 11.8 KB
Raw
1 //! What the editor shows when the endpoint refuses a turn.
2 //!
3 //! Every endpoint siGit talks to reports failures in the OpenAI error envelope,
4 //! and siGit Code Cloud writes those messages for the person reading them
5 //! ("Monthly siGit Code Cloud allowance reached…"). The agent used to hand the
6 //! raw status line and JSON body to the client, so Zed's error banner showed
7 //! the envelope rather than the sentence inside it.
8 //!
9 //! Two shapes to cover: a failing HTTP status, and an endpoint that only
10 //! discovers the problem after the status line is out and reports it as a
11 //! `data:` frame in the stream.
12
13 use std::collections::VecDeque;
14 use std::io::{BufRead, BufReader, Read, Write};
15 use std::net::TcpListener;
16 use std::process::{Child, ChildStdin, Command, Stdio};
17 use std::sync::Mutex;
18 use std::sync::mpsc::{Receiver, channel};
19 use std::time::{Duration, Instant};
20
21 use serde_json::{Value, json};
22
23 const TIMEOUT: Duration = Duration::from_secs(30);
24
25 const ALLOWANCE_MESSAGE: &str = "Monthly siGit Code Cloud allowance reached. It resets at the start of your next billing period.";
26
27 /// One scripted reply per request: an HTTP status plus a body.
28 struct Reply {
29 status: &'static str,
30 content_type: &'static str,
31 body: String,
32 }
33
34 fn error_status_reply() -> Reply {
35 Reply {
36 status: "429 Too Many Requests",
37 content_type: "application/json",
38 body: json!({"error": {"message": ALLOWANCE_MESSAGE, "type": "server_error"}}).to_string(),
39 }
40 }
41
42 /// A 200 that turns into an error partway through, the way an endpoint has to
43 /// report a failure it only sees once the response is already open.
44 fn error_in_stream_reply() -> Reply {
45 let frame = json!({"error": {"message": ALLOWANCE_MESSAGE, "type": "server_error"}});
46 Reply {
47 status: "200 OK",
48 content_type: "text/event-stream",
49 body: format!("data: {frame}\n\n"),
50 }
51 }
52
53 fn sse_tool_call(id: &str, name: &str, arguments: &str) -> Reply {
54 Reply {
55 status: "200 OK",
56 content_type: "text/event-stream",
57 body: format!(
58 "data: {}\n\ndata: [DONE]\n\n",
59 json!({
60 "choices": [{"delta": {"tool_calls": [{
61 "index": 0,
62 "id": id,
63 "function": {"name": name, "arguments": arguments},
64 }]}}]
65 })
66 ),
67 }
68 }
69
70 fn upstream_500_reply() -> Reply {
71 Reply {
72 status: "500 Internal Server Error",
73 content_type: "application/json",
74 body: json!({
75 "error": {"message": "Onde Cloud upstream error (500)", "type": "server_error"}
76 })
77 .to_string(),
78 }
79 }
80
81 fn start_fake_endpoint(replies: Vec<Reply>) -> u16 {
82 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
83 let port = listener.local_addr().unwrap().port();
84 let queue = Mutex::new(VecDeque::from(replies));
85
86 std::thread::spawn(move || {
87 for stream in listener.incoming() {
88 let Ok(mut stream) = stream else { continue };
89 let mut reader = BufReader::new(match stream.try_clone() {
90 Ok(clone) => clone,
91 Err(_) => continue,
92 });
93 let mut content_length = 0usize;
94 loop {
95 let mut line = String::new();
96 if reader.read_line(&mut line).unwrap_or(0) == 0 {
97 break;
98 }
99 let line = line.trim();
100 if line.is_empty() {
101 break;
102 }
103 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
104 content_length = length.trim().parse().unwrap_or(0);
105 }
106 }
107 let mut body = vec![0u8; content_length];
108 if reader.read_exact(&mut body).is_err() {
109 continue;
110 }
111 let reply = queue.lock().unwrap().pop_front().unwrap_or_else(|| Reply {
112 status: "200 OK",
113 content_type: "text/event-stream",
114 body: "data: [DONE]\n\n".to_string(),
115 });
116 let response = format!(
117 "HTTP/1.1 {}\r\ncontent-type: {}\r\ncontent-length: {}\r\n\
118 connection: close\r\n\r\n{}",
119 reply.status,
120 reply.content_type,
121 reply.body.len(),
122 reply.body
123 );
124 let _ = stream.write_all(response.as_bytes());
125 let _ = stream.flush();
126 }
127 });
128
129 port
130 }
131
132 struct AgentUnderTest {
133 child: Child,
134 stdin: ChildStdin,
135 incoming: Receiver<Value>,
136 next_id: u64,
137 }
138
139 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
140 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
141 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
142 .env("OPENAI_API_KEY", "test-key")
143 .env("SIGIT_MODEL", "scripted-model")
144 .env("SIGIT_CONFIG_DIR", config_dir)
145 .env("SIGIT_MCP", "off")
146 .env("SIGIT_PERMISSIONS", "allow")
147 .env_remove("SIGIT_LOCAL_INFERENCE")
148 .stdin(Stdio::piped())
149 .stdout(Stdio::piped())
150 .stderr(Stdio::null())
151 .spawn()
152 .expect("spawn sigit in ACP mode");
153
154 let stdout = child.stdout.take().unwrap();
155 let (message_tx, incoming) = channel();
156 std::thread::spawn(move || {
157 for line in BufReader::new(stdout).lines() {
158 let Ok(line) = line else { break };
159 if let Ok(message) = serde_json::from_str::<Value>(&line)
160 && message_tx.send(message).is_err()
161 {
162 break;
163 }
164 }
165 });
166
167 let stdin = child.stdin.take().unwrap();
168 AgentUnderTest {
169 child,
170 stdin,
171 incoming,
172 next_id: 0,
173 }
174 }
175
176 impl AgentUnderTest {
177 fn request(&mut self, method: &str, params: Value) -> u64 {
178 self.next_id += 1;
179 let id = self.next_id;
180 let mut line =
181 json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
182 line.push('\n');
183 self.stdin.write_all(line.as_bytes()).expect("write stdin");
184 self.stdin.flush().expect("flush stdin");
185 id
186 }
187
188 fn wait_for_message(&mut self, id: u64) -> Value {
189 let deadline = Instant::now() + TIMEOUT;
190 loop {
191 let remaining = deadline.saturating_duration_since(Instant::now());
192 let Ok(message) = self.incoming.recv_timeout(remaining) else {
193 panic!("timed out waiting for the response to request {id}");
194 };
195 if message["id"] == id && message.get("method").is_none() {
196 return message;
197 }
198 }
199 }
200
201 fn wait_for_response(&mut self, id: u64) -> Value {
202 let message = self.wait_for_message(id);
203 assert!(
204 message.get("error").is_none(),
205 "request {id} failed: {message}"
206 );
207 message
208 }
209
210 fn wait_for_prompt_updates(&mut self, id: u64) -> (Value, Vec<Value>) {
211 let mut updates = Vec::new();
212 let deadline = Instant::now() + TIMEOUT;
213 loop {
214 let remaining = deadline.saturating_duration_since(Instant::now());
215 let Ok(message) = self.incoming.recv_timeout(remaining) else {
216 panic!("timed out waiting for the response to request {id}");
217 };
218 if message["method"] == "session/update" {
219 updates.push(message["params"]["update"].clone());
220 }
221 if message["id"] == id && message.get("method").is_none() {
222 return (message, updates);
223 }
224 }
225 }
226
227 fn open_session(&mut self, cwd: &std::path::Path) -> String {
228 let id = self.request(
229 "initialize",
230 json!({"protocolVersion": 1, "clientCapabilities": {}}),
231 );
232 self.wait_for_response(id);
233
234 let id = self.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
235 self.wait_for_response(id)["result"]["sessionId"]
236 .as_str()
237 .expect("session id")
238 .to_string()
239 }
240 }
241
242 impl Drop for AgentUnderTest {
243 fn drop(&mut self) {
244 let _ = self.child.kill();
245 let _ = self.child.wait();
246 }
247 }
248
249 fn scratch(name: &str) -> std::path::PathBuf {
250 let dir = std::env::temp_dir().join(format!("sigit_acp_err_{name}_{}", std::process::id()));
251 std::fs::create_dir_all(dir.join("config")).unwrap();
252 std::fs::create_dir_all(dir.join("work")).unwrap();
253 dir
254 }
255
256 /// The message the endpoint wrote is the message the client gets: no status
257 /// line in front of it, no JSON envelope around it.
258 fn assert_is_the_endpoints_message(error: &Value) {
259 let message = error["error"]["message"].as_str().unwrap_or_default();
260 assert_eq!(message, ALLOWANCE_MESSAGE, "full error: {error}");
261 }
262
263 #[test]
264 fn a_refused_turn_reports_the_endpoints_own_message() {
265 let dir = scratch("status");
266 let port = start_fake_endpoint(vec![error_status_reply()]);
267 let mut agent = spawn_agent(port, &dir.join("config"));
268 let session_id = agent.open_session(&dir.join("work"));
269
270 let id = agent.request(
271 "session/prompt",
272 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": "hello"}]}),
273 );
274 let response = agent.wait_for_message(id);
275
276 assert!(
277 response.get("error").is_some(),
278 "expected a failure: {response}"
279 );
280 assert_is_the_endpoints_message(&response);
281
282 std::fs::remove_dir_all(&dir).ok();
283 }
284
285 /// An error that arrives as a stream frame used to parse as a chunk with no
286 /// choices and get skipped, so the turn ended as if the model had said nothing.
287 #[test]
288 fn an_error_frame_mid_stream_is_not_swallowed() {
289 let dir = scratch("stream");
290 let port = start_fake_endpoint(vec![error_in_stream_reply()]);
291 let mut agent = spawn_agent(port, &dir.join("config"));
292 let session_id = agent.open_session(&dir.join("work"));
293
294 let id = agent.request(
295 "session/prompt",
296 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": "hello"}]}),
297 );
298 let response = agent.wait_for_message(id);
299
300 assert!(
301 response.get("error").is_some(),
302 "expected a failure: {response}"
303 );
304 assert_is_the_endpoints_message(&response);
305
306 std::fs::remove_dir_all(&dir).ok();
307 }
308
309 #[test]
310 fn failed_compaction_in_an_oversized_acp_turn_is_reported_as_chat_text() {
311 let dir = scratch("compact");
312 let port = start_fake_endpoint(vec![
313 sse_tool_call(
314 "call_1",
315 "run_command",
316 r#"{"command":"echo should-not-run"}"#,
317 ),
318 upstream_500_reply(),
319 ]);
320 let mut agent = spawn_agent(port, &dir.join("config"));
321 let session_id = agent.open_session(&dir.join("work"));
322
323 let oversized_prompt = "x".repeat(120_000);
324 let id = agent.request(
325 "session/prompt",
326 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": oversized_prompt}]}),
327 );
328 let (response, updates) = agent.wait_for_prompt_updates(id);
329
330 assert!(
331 response.get("error").is_none(),
332 "compaction failure should be a readable chat message, not an ACP error: {response}"
333 );
334 assert_eq!(response["result"]["stopReason"], "end_turn");
335
336 let rendered = updates
337 .iter()
338 .filter_map(|update| {
339 (update["sessionUpdate"] == "agent_message_chunk")
340 .then(|| update["content"]["text"].as_str())
341 .flatten()
342 })
343 .collect::<String>();
344 assert!(
345 rendered.contains("could not compact it: Onde Cloud upstream error (500)"),
346 "wrong rendered message: {rendered:?}"
347 );
348 assert!(
349 !updates
350 .iter()
351 .any(|update| update["sessionUpdate"] == "tool_call"),
352 "tools should not run after compaction fails: {updates:?}"
353 );
354
355 std::fs::remove_dir_all(&dir).ok();
356 }