@sigitcodereview / sigit
Code Commits Issues Pulls Cloud Agent CI/CD
claude/sigit-cloud-agent-expansion-reox0a
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 262 lines 8.77 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 start_fake_endpoint(replies: Vec<Reply>) -> u16 {
54 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
55 let port = listener.local_addr().unwrap().port();
56 let queue = Mutex::new(VecDeque::from(replies));
57
58 std::thread::spawn(move || {
59 for stream in listener.incoming() {
60 let Ok(mut stream) = stream else { continue };
61 let mut reader = BufReader::new(match stream.try_clone() {
62 Ok(clone) => clone,
63 Err(_) => continue,
64 });
65 let mut content_length = 0usize;
66 loop {
67 let mut line = String::new();
68 if reader.read_line(&mut line).unwrap_or(0) == 0 {
69 break;
70 }
71 let line = line.trim();
72 if line.is_empty() {
73 break;
74 }
75 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
76 content_length = length.trim().parse().unwrap_or(0);
77 }
78 }
79 let mut body = vec![0u8; content_length];
80 if reader.read_exact(&mut body).is_err() {
81 continue;
82 }
83 let reply = queue.lock().unwrap().pop_front().unwrap_or_else(|| Reply {
84 status: "200 OK",
85 content_type: "text/event-stream",
86 body: "data: [DONE]\n\n".to_string(),
87 });
88 let response = format!(
89 "HTTP/1.1 {}\r\ncontent-type: {}\r\ncontent-length: {}\r\n\
90 connection: close\r\n\r\n{}",
91 reply.status,
92 reply.content_type,
93 reply.body.len(),
94 reply.body
95 );
96 let _ = stream.write_all(response.as_bytes());
97 let _ = stream.flush();
98 }
99 });
100
101 port
102 }
103
104 struct AgentUnderTest {
105 child: Child,
106 stdin: ChildStdin,
107 incoming: Receiver<Value>,
108 next_id: u64,
109 }
110
111 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
112 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
113 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
114 .env("OPENAI_API_KEY", "test-key")
115 .env("SIGIT_MODEL", "scripted-model")
116 .env("SIGIT_CONFIG_DIR", config_dir)
117 .env("SIGIT_MCP", "off")
118 .env("SIGIT_PERMISSIONS", "allow")
119 .env_remove("SIGIT_LOCAL_INFERENCE")
120 .stdin(Stdio::piped())
121 .stdout(Stdio::piped())
122 .stderr(Stdio::null())
123 .spawn()
124 .expect("spawn sigit in ACP mode");
125
126 let stdout = child.stdout.take().unwrap();
127 let (message_tx, incoming) = channel();
128 std::thread::spawn(move || {
129 for line in BufReader::new(stdout).lines() {
130 let Ok(line) = line else { break };
131 if let Ok(message) = serde_json::from_str::<Value>(&line)
132 && message_tx.send(message).is_err()
133 {
134 break;
135 }
136 }
137 });
138
139 let stdin = child.stdin.take().unwrap();
140 AgentUnderTest {
141 child,
142 stdin,
143 incoming,
144 next_id: 0,
145 }
146 }
147
148 impl AgentUnderTest {
149 fn request(&mut self, method: &str, params: Value) -> u64 {
150 self.next_id += 1;
151 let id = self.next_id;
152 let mut line =
153 json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
154 line.push('\n');
155 self.stdin.write_all(line.as_bytes()).expect("write stdin");
156 self.stdin.flush().expect("flush stdin");
157 id
158 }
159
160 fn wait_for_message(&mut self, id: u64) -> Value {
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["id"] == id && message.get("method").is_none() {
168 return message;
169 }
170 }
171 }
172
173 fn wait_for_response(&mut self, id: u64) -> Value {
174 let message = self.wait_for_message(id);
175 assert!(
176 message.get("error").is_none(),
177 "request {id} failed: {message}"
178 );
179 message
180 }
181
182 fn open_session(&mut self, cwd: &std::path::Path) -> String {
183 let id = self.request(
184 "initialize",
185 json!({"protocolVersion": 1, "clientCapabilities": {}}),
186 );
187 self.wait_for_response(id);
188
189 let id = self.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
190 self.wait_for_response(id)["result"]["sessionId"]
191 .as_str()
192 .expect("session id")
193 .to_string()
194 }
195 }
196
197 impl Drop for AgentUnderTest {
198 fn drop(&mut self) {
199 let _ = self.child.kill();
200 let _ = self.child.wait();
201 }
202 }
203
204 fn scratch(name: &str) -> std::path::PathBuf {
205 let dir = std::env::temp_dir().join(format!("sigit_acp_err_{name}_{}", std::process::id()));
206 std::fs::create_dir_all(dir.join("config")).unwrap();
207 std::fs::create_dir_all(dir.join("work")).unwrap();
208 dir
209 }
210
211 /// The message the endpoint wrote is the message the client gets: no status
212 /// line in front of it, no JSON envelope around it.
213 fn assert_is_the_endpoints_message(error: &Value) {
214 let message = error["error"]["message"].as_str().unwrap_or_default();
215 assert_eq!(message, ALLOWANCE_MESSAGE, "full error: {error}");
216 }
217
218 #[test]
219 fn a_refused_turn_reports_the_endpoints_own_message() {
220 let dir = scratch("status");
221 let port = start_fake_endpoint(vec![error_status_reply()]);
222 let mut agent = spawn_agent(port, &dir.join("config"));
223 let session_id = agent.open_session(&dir.join("work"));
224
225 let id = agent.request(
226 "session/prompt",
227 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": "hello"}]}),
228 );
229 let response = agent.wait_for_message(id);
230
231 assert!(
232 response.get("error").is_some(),
233 "expected a failure: {response}"
234 );
235 assert_is_the_endpoints_message(&response);
236
237 std::fs::remove_dir_all(&dir).ok();
238 }
239
240 /// An error that arrives as a stream frame used to parse as a chunk with no
241 /// choices and get skipped, so the turn ended as if the model had said nothing.
242 #[test]
243 fn an_error_frame_mid_stream_is_not_swallowed() {
244 let dir = scratch("stream");
245 let port = start_fake_endpoint(vec![error_in_stream_reply()]);
246 let mut agent = spawn_agent(port, &dir.join("config"));
247 let session_id = agent.open_session(&dir.join("work"));
248
249 let id = agent.request(
250 "session/prompt",
251 json!({"sessionId": session_id, "prompt": [{"type": "text", "text": "hello"}]}),
252 );
253 let response = agent.wait_for_message(id);
254
255 assert!(
256 response.get("error").is_some(),
257 "expected a failure: {response}"
258 );
259 assert_is_the_endpoints_message(&response);
260
261 std::fs::remove_dir_all(&dir).ok();
262 }