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
+}