1
+//! The stop reason a prompt turn ends with (issue #140).
2
+//!
3
+//! ACP defines five and the agent only ever sent `end_turn` and `cancelled`,
4
+//! so a client could not tell a finished answer from one cut off at the token
5
+//! limit, withheld by the endpoint, or stopped at the tool-round cap. The
6
+//! `cancelled` case is covered in `acp_permissions.rs`; this file drives the
7
+//! other three through a scripted endpoint.
8
+
9
+use std::collections::VecDeque;
10
+use std::io::{BufRead, BufReader, Read, Write};
11
+use std::net::TcpListener;
12
+use std::process::{Child, ChildStdin, Command, Stdio};
13
+use std::sync::mpsc::{Receiver, channel};
14
+use std::sync::{Arc, Mutex};
15
+use std::time::{Duration, Instant};
16
+
17
+use serde_json::{Value, json};
18
+
19
+const TIMEOUT: Duration = Duration::from_secs(60);
20
+
21
+/// The built-in tool-round cap (`MAX_TOOL_ROUNDS` in `main.rs`).
22
+const TOOL_ROUND_CAP: usize = 24;
23
+
24
+/// A streamed text reply that ends with `finish_reason`.
25
+fn sse_text(text: &str, finish_reason: &str) -> String {
26
+ format!(
27
+ "data: {}\n\ndata: {}\n\ndata: [DONE]\n\n",
28
+ json!({"choices": [{"delta": {"content": text}}]}),
29
+ json!({"choices": [{"delta": {}, "finish_reason": finish_reason}]}),
30
+ )
31
+}
32
+
33
+fn sse_tool_call(id: &str, name: &str, arguments: &str) -> String {
34
+ format!(
35
+ "data: {}\n\ndata: {}\n\ndata: [DONE]\n\n",
36
+ json!({
37
+ "choices": [{"delta": {"tool_calls": [{
38
+ "index": 0,
39
+ "id": id,
40
+ "function": {"name": name, "arguments": arguments},
41
+ }]}}]
42
+ }),
43
+ json!({"choices": [{"delta": {}, "finish_reason": "tool_calls"}]}),
44
+ )
45
+}
46
+
47
+/// A scripted endpoint: one SSE body per request, in order, then empty
48
+/// replies. Every request body it receives is kept for the test to read.
49
+struct FakeEndpoint {
50
+ port: u16,
51
+ requests: Arc<Mutex<Vec<Value>>>,
52
+}
53
+
54
+fn start_fake_endpoint(replies: Vec<String>) -> FakeEndpoint {
55
+ let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
56
+ let port = listener.local_addr().unwrap().port();
57
+ let queue = Mutex::new(VecDeque::from(replies));
58
+ let requests = Arc::new(Mutex::new(Vec::new()));
59
+ let seen = Arc::clone(&requests);
60
+
61
+ std::thread::spawn(move || {
62
+ for stream in listener.incoming() {
63
+ let Ok(mut stream) = stream else { continue };
64
+ let mut reader = BufReader::new(match stream.try_clone() {
65
+ Ok(clone) => clone,
66
+ Err(_) => continue,
67
+ });
68
+ let mut content_length = 0usize;
69
+ loop {
70
+ let mut line = String::new();
71
+ if reader.read_line(&mut line).unwrap_or(0) == 0 {
72
+ break;
73
+ }
74
+ let line = line.trim();
75
+ if line.is_empty() {
76
+ break;
77
+ }
78
+ if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
79
+ content_length = length.trim().parse().unwrap_or(0);
80
+ }
81
+ }
82
+ let mut body = vec![0u8; content_length];
83
+ if reader.read_exact(&mut body).is_err() {
84
+ continue;
85
+ }
86
+ if let Ok(request) = serde_json::from_slice::<Value>(&body) {
87
+ seen.lock().unwrap().push(request);
88
+ }
89
+ let reply = queue
90
+ .lock()
91
+ .unwrap()
92
+ .pop_front()
93
+ .unwrap_or_else(|| "data: [DONE]\n\n".to_string());
94
+ let response = format!(
95
+ "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: {}\r\n\
96
+ connection: close\r\n\r\n{}",
97
+ reply.len(),
98
+ reply
99
+ );
100
+ let _ = stream.write_all(response.as_bytes());
101
+ let _ = stream.flush();
102
+ }
103
+ });
104
+
105
+ FakeEndpoint { port, requests }
106
+}
107
+
108
+struct AgentUnderTest {
109
+ child: Child,
110
+ stdin: ChildStdin,
111
+ incoming: Receiver<Value>,
112
+ next_id: u64,
113
+}
114
+
115
+fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
116
+ let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
117
+ .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
118
+ .env("OPENAI_API_KEY", "test-key")
119
+ .env("SIGIT_MODEL", "scripted-model")
120
+ .env("SIGIT_CONFIG_DIR", config_dir)
121
+ .env("SIGIT_MCP", "off")
122
+ .env("SIGIT_PERMISSIONS", "allow")
123
+ .env_remove("SIGIT_LOCAL_INFERENCE")
124
+ .stdin(Stdio::piped())
125
+ .stdout(Stdio::piped())
126
+ .stderr(Stdio::null())
127
+ .spawn()
128
+ .expect("spawn sigit in ACP mode");
129
+
130
+ let stdout = child.stdout.take().unwrap();
131
+ let (message_tx, incoming) = channel();
132
+ std::thread::spawn(move || {
133
+ for line in BufReader::new(stdout).lines() {
134
+ let Ok(line) = line else { break };
135
+ if let Ok(message) = serde_json::from_str::<Value>(&line)
136
+ && message_tx.send(message).is_err()
137
+ {
138
+ break;
139
+ }
140
+ }
141
+ });
142
+
143
+ let stdin = child.stdin.take().unwrap();
144
+ AgentUnderTest {
145
+ child,
146
+ stdin,
147
+ incoming,
148
+ next_id: 0,
149
+ }
150
+}
151
+
152
+impl AgentUnderTest {
153
+ fn request(&mut self, method: &str, params: Value) -> u64 {
154
+ self.next_id += 1;
155
+ let id = self.next_id;
156
+ let mut line =
157
+ json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
158
+ line.push('\n');
159
+ self.stdin.write_all(line.as_bytes()).expect("write stdin");
160
+ self.stdin.flush().expect("flush stdin");
161
+ id
162
+ }
163
+
164
+ fn wait_for_response(&mut self, id: u64) -> Value {
165
+ let deadline = Instant::now() + TIMEOUT;
166
+ loop {
167
+ let remaining = deadline.saturating_duration_since(Instant::now());
168
+ let Ok(message) = self.incoming.recv_timeout(remaining) else {
169
+ panic!("timed out waiting for the response to request {id}");
170
+ };
171
+ if message["id"] == id && message.get("method").is_none() {
172
+ assert!(
173
+ message.get("error").is_none(),
174
+ "request {id} failed: {message}"
175
+ );
176
+ return message;
177
+ }
178
+ }
179
+ }
180
+
181
+ fn open_session(&mut self, cwd: &std::path::Path) -> String {
182
+ let id = self.request(
183
+ "initialize",
184
+ json!({"protocolVersion": 1, "clientCapabilities": {}}),
185
+ );
186
+ self.wait_for_response(id);
187
+
188
+ let id = self.request("session/new", json!({"cwd": cwd, "mcpServers": []}));
189
+ self.wait_for_response(id)["result"]["sessionId"]
190
+ .as_str()
191
+ .expect("session id")
192
+ .to_string()
193
+ }
194
+
195
+ /// Send one text prompt and return the stop reason the turn ended with.
196
+ fn prompt(&mut self, session_id: &str, text: &str) -> String {
197
+ let id = self.request(
198
+ "session/prompt",
199
+ json!({"sessionId": session_id, "prompt": [{"type": "text", "text": text}]}),
200
+ );
201
+ self.wait_for_response(id)["result"]["stopReason"]
202
+ .as_str()
203
+ .expect("stop reason")
204
+ .to_string()
205
+ }
206
+}
207
+
208
+impl Drop for AgentUnderTest {
209
+ fn drop(&mut self) {
210
+ let _ = self.child.kill();
211
+ let _ = self.child.wait();
212
+ }
213
+}
214
+
215
+fn scratch(name: &str) -> std::path::PathBuf {
216
+ let dir = std::env::temp_dir().join(format!("sigit_acp_stop_{name}_{}", std::process::id()));
217
+ std::fs::create_dir_all(dir.join("config")).unwrap();
218
+ std::fs::create_dir_all(dir.join("work")).unwrap();
219
+ dir
220
+}
221
+
222
+/// Whether any message of a chat-completion request mentions `needle`.
223
+fn request_mentions(request: &Value, needle: &str) -> bool {
224
+ request["messages"]
225
+ .as_array()
226
+ .is_some_and(|messages| messages.iter().any(|m| m.to_string().contains(needle)))
227
+}
228
+
229
+#[test]
230
+fn a_reply_cut_off_at_the_token_limit_ends_with_max_tokens() {
231
+ let dir = scratch("length");
232
+ let endpoint = start_fake_endpoint(vec![sse_text("The answer is", "length")]);
233
+ let mut agent = spawn_agent(endpoint.port, &dir.join("config"));
234
+ let session_id = agent.open_session(&dir.join("work"));
235
+
236
+ assert_eq!(agent.prompt(&session_id, "explain"), "max_tokens");
237
+
238
+ std::fs::remove_dir_all(&dir).ok();
239
+}
240
+
241
+#[test]
242
+fn a_finished_reply_still_ends_with_end_turn() {
243
+ let dir = scratch("stop");
244
+ let endpoint = start_fake_endpoint(vec![sse_text("Done.", "stop")]);
245
+ let mut agent = spawn_agent(endpoint.port, &dir.join("config"));
246
+ let session_id = agent.open_session(&dir.join("work"));
247
+
248
+ assert_eq!(agent.prompt(&session_id, "explain"), "end_turn");
249
+
250
+ std::fs::remove_dir_all(&dir).ok();
251
+}
252
+
253
+/// A refusal is more than a label: ACP says the refused prompt and what
254
+/// followed it are not part of the next prompt, and the client drops them from
255
+/// its view on that promise. The agent has to drop them from history too.
256
+#[test]
257
+fn a_withheld_reply_ends_with_refusal_and_leaves_history() {
258
+ let dir = scratch("refusal");
259
+ let endpoint = start_fake_endpoint(vec![
260
+ sse_text("Sure.", "stop"),
261
+ sse_text("", "content_filter"),
262
+ sse_text("Here you go.", "stop"),
263
+ ]);
264
+ let mut agent = spawn_agent(endpoint.port, &dir.join("config"));
265
+ let session_id = agent.open_session(&dir.join("work"));
266
+
267
+ assert_eq!(agent.prompt(&session_id, "first-kept-prompt"), "end_turn");
268
+ assert_eq!(
269
+ agent.prompt(&session_id, "second-refused-prompt"),
270
+ "refusal"
271
+ );
272
+ assert_eq!(agent.prompt(&session_id, "third-prompt"), "end_turn");
273
+
274
+ let requests = endpoint.requests.lock().unwrap();
275
+ let last = requests
276
+ .last()
277
+ .expect("the third prompt reached the endpoint");
278
+ assert!(
279
+ request_mentions(last, "first-kept-prompt"),
280
+ "the turn before the refusal stays in history: {last}"
281
+ );
282
+ assert!(
283
+ !request_mentions(last, "second-refused-prompt"),
284
+ "the refused prompt must not be replayed: {last}"
285
+ );
286
+
287
+ std::fs::remove_dir_all(&dir).ok();
288
+}
289
+
290
+/// The case the reason exists for: the model keeps asking for tools until the
291
+/// agent stops offering them.
292
+#[test]
293
+fn a_turn_stopped_at_the_tool_round_cap_ends_with_max_turn_requests() {
294
+ let dir = scratch("cap");
295
+ // Each call differs, or the repeated-call guard ends the turn first. The
296
+ // commands stay quote-free: `cmd /C` mangles quotes on Windows.
297
+ let replies = (1..=TOOL_ROUND_CAP)
298
+ .map(|round| {
299
+ sse_tool_call(
300
+ &format!("call_{round}"),
301
+ "run_command",
302
+ &json!({"command": format!("echo round-{round}")}).to_string(),
303
+ )
304
+ })
305
+ .chain([sse_text("Out of rounds.", "stop")])
306
+ .collect();
307
+ let endpoint = start_fake_endpoint(replies);
308
+ let mut agent = spawn_agent(endpoint.port, &dir.join("config"));
309
+ let session_id = agent.open_session(&dir.join("work"));
310
+
311
+ assert_eq!(agent.prompt(&session_id, "keep going"), "max_turn_requests");
312
+ assert_eq!(
313
+ endpoint.requests.lock().unwrap().len(),
314
+ TOOL_ROUND_CAP + 1,
315
+ "one request per tool round, plus the forced closing reply"
316
+ );
317
+
318
+ std::fs::remove_dir_all(&dir).ok();
319
+}