1 //! End-to-end check that saved threads can be imported through `session/list`.
2 //!
3 //! Editors offer an "Import Threads" picker only to an agent that advertises
4 //! ACP's `sessionCapabilities.list` and answers `session/list` with the
5 //! sessions it has stored — before this, Zed said siGit Code doesn't support
6 //! the capability (issue #93).
7 //!
8 //! This drives the real binary against a scripted OpenAI-compatible endpoint:
9 //! run one prompt so a session gets saved, then list and assert the thread
10 //! comes back with its directory, title, and timestamp — and that the `cwd`
11 //! filter keeps another project's threads out.
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(60);
24
25 fn sse_text(text: &str) -> String {
26 format!(
27 "data: {}\n\ndata: [DONE]\n\n",
28 json!({"choices": [{"delta": {"content": text}}]})
29 )
30 }
31
32 /// Serves one scripted SSE response per request.
33 fn start_fake_endpoint(responses: Vec<String>) -> u16 {
34 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
35 let port = listener.local_addr().unwrap().port();
36 let queue = Mutex::new(VecDeque::from(responses));
37
38 std::thread::spawn(move || {
39 for stream in listener.incoming() {
40 let Ok(mut stream) = stream else { continue };
41 let mut reader = BufReader::new(match stream.try_clone() {
42 Ok(clone) => clone,
43 Err(_) => continue,
44 });
45 let mut content_length = 0usize;
46 loop {
47 let mut line = String::new();
48 if reader.read_line(&mut line).unwrap_or(0) == 0 {
49 break;
50 }
51 let line = line.trim();
52 if line.is_empty() {
53 break;
54 }
55 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
56 content_length = length.trim().parse().unwrap_or(0);
57 }
58 }
59 let mut body = vec![0u8; content_length];
60 if reader.read_exact(&mut body).is_err() {
61 continue;
62 }
63 let payload = queue
64 .lock()
65 .unwrap()
66 .pop_front()
67 .unwrap_or_else(|| "data: [DONE]\n\n".to_string());
68 let response = format!(
69 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\
70 content-length: {}\r\nconnection: close\r\n\r\n{}",
71 payload.len(),
72 payload
73 );
74 let _ = stream.write_all(response.as_bytes());
75 }
76 });
77
78 port
79 }
80
81 struct AgentUnderTest {
82 child: Child,
83 stdin: ChildStdin,
84 incoming: Receiver<Value>,
85 next_id: u64,
86 }
87
88 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
89 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
90 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
91 .env("OPENAI_API_KEY", "test-key")
92 .env("SIGIT_MODEL", "scripted-model")
93 .env("SIGIT_CONFIG_DIR", config_dir)
94 .env("SIGIT_MCP", "off")
95 .env("SIGIT_PERMISSIONS", "allow")
96 .env_remove("SIGIT_LOCAL_INFERENCE")
97 .stdin(Stdio::piped())
98 .stdout(Stdio::piped())
99 .stderr(Stdio::null())
100 .spawn()
101 .expect("spawn sigit in ACP mode");
102
103 let stdout = child.stdout.take().unwrap();
104 let (message_tx, incoming) = channel();
105 std::thread::spawn(move || {
106 for line in BufReader::new(stdout).lines() {
107 let Ok(line) = line else { break };
108 if let Ok(message) = serde_json::from_str::<Value>(&line)
109 && message_tx.send(message).is_err()
110 {
111 break;
112 }
113 }
114 });
115
116 let stdin = child.stdin.take().unwrap();
117 AgentUnderTest {
118 child,
119 stdin,
120 incoming,
121 next_id: 0,
122 }
123 }
124
125 impl AgentUnderTest {
126 fn request(&mut self, method: &str, params: Value) -> u64 {
127 self.next_id += 1;
128 let id = self.next_id;
129 let mut line =
130 json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
131 line.push('\n');
132 self.stdin
133 .write_all(line.as_bytes())
134 .expect("write to agent stdin");
135 self.stdin.flush().expect("flush agent stdin");
136 id
137 }
138
139 fn wait_for_response(&mut self, id: u64) -> Value {
140 let deadline = Instant::now() + TIMEOUT;
141 loop {
142 let remaining = deadline.saturating_duration_since(Instant::now());
143 let Ok(message) = self.incoming.recv_timeout(remaining) else {
144 panic!("timed out waiting for the response to request {id}");
145 };
146 if message["id"] == id && message.get("method").is_none() {
147 assert!(
148 message.get("error").is_none(),
149 "request {id} failed: {message}"
150 );
151 return message;
152 }
153 }
154 }
155 }
156
157 impl Drop for AgentUnderTest {
158 fn drop(&mut self) {
159 let _ = self.child.kill();
160 let _ = self.child.wait();
161 }
162 }
163
164 #[test]
165 fn saved_sessions_are_listed_for_their_project() {
166 let scratch = std::env::temp_dir().join(format!("sigit_acp_list_{}", std::process::id()));
167 let config_dir = scratch.join("config");
168 let project = scratch.join("project");
169 let other = scratch.join("other");
170 std::fs::create_dir_all(&config_dir).unwrap();
171 std::fs::create_dir_all(&project).unwrap();
172 std::fs::create_dir_all(&other).unwrap();
173
174 let port = start_fake_endpoint(vec![sse_text("Sure thing.")]);
175 let mut agent = spawn_agent(port, &config_dir);
176
177 let id = agent.request(
178 "initialize",
179 json!({"protocolVersion": 1, "clientCapabilities": {}}),
180 );
181 let initialize = agent.wait_for_response(id);
182 assert!(
183 initialize["result"]["agentCapabilities"]["sessionCapabilities"]["list"].is_object(),
184 "the import picker is gated on the session/list capability: {initialize}"
185 );
186
187 // Nothing saved yet.
188 let id = agent.request("session/list", json!({}));
189 let empty = agent.wait_for_response(id);
190 assert_eq!(
191 empty["result"]["sessions"].as_array().map(Vec::len),
192 Some(0),
193 "no sessions have been saved yet: {empty}"
194 );
195
196 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
197 let new_session = agent.wait_for_response(id);
198 let session_id = new_session["result"]["sessionId"]
199 .as_str()
200 .expect("session id")
201 .to_string();
202
203 let id = agent.request(
204 "session/prompt",
205 json!({
206 "sessionId": session_id,
207 "prompt": [{"type": "text", "text": "rename the parser module"}],
208 }),
209 );
210 agent.wait_for_response(id);
211
212 // The saved thread shows up for its own project …
213 let id = agent.request("session/list", json!({"cwd": project}));
214 let listed = agent.wait_for_response(id);
215 let sessions = listed["result"]["sessions"]
216 .as_array()
217 .unwrap_or_else(|| panic!("sessions array: {listed}"));
218 assert_eq!(sessions.len(), 1, "one saved thread expected: {listed}");
219 let info = &sessions[0];
220 assert_eq!(info["sessionId"], json!(session_id));
221 assert_eq!(
222 std::fs::canonicalize(info["cwd"].as_str().unwrap()).unwrap(),
223 std::fs::canonicalize(&project).unwrap()
224 );
225 assert_eq!(
226 info["title"], "rename the parser module",
227 "the title comes from the first user message: {info}"
228 );
229 let updated_at = info["updatedAt"].as_str().unwrap_or_default();
230 assert!(
231 updated_at.len() == 24 && updated_at.ends_with('Z') && updated_at.contains('.'),
232 "updatedAt should be an ISO 8601 UTC instant with milliseconds, got {updated_at:?}"
233 );
234
235 // … and not for a different one.
236 let id = agent.request("session/list", json!({"cwd": other}));
237 let elsewhere = agent.wait_for_response(id);
238 assert_eq!(
239 elsewhere["result"]["sessions"].as_array().map(Vec::len),
240 Some(0),
241 "another project's threads must not be offered: {elsewhere}"
242 );
243
244 drop(agent);
245 let _ = std::fs::remove_dir_all(&scratch);
246 }
247
248 /// A thread the user removes in the editor is gone for good: it no longer
249 /// lists or reopens, and deleting it again still succeeds (issue #192).
250 #[test]
251 fn a_deleted_session_no_longer_lists_or_loads() {
252 let scratch = std::env::temp_dir().join(format!("sigit_acp_delete_{}", std::process::id()));
253 let config_dir = scratch.join("config");
254 let project = scratch.join("project");
255 std::fs::create_dir_all(&config_dir).unwrap();
256 std::fs::create_dir_all(&project).unwrap();
257
258 let port = start_fake_endpoint(vec![sse_text("Done.")]);
259 let mut agent = spawn_agent(port, &config_dir);
260
261 let id = agent.request(
262 "initialize",
263 json!({"protocolVersion": 1, "clientCapabilities": {}}),
264 );
265 let initialize = agent.wait_for_response(id);
266 assert!(
267 initialize["result"]["agentCapabilities"]["sessionCapabilities"]["delete"].is_object(),
268 "session/delete is gated on its capability: {initialize}"
269 );
270
271 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
272 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
273 .as_str()
274 .expect("session id")
275 .to_string();
276 let id = agent.request(
277 "session/prompt",
278 json!({
279 "sessionId": session_id,
280 "prompt": [{"type": "text", "text": "say done"}],
281 }),
282 );
283 agent.wait_for_response(id);
284
285 let id = agent.request("session/list", json!({"cwd": project}));
286 let listed = agent.wait_for_response(id);
287 assert_eq!(
288 listed["result"]["sessions"].as_array().map(Vec::len),
289 Some(1),
290 "{listed}"
291 );
292
293 // The live thread is deleted: closed first, then removed from disk.
294 let id = agent.request("session/delete", json!({"sessionId": session_id}));
295 agent.wait_for_response(id);
296
297 let id = agent.request("session/list", json!({"cwd": project}));
298 let listed = agent.wait_for_response(id);
299 assert_eq!(
300 listed["result"]["sessions"].as_array().map(Vec::len),
301 Some(0),
302 "a deleted thread must not come back in the picker: {listed}"
303 );
304 let saved: Vec<_> = std::fs::read_dir(config_dir.join("sessions"))
305 .map(|entries| entries.flatten().map(|entry| entry.file_name()).collect())
306 .unwrap_or_default();
307 assert!(saved.is_empty(), "history and sidecar removed: {saved:?}");
308
309 // Nothing is left to reopen.
310 let id = agent.request(
311 "session/load",
312 json!({"sessionId": session_id, "cwd": project, "mcpServers": []}),
313 );
314 let deadline = Instant::now() + TIMEOUT;
315 let load = loop {
316 let remaining = deadline.saturating_duration_since(Instant::now());
317 let message = agent
318 .incoming
319 .recv_timeout(remaining)
320 .expect("session/load response");
321 if message["id"] == id && message.get("method").is_none() {
322 break message;
323 }
324 };
325 assert!(load.get("error").is_some(), "{load}");
326
327 // Deleting it again, or an id that never existed, succeeds quietly.
328 let id = agent.request("session/delete", json!({"sessionId": session_id}));
329 agent.wait_for_response(id);
330 let id = agent.request("session/delete", json!({"sessionId": "never-existed"}));
331 agent.wait_for_response(id);
332
333 drop(agent);
334 let _ = std::fs::remove_dir_all(&scratch);
335 }
336
337 /// Every notification for `session_id` that arrives before the response to
338 /// request `id`, in order.
339 fn updates_until_response(agent: &mut AgentUnderTest, id: u64) -> Vec<Value> {
340 let mut updates = Vec::new();
341 let deadline = Instant::now() + TIMEOUT;
342 loop {
343 let remaining = deadline.saturating_duration_since(Instant::now());
344 let message = agent
345 .incoming
346 .recv_timeout(remaining)
347 .unwrap_or_else(|_| panic!("timed out waiting for the response to request {id}"));
348 if message["method"] == "session/update" {
349 updates.push(message["params"]["update"].clone());
350 }
351 if message["id"] == id && message.get("method").is_none() {
352 assert!(message.get("error").is_none(), "{message}");
353 return updates;
354 }
355 }
356 }
357
358 /// A saved turn reaches the editor's sidebar right away: the title the first
359 /// time there is one, and a fresh `updatedAt` on every save (issue #193).
360 #[test]
361 fn a_saved_turn_sends_session_info_to_the_client() {
362 let scratch = std::env::temp_dir().join(format!("sigit_acp_info_{}", std::process::id()));
363 let config_dir = scratch.join("config");
364 let project = scratch.join("project");
365 std::fs::create_dir_all(&config_dir).unwrap();
366 std::fs::create_dir_all(&project).unwrap();
367
368 let port = start_fake_endpoint(vec![sse_text("First."), sse_text("Second.")]);
369 let mut agent = spawn_agent(port, &config_dir);
370
371 let id = agent.request(
372 "initialize",
373 json!({"protocolVersion": 1, "clientCapabilities": {}}),
374 );
375 agent.wait_for_response(id);
376 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
377 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
378 .as_str()
379 .expect("session id")
380 .to_string();
381
382 let info_updates = |updates: Vec<Value>| -> Vec<Value> {
383 updates
384 .into_iter()
385 .filter(|update| update["sessionUpdate"] == "session_info_update")
386 .collect()
387 };
388
389 let id = agent.request(
390 "session/prompt",
391 json!({
392 "sessionId": session_id,
393 "prompt": [{"type": "text", "text": "tidy up the build script"}],
394 }),
395 );
396 let first = info_updates(updates_until_response(&mut agent, id));
397 assert_eq!(first.len(), 1, "{first:?}");
398 assert_eq!(first[0]["title"], "tidy up the build script");
399 let updated_at = first[0]["updatedAt"].as_str().unwrap_or_default();
400 assert!(
401 updated_at.len() == 24 && updated_at.ends_with('Z'),
402 "updatedAt is an ISO 8601 instant: {updated_at:?}"
403 );
404 for field in ["sessionId", "cwd", "additionalDirectories"] {
405 assert!(first[0].get(field).is_none(), "{field} must not be sent");
406 }
407
408 let id = agent.request(
409 "session/prompt",
410 json!({
411 "sessionId": session_id,
412 "prompt": [{"type": "text", "text": "and the tests"}],
413 }),
414 );
415 let second = info_updates(updates_until_response(&mut agent, id));
416 assert_eq!(second.len(), 1, "{second:?}");
417 assert!(
418 second[0].get("title").is_none(),
419 "an unchanged title is not sent again: {second:?}"
420 );
421 assert!(second[0]["updatedAt"].is_string(), "{second:?}");
422
423 drop(agent);
424 let _ = std::fs::remove_dir_all(&scratch);
425 }