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 AgentUnderTest {
158 /// Like `wait_for_response`, for a request that is expected to fail.
159 fn wait_for_reply(&mut self, id: u64) -> Value {
160 let deadline = Instant::now() + TIMEOUT;
161 loop {
162 let remaining = deadline.saturating_duration_since(Instant::now());
163 let Ok(message) = self.incoming.recv_timeout(remaining) else {
164 panic!("timed out waiting for the reply to request {id}");
165 };
166 if message["id"] == id && message.get("method").is_none() {
167 return message;
168 }
169 }
170 }
171 }
172
173 impl Drop for AgentUnderTest {
174 fn drop(&mut self) {
175 let _ = self.child.kill();
176 let _ = self.child.wait();
177 }
178 }
179
180 #[test]
181 fn saved_sessions_are_listed_for_their_project() {
182 let scratch = std::env::temp_dir().join(format!("sigit_acp_list_{}", std::process::id()));
183 let config_dir = scratch.join("config");
184 let project = scratch.join("project");
185 let other = scratch.join("other");
186 std::fs::create_dir_all(&config_dir).unwrap();
187 std::fs::create_dir_all(&project).unwrap();
188 std::fs::create_dir_all(&other).unwrap();
189
190 let port = start_fake_endpoint(vec![sse_text("Sure thing.")]);
191 let mut agent = spawn_agent(port, &config_dir);
192
193 let id = agent.request(
194 "initialize",
195 json!({"protocolVersion": 1, "clientCapabilities": {}}),
196 );
197 let initialize = agent.wait_for_response(id);
198 assert!(
199 initialize["result"]["agentCapabilities"]["sessionCapabilities"]["list"].is_object(),
200 "the import picker is gated on the session/list capability: {initialize}"
201 );
202
203 // Nothing saved yet.
204 let id = agent.request("session/list", json!({}));
205 let empty = agent.wait_for_response(id);
206 assert_eq!(
207 empty["result"]["sessions"].as_array().map(Vec::len),
208 Some(0),
209 "no sessions have been saved yet: {empty}"
210 );
211
212 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
213 let new_session = agent.wait_for_response(id);
214 let session_id = new_session["result"]["sessionId"]
215 .as_str()
216 .expect("session id")
217 .to_string();
218
219 let id = agent.request(
220 "session/prompt",
221 json!({
222 "sessionId": session_id,
223 "prompt": [{"type": "text", "text": "rename the parser module"}],
224 }),
225 );
226 agent.wait_for_response(id);
227
228 // The saved thread shows up for its own project …
229 let id = agent.request("session/list", json!({"cwd": project}));
230 let listed = agent.wait_for_response(id);
231 let sessions = listed["result"]["sessions"]
232 .as_array()
233 .unwrap_or_else(|| panic!("sessions array: {listed}"));
234 assert_eq!(sessions.len(), 1, "one saved thread expected: {listed}");
235 let info = &sessions[0];
236 assert_eq!(info["sessionId"], json!(session_id));
237 assert_eq!(
238 std::fs::canonicalize(info["cwd"].as_str().unwrap()).unwrap(),
239 std::fs::canonicalize(&project).unwrap()
240 );
241 assert_eq!(
242 info["title"], "rename the parser module",
243 "the title comes from the first user message: {info}"
244 );
245 let updated_at = info["updatedAt"].as_str().unwrap_or_default();
246 assert!(
247 updated_at.len() == 24 && updated_at.ends_with('Z') && updated_at.contains('.'),
248 "updatedAt should be an ISO 8601 UTC instant with milliseconds, got {updated_at:?}"
249 );
250
251 // … and not for a different one.
252 let id = agent.request("session/list", json!({"cwd": other}));
253 let elsewhere = agent.wait_for_response(id);
254 assert_eq!(
255 elsewhere["result"]["sessions"].as_array().map(Vec::len),
256 Some(0),
257 "another project's threads must not be offered: {elsewhere}"
258 );
259
260 drop(agent);
261 let _ = std::fs::remove_dir_all(&scratch);
262 }
263
264 /// A thread the user removes in the editor is gone for good: it no longer
265 /// lists or reopens, and deleting it again still succeeds (issue #192).
266 #[test]
267 fn a_deleted_session_no_longer_lists_or_loads() {
268 let scratch = std::env::temp_dir().join(format!("sigit_acp_delete_{}", std::process::id()));
269 let config_dir = scratch.join("config");
270 let project = scratch.join("project");
271 std::fs::create_dir_all(&config_dir).unwrap();
272 std::fs::create_dir_all(&project).unwrap();
273
274 let port = start_fake_endpoint(vec![sse_text("Done.")]);
275 let mut agent = spawn_agent(port, &config_dir);
276
277 let id = agent.request(
278 "initialize",
279 json!({"protocolVersion": 1, "clientCapabilities": {}}),
280 );
281 let initialize = agent.wait_for_response(id);
282 assert!(
283 initialize["result"]["agentCapabilities"]["sessionCapabilities"]["delete"].is_object(),
284 "session/delete is gated on its capability: {initialize}"
285 );
286
287 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
288 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
289 .as_str()
290 .expect("session id")
291 .to_string();
292 let id = agent.request(
293 "session/prompt",
294 json!({
295 "sessionId": session_id,
296 "prompt": [{"type": "text", "text": "say done"}],
297 }),
298 );
299 agent.wait_for_response(id);
300
301 let id = agent.request("session/list", json!({"cwd": project}));
302 let listed = agent.wait_for_response(id);
303 assert_eq!(
304 listed["result"]["sessions"].as_array().map(Vec::len),
305 Some(1),
306 "{listed}"
307 );
308
309 // The live thread is deleted: closed first, then removed from disk.
310 let id = agent.request("session/delete", json!({"sessionId": session_id}));
311 agent.wait_for_response(id);
312
313 let id = agent.request("session/list", json!({"cwd": project}));
314 let listed = agent.wait_for_response(id);
315 assert_eq!(
316 listed["result"]["sessions"].as_array().map(Vec::len),
317 Some(0),
318 "a deleted thread must not come back in the picker: {listed}"
319 );
320 let saved: Vec<_> = std::fs::read_dir(config_dir.join("sessions"))
321 .map(|entries| entries.flatten().map(|entry| entry.file_name()).collect())
322 .unwrap_or_default();
323 assert!(saved.is_empty(), "history and sidecar removed: {saved:?}");
324
325 // Nothing is left to reopen.
326 let id = agent.request(
327 "session/load",
328 json!({"sessionId": session_id, "cwd": project, "mcpServers": []}),
329 );
330 let deadline = Instant::now() + TIMEOUT;
331 let load = loop {
332 let remaining = deadline.saturating_duration_since(Instant::now());
333 let message = agent
334 .incoming
335 .recv_timeout(remaining)
336 .expect("session/load response");
337 if message["id"] == id && message.get("method").is_none() {
338 break message;
339 }
340 };
341 assert!(load.get("error").is_some(), "{load}");
342
343 // Deleting it again, or an id that never existed, succeeds quietly.
344 let id = agent.request("session/delete", json!({"sessionId": session_id}));
345 agent.wait_for_response(id);
346 let id = agent.request("session/delete", json!({"sessionId": "never-existed"}));
347 agent.wait_for_response(id);
348
349 drop(agent);
350 let _ = std::fs::remove_dir_all(&scratch);
351 }
352
353 /// Every notification for `session_id` that arrives before the response to
354 /// request `id`, in order.
355 fn updates_until_response(agent: &mut AgentUnderTest, id: u64) -> Vec<Value> {
356 let mut updates = Vec::new();
357 let deadline = Instant::now() + TIMEOUT;
358 loop {
359 let remaining = deadline.saturating_duration_since(Instant::now());
360 let message = agent
361 .incoming
362 .recv_timeout(remaining)
363 .unwrap_or_else(|_| panic!("timed out waiting for the response to request {id}"));
364 if message["method"] == "session/update" {
365 updates.push(message["params"]["update"].clone());
366 }
367 if message["id"] == id && message.get("method").is_none() {
368 assert!(message.get("error").is_none(), "{message}");
369 return updates;
370 }
371 }
372 }
373
374 /// A saved turn reaches the editor's sidebar right away: the title the first
375 /// time there is one, and a fresh `updatedAt` on every save (issue #193).
376 #[test]
377 fn a_saved_turn_sends_session_info_to_the_client() {
378 let scratch = std::env::temp_dir().join(format!("sigit_acp_info_{}", std::process::id()));
379 let config_dir = scratch.join("config");
380 let project = scratch.join("project");
381 std::fs::create_dir_all(&config_dir).unwrap();
382 std::fs::create_dir_all(&project).unwrap();
383
384 let port = start_fake_endpoint(vec![sse_text("First."), sse_text("Second.")]);
385 let mut agent = spawn_agent(port, &config_dir);
386
387 let id = agent.request(
388 "initialize",
389 json!({"protocolVersion": 1, "clientCapabilities": {}}),
390 );
391 agent.wait_for_response(id);
392 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
393 let session_id = agent.wait_for_response(id)["result"]["sessionId"]
394 .as_str()
395 .expect("session id")
396 .to_string();
397
398 let info_updates = |updates: Vec<Value>| -> Vec<Value> {
399 updates
400 .into_iter()
401 .filter(|update| update["sessionUpdate"] == "session_info_update")
402 .collect()
403 };
404
405 let id = agent.request(
406 "session/prompt",
407 json!({
408 "sessionId": session_id,
409 "prompt": [{"type": "text", "text": "tidy up the build script"}],
410 }),
411 );
412 let first = info_updates(updates_until_response(&mut agent, id));
413 assert_eq!(first.len(), 1, "{first:?}");
414 assert_eq!(first[0]["title"], "tidy up the build script");
415 let updated_at = first[0]["updatedAt"].as_str().unwrap_or_default();
416 assert!(
417 updated_at.len() == 24 && updated_at.ends_with('Z'),
418 "updatedAt is an ISO 8601 instant: {updated_at:?}"
419 );
420 for field in ["sessionId", "cwd", "additionalDirectories"] {
421 assert!(first[0].get(field).is_none(), "{field} must not be sent");
422 }
423
424 let id = agent.request(
425 "session/prompt",
426 json!({
427 "sessionId": session_id,
428 "prompt": [{"type": "text", "text": "and the tests"}],
429 }),
430 );
431 let second = info_updates(updates_until_response(&mut agent, id));
432 assert_eq!(second.len(), 1, "{second:?}");
433 assert!(
434 second[0].get("title").is_none(),
435 "an unchanged title is not sent again: {second:?}"
436 );
437 assert!(second[0]["updatedAt"].is_string(), "{second:?}");
438
439 drop(agent);
440 let _ = std::fs::remove_dir_all(&scratch);
441 }
442
443 /// Write a saved thread straight into the session store, the way a finished
444 /// turn leaves one.
445 fn seed_session(config_dir: &std::path::Path, id: &str, cwd: &std::path::Path) {
446 let sessions = config_dir.join("sessions");
447 std::fs::create_dir_all(&sessions).unwrap();
448 std::fs::write(
449 sessions.join(format!("{id}.jsonl")),
450 format!(
451 "{}\n",
452 json!({"role": "user", "content": format!("thread {id}")})
453 ),
454 )
455 .unwrap();
456 std::fs::write(
457 sessions.join(format!("{id}.meta.json")),
458 json!({"cwd": cwd}).to_string(),
459 )
460 .unwrap();
461 }
462
463 #[test]
464 fn a_long_session_list_is_paged_with_a_cursor() {
465 let scratch = std::env::temp_dir().join(format!("sigit_acp_list_pages_{}", std::process::id()));
466 let config_dir = scratch.join("config");
467 let project = scratch.join("project");
468 std::fs::create_dir_all(&project).unwrap();
469 // One more than a page holds.
470 for index in 0..101 {
471 seed_session(&config_dir, &format!("thread-{index:03}"), &project);
472 }
473
474 let port = start_fake_endpoint(Vec::new());
475 let mut agent = spawn_agent(port, &config_dir);
476 let id = agent.request(
477 "initialize",
478 json!({"protocolVersion": 1, "clientCapabilities": {}}),
479 );
480 agent.wait_for_response(id);
481
482 let id = agent.request("session/list", json!({"cwd": project}));
483 let first = agent.wait_for_response(id);
484 let first_page = first["result"]["sessions"].as_array().expect("sessions");
485 assert_eq!(first_page.len(), 100);
486 let cursor = first["result"]["nextCursor"]
487 .as_str()
488 .expect("a full page with more behind it carries nextCursor")
489 .to_string();
490
491 let id = agent.request("session/list", json!({"cwd": project, "cursor": cursor}));
492 let second = agent.wait_for_response(id);
493 let second_page = second["result"]["sessions"].as_array().expect("sessions");
494 assert_eq!(second_page.len(), 1);
495 assert!(
496 second["result"].get("nextCursor").is_none(),
497 "the last page has no cursor: {second}"
498 );
499
500 let mut seen: Vec<&str> = first_page
501 .iter()
502 .chain(second_page)
503 .map(|session| session["sessionId"].as_str().unwrap())
504 .collect();
505 seen.sort_unstable();
506 seen.dedup();
507 assert_eq!(seen.len(), 101, "every thread is listed exactly once");
508
509 let id = agent.request("session/list", json!({"cursor": "not-a-cursor"}));
510 let rejected = agent.wait_for_reply(id);
511 assert_eq!(rejected["error"]["code"], -32602, "{rejected}");
512
513 drop(agent);
514 let _ = std::fs::remove_dir_all(&scratch);
515 }