1 //! End-to-end check that reopening a saved session redraws it in the client.
2 //!
3 //! ACP's `session/load` is a replay: the client renders the reopened thread
4 //! from the `session/update` notifications the agent streams while the request
5 //! is in flight, and keeps nothing of its own. Restoring the history into the
6 //! backend is what makes the *model* remember, but it puts nothing on screen —
7 //! before this, clicking a thread in Zed's history opened an empty one that
8 //! looked brand new (issue #77).
9 //!
10 //! This drives the real binary against a scripted OpenAI-compatible endpoint:
11 //! run one prompt that talks and calls a tool, then load the same session id
12 //! and assert the conversation comes back over the wire.
13
14 use std::collections::VecDeque;
15 use std::io::{BufRead, BufReader, Read, Write};
16 use std::net::TcpListener;
17 use std::process::{Child, ChildStdin, Command, Stdio};
18 use std::sync::Mutex;
19 use std::sync::mpsc::{Receiver, channel};
20 use std::time::{Duration, Instant};
21
22 use serde_json::{Value, json};
23
24 const TIMEOUT: Duration = Duration::from_secs(60);
25
26 // ── Scripted OpenAI-compatible endpoint ─────────────────────────────────────
27
28 fn sse_body(events: &[Value]) -> String {
29 let mut body = String::new();
30 for event in events {
31 body.push_str("data: ");
32 body.push_str(&event.to_string());
33 body.push_str("\n\n");
34 }
35 body.push_str("data: [DONE]\n\n");
36 body
37 }
38
39 fn sse_text(text: &str) -> String {
40 sse_body(&[json!({"choices": [{"delta": {"content": text}}]})])
41 }
42
43 fn sse_text_then_tool_call(text: &str, id: &str, name: &str, arguments: &str) -> String {
44 sse_body(&[
45 json!({"choices": [{"delta": {"content": text}}]}),
46 json!({
47 "choices": [{"delta": {"tool_calls": [{
48 "index": 0,
49 "id": id,
50 "function": {"name": name, "arguments": arguments},
51 }]}}]
52 }),
53 ])
54 }
55
56 /// Serves one scripted SSE response per request.
57 fn start_fake_endpoint(responses: Vec<String>) -> u16 {
58 let listener = TcpListener::bind("127.0.0.1:0").expect("bind fake endpoint");
59 let port = listener.local_addr().unwrap().port();
60 let queue = Mutex::new(VecDeque::from(responses));
61
62 std::thread::spawn(move || {
63 for stream in listener.incoming() {
64 let Ok(mut stream) = stream else { continue };
65 let mut reader = BufReader::new(match stream.try_clone() {
66 Ok(clone) => clone,
67 Err(_) => continue,
68 });
69 let mut content_length = 0usize;
70 loop {
71 let mut line = String::new();
72 if reader.read_line(&mut line).unwrap_or(0) == 0 {
73 break;
74 }
75 let line = line.trim();
76 if line.is_empty() {
77 break;
78 }
79 if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
80 content_length = length.trim().parse().unwrap_or(0);
81 }
82 }
83 let mut body = vec![0u8; content_length];
84 if reader.read_exact(&mut body).is_err() {
85 continue;
86 }
87 let payload = queue
88 .lock()
89 .unwrap()
90 .pop_front()
91 .unwrap_or_else(|| "data: [DONE]\n\n".to_string());
92 let response = format!(
93 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\
94 content-length: {}\r\nconnection: close\r\n\r\n{}",
95 payload.len(),
96 payload
97 );
98 let _ = stream.write_all(response.as_bytes());
99 }
100 });
101
102 port
103 }
104
105 // ── The agent under test ────────────────────────────────────────────────────
106
107 struct AgentUnderTest {
108 child: Child,
109 stdin: ChildStdin,
110 incoming: Receiver<Value>,
111 next_id: u64,
112 }
113
114 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {
115 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
116 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))
117 .env("OPENAI_API_KEY", "test-key")
118 .env("SIGIT_MODEL", "scripted-model")
119 .env("SIGIT_CONFIG_DIR", config_dir)
120 .env("SIGIT_MCP", "off")
121 .env("SIGIT_PERMISSIONS", "allow")
122 .env_remove("SIGIT_LOCAL_INFERENCE")
123 .stdin(Stdio::piped())
124 .stdout(Stdio::piped())
125 .stderr(Stdio::null())
126 .spawn()
127 .expect("spawn sigit in ACP mode");
128
129 let stdout = child.stdout.take().unwrap();
130 let (message_tx, incoming) = channel();
131 std::thread::spawn(move || {
132 for line in BufReader::new(stdout).lines() {
133 let Ok(line) = line else { break };
134 if let Ok(message) = serde_json::from_str::<Value>(&line)
135 && message_tx.send(message).is_err()
136 {
137 break;
138 }
139 }
140 });
141
142 let stdin = child.stdin.take().unwrap();
143 AgentUnderTest {
144 child,
145 stdin,
146 incoming,
147 next_id: 0,
148 }
149 }
150
151 impl AgentUnderTest {
152 fn request(&mut self, method: &str, params: Value) -> u64 {
153 self.next_id += 1;
154 let id = self.next_id;
155 let mut line =
156 json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
157 line.push('\n');
158 self.stdin
159 .write_all(line.as_bytes())
160 .expect("write to agent stdin");
161 self.stdin.flush().expect("flush agent stdin");
162 id
163 }
164
165 /// The response to one of our requests, plus every `session/update` that
166 /// arrived before it — which for `session/load` is the whole replay.
167 fn wait_for_response_with_updates(&mut self, id: u64) -> (Value, Vec<Value>) {
168 let mut updates = Vec::new();
169 let deadline = Instant::now() + TIMEOUT;
170 loop {
171 let remaining = deadline.saturating_duration_since(Instant::now());
172 let Ok(message) = self.incoming.recv_timeout(remaining) else {
173 panic!("timed out waiting for the response to request {id}");
174 };
175 if message["method"] == "session/update" {
176 updates.push(message["params"]["update"].clone());
177 }
178 if message["id"] == id && message.get("method").is_none() {
179 assert!(
180 message.get("error").is_none(),
181 "request {id} failed: {message}"
182 );
183 return (message, updates);
184 }
185 }
186 }
187 }
188
189 impl AgentUnderTest {
190 /// The raw response to one of our requests, error or not.
191 fn wait_for_raw_response(&mut self, id: u64) -> Value {
192 let deadline = Instant::now() + TIMEOUT;
193 loop {
194 let remaining = deadline.saturating_duration_since(Instant::now());
195 let Ok(message) = self.incoming.recv_timeout(remaining) else {
196 panic!("timed out waiting for the response to request {id}");
197 };
198 if message["id"] == id && message.get("method").is_none() {
199 return message;
200 }
201 }
202 }
203 }
204
205 impl Drop for AgentUnderTest {
206 fn drop(&mut self) {
207 let _ = self.child.kill();
208 let _ = self.child.wait();
209 }
210 }
211
212 fn text_of(update: &Value) -> &str {
213 update["content"]["text"].as_str().unwrap_or_default()
214 }
215
216 #[test]
217 fn loading_a_saved_session_replays_it_to_the_client() {
218 let scratch = std::env::temp_dir().join(format!("sigit_acp_load_{}", std::process::id()));
219 let config_dir = scratch.join("config");
220 let project = scratch.join("project");
221 std::fs::create_dir_all(&config_dir).unwrap();
222 std::fs::create_dir_all(&project).unwrap();
223 let notes = project.join("notes.md");
224 std::fs::write(&notes, "hello from the notes\n").unwrap();
225
226 // Round 1 talks and reads that file; round 2 answers.
227 let port = start_fake_endpoint(vec![
228 sse_text_then_tool_call(
229 "Let me look at the notes.",
230 "call_1",
231 "read_file",
232 &json!({"path": notes.to_string_lossy()}).to_string(),
233 ),
234 sse_text("The notes say hello."),
235 ]);
236
237 let mut agent = spawn_agent(port, &config_dir);
238
239 let id = agent.request(
240 "initialize",
241 json!({"protocolVersion": 1, "clientCapabilities": {}}),
242 );
243 let (initialize, _) = agent.wait_for_response_with_updates(id);
244 assert_eq!(
245 initialize["result"]["agentCapabilities"]["loadSession"], true,
246 "a client only offers to reopen threads for an agent that advertises loadSession: \
247 {initialize}"
248 );
249
250 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
251 let (new_session, _) = agent.wait_for_response_with_updates(id);
252 let session_id = new_session["result"]["sessionId"]
253 .as_str()
254 .expect("session id")
255 .to_string();
256
257 let id = agent.request(
258 "session/prompt",
259 json!({
260 "sessionId": session_id,
261 "prompt": [{"type": "text", "text": "what do the notes say?"}],
262 }),
263 );
264 agent.wait_for_response_with_updates(id);
265
266 // Reopen it, the way clicking the thread in the editor's history does.
267 let id = agent.request(
268 "session/load",
269 json!({"sessionId": session_id, "cwd": project, "mcpServers": []}),
270 );
271 let (_, replay) = agent.wait_for_response_with_updates(id);
272
273 let user: Vec<&str> = replay
274 .iter()
275 .filter(|update| update["sessionUpdate"] == "user_message_chunk")
276 .map(text_of)
277 .collect();
278 assert_eq!(
279 user,
280 vec!["what do the notes say?"],
281 "the user's turn must come back: {replay:#?}"
282 );
283
284 let agent_text: String = replay
285 .iter()
286 .filter(|update| update["sessionUpdate"] == "agent_message_chunk")
287 .map(text_of)
288 .collect();
289 assert!(
290 agent_text.contains("Let me look at the notes.")
291 && agent_text.contains("The notes say hello."),
292 "both rounds of the reply must come back, got {agent_text:?}"
293 );
294
295 let tool_call = replay
296 .iter()
297 .find(|update| update["sessionUpdate"] == "tool_call")
298 .unwrap_or_else(|| panic!("the tool call must come back: {replay:#?}"));
299 // The full path can run past the title's truncation cap under a long
300 // scratch dir, so check the readable prefix rather than an exact match.
301 assert!(
302 tool_call["title"]
303 .as_str()
304 .unwrap_or_default()
305 .starts_with("read_file · "),
306 "title should read '<tool> · <path>': {tool_call}"
307 );
308 // Nothing is still running in a saved session, and the result is folded
309 // back into its call rather than replayed as a loose message.
310 assert_eq!(tool_call["status"], "completed");
311 assert!(
312 tool_call["rawOutput"]
313 .as_str()
314 .unwrap_or_default()
315 .contains("hello from the notes"),
316 "the tool result must ride along with its call: {tool_call}"
317 );
318 // The card must be expandable on replay, not just carry raw_output —
319 // that's the point of issue #78.
320 assert!(
321 format!("{:?}", tool_call["content"]).contains("hello from the notes"),
322 "the replayed card must show its output as content: {tool_call}"
323 );
324
325 // The seeded system context is the model's, not the user's — it must never
326 // surface as a message in the thread.
327 for update in &replay {
328 assert!(
329 !text_of(update).contains("project working directory"),
330 "system context leaked into the replay: {update}"
331 );
332 }
333
334 drop(agent);
335 let _ = std::fs::remove_dir_all(&scratch);
336 }
337
338 /// A session id that was never saved and that this process has never opened
339 /// names nothing. Answering with an empty session would leave the client
340 /// showing a blank thread under the old title (issue #137).
341 #[test]
342 fn loading_an_unknown_session_is_an_error() {
343 let scratch =
344 std::env::temp_dir().join(format!("sigit_acp_load_unknown_{}", std::process::id()));
345 let config_dir = scratch.join("config");
346 let project = scratch.join("project");
347 std::fs::create_dir_all(&config_dir).unwrap();
348 std::fs::create_dir_all(&project).unwrap();
349
350 let port = start_fake_endpoint(Vec::new());
351 let mut agent = spawn_agent(port, &config_dir);
352
353 let id = agent.request(
354 "initialize",
355 json!({"protocolVersion": 1, "clientCapabilities": {}}),
356 );
357 agent.wait_for_response_with_updates(id);
358
359 let id = agent.request(
360 "session/load",
361 json!({"sessionId": "does-not-exist", "cwd": project, "mcpServers": []}),
362 );
363 let response = agent.wait_for_raw_response(id);
364 assert_eq!(
365 response["error"]["code"], -32002,
366 "an unknown id must be reported as not found: {response}"
367 );
368
369 // The failed load must not have registered the id as a side effect.
370 let id = agent.request(
371 "session/prompt",
372 json!({
373 "sessionId": "does-not-exist",
374 "prompt": [{"type": "text", "text": "/help"}],
375 }),
376 );
377 let response = agent.wait_for_raw_response(id);
378 assert!(
379 response.get("error").is_some(),
380 "the id must still be unknown after a failed load: {response}"
381 );
382
383 // A thread opened in this process but never spoken in has nothing on disk
384 // yet. It is still a session, so loading it keeps working.
385 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
386 let (new_session, _) = agent.wait_for_response_with_updates(id);
387 let session_id = new_session["result"]["sessionId"]
388 .as_str()
389 .expect("session id")
390 .to_string();
391 let id = agent.request(
392 "session/load",
393 json!({"sessionId": session_id, "cwd": project, "mcpServers": []}),
394 );
395 let response = agent.wait_for_raw_response(id);
396 assert!(
397 response.get("error").is_none(),
398 "a session this process holds must load without a saved copy: {response}"
399 );
400
401 drop(agent);
402 let _ = std::fs::remove_dir_all(&scratch);
403 }