@sigitcodereview / sigit
Code Commits Issues Pulls Cloud Agent CI/CD
fix/acp-replay-loaded-session-history
claude/github-issue-40-0au5ua claude/linked-issue-acp-tui-u4vvpa claude/minor-release-cut-vkugow claude/sigit-acp-local-chat-cx6380 claude/sigit-cloud-agent-expansion-reox0a claude/tool-permission-system claude/zen-feynman-0u78dk development feature/acp-cancel-request feature/acp-session-list-cursor feature/acp-v1-conformance feature/agent-tools-multiedit-glob-todos-remember feature/cargo-workspace feature/hooks feature/load-local-model-explicitly feature/sigit-agent-profile feature/sigit-code-cloud feature/xcode-mcp-server fix/acp-replay-loaded-session-history fix/acp-replay-reasoning-and-attachments fix/dsml-doubled-bar-markers main release/v1.5.10 release/v1.5.6 release/v1.6.1
rs 306 lines 10.5 KB
Raw
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 Drop for AgentUnderTest {
190 fn drop(&mut self) {
191 let _ = self.child.kill();
192 let _ = self.child.wait();
193 }
194 }
195
196 fn text_of(update: &Value) -> &str {
197 update["content"]["text"].as_str().unwrap_or_default()
198 }
199
200 #[test]
201 fn loading_a_saved_session_replays_it_to_the_client() {
202 let scratch = std::env::temp_dir().join(format!("sigit_acp_load_{}", std::process::id()));
203 let config_dir = scratch.join("config");
204 let project = scratch.join("project");
205 std::fs::create_dir_all(&config_dir).unwrap();
206 std::fs::create_dir_all(&project).unwrap();
207 let notes = project.join("notes.md");
208 std::fs::write(&notes, "hello from the notes\n").unwrap();
209
210 // Round 1 talks and reads that file; round 2 answers.
211 let port = start_fake_endpoint(vec![
212 sse_text_then_tool_call(
213 "Let me look at the notes.",
214 "call_1",
215 "read_file",
216 &json!({"path": notes.to_string_lossy()}).to_string(),
217 ),
218 sse_text("The notes say hello."),
219 ]);
220
221 let mut agent = spawn_agent(port, &config_dir);
222
223 let id = agent.request(
224 "initialize",
225 json!({"protocolVersion": 1, "clientCapabilities": {}}),
226 );
227 let (initialize, _) = agent.wait_for_response_with_updates(id);
228 assert_eq!(
229 initialize["result"]["agentCapabilities"]["loadSession"], true,
230 "a client only offers to reopen threads for an agent that advertises loadSession: \
231 {initialize}"
232 );
233
234 let id = agent.request("session/new", json!({"cwd": project, "mcpServers": []}));
235 let (new_session, _) = agent.wait_for_response_with_updates(id);
236 let session_id = new_session["result"]["sessionId"]
237 .as_str()
238 .expect("session id")
239 .to_string();
240
241 let id = agent.request(
242 "session/prompt",
243 json!({
244 "sessionId": session_id,
245 "prompt": [{"type": "text", "text": "what do the notes say?"}],
246 }),
247 );
248 agent.wait_for_response_with_updates(id);
249
250 // Reopen it, the way clicking the thread in the editor's history does.
251 let id = agent.request(
252 "session/load",
253 json!({"sessionId": session_id, "cwd": project, "mcpServers": []}),
254 );
255 let (_, replay) = agent.wait_for_response_with_updates(id);
256
257 let user: Vec<&str> = replay
258 .iter()
259 .filter(|update| update["sessionUpdate"] == "user_message_chunk")
260 .map(text_of)
261 .collect();
262 assert_eq!(
263 user,
264 vec!["what do the notes say?"],
265 "the user's turn must come back: {replay:#?}"
266 );
267
268 let agent_text: String = replay
269 .iter()
270 .filter(|update| update["sessionUpdate"] == "agent_message_chunk")
271 .map(text_of)
272 .collect();
273 assert!(
274 agent_text.contains("Let me look at the notes.")
275 && agent_text.contains("The notes say hello."),
276 "both rounds of the reply must come back, got {agent_text:?}"
277 );
278
279 let tool_call = replay
280 .iter()
281 .find(|update| update["sessionUpdate"] == "tool_call")
282 .unwrap_or_else(|| panic!("the tool call must come back: {replay:#?}"));
283 assert_eq!(tool_call["title"], "read_file");
284 // Nothing is still running in a saved session, and the result is folded
285 // back into its call rather than replayed as a loose message.
286 assert_eq!(tool_call["status"], "completed");
287 assert!(
288 tool_call["rawOutput"]
289 .as_str()
290 .unwrap_or_default()
291 .contains("hello from the notes"),
292 "the tool result must ride along with its call: {tool_call}"
293 );
294
295 // The seeded system context is the model's, not the user's — it must never
296 // surface as a message in the thread.
297 for update in &replay {
298 assert!(
299 !text_of(update).contains("project working directory"),
300 "system context leaked into the replay: {update}"
301 );
302 }
303
304 drop(agent);
305 let _ = std::fs::remove_dir_all(&scratch);
306 }