@sigitcodereview / sigit / commits / 779db22

Replay a loaded session so the thread comes back

Clicking a saved thread in Zed sends session/load, and the client draws the reopened thread purely from the session/update notifications the agent streams while that request is in flight. We restored the saved history into the backend, which is what makes the model remember, but sent the client nothing, so the thread opened empty and looked like a brand new conversation. The handler now turns the saved snapshot into updates before restoring it. User and assistant text come back as message chunks with reasoning stripped, each tool call comes back completed with its result folded into it, and write_todos comes back as a plan the way it renders live. System messages stay out: they seeded the model and were never on screen. Fixes #77 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Seto Elkahfi committed Sep 8, 2026 at 21:13 UTC 779db220678eae90def0bbda005775d5d7a86846

2 files changed +533 -1

src/main.rs

+227 -1

@@ -349,6 +349,107 @@ fn tool_kind_for(tool_name: &str) -> ToolKind {

349 }

350 }

351

352 +/// The visible text of one saved history message. `content` is normally a

353 +/// string, but an OpenAI-compatible endpoint may hand back the block form, and

354 +/// tool-call-only assistant turns carry `null`.

355 +fn history_message_text(message: &serde_json::Value) -> String {

356 + match &message["content"] {

357 + serde_json::Value::String(text) => text.clone(),

358 + serde_json::Value::Array(blocks) => blocks

359 + .iter()

360 + .filter_map(|block| block["text"].as_str())

361 + .collect::<Vec<_>>()

362 + .join(""),

363 + _ => String::new(),

364 + }

365 +}

366 +

367 +/// Rebuild the `session/update` notifications that re-draw a saved conversation

368 +/// in the client.

369 +///

370 +/// ACP's `session/load` is a replay: the client renders the reopened thread

371 +/// purely from the updates the agent streams while the request is in flight.

372 +/// Restoring history into the backend is what makes the *model* remember, but

373 +/// it puts nothing on screen, so a reopened session came up blank and looked

374 +/// like a brand-new one (issue #77).

375 +fn history_replay_updates(history: &[serde_json::Value]) -> Vec<SessionUpdate> {

376 + // Match results to their calls up front so each call replays as one

377 + // finished tool call instead of a call followed by a loose result.

378 + let mut outputs: std::collections::HashMap<&str, &str> = std::collections::HashMap::new();

379 + for message in history {

380 + if message["role"] == "tool"

381 + && let (Some(id), Some(content)) = (

382 + message["tool_call_id"].as_str(),

383 + message["content"].as_str(),

384 + )

385 + {

386 + outputs.insert(id, content);

387 + }

388 + }

389 +

390 + let mut updates = Vec::new();

391 + for message in history {

392 + match message["role"].as_str().unwrap_or_default() {

393 + // Seeded context and project instructions were never on screen.

394 + "system" => {}

395 + "user" => {

396 + let text = history_message_text(message);

397 + if !text.trim().is_empty() {

398 + updates.push(SessionUpdate::UserMessageChunk(ContentChunk::new(

399 + ContentBlock::from(text),

400 + )));

401 + }

402 + }

403 + "assistant" => {

404 + let (_think, visible) = chat::strip_think_blocks(&history_message_text(message));

405 + if !visible.trim().is_empty() {

406 + updates.push(SessionUpdate::AgentMessageChunk(ContentChunk::new(

407 + ContentBlock::from(visible),

408 + )));

409 + }

410 + for call in message["tool_calls"].as_array().into_iter().flatten() {

411 + let name = call["function"]["name"].as_str().unwrap_or("tool");

412 + let arguments = call["function"]["arguments"].as_str().unwrap_or("");

413 +

414 + // `write_todos` renders as a plan while it runs; replay it

415 + // the same way, and let a later list replace an earlier one

416 + // exactly as it did live.

417 + if name == "write_todos"

418 + && let Some(plan) = todos_arguments_to_plan(arguments)

419 + {

420 + updates.push(SessionUpdate::Plan(plan));

421 + continue;

422 + }

423 +

424 + // A saved call is always finished — nothing is still

425 + // running when the session file is written.

426 + let id = call["id"].as_str().unwrap_or_default();

427 + // A call saved without an id would otherwise collapse into

428 + // one entry per replay; give each its own.

429 + let replay_id = if id.is_empty() {

430 + format!("replay-{}", uuid::Uuid::new_v4())

431 + } else {

432 + id.to_string()

433 + };

434 + let mut tool_call = ToolCall::new(replay_id, name)

435 + .kind(tool_kind_for(name))

436 + .status(ToolCallStatus::Completed)

437 + .raw_input(serde_json::from_str::<serde_json::Value>(arguments).ok());

438 + if let Some(output) = outputs.get(id) {

439 + tool_call =

440 + tool_call.raw_output(serde_json::Value::String((*output).to_string()));

441 + }

442 + updates.push(SessionUpdate::ToolCall(tool_call));

443 + }

444 + }

445 + // Folded into the tool call above.

446 + "tool" => {}

447 + other => log::debug!("session replay: skipping unknown history role '{other}'"),

448 + }

449 + }

450 + updates

451 +}

452 +

453 fn todos_arguments_to_plan(arguments: &str) -> Option<Plan> {

454 let args: serde_json::Value = serde_json::from_str(arguments).ok()?;

455 let todos = args.get("todos")?.as_array()?;

@@ -1273,10 +1374,21 @@ impl SiGitAgent {

1374 // freshly seeded state wholesale.

1375 if let Some(history) = session_store::load(&args.session_id.to_string()) {

1376 let restored = history.len();

1377 +

1378 + // Replay before restoring: the client draws the reopened thread

1379 + // from these notifications alone, and `restore_history` consumes

1380 + // the snapshot.

1381 + let updates = history_replay_updates(&history);

1382 + let replayed = updates.len();

1383 + for update in updates {

1384 + cx.send_notification(SessionNotification::new(args.session_id.clone(), update))

1385 + .ok();

1386 + }

1387 +

1388 let backend = self.backend.lock().await.clone();

1389 backend.restore_history(history).await;

1390 log::info!(

1279 - "load_session: restored {restored} message(s) for {}",

1391 + "load_session: restored {restored} message(s), replayed {replayed} update(s) for {}",

1392 args.session_id

1393 );

1394 }

@@ -3854,6 +3966,120 @@ mod tests {

3966 assert!(multi.contains("absolute paths"));

3967 }

3968

3969 + #[test]

3970 + fn history_replay_redraws_the_conversation_for_the_client() {

3971 + let history = vec![

3972 + serde_json::json!({ "role": "system", "content": "project context" }),

3973 + serde_json::json!({ "role": "user", "content": "read the changelog" }),

3974 + serde_json::json!({

3975 + "role": "assistant",

3976 + "content": "<think>which file?</think>Reading it now.",

3977 + "tool_calls": [{

3978 + "id": "call_1", "type": "function",

3979 + "function": { "name": "read_file", "arguments": "{\"path\":\"/tmp/CHANGELOG.md\"}" },

3980 + }],

3981 + }),

3982 + serde_json::json!({ "role": "tool", "tool_call_id": "call_1", "content": "# Changelog" }),

3983 + serde_json::json!({ "role": "assistant", "content": "It starts at v1.0." }),

3984 + ];

3985 +

3986 + let updates = history_replay_updates(&history);

3987 +

3988 + // The system message seeded the model; it was never on screen.

3989 + assert_eq!(updates.len(), 4, "{updates:#?}");

3990 +

3991 + match &updates[0] {

3992 + SessionUpdate::UserMessageChunk(chunk) => {

3993 + assert!(format!("{:?}", chunk.content).contains("read the changelog"));

3994 + }

3995 + other => panic!("expected the user message first, got {other:?}"),

3996 + }

3997 + match &updates[1] {

3998 + SessionUpdate::AgentMessageChunk(chunk) => {

3999 + let rendered = format!("{:?}", chunk.content);

4000 + assert!(rendered.contains("Reading it now."));

4001 + // Reasoning stays hidden on replay, exactly as it did live.

4002 + assert!(!rendered.contains("which file?"));

4003 + }

4004 + other => panic!("expected the assistant reply, got {other:?}"),

4005 + }

4006 + match &updates[2] {

4007 + SessionUpdate::ToolCall(call) => {

4008 + assert_eq!(call.title, "read_file");

4009 + assert_eq!(call.kind, ToolKind::Read);

4010 + // Nothing is still running in a saved session.

4011 + assert_eq!(call.status, ToolCallStatus::Completed);

4012 + assert_eq!(

4013 + call.raw_input,

4014 + Some(serde_json::json!({"path": "/tmp/CHANGELOG.md"}))

4015 + );

4016 + // The result is folded into its call rather than replayed loose.

4017 + assert_eq!(

4018 + call.raw_output,

4019 + Some(serde_json::Value::String("# Changelog".to_string()))

4020 + );

4021 + }

4022 + other => panic!("expected the tool call, got {other:?}"),

4023 + }

4024 + assert!(matches!(updates[3], SessionUpdate::AgentMessageChunk(_)));

4025 + }

4026 +

4027 + #[test]

4028 + fn history_replay_renders_write_todos_as_a_plan() {

4029 + let history = vec![serde_json::json!({

4030 + "role": "assistant",

4031 + "content": null,

4032 + "tool_calls": [{

4033 + "id": "call_1", "type": "function",

4034 + "function": {

4035 + "name": "write_todos",

4036 + "arguments": "{\"todos\":[{\"content\":\"Ship it\",\"status\":\"completed\"}]}",

4037 + },

4038 + }],

4039 + })];

4040 +

4041 + match history_replay_updates(&history).as_slice() {

4042 + [SessionUpdate::Plan(plan)] => {

4043 + assert_eq!(plan.entries.len(), 1);

4044 + assert_eq!(plan.entries[0].content, "Ship it");

4045 + assert_eq!(plan.entries[0].status, PlanEntryStatus::Completed);

4046 + }

4047 + other => panic!("expected one plan update, got {other:?}"),

4048 + }

4049 + }

4050 +

4051 + #[test]

4052 + fn history_replay_skips_messages_with_nothing_to_show() {

4053 + // A tool-call-only assistant turn (`content: null`), an empty reply and

4054 + // a blank user message must not become empty bubbles in the client.

4055 + let history = vec![

4056 + serde_json::json!({ "role": "user", "content": " " }),

4057 + serde_json::json!({ "role": "assistant", "content": null }),

4058 + serde_json::json!({ "role": "assistant", "content": "" }),

4059 + ];

4060 + assert!(history_replay_updates(&history).is_empty());

4061 + }

4062 +

4063 + #[test]

4064 + fn history_replay_reads_block_form_content() {

4065 + // Some OpenAI-compatible endpoints hand back content blocks rather than

4066 + // a plain string; both reach the session file.

4067 + let history = vec![serde_json::json!({

4068 + "role": "user",

4069 + "content": [

4070 + { "type": "text", "text": "check " },

4071 + { "type": "text", "text": "this file" },

4072 + ],

4073 + })];

4074 +

4075 + match history_replay_updates(&history).as_slice() {

4076 + [SessionUpdate::UserMessageChunk(chunk)] => {

4077 + assert!(format!("{:?}", chunk.content).contains("check this file"));

4078 + }

4079 + other => panic!("expected one user message, got {other:?}"),

4080 + }

4081 + }

4082 +

4083 #[test]

4084 fn streamed_reply_sends_each_fragment_as_it_is_revealed() {

4085 let mut reply = StreamedReply::default();

tests/acp_session_load.rs new

+306

@@ -0,0 +1,306 @@

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 +}