From bd8a161c2efe91a15933214bc7f0dd0bda8900a7 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Fri, 2 Oct 2026 09:37:07 -0700 Subject: [PATCH] refactor: replay session history on the experimental ACP v2 surface (#1237) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Last part of step 4's session baseline for opt-in ACP v2 support (after #1231–#1235): a v2 `session/resume` can now replay the session's history. Nothing changes on ACP v1. ## Change - **Routing.** v2 restores a session with `session/resume` and replays only when `replayFrom` asks for it. v1 splits that into `session/resume`, which never replays, and `session/load`, which replays everything before it responds. That's exactly what v2 requires of `replayFrom: { type: "start" }`. So `v1RestoreSessionRequest` maps the v2 request to one of the two. A cursor that starts elsewhere is rejected, since v1 can only replay from the start. - **Clearing replayed messages.** The agent replays messages as chunks, and chunks append. A v2 client may still hold a replayed message, so before the first replayed chunk of each message, the v2 surface sends that message with `content: []`. The draft requires that of a message replayed as chunks. It happens only while that session replays (`V2ClientConnection.replaying`), and the clear and the chunk go to the connection together, so nothing comes between them. - **Prompt ids.** The draft requires a prompt's user message to replay under the id its `session/prompt` response returned. Since #1226 that id is the user message's uuid, which Claude Code keeps in the transcript, so no new code is needed. A test checks it end to end. ## Scenario suite `session-load-replay` passes, so every scenario now passes on v2. The status table becomes `V2_STOPS`: the scenarios that stop on v2, which is empty today and stays as the place for a scenario v2 can't run yet. The replay golden file shows each message cleared before its chunks, and the `session/resume` response after the last replayed update. A replayed Edit or Write shows no diff yet on v2: replay has no PostToolUse hook. Building it from the transcript is the next PR. ## Verification - **Replay test:** a live prompt, then resume with replay from a transcript holding the uuid Claude Code saved. The replayed user message carries the prompt response's id, each message is cleared before its chunks, everything arrives before the response, and a live prompt afterwards gets no clears. - **Mutations:** each rule fails a test when removed: no clears, clears that stay on after the replay, a replay served as a plain resume, and one clear per id rather than per message. - Build, lint and all 2385 tests pass. --- src/tests/acp-scenarios-v2.test.ts | 68 +++----------- .../v2/session-load-replay.jsonl | 17 ++++ src/tests/acp-v2.test.ts | 80 ++++++++++++++++- src/v2/agent.ts | 90 ++++++++++++++++--- src/v2/session.ts | 31 +++++-- 5 files changed, 211 insertions(+), 75 deletions(-) create mode 100644 src/tests/acp-scenarios/__snapshots__/v2/session-load-replay.jsonl diff --git a/src/tests/acp-scenarios-v2.test.ts b/src/tests/acp-scenarios-v2.test.ts index a1e76cba..8488abe9 100644 --- a/src/tests/acp-scenarios-v2.test.ts +++ b/src/tests/acp-scenarios-v2.test.ts @@ -2,10 +2,9 @@ * Every scenario of `acp-scenarios/scenarios.ts` on the experimental ACP v2 * surface: through the protocol router, with the SDK's v2 client. * - * - Status: each scenario either passes, every prompt turn ending with a stop - * reason, or stops for the reason in {@link V2_STATUS}. A step of the v2 - * support that translates more of the agent's messages turns scenarios from - * stopped into passing, and updates the table. + * - Status: each scenario passes, every prompt turn ending with a stop reason, + * unless {@link V2_STOPS} names why it stops. Every scenario passes today; + * a scenario that the v2 surface cannot run yet goes in the table. * - Schema: every message the agent sends, also in a scenario that stops, is * valid against the draft v2 schema of the SDK. * - Tool calls: the first report of each tool call has a title. v2 has no @@ -59,55 +58,14 @@ vi.mock("node:crypto", async (importOriginal) => { const here = path.dirname(fileURLToPath(import.meta.url)); const PASSES = "passes"; -/** Step 4 of `docs/acp-v2.md`. */ -const REPLAY = "Invalid params: The ACP v2 surface does not replay session history yet"; /** - * How each scenario runs on the v2 surface: it passes, or stops for a reason - * that names the step of `docs/acp-v2.md` that removes it. + * The scenarios that stop on the v2 surface, each with the reason, which names + * the step of `docs/acp-v2.md` that removes it. Every other scenario passes. */ -const V2_STATUS: Record = { - "session-setup": PASSES, - "text-and-thinking": PASSES, - "bash-foreground": PASSES, - "bash-error": PASSES, - "bash-background": PASSES, - "async-task-held-until-tool-id": PASSES, - "async-task-ends-before-tool-id": PASSES, - "async-task-released-at-turn-end": PASSES, - read: PASSES, - "write-new": PASSES, - "write-existing": PASSES, - "edit-with-permission": PASSES, - "edit-rejected": PASSES, - "notebook-edit": PASSES, - "grep-and-glob": PASSES, - "web-fetch-and-search": PASSES, - "subagent-task-legacy": PASSES, - "subagent-agent-async": PASSES, - "subagent-native-sessions": PASSES, - "subagent-nested": PASSES, - "subagent-late-child-update": PASSES, - "subagent-transcript-extension": PASSES, - "todo-write": PASSES, - "task-create-update": PASSES, - goal: PASSES, - "network-permission": PASSES, - "exit-plan-approve": PASSES, - "exit-plan-reject": PASSES, - "ask-user-question": PASSES, - skill: PASSES, - "mcp-tool": PASSES, - "task-output-and-stop": PASSES, - "memory-recall": PASSES, - "permission-denied": PASSES, - "tool-progress": PASSES, - "rate-limit-and-origin": PASSES, - "compaction-legacy": PASSES, - "compaction-update": PASSES, - "compaction-failed-legacy": PASSES, - "session-load-replay": REPLAY, -}; +const V2_STOPS: Record = {}; + +const statusOf = (scenario: string) => V2_STOPS[scenario] ?? PASSES; const runs = new Map(); let configDir: string; @@ -156,10 +114,10 @@ function toJsonLines(recorded: WireRecorded[]): string { } describe("ACP v2 scenarios", () => { - it("run as the status table says", () => { - expect( - Object.fromEntries(SCENARIOS.map((scenario) => [scenario.name, run(scenario.name).status])), - ).toEqual(V2_STATUS); + it("pass, or stop as the table says", () => { + const table = (status: (scenario: string) => string) => + Object.fromEntries(SCENARIOS.map((scenario) => [scenario.name, status(scenario.name)])); + expect(table((scenario) => run(scenario).status)).toEqual(table(statusOf)); }); describe("schema check", () => { @@ -202,7 +160,7 @@ describe("ACP v2 scenarios", () => { describe("v2 golden files", () => { const passing = SCENARIOS.map((scenario) => scenario.name).filter( - (scenario) => V2_STATUS[scenario] === PASSES, + (scenario) => statusOf(scenario) === PASSES, ); it.each(passing)("%s", async (scenario) => { await expect(toJsonLines(run(scenario).normalized)).toMatchFileSnapshot( diff --git a/src/tests/acp-scenarios/__snapshots__/v2/session-load-replay.jsonl b/src/tests/acp-scenarios/__snapshots__/v2/session-load-replay.jsonl new file mode 100644 index 00000000..c0cbb3ad --- /dev/null +++ b/src/tests/acp-scenarios/__snapshots__/v2/session-load-replay.jsonl @@ -0,0 +1,17 @@ +{"kind":"response","method":"initialize","payload":{"_meta":{"steering":{"supported":true}},"authMethods":[],"capabilities":{"_meta":{"authStatus":{},"claudeCode":{"promptQueueing":true}}},"info":{"name":"@agentclientprotocol/claude-agent-acp","title":"Claude Agent","version":""},"protocolVersion":2}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"content":[],"messageId":"00000000-0000-4000-8000-00000000c001","sessionUpdate":"user_message"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"content":{"text":"Fix the bug","type":"text"},"messageId":"00000000-0000-4000-8000-00000000c001","sessionUpdate":"user_message_chunk"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"content":[],"messageId":"msg_r1","sessionUpdate":"agent_thought"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"content":{"text":"Check the file.","type":"text"},"messageId":"msg_r1","sessionUpdate":"agent_thought_chunk"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"content":[],"messageId":"msg_r1","sessionUpdate":"agent_message"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"content":{"text":"I will edit it.","type":"text"},"messageId":"msg_r1","sessionUpdate":"agent_message_chunk"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"_meta":{"claudeCode":{"toolName":"Bash"}},"content":[{"content":{"text":"List","type":"text"},"type":"content"}],"kind":"execute","name":"Bash","rawInput":{"command":"ls","description":"List"},"sessionUpdate":"tool_call_update","status":"pending","title":"ls","toolCallId":"toolu_r_bash"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"_meta":{"claudeCode":{"toolName":"Edit"}},"content":[],"kind":"edit","locations":[{"path":"/src/app.ts"}],"name":"Edit","rawInput":{"file_path":"/src/app.ts","new_string":"b","old_string":"a"},"sessionUpdate":"tool_call_update","status":"pending","title":"Edit src/app.ts","toolCallId":"toolu_r_edit"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"_meta":{"claudeCode":{"toolName":"Task"}},"content":[{"content":{"text":"Look around","type":"text"},"type":"content"}],"kind":"think","name":"Task","rawInput":{"description":"Explore","prompt":"Look around"},"sessionUpdate":"tool_call_update","status":"pending","title":"Explore","toolCallId":"toolu_r_task"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"_meta":{"claudeCode":{"toolName":"Bash"}},"content":[{"content":{"text":"```console\napp.ts\n```","type":"text"},"type":"content"}],"rawOutput":"app.ts","sessionUpdate":"tool_call_update","status":"completed","toolCallId":"toolu_r_bash"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"_meta":{"claudeCode":{"toolName":"Edit"}},"rawOutput":"The file /src/app.ts has been updated successfully.","sessionUpdate":"tool_call_update","status":"completed","toolCallId":"toolu_r_edit"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"_meta":{"claudeCode":{"toolName":"Task"}},"content":[{"content":{"text":"Report.","type":"text"},"type":"content"}],"rawOutput":"Report.","sessionUpdate":"tool_call_update","status":"completed","toolCallId":"toolu_r_task"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"content":[],"messageId":"msg_r2","sessionUpdate":"agent_message"}}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"content":{"text":"Fixed.","type":"text"},"messageId":"msg_r2","sessionUpdate":"agent_message_chunk"}}} +{"kind":"response","method":"session/resume","payload":{"configOptions":[{"category":"mode","configId":"mode","currentValue":"default","description":"Session permission mode","name":"Mode","options":[{"description":"Always ask before making changes","name":"Manual","value":"default"},{"description":"Automatically accept all file edits","name":"Accept edits","value":"acceptEdits"},{"description":"Create a plan before making changes","name":"Plan","value":"plan"},{"description":"Claude handles permission decisions","name":"Auto","value":"auto"},{"description":"Accepts all permissions","name":"Bypass permissions","value":"bypassPermissions"}],"type":"select"},{"category":"model","configId":"model","currentValue":"claude-sonnet-4-6","description":"AI model to use","name":"Model","options":[{"description":"Fast","name":"Claude Sonnet 4.6","value":"claude-sonnet-4-6"}],"type":"select"}]}} +{"kind":"notification","method":"session/update","payload":{"sessionId":"","update":{"availableCommands":[{"description":"Show the MCP servers and their status, or reconnect, enable, or disable a server","input":{"hint":"[reconnect|enable|disable [|all]]","type":"text"},"name":"mcp"}],"sessionUpdate":"available_commands_update"}}} diff --git a/src/tests/acp-v2.test.ts b/src/tests/acp-v2.test.ts index dd5db898..f88bd400 100644 --- a/src/tests/acp-v2.test.ts +++ b/src/tests/acp-v2.test.ts @@ -11,7 +11,11 @@ import * as path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import * as v1 from "@agentclientprotocol/sdk"; import * as v2 from "@agentclientprotocol/sdk/experimental/v2"; -import type { Options, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk"; +import { + getSessionMessages, + type Options, + type SDKUserMessage, +} from "@anthropic-ai/claude-agent-sdk"; import type { ClaudeAcpAgent } from "../acp-agent.js"; import type { AuthStatusUpdateNotification } from "../auth-status.js"; import { v2DiffContent } from "../diff.js"; @@ -363,11 +367,12 @@ describe("ACP v2 sessions", () => { expect(resumed).not.toHaveProperty("modes"); expect(resumed.configOptions?.[0].configId).toBe("mode"); + // v1 replays only from the start. await expect( agent.request(v2.methods.agent.session.resume, { sessionId, cwd, - replayFrom: { type: "start" }, + replayFrom: { type: "_message", messageId: "m" }, }), ).rejects.toMatchObject({ code: -32602 }); @@ -473,6 +478,11 @@ function scriptTurns(...turns: TurnScript[]): string[] { return echoed; } +/** An entry of a Claude Code transcript, as `getSessionMessages` returns it. */ +function transcriptEntry(type: "user" | "assistant", uuid: string, message: unknown) { + return { type, uuid, session_id: "session", parent_tool_use_id: null, message }; +} + function stream(options: Options, event: Record) { return { type: "stream_event", @@ -493,11 +503,11 @@ function idleState(options: Options) { }; } -function assistantText(options: Options, text: string) { +function assistantText(options: Options, text: string, id = "msg_answer") { return { type: "assistant", message: { - id: "msg_answer", + id, type: "message", role: "assistant", model: "claude-sonnet-4-6", @@ -730,6 +740,68 @@ describe("ACP v2 prompts", () => { ]); }); + it("replays the history before it answers session/resume, clearing each message first", async () => { + const echoed = scriptTurns( + async function* (options) { + yield assistantText(options, "Hello!"); + yield result(options); + }, + async function* (options) { + yield assistantText(options, "Again.", "msg_again"); + yield result(options); + }, + ); + const client = v2Client(); + await client.app.connectWith(connectRouter(), async (agent) => { + await initializeV2(agent); + const { sessionId } = await agent.request(v2.methods.agent.session.new, { cwd }); + const { messageId } = await agent.request(v2.methods.agent.session.prompt, { + sessionId, + prompt: text("hi"), + }); + await vi.waitFor(() => expect(turnTrace(client.sessionUpdates)).toContain("idle end_turn")); + + // Claude Code keeps the uuid of the prompt's user message in the transcript. + vi.mocked(getSessionMessages).mockResolvedValueOnce([ + transcriptEntry("user", echoed[0], { role: "user", content: text("hi") }), + transcriptEntry("assistant", randomUUID(), { + id: "msg_answer", + role: "assistant", + content: text("Hello!"), + }), + ] as unknown as Awaited>); + const before = client.sessionUpdates.length; + await agent.request(v2.methods.agent.session.resume, { + sessionId, + cwd, + replayFrom: { type: "start" }, + }); + const replayed = client.sessionUpdates.slice(before).map(({ update }) => update); + // The prompt's user message replays under the id that its response returned. + expect(replayed).toEqual([ + { sessionUpdate: "user_message", messageId, content: [] }, + { sessionUpdate: "user_message_chunk", messageId, content: text("hi")[0] }, + { sessionUpdate: "agent_message", messageId: "msg_answer", content: [] }, + { + sessionUpdate: "agent_message_chunk", + messageId: "msg_answer", + content: text("Hello!")[0], + }, + ]); + + // Live chunks after the replay get no clear. + const afterReplay = client.sessionUpdates.length; + await agent.request(v2.methods.agent.session.prompt, { sessionId, prompt: text("again") }); + await vi.waitFor(() => + expect(turnTrace(client.sessionUpdates.slice(afterReplay))).toContain("idle end_turn"), + ); + expect( + client.sessionUpdates.slice(afterReplay).map(({ update }) => update.sessionUpdate), + ).not.toContain("agent_message"); + await client.authUpdate(1); + }); + }); + it("answers a queued prompt once Claude Code takes it in, after the turn before it", async () => { const firstAnswer = Promise.withResolvers(); const echoed = scriptTurns( diff --git a/src/v2/agent.ts b/src/v2/agent.ts index f61af0fb..d6b9470d 100644 --- a/src/v2/agent.ts +++ b/src/v2/agent.ts @@ -22,7 +22,7 @@ import { v1PermissionResponse, v2PermissionRequest } from "./permission.js"; import { v2Prompt } from "./prompt.js"; import { v1NewSessionRequest, - v1ResumeSessionRequest, + v1RestoreSessionRequest, v1SetSessionConfigOptionRequest, v2ConfigOptions, v2NewSessionResponse, @@ -59,9 +59,14 @@ export function v2AgentApp( v2NewSessionResponse(await agent.newSession(v1NewSessionRequest(params))), ) .onRequest(v2.methods.agent.session.list, ({ params }) => agent.listSessions(params)) - .onRequest(v2.methods.agent.session.resume, async ({ params }) => - v2ResumeSessionResponse(await agent.resumeSession(v1ResumeSessionRequest(params))), - ) + .onRequest(v2.methods.agent.session.resume, async ({ params }) => { + const restore = v1RestoreSessionRequest(params); + return v2ResumeSessionResponse( + restore.method === "resume" + ? await agent.resumeSession(restore.request) + : await client.replaying(params.sessionId, () => agent.loadSession(restore.request)), + ); + }) .onRequest(v2.methods.agent.session.close, ({ params }) => agent.closeSession(params)) .onRequest(v2.methods.agent.session.delete, ({ params }) => agent.deleteSession(params)) .onRequest(v2.methods.agent.session.setConfigOption, async ({ params }) => { @@ -83,11 +88,34 @@ export function v2AgentApp( * the agent as v2 messages. */ class V2ClientConnection implements AcpClient { + /** + * The sessions whose history replays, each with the messages that the + * replay has cleared so far. + */ + private readonly replays = new Map>(); + constructor( private readonly ctx: v2.AgentContext, private readonly logger: Logger, ) {} + /** + * Runs `restore`, which replays the history of `sessionId`. + * + * The agent replays messages as chunks. A v2 client may still hold a + * replayed message, and chunks append, so before the first replayed chunk of + * each message the client gets the message with no content, which clears it, + * as v2 requires of a message replayed as chunks. + */ + async replaying(sessionId: string, restore: () => Promise): Promise { + this.replays.set(sessionId, new Set()); + try { + return await restore(); + } finally { + this.replays.delete(sessionId); + } + } + /** * Sends a v2 session update, and logs rather than rejects when that fails. * It and {@link sessionUpdate} hand updates to the connection synchronously, @@ -112,12 +140,32 @@ class V2ClientConnection implements AcpClient { async sessionUpdate({ update, ...notification }: AcpSessionNotification): Promise { const v2Update = v2SessionUpdate(update); - if (v2Update) { - await this.ctx.notify(v2.methods.client.session.update, { - ...notification, - update: v2Update, - }); - } + if (!v2Update) return; + const clear = this.replayClear(notification.sessionId, v2Update); + // Both go to the connection before either is awaited, so a chunk that + // another call sends meanwhile cannot come between them. + await Promise.all([ + ...(clear + ? [this.ctx.notify(v2.methods.client.session.update, { ...notification, update: clear })] + : []), + this.ctx.notify(v2.methods.client.session.update, { ...notification, update: v2Update }), + ]); + } + + /** + * The update that clears the message of a replayed chunk, when it is the + * first replayed chunk of that message (see {@link replaying}). + */ + private replayClear(sessionId: string, update: v2.SessionUpdate): v2.SessionUpdate | undefined { + const cleared = this.replays.get(sessionId); + if (!cleared) return undefined; + const kind = MESSAGE_OF_CHUNK.get(update.sessionUpdate); + if (!kind) return undefined; + const { messageId } = update as { messageId: string }; + const key = `${kind} ${messageId}`; + if (cleared.has(key)) return undefined; + cleared.add(key); + return emptyMessage(kind, messageId); } async requestPermission( @@ -162,6 +210,28 @@ class V2ClientConnection implements AcpClient { } } +/** The message update of each message chunk. */ +const MESSAGE_OF_CHUNK = new Map([ + ["user_message_chunk", "user_message"], + ["agent_message_chunk", "agent_message"], + ["agent_thought_chunk", "agent_thought"], +]); + +function emptyMessage( + kind: "user_message" | "agent_message" | "agent_thought", + messageId: string, +): v2.SessionUpdate { + const message = { messageId, content: [] }; + switch (kind) { + case "user_message": + return { ...message, sessionUpdate: kind }; + case "agent_message": + return { ...message, sessionUpdate: kind }; + default: + return { ...message, sessionUpdate: kind }; + } +} + function isExtensionMethod(method: string): method is v2.ExtensionMethod { return method.startsWith("_"); } diff --git a/src/v2/session.ts b/src/v2/session.ts index 488d6b81..6bcd6487 100644 --- a/src/v2/session.ts +++ b/src/v2/session.ts @@ -6,6 +6,7 @@ import { RequestError, type AvailableCommand, + type LoadSessionRequest, type McpServer, type NewSessionRequest, type NewSessionResponse, @@ -25,18 +26,36 @@ export function v1NewSessionRequest(params: v2.NewSessionRequest): NewSessionReq } /** - * `replayFrom` is rejected until the v2 surface can translate the replayed - * history, which needs the message and tool call updates. + * The v1 request that a v2 `session/resume` means. + * + * v2 restores a session with `session/resume` and replays its history only + * when `replayFrom` asks for it. v1 has two methods: `session/resume`, which + * never replays, and `session/load`, which replays the whole history before it + * responds. `replayFrom: { type: "start" }` is the second; a cursor that + * starts elsewhere is rejected. */ -export function v1ResumeSessionRequest(params: v2.ResumeSessionRequest): ResumeSessionRequest { +export function v1RestoreSessionRequest( + params: v2.ResumeSessionRequest, +): + | { method: "resume"; request: ResumeSessionRequest } + | { method: "load"; request: LoadSessionRequest } { const { replayFrom, mcpServers, ...request } = params; - if (replayFrom != null) { + if (replayFrom == null) { + return { + method: "resume", + request: { ...request, ...(mcpServers ? { mcpServers: mcpServers.map(v1McpServer) } : {}) }, + }; + } + if (replayFrom.type !== "start") { throw RequestError.invalidParams( { replayFrom }, - "The ACP v2 surface does not replay session history yet", + `Replaying from a ${replayFrom.type} cursor is not supported`, ); } - return { ...request, ...(mcpServers ? { mcpServers: mcpServers.map(v1McpServer) } : {}) }; + return { + method: "load", + request: { ...request, mcpServers: (mcpServers ?? []).map(v1McpServer) }, + }; } /** v2 drops `modes`: the session mode is the `mode` config option, which the agent always lists. */