diff --git a/src/fork-session.ts b/src/fork-session.ts index e4bd5f11..245d2edd 100644 --- a/src/fork-session.ts +++ b/src/fork-session.ts @@ -6,11 +6,10 @@ import { import { forkSession as forkClaudeSession, getSessionMessages, - importSessionToStore, - type SessionStore, type SessionStoreEntry, } from "@anthropic-ai/claude-agent-sdk"; import { createHash } from "node:crypto"; +import { loadFullSessionHistory } from "./resumed-session.js"; import { assistantMessageText } from "./session-failure-extension.js"; import { SessionTiming } from "./session-timing.js"; @@ -69,19 +68,6 @@ function forkPointMessageIdCandidates(messageId: string): string[] { return protocolMessageId === messageId ? [messageId] : [messageId, protocolMessageId]; } -async function loadFullSessionHistory(sessionId: string): Promise { - // getSessionMessages returns only the active parentUuid chain. The import keeps inactive branches that AIR can still reference. - const entries: SessionStoreEntry[] = []; - const store: SessionStore = { - append: async (_key, batch) => { - entries.push(...batch); - }, - load: async () => null, - }; - await importSessionToStore(sessionId, store, { includeSubagents: false }); - return entries; -} - function assistantGroups( entries: SessionStoreEntry[], messageIdForGrouping: ForkSessionDependencies["messageIdForGrouping"], @@ -235,6 +221,7 @@ async function forkTranscript( .find((message) => dependencies.messageIdForGrouping(message) === candidateId)?.uuid, ) .find(Boolean); + // getSessionMessages returns only the active parentUuid chain. The full history keeps inactive branches that AIR can still reference. const fullHistoryUuid = messageUuid ? undefined : resolveFromFullHistory( diff --git a/src/resumed-session.ts b/src/resumed-session.ts index bc50df27..93f069e1 100644 --- a/src/resumed-session.ts +++ b/src/resumed-session.ts @@ -1,6 +1,13 @@ -import { getSessionMessages, type SessionMessage } from "@anthropic-ai/claude-agent-sdk"; +import { + getSessionMessages, + importSessionToStore, + type SessionMessage, + type SessionStore, + type SessionStoreEntry, +} from "@anthropic-ai/claude-agent-sdk"; import { access, open, readdir } from "node:fs/promises"; import path from "node:path"; +import { isCompactSummaryMessage } from "./context-compaction.js"; import { claudeConfigDir } from "./paths.js"; import { SessionTiming } from "./session-timing.js"; @@ -68,6 +75,97 @@ export async function readResumedSession( } } +/** + * Read the messages of a session from its first prompt on, given the + * `getSessionMessages` chain of the session. + * + * `getSessionMessages` follows `parentUuid` links. Claude Code writes each + * `compact_boundary` with `parentUuid: null`, so after a compaction the chain + * starts at its summary. The boundary keeps the last message before it in + * `logicalParentUuid`. For a chain that starts at a summary, this links each + * boundary to that message and reads the transcript through + * `getSessionMessages` again. For a session compacted twice, the result has + * both summaries and the messages before, between and after them. The result + * ends at the last message of `messages`. Any other chain is returned as it + * is, and so is the chain when the read fails or when its result does not + * reach that last message (e.g. the transcript changed between the reads). + * + * `dir` is the project directory of the first read: both reads then resolve + * the same transcript file. Without it, all project directories are searched. + * + * The preserved-message metadata of each boundary is dropped. With it, + * `getSessionMessages` moves the messages that a compaction kept to after the + * summary, and the relinked boundary then leads from them back to the summary: + * the chain loops and ends there. Without it, the kept messages stay where + * they were said, before the boundary. + */ +export async function readHistoryAcrossCompactions( + sessionId: string, + messages: SessionMessage[], + logger?: ResumeLogger, + dir?: string, +): Promise { + if (!isCompactSummaryMessage(messages[0])) return messages; + try { + const entries = (await loadFullSessionHistory(sessionId, dir)).map(linkCompactBoundary); + const history = await getSessionMessages(sessionId, { + dir, + sessionStore: { append: async () => {}, load: async () => entries }, + }); + const leaf = messages.at(-1)?.uuid; + const end = history.findIndex((message) => message.uuid === leaf); + if (end < 0) { + logger?.error( + `The history before the last compaction of session ${sessionId} does not reach message ${leaf}: replaying the messages after it`, + ); + return messages; + } + return history.slice(0, end + 1); + } catch (error) { + logger?.error( + `Failed to read the history before the last compaction of session ${sessionId}:`, + error, + ); + return messages; + } +} + +/** Every record of a session's transcript, including the inactive branches + * and the history before a compaction that `getSessionMessages` leaves out. + * Without `dir`, all project directories are searched. */ +export async function loadFullSessionHistory( + sessionId: string, + dir?: string, +): Promise { + const entries: SessionStoreEntry[] = []; + const store: SessionStore = { + append: async (_key, batch) => { + entries.push(...batch); + }, + load: async () => null, + }; + await importSessionToStore(sessionId, store, { dir, includeSubagents: false }); + return entries; +} + +/** The record unchanged, or a compaction boundary linked to the message before it. */ +function linkCompactBoundary(entry: SessionStoreEntry): SessionStoreEntry { + if ( + entry.type !== "system" || + entry.subtype !== "compact_boundary" || + typeof entry.logicalParentUuid !== "string" + ) { + return entry; + } + const compactMetadata: Record = + typeof entry.compactMetadata === "object" && entry.compactMetadata !== null + ? { ...entry.compactMetadata } + : {}; + delete compactMetadata.preservedSegment; + delete compactMetadata.preservedMessages; + return { ...entry, parentUuid: entry.logicalParentUuid, compactMetadata }; +} + /** What a resume restores from the end of the local transcript. */ export type ResumedTail = { model?: string; diff --git a/src/tests/helpers.ts b/src/tests/helpers.ts index df3a7f5e..a8608528 100644 --- a/src/tests/helpers.ts +++ b/src/tests/helpers.ts @@ -9,6 +9,8 @@ import { type ClientCapabilities, type InitializeRequest, } from "@agentclientprotocol/sdk/unstable"; +import { mkdir, writeFile } from "node:fs/promises"; +import path from "node:path"; /** The context-usage report the base mock query returns. `rawMaxTokens` * matches the agent's DEFAULT_CONTEXT_WINDOW so window-related assertions @@ -82,3 +84,73 @@ export function heldRequest(_params?: unknown, signal?: AbortSignal): Promise else signal?.addEventListener("abort", cancel, { once: true }); }); } + +/** A transcript record of a user message. */ +export function transcriptUser(uuid: string, content: string, extra: object = {}) { + return { type: "user", uuid, message: { role: "user", content }, ...extra }; +} + +/** A transcript record of an assistant text response. */ +export function transcriptAssistant(uuid: string, text: string) { + return { + type: "assistant", + uuid, + message: { + id: `msg_${uuid}`, + role: "assistant", + model: "claude-opus-5", + content: [{ type: "text", text }], + }, + }; +} + +/** A `compact_boundary` as Claude Code writes it: no parent, and the last + * message before it in `logicalParentUuid`. */ +export function transcriptCompactBoundary( + uuid: string, + logicalParentUuid: string, + compactMetadata: object = {}, +) { + return { + type: "system", + subtype: "compact_boundary", + uuid, + parentUuid: null, + logicalParentUuid, + compactMetadata: { trigger: "manual", preTokens: 1000, ...compactMetadata }, + }; +} + +/** The summary message that follows a `compact_boundary`. */ +export function transcriptCompactSummary( + uuid: string, + content = "This session is being continued from a previous conversation that ran out of context. The summary below covers the earlier portion of the conversation.\n\n" + + "Summary:\n1. Primary Request and Intent:\n Say hi.\n\n" + + "Continue the conversation from where it left off without asking the user any further questions. Resume directly — do not acknowledge the summary.", +) { + return transcriptUser(uuid, content, { isCompactSummary: true }); +} + +/** Writes `/.jsonl` as Claude Code does: each record + * links to the record before it, except a record that sets its own + * `parentUuid`. */ +export async function writeTranscript( + directory: string, + sessionId: string, + records: Record[], +): Promise { + await mkdir(directory, { recursive: true }); + let parentUuid: unknown = null; + const lines = records.map((record) => { + const line = { + parentUuid, + sessionId, + isSidechain: false, + timestamp: "2026-10-11T00:00:00.000Z", + ...record, + }; + parentUuid = record.uuid; + return JSON.stringify(line); + }); + await writeFile(path.join(directory, `${sessionId}.jsonl`), lines.join("\n") + "\n"); +} diff --git a/src/tests/resumed-session.test.ts b/src/tests/resumed-session.test.ts index 930aa284..378cf382 100644 --- a/src/tests/resumed-session.test.ts +++ b/src/tests/resumed-session.test.ts @@ -1,9 +1,20 @@ -import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import type { SessionMessage } from "@anthropic-ai/claude-agent-sdk"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { getSessionMessages, type SessionMessage } from "@anthropic-ai/claude-agent-sdk"; import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { + transcriptAssistant, + transcriptCompactBoundary, + transcriptCompactSummary, + transcriptUser, + writeTranscript, +} from "./helpers.js"; import os from "node:os"; import path from "node:path"; -import { readResumedTail, resumedModelFromTranscript } from "../resumed-session.js"; +import { + readHistoryAcrossCompactions, + readResumedTail, + resumedModelFromTranscript, +} from "../resumed-session.js"; function assistant( model: unknown, @@ -203,3 +214,187 @@ describe("readResumedTail", () => { expect(await readResumedTail("no-mode")).toEqual({ model: "claude-opus-5" }); }); }); + +describe("readHistoryAcrossCompactions", () => { + const sessionId = "33333333-3333-4333-8333-333333333333"; + let configDir: string; + let originalConfigDir: string | undefined; + + beforeEach(async () => { + configDir = await mkdtemp(path.join(os.tmpdir(), "claude-acp-compacted-")); + originalConfigDir = process.env.CLAUDE_CONFIG_DIR; + process.env.CLAUDE_CONFIG_DIR = configDir; + }); + + afterEach(async () => { + if (originalConfigDir === undefined) delete process.env.CLAUDE_CONFIG_DIR; + else process.env.CLAUDE_CONFIG_DIR = originalConfigDir; + await rm(configDir, { recursive: true, force: true }); + }); + + const transcript = (records: Record[], project = "-workspace") => + writeTranscript(path.join(configDir, "projects", project), sessionId, records); + const user = transcriptUser; + const assistant = transcriptAssistant; + const boundary = transcriptCompactBoundary; + const summary = (uuid: string) => transcriptCompactSummary(uuid, `Summary ${uuid}`); + + const texts = (messages: SessionMessage[]) => + messages.map((message) => { + const content = (message.message as { content: unknown }).content; + return typeof content === "string" ? content : (content as { text: string }[])[0]!.text; + }); + + it("reads the messages before each compaction", async () => { + await transcript([ + user("u1", "hi"), + assistant("a1", "Hello"), + boundary("b1", "a1"), + summary("s1"), + user("u2", "next"), + assistant("a2", "Next"), + boundary("b2", "a2"), + summary("s2"), + user("u3", "identify yourself"), + assistant("a3", "Claude"), + ]); + + // getSessionMessages stops at the last boundary. + const chain = await getSessionMessages(sessionId); + expect(texts(chain)).toEqual(["Summary s2", "identify yourself", "Claude"]); + expect(texts(await readHistoryAcrossCompactions(sessionId, chain))).toEqual([ + "hi", + "Hello", + "Summary s1", + "next", + "Next", + "Summary s2", + "identify yourself", + "Claude", + ]); + }); + + it("keeps the messages that a compaction preserved where they were said", async () => { + // Claude Code 2.1.293 keeps u2 and a2 in the context after the summary, + // and records them in both forms of preserved-message metadata. + await transcript([ + user("u1", "hi"), + assistant("a1", "Hello"), + user("u2", "keep this"), + assistant("a2", "Kept"), + boundary("b1", "a2", { + preservedSegment: { headUuid: "u2", anchorUuid: "s1", tailUuid: "a2" }, + preservedMessages: { anchorUuid: "s1", uuids: ["u2", "a2"] }, + }), + summary("s1"), + user("u3", "after"), + assistant("a3", "After"), + ]); + + // getSessionMessages moves the preserved messages after the summary. + const chain = await getSessionMessages(sessionId); + expect(texts(chain)).toEqual(["Summary s1", "keep this", "Kept", "after", "After"]); + expect(texts(await readHistoryAcrossCompactions(sessionId, chain))).toEqual([ + "hi", + "Hello", + "keep this", + "Kept", + "Summary s1", + "after", + "After", + ]); + }); + + it("returns the chain of a session without a compaction as it is", async () => { + await transcript([user("u1", "hi"), assistant("a1", "Hello")]); + + const chain = await getSessionMessages(sessionId); + expect(await readHistoryAcrossCompactions(sessionId, chain)).toBe(chain); + }); + + it("ends at the last message of the chain it was given", async () => { + await transcript([ + user("u1", "hi"), + assistant("a1", "Hello"), + boundary("b1", "a1"), + summary("s1"), + user("u2", "identify yourself"), + ]); + const chain = await getSessionMessages(sessionId); + // Claude Code appends a response after the first read. + await transcript([ + user("u1", "hi"), + assistant("a1", "Hello"), + boundary("b1", "a1"), + summary("s1"), + user("u2", "identify yourself"), + assistant("a2", "Claude"), + ]); + + expect(texts(await readHistoryAcrossCompactions(sessionId, chain))).toEqual([ + "hi", + "Hello", + "Summary s1", + "identify yourself", + ]); + }); + + it.each([ + ["the transcript is gone", async () => rm(configDir, { recursive: true, force: true })], + [ + "the transcript no longer has the last message", + () => + transcript([ + user("u1", "hi"), + assistant("a1", "Hello"), + boundary("b1", "a1"), + summary("s1"), + user("u9", "other"), + ]), + ], + ])("returns the chain and logs when %s", async (_case, change) => { + await transcript([ + user("u1", "hi"), + assistant("a1", "Hello"), + boundary("b1", "a1"), + summary("s1"), + user("u2", "identify yourself"), + ]); + const chain = await getSessionMessages(sessionId); + await change(); + const error = vi.fn(); + + expect(await readHistoryAcrossCompactions(sessionId, chain, { log: () => {}, error })).toBe( + chain, + ); + expect(error).toHaveBeenCalledOnce(); + }); + + it("reads the transcript in the project directory it is given", async () => { + await transcript( + [ + user("u1", "elsewhere"), + assistant("a1", "Elsewhere"), + boundary("b1", "a1"), + summary("s1"), + user("u2", "identify yourself"), + ], + "-a-other", + ); + await transcript( + [ + user("u1", "hi"), + assistant("a1", "Hello"), + boundary("b1", "a1"), + summary("s1"), + user("u2", "identify yourself"), + ], + "-z-workspace", + ); + const chain = await getSessionMessages(sessionId, { dir: "/z/workspace" }); + + expect( + texts(await readHistoryAcrossCompactions(sessionId, chain, undefined, "/z/workspace")), + ).toEqual(["hi", "Hello", "Summary s1", "identify yourself"]); + }); +}); diff --git a/src/tests/v1/acp-agent.test.ts b/src/tests/v1/acp-agent.test.ts index 8539a87d..d53360d5 100644 --- a/src/tests/v1/acp-agent.test.ts +++ b/src/tests/v1/acp-agent.test.ts @@ -58,7 +58,16 @@ import { parseUsageResponse, } from "../../usage-markdown.js"; import { Pushable } from "../../utils.js"; -import { heldRequest, initializeClient, makeMockQuery } from "../helpers.js"; +import { + heldRequest, + initializeClient, + makeMockQuery, + transcriptAssistant, + transcriptCompactBoundary, + transcriptCompactSummary, + transcriptUser, + writeTranscript, +} from "../helpers.js"; import { acpProtocolRouter } from "../../serve.js"; import * as agentSdk from "@anthropic-ai/claude-agent-sdk"; import { @@ -3629,6 +3638,63 @@ describe("usage-limit failure replay", () => { }); }); +describe("compacted session replay", () => { + it("replays the messages before the compaction, then the compaction", async () => { + const configDir = await mkdtemp(path.join(os.tmpdir(), "claude-acp-compacted-replay-")); + const originalConfigDir = process.env.CLAUDE_CONFIG_DIR; + process.env.CLAUDE_CONFIG_DIR = configDir; + try { + const sessionId = "44444444-4444-4444-8444-444444444444"; + // The records of `/compact` between two prompts, as Claude Code writes + // them: the boundary has no parent and links "Hello" logically. + await writeTranscript(path.join(configDir, "projects", "-tmp-proj"), sessionId, [ + transcriptUser("u1", "hi"), + transcriptAssistant("a1", "Hello"), + transcriptCompactBoundary("b1", "a1"), + transcriptCompactSummary("s1"), + transcriptUser("u2", "identify yourself"), + transcriptAssistant("a2", "Claude"), + ]); + const updates: SessionNotification[] = []; + const client = { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient; + const agent = new ClaudeAcpAgent(client, { log: () => {}, error: () => {} }); + await initializeClient(agent, { session: { compaction: {} } } as ClientCapabilities); + + await ( + agent as unknown as { replaySessionHistory(sessionId: string): Promise } + ).replaySessionHistory(sessionId); + + expect( + updates.flatMap(({ update }) => { + if (update.sessionUpdate === "compaction_update") return [["compaction", update.status]]; + if ( + (update.sessionUpdate === "user_message_chunk" || + update.sessionUpdate === "agent_message_chunk") && + update.content.type === "text" + ) { + return [[update.sessionUpdate, update.content.text]]; + } + return []; + }), + ).toEqual([ + ["user_message_chunk", "hi"], + ["agent_message_chunk", "Hello"], + ["compaction", "completed"], + ["user_message_chunk", "identify yourself"], + ["agent_message_chunk", "Claude"], + ]); + } finally { + if (originalConfigDir === undefined) delete process.env.CLAUDE_CONFIG_DIR; + else process.env.CLAUDE_CONFIG_DIR = originalConfigDir; + await rm(configDir, { recursive: true, force: true }); + } + }); +}); + describe("session/list", () => { const sdkSession = (index: number) => ({ sessionId: `session-${index}`, diff --git a/src/tests/v2/v2.test.ts b/src/tests/v2/v2.test.ts index 120b7be1..761df604 100644 --- a/src/tests/v2/v2.test.ts +++ b/src/tests/v2/v2.test.ts @@ -24,6 +24,13 @@ import { acpProtocolRouter } from "../../serve.js"; import { SettingsManager } from "../../settings.js"; import type { AgentHandle } from "../../v2/agent.js"; import { StoppedWork } from "../../v2/stopped-work.js"; +import { + transcriptAssistant, + transcriptCompactBoundary, + transcriptCompactSummary, + transcriptUser, + writeTranscript, +} from "../helpers.js"; /** One Claude Code process: the prompts it reads and the messages it sends. */ type FakeClaudeCode = { @@ -1964,6 +1971,49 @@ describe("ACP v2 core", () => { ]); }); + it("replays the messages before a compaction from the transcript", async () => { + const sessionId = randomUUID(); + // The records of `/compact` between two prompts, as Claude Code writes + // them: the boundary has no parent and links "Hello" logically. The SDK + // looks the project up by the real path of `cwd` (`/private/var` on macOS). + const project = (await realpath(cwd)).replace(/[^a-zA-Z0-9]/g, "-"); + await writeTranscript(path.join(tempDir, "claude", "projects", project), sessionId, [ + transcriptUser("u1", "hi"), + transcriptAssistant("a1", "Hello"), + transcriptCompactBoundary("b1", "a1"), + transcriptCompactSummary("s1"), + transcriptUser("u2", "identify yourself"), + ]); + // The transcript read and the read across the compaction both go to disk. + const actual = await vi.importActual( + "@anthropic-ai/claude-agent-sdk", + ); + vi.mocked(getSessionMessages) + .mockImplementationOnce(actual.getSessionMessages) + .mockImplementationOnce(actual.getSessionMessages); + const client = v2Client(); + await client.connect(async (agent) => { + await agent.request(v2.methods.agent.initialize, { + protocolVersion: v2.PROTOCOL_VERSION, + info: { name: "v2-test-client", version: "1.0.0" }, + capabilities: {}, + }); + await agent.request(v2.methods.agent.session.resume, { + sessionId, + cwd, + replayFrom: { type: "start" }, + }); + }); + + expect( + client.sessionUpdates.flatMap((update) => + update.sessionUpdate === "user_message" || update.sessionUpdate === "compaction_update" + ? [update.sessionUpdate] + : [], + ), + ).toEqual(["user_message", "compaction_update", "user_message"]); + }); + it.each([ [ "a prompt that Claude Code started after the cancel answered it", diff --git a/src/v1/acp-agent.ts b/src/v1/acp-agent.ts index d6e6db3c..b468445f 100644 --- a/src/v1/acp-agent.ts +++ b/src/v1/acp-agent.ts @@ -203,6 +203,7 @@ import { replaySubagentTerminalState, } from "../subagent-replay.js"; import { + readHistoryAcrossCompactions, readResumedSession, readResumedTail, type ResumedSessionSnapshot, @@ -7957,7 +7958,12 @@ export class ClaudeAcpAgent { ): Promise { const replayStartedAt = performance.now(); const toolUseCache: ToolUseCache = {}; - const messages = resumedMessages ?? (await getSessionMessages(sessionId)); + // A load replays the messages before each compaction too, unless reading them fails. + const messages = await readHistoryAcrossCompactions( + sessionId, + resumedMessages ?? (await getSessionMessages(sessionId)), + this.logger, + ); const historyLoadedAt = performance.now(); this.logger.log( `[session/replay] sessionId=${sessionId} phase=read durationMs=${Math.round(historyLoadedAt - replayStartedAt)} messages=${messages.length}`, diff --git a/src/v2/agent.ts b/src/v2/agent.ts index f6d405d9..3efd1a28 100644 --- a/src/v2/agent.ts +++ b/src/v2/agent.ts @@ -39,7 +39,11 @@ import { forkSession } from "../fork-session.js"; import { supportsMcpOAuth } from "../mcp-oauth.js"; import { sdkMcpServers } from "../mcp-servers.js"; import { ALLOW_BYPASS, autoModeDisabled } from "../permissions/modes.js"; -import { readResumedTail, resumedModelFromTranscript } from "../resumed-session.js"; +import { + readHistoryAcrossCompactions, + readResumedTail, + resumedModelFromTranscript, +} from "../resumed-session.js"; import { mergeEffortSettings } from "../session-effort.js"; import { configuredSettings, resolveThinkingConfig } from "../session-env.js"; import { resolveInitialPermissionMode } from "../session-mode.js"; @@ -376,6 +380,8 @@ class V2Agent implements AgentHandle, SessionConnection { await this.startSettled(sessionId); let session = this.sessions.get(sessionId); let history: Promise | undefined; + // The replay: the messages before each compaction too, unless reading them fails. + let replayed: Promise | undefined; if (session) { // Resuming the active session synchronizes it again. Its work goes on, // and the request's MCP servers and additional directories replace the @@ -402,13 +408,22 @@ class V2Agent implements AgentHandle, SessionConnection { } else { // One read of the transcript serves the replay and the resumed model. history = replayFrom != null ? getSessionMessages(sessionId, { dir: params.cwd }) : undefined; - // A failed start leaves it unawaited. + // The read across compactions overlaps the start of the query. + replayed = history?.then((messages) => + readHistoryAcrossCompactions(sessionId, messages, this.logger, params.cwd), + ); + // A failed start leaves them unawaited. history?.catch(() => {}); + replayed?.catch(() => {}); session = await this.startSession(sessionId, params, sessionId, history); } let subagents: SubagentReplay | undefined; if (replayFrom != null) { - const messages = await (history ?? getSessionMessages(sessionId, { dir: session.cwd })); + const cwd = session.cwd; + const messages = await (replayed ?? + getSessionMessages(sessionId, { dir: cwd }).then((chain) => + readHistoryAcrossCompactions(sessionId, chain, this.logger, cwd), + )); subagents = await session.tasks.loadReplay(messages); session.output.replay( messages,