From b117e3fb0f814730546346ec619720b100f03590 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Sat, 31 Oct 2026 13:03:27 +0000 Subject: [PATCH] fix(claude): allow conversation rewind to messages sent during a turn (#8372) * fix(server): let a message steered into a running Claude turn be rewound to A steered message never got a provider message id, because Claude takes it into the running turn without echoing it back, so rewinding to it refused with "Cannot rewind before the provider acknowledges the submitted prompt". The steer admission now reports the uuid Claude minted for the message, and a Claude conversation rewind to it forks at the entry Claude read it after. Co-Authored-By: Claude Opus 5.5 * fix(server): keep messages steered into a running Claude turn in the replayed timeline Claude records a steered message as a queued_command attachment rather than a user entry, and the history replay skipped it, so every such message, replies from other agents included, vanished from the timeline after a daemon restart. Co-Authored-By: Claude Opus 5.5 * test(claude): verify steer identities across lifecycle tests Match the accepted-steer contract used by conversation rewind and verify that its identity names the message delivered to the provider. Co-authored-by: tieo <65707274+tieo@users.noreply.github.com> * fix(claude): release transcript streams after rewind lookup Closing readline leaves a partially read transcript open after an early match. Destroy the owned stream on exit and cover repeated lookups with unread data. --------- Co-authored-by: tieo <65707274+tieo@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../src/server/agent/agent-manager.test.ts | 68 ++++++++++++ .../server/src/server/agent/agent-manager.ts | 18 ++- .../src/server/agent/agent-sdk-types.ts | 6 +- .../claude/agent.history-load.test.ts | 37 +++++++ ...agent.interrupt-restart-regression.test.ts | 10 +- .../claude/agent.rewind-anchors.test.ts | 50 +++++++++ .../agent/providers/claude/agent.test.ts | 17 +-- .../server/agent/providers/claude/agent.ts | 38 ++++++- .../agent/providers/claude/rewind.test.ts | 103 +++++++++++++++++- .../server/agent/providers/claude/rewind.ts | 50 +++++++++ 10 files changed, 380 insertions(+), 17 deletions(-) diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 4b3fb5ea46..ccf29b17b0 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1220,6 +1220,74 @@ test("orders buffered pre-steer output before an immediately accepted steer", as } }); +test("a steer the provider names without echoing it can be rewound to", async () => { + // Claude takes a steered message into the running turn without echoing it back as a user + // message, so the id it minted for the message reaches the manager only through the admission. + class SilentSteeringSession extends SteeringTestSession { + override readonly capabilities = { + ...TEST_CAPABILITIES, + supportsRewindFiles: true, + }; + rewindMessageIds: string[] = []; + + override async steerActiveTurn( + _prompt: AgentPromptInput, + options: import("./agent-sdk-types.js").SteerActiveTurnOptions, + ): Promise { + if (options.expectedTurnId !== `active-turn-${this.startCount}`) { + return { status: "unavailable" }; + } + return { status: "accepted", providerMessageId: `provider-${options.clientMessageId}` }; + } + + override async revertFiles({ messageId }: { messageId: string }): Promise { + this.rewindMessageIds.push(messageId); + } + } + const session = new SilentSteeringSession({ provider: "codex", cwd: process.cwd() }); + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-steer-rewind-")); + const manager = new AgentManager({ + clients: { + codex: new (class extends TestAgentClient { + override async createSession(): Promise { + return session; + } + })(), + }, + logger, + }); + let agentId: string | null = null; + try { + const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + agentId = agent.id; + const run = manager.streamAgent(agent.id, "initial"); + void (async () => { + for await (const _event of run) { + } + })(); + await manager.waitForAgentRunStart(agent.id); + + await expect( + manager.steerAgentRun(agent.id, "steered", { clientMessageId: "steer-client" }), + ).resolves.toMatchObject({ status: "accepted" }); + const row = manager + .fetchTimeline(agent.id, { limit: 0 }) + .rows.find((candidate) => candidate.item.type === "user_message"); + expect(row).toMatchObject({ + providerMessageId: "provider-steer-client", + item: { text: "steered", clientMessageId: "steer-client" }, + }); + + await manager.rewind(agent.id, "steer-client", "files"); + expect(session.rewindMessageIds).toEqual(["provider-steer-client"]); + } finally { + if (agentId) await manager.closeAgent(agentId).catch(() => undefined); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("orders a concurrent replacement after a pending accepted steer", async () => { const entered = deferred(); const release = deferred(); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index c69a574849..94250e5cdd 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -3029,7 +3029,13 @@ export class AgentManager { expectedTurnId, }); if (admission.status === "accepted") { - await this.recordAcceptedSteer(agent, prompt, options?.clientMessageId, expectedTurnId); + await this.recordAcceptedSteer( + agent, + prompt, + options?.clientMessageId, + expectedTurnId, + admission.providerMessageId, + ); } return admission; }); @@ -3062,7 +3068,13 @@ export class AgentManager { expectedTurnId, }); if (admission.status === "accepted") { - await this.recordAcceptedSteer(agent, prompt, options?.clientMessageId, expectedTurnId); + await this.recordAcceptedSteer( + agent, + prompt, + options?.clientMessageId, + expectedTurnId, + admission.providerMessageId, + ); } return admission; }) @@ -3167,6 +3179,7 @@ export class AgentManager { prompt: AgentPromptInput, clientMessageId: string | undefined, expectedTurnId: string, + providerMessageId: string | undefined, ): Promise { if (!clientMessageId) { return; @@ -3174,6 +3187,7 @@ export class AgentManager { this.recordSubmittedPrompt(agent, prompt, clientMessageId, "accepted", { messageId: clientMessageId, turnId: expectedTurnId, + ...(providerMessageId ? { providerMessageId } : {}), }); this.emitState(agent); } diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index 6570392311..c31d1afb86 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -208,7 +208,11 @@ export interface AgentSteerOptions extends AgentRunOptions { clearPendingPermissions?: boolean; } -export type SteerResult = { status: "accepted" } | { status: "unavailable" }; +// providerMessageId names the steered message in the provider's own history, when the provider +// mints one. Rewinding to the message needs it. +export type SteerResult = + | { status: "accepted"; providerMessageId?: string } + | { status: "unavailable" }; export interface SteerActiveTurnOptions extends AgentSteerOptions { expectedTurnId: string; diff --git a/packages/server/src/server/agent/providers/claude/agent.history-load.test.ts b/packages/server/src/server/agent/providers/claude/agent.history-load.test.ts index 393fda60de..e2ea6850bb 100644 --- a/packages/server/src/server/agent/providers/claude/agent.history-load.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.history-load.test.ts @@ -614,6 +614,43 @@ describe("ClaudeAgentSession persisted history load", () => { ); }); + test("replays a message steered into a running turn as the user's message", async () => { + const queued = (commandMode: string, prompt: unknown, sourceUuid?: string) => ({ + type: "attachment", + uuid: `attachment-${commandMode}`, + parentUuid: "a1", + attachment: { + type: "queued_command", + commandMode, + prompt, + ...(sourceUuid ? { source_uuid: sourceUuid } : {}), + }, + }); + const entries = [ + { type: "user", uuid: "u1", parentUuid: null, message: { content: "first" } }, + { type: "assistant", uuid: "a1", parentUuid: "u1", message: { content: "working" } }, + queued("prompt", [{ type: "text", text: "steered while working" }], "steered-1"), + queued("task-notification", "t1"), + { type: "assistant", uuid: "a2", parentUuid: "a1", message: { content: "answer" } }, + ]; + mkdirSync(path.dirname(transcriptPath()), { recursive: true }); + writeFileSync(transcriptPath(), entries.map((entry) => JSON.stringify(entry)).join("\n")); + + const session = await resume(); + try { + const users = []; + for await (const event of session.streamHistory()) { + if (event.type === "timeline" && event.item.type === "user_message") users.push(event.item); + } + expect(users).toEqual([ + { type: "user_message", text: "first", messageId: "u1" }, + { type: "user_message", text: "steered while working", messageId: "steered-1" }, + ]); + } finally { + await session.close(); + } + }); + test("loads readable history when the temporary directory is unavailable", async () => { const file = transcriptPath(); const entry = { type: "assistant", message: { content: [{ type: "text", text: "saved" }] } }; diff --git a/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts b/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts index 02a09ba44b..fa0266c355 100644 --- a/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts @@ -349,8 +349,11 @@ async function startSteeredTurn(sessionId: string): Promise<{ expectedTurnId: start.value.turnId, clientMessageId: "steer-client", }); - expect(steered).toEqual({ status: "accepted" }); await waitFor(() => query?.prompts.length === 2); + expect(steered).toEqual({ + status: "accepted", + providerMessageId: query?.prompts[1]?.uuid, + }); return { session, query: () => query, turn }; } @@ -972,7 +975,10 @@ test("steers an autonomous turn through its existing query without restarting it const steeredTimeline = await autonomousEvents.next(); const completion = await autonomousEvents.next(); - expect(steer).toEqual({ status: "accepted" }); + expect(steer).toEqual({ + status: "accepted", + providerMessageId: queryRef?.prompts[1]?.uuid, + }); expect(queryFactory).toHaveBeenCalledTimes(1); expect(queryRef?.interrupt).not.toHaveBeenCalled(); expect(queryRef?.prompts.map((prompt) => prompt.text)).toEqual(["seed prompt", "steer prompt"]); diff --git a/packages/server/src/server/agent/providers/claude/agent.rewind-anchors.test.ts b/packages/server/src/server/agent/providers/claude/agent.rewind-anchors.test.ts index 1b1932e9cd..ed9b324ff4 100644 --- a/packages/server/src/server/agent/providers/claude/agent.rewind-anchors.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.rewind-anchors.test.ts @@ -1,7 +1,11 @@ +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; import { describe, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; import { ClaudeAgentClient } from "./agent.js"; +import { claudeProjectDir } from "./project-dir.js"; import { FakeClaudeSdk } from "./test-rewind-claude-sdk.js"; import { streamSession } from "../test-utils/session-stream-adapter.js"; import type { AgentSession } from "../../agent-sdk-types.js"; @@ -267,3 +271,49 @@ describe("Claude rewind after a turn whose last assistant message came from a su expect(rewindSdk.recordedForks).toEqual([{ upToMessageId: "assistant-1" }]); }); }); + +describe("Claude rewind to a message steered into a running turn", () => { + test("forks at the entry Claude read the message after", async () => { + const configDir = mkdtempSync(path.join(tmpdir(), "claude-rewind-steer-")); + const previousConfigDir = process.env.CLAUDE_CONFIG_DIR; + process.env.CLAUDE_CONFIG_DIR = configDir; + const conversation = createConversation([{ assistantMessageId: "assistant-1" }]); + const rewindSdk = new FakeClaudeSdk(); + const session = await createSession(conversation, rewindSdk); + + try { + await runTurns(session, 1); + // A steered message opens no turn: Claude Code records it as a queued_command attachment + // under the uuid Paseo minted for it, hung off the entry it was read after. + const projectDir = await claudeProjectDir(process.cwd(), { configDir }); + mkdirSync(projectDir, { recursive: true }); + const entries = [ + { type: "user", uuid: conversation.userMessageIds[0], parentUuid: null }, + { type: "assistant", uuid: "assistant-1", parentUuid: conversation.userMessageIds[0] }, + { type: "user", uuid: "tool-result-1", parentUuid: "assistant-1" }, + { + type: "attachment", + uuid: "queued-command-1", + parentUuid: "tool-result-1", + attachment: { type: "queued_command", source_uuid: "steered-message-1" }, + }, + ]; + writeFileSync( + path.join(projectDir, `${SESSION_ID}.jsonl`), + entries.map((entry) => `${JSON.stringify(entry)}\n`).join(""), + ); + + await session.revertConversation?.({ messageId: "steered-message-1" }); + } finally { + await session.close(); + if (previousConfigDir === undefined) { + delete process.env.CLAUDE_CONFIG_DIR; + } else { + process.env.CLAUDE_CONFIG_DIR = previousConfigDir; + } + rmSync(configDir, { recursive: true, force: true }); + } + + expect(rewindSdk.recordedForks).toEqual([{ upToMessageId: "tool-result-1" }]); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/agent.test.ts b/packages/server/src/server/agent/providers/claude/agent.test.ts index deb52f9a6b..758fad2e8f 100644 --- a/packages/server/src/server/agent/providers/claude/agent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.test.ts @@ -1224,16 +1224,17 @@ describe("ClaudeAgentSession features", () => { const iterator = input[Symbol.asyncIterator](); await expect(iterator.next()).resolves.toMatchObject({ value: { type: "user" } }); - await expect( - session.steerActiveTurn?.("same turn follow-up", { - expectedTurnId: turnId, - clientMessageId: "steer-client-id", - }), - ).resolves.toEqual({ status: "accepted" }); + const admission = await session.steerActiveTurn?.("same turn follow-up", { + expectedTurnId: turnId, + clientMessageId: "steer-client-id", + }); + expect(admission).toEqual({ status: "accepted", providerMessageId: expect.any(String) }); + if (admission?.status !== "accepted") throw new Error("Expected steer admission"); await expect(iterator.next()).resolves.toMatchObject({ value: { type: "user", priority: "next", + uuid: admission.providerMessageId, message: { content: [{ type: "text", text: "same turn follow-up" }] }, }, }); @@ -1291,7 +1292,7 @@ describe("ClaudeAgentSession features", () => { expectedTurnId: turnId, clearPendingPermissions: true, }), - ).resolves.toEqual({ status: "accepted" }); + ).resolves.toEqual({ status: "accepted", providerMessageId: expect.any(String) }); await expect(firstPermission).resolves.toMatchObject({ behavior: "deny", interrupt: undefined, @@ -1350,7 +1351,7 @@ describe("ClaudeAgentSession features", () => { const permission = internal.handlePermissionRequest("Write", {}, { toolUseID: "tool-1" }); await expect( session.steerActiveTurn?.("system notification", { expectedTurnId: turnId }), - ).resolves.toEqual({ status: "accepted" }); + ).resolves.toEqual({ status: "accepted", providerMessageId: expect.any(String) }); expect(session.getPendingPermissions()).toHaveLength(1); const requestId = session.getPendingPermissions()[0]!.id; await session.respondToPermission(requestId, { behavior: "deny", message: "test cleanup" }); diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index c675b1ed62..ae48883654 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -82,6 +82,7 @@ import { renderPromptAttachmentAsText } from "../../prompt-attachments.js"; import { discoverClaudeModels } from "./model-discovery.js"; import { claudeQuery, type ClaudeOptions, type ClaudeQueryFactory } from "./query.js"; import { + findSteeredMessageParent, realClaudeRewindSdk, revertClaudeConversation, revertClaudeFiles, @@ -2627,7 +2628,7 @@ class ClaudeAgentSession implements AgentSession { return { status: "unavailable" }; } this.enqueueSteer(input, message, options.clearPendingPermissions === true); - return { status: "accepted" }; + return { status: "accepted", ...(message.uuid ? { providerMessageId: message.uuid } : {}) }; } private enqueueSteer( @@ -3094,7 +3095,7 @@ class ClaudeAgentSession implements AgentSession { async revertConversation(input: { messageId: string }): Promise { await this.ensureHistoryReady(); - const target = this.resolveConversationRewindTarget(input.messageId); + const target = await this.resolveConversationRewindTarget(input.messageId); if (target.kind === "fresh-session") { this.startFreshConversationSession(); return; @@ -3429,12 +3430,24 @@ class ClaudeAgentSession implements AgentSession { return messageId; } - private resolveConversationRewindTarget(messageId: string): ClaudeConversationRewindTarget { + private async resolveConversationRewindTarget( + messageId: string, + ): Promise { const targetUserMessageId = this.resolveClaudeMessageId(messageId); const index = this.rewindTurnAnchors.findIndex( (anchor) => anchor.userMessageId === targetUserMessageId, ); if (index < 0) { + // A message steered into a running turn opens no turn of its own. + const transcriptPath = this.claudeSessionId + ? await this.resolveHistoryPath(this.claudeSessionId) + : null; + const steerParent = transcriptPath + ? await findSteeredMessageParent(transcriptPath, targetUserMessageId) + : null; + if (steerParent) { + return { kind: "fork", messageId: steerParent }; + } throw new Error(`Claude rewind target ${messageId} is not in the tracked conversation`); } @@ -6431,6 +6444,10 @@ function convertClaudeHistoryEntryPreamble( if (entry.type === "user" && isSyntheticHistoryUserEntry(entry)) { return { shortCircuit: [] }; } + const steeredPrompt = readSteeredPromptHistoryItem(entry); + if (steeredPrompt) { + return { shortCircuit: [steeredPrompt] }; + } // Only user and assistant records are conversation turns. Claude Code 2.1.x also journals each // API request as `api-request-blob` records whose message replays the system context. if (entry.type !== "user" && entry.type !== "assistant") { @@ -6457,6 +6474,21 @@ function isProviderImageMessage(item: AgentTimelineItem): boolean { return item.type === "assistant_message" && isProviderImageMarkdown(item.text); } +/** + * A message steered into a running turn. Claude Code gives it no user entry: it records a + * queued_command attachment in prompt mode, under the uuid Paseo minted for the message, which is + * also what a rewind to it resolves. The other modes carry task notifications, not user messages. + */ +function readSteeredPromptHistoryItem(entry: ClaudeHistoryEntry): AgentTimelineItem | null { + if (entry.type !== "attachment") return null; + const attachment = toObjectRecord(entry.attachment); + if (attachment?.type !== "queued_command" || attachment.commandMode !== "prompt") return null; + const text = extractUserMessageText(attachment.prompt); + if (!text) return null; + const messageId = readTrimmedString(attachment.source_uuid); + return { type: "user_message", text, ...(messageId ? { messageId } : {}) }; +} + export async function convertClaudeHistoryEntry( entry: ClaudeHistoryEntry, mapBlocks: ( diff --git a/packages/server/src/server/agent/providers/claude/rewind.test.ts b/packages/server/src/server/agent/providers/claude/rewind.test.ts index 2d6d3853c2..226f41d39c 100644 --- a/packages/server/src/server/agent/providers/claude/rewind.test.ts +++ b/packages/server/src/server/agent/providers/claude/rewind.test.ts @@ -1,11 +1,20 @@ import { randomUUID } from "node:crypto"; -import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { + mkdirSync, + mkdtempSync, + readFileSync, + readdirSync, + readlinkSync, + rmSync, + writeFileSync, +} from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; import { afterEach, describe, expect, test } from "vitest"; import type { Query } from "@anthropic-ai/claude-agent-sdk"; import { + findSteeredMessageParent, revertClaudeConversation, revertClaudeConversationAndFiles, revertClaudeFiles, @@ -217,4 +226,96 @@ describe("Claude rewind against the Claude SDK", () => { "first reply", ]); }); + + test("forks just before a message steered into a running turn", async () => { + tempRoot = mkdtempSync(path.join(tmpdir(), "claude-rewind-steer-")); + delete process.env.CLAUDE_CONFIG_DIR; + const projectDir = path.join(tempRoot, "projects", "-repo"); + mkdirSync(projectDir, { recursive: true }); + const sessionId = randomUUID(); + const firstUser = randomUUID(); + const toolUse = randomUUID(); + const toolResult = randomUUID(); + const steeredMessage = randomUUID(); + const queuedCommand = randomUUID(); + const transcriptPath = path.join(projectDir, `${sessionId}.jsonl`); + writeTranscript(transcriptPath, sessionId, [ + { type: "user", uuid: firstUser, parentUuid: null, text: "first" }, + { type: "assistant", uuid: toolUse, parentUuid: firstUser, text: "running a tool" }, + { type: "user", uuid: toolResult, parentUuid: toolUse, text: "tool output" }, + // What Claude Code writes when it reads a steered message between two tool calls. + JSON.stringify({ + type: "attachment", + uuid: queuedCommand, + parentUuid: toolResult, + sessionId, + attachment: { + type: "queued_command", + prompt: [{ type: "text", text: "steered" }], + source_uuid: steeredMessage, + }, + }), + { type: "assistant", uuid: randomUUID(), parentUuid: queuedCommand, text: "reply to steer" }, + ]); + + const parent = await findSteeredMessageParent(transcriptPath, steeredMessage); + expect(parent).toBe(toolResult); + expect(await findSteeredMessageParent(transcriptPath, randomUUID())).toBeNull(); + + let currentSessionId = sessionId; + await revertClaudeConversation({ + sdk: realClaudeRewindSdk, + sessionId, + transcriptPath, + messageId: parent!, + setSessionId: (nextSessionId) => { + currentSessionId = nextSessionId; + }, + }); + expect(readMessageTexts(path.join(projectDir, `${currentSessionId}.jsonl`))).toEqual([ + "first", + "running a tool", + "tool output", + ]); + }); }); + +test.runIf(process.platform === "linux")( + "releases the transcript descriptor after an early steer match", + async () => { + const artifacts = path.resolve(".artifacts"); + mkdirSync(artifacts, { recursive: true }); + const root = mkdtempSync(path.join(artifacts, "claude-rewind-descriptor-")); + const transcriptPath = path.join(root, "session.jsonl"); + const matchingEntry = { + type: "attachment", + parentUuid: "before-steer", + attachment: { type: "queued_command", source_uuid: "steered-message" }, + }; + const unreadEntry = JSON.stringify({ type: "assistant", message: "x".repeat(1024) }); + writeFileSync( + transcriptPath, + JSON.stringify(matchingEntry) + "\n" + (unreadEntry + "\n").repeat(1200), + ); + const openDescriptors = () => + readdirSync("/proc/self/fd").filter((descriptor) => { + try { + return readlinkSync(path.join("/proc/self/fd", descriptor)) === transcriptPath; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; + throw error; + } + }).length; + try { + expect(openDescriptors()).toBe(0); + for (let attempt = 0; attempt < 5; attempt++) { + await expect(findSteeredMessageParent(transcriptPath, "steered-message")).resolves.toBe( + "before-steer", + ); + } + await expect.poll(openDescriptors, { timeout: 1000 }).toBe(0); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, +); diff --git a/packages/server/src/server/agent/providers/claude/rewind.ts b/packages/server/src/server/agent/providers/claude/rewind.ts index 91183687e0..75607c184b 100644 --- a/packages/server/src/server/agent/providers/claude/rewind.ts +++ b/packages/server/src/server/agent/providers/claude/rewind.ts @@ -1,5 +1,7 @@ +import { createReadStream } from "node:fs"; import { appendFile, readFile } from "node:fs/promises"; import path from "node:path"; +import { createInterface } from "node:readline"; import { forkSession as claudeForkSession, type Query, @@ -70,6 +72,54 @@ function isSessionStoreEntry(value: unknown): value is SessionStoreEntry { ); } +/** + * The entry a message steered into a running turn follows. Claude Code gives such a message no + * entry of its own: it records a queued_command attachment, under the uuid Paseo minted for the + * message, hung off the entry it was read after. Forking at that entry drops the message and + * everything after it. Null when the transcript holds no such attachment. + */ +export async function findSteeredMessageParent( + transcriptPath: string, + messageId: string, +): Promise { + const input = createReadStream(transcriptPath, { encoding: "utf8" }); + const lines = createInterface({ + input, + crlfDelay: Infinity, + }); + try { + for await (const line of lines) { + if (!line.includes(messageId) || !line.includes('"queued_command"')) continue; + let entry: unknown; + try { + entry = JSON.parse(line); + } catch { + continue; + } + const record = entry as { + type?: unknown; + parentUuid?: unknown; + attachment?: { type?: unknown; source_uuid?: unknown }; + }; + if ( + record.type === "attachment" && + record.attachment?.type === "queued_command" && + record.attachment.source_uuid === messageId && + typeof record.parentUuid === "string" + ) { + return record.parentUuid; + } + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; + } finally { + lines.close(); + input.destroy(); + } + return null; +} + export async function revertClaudeConversation(input: { sdk: ClaudeRewindSdk; sessionId: string | null;