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;