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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>
Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

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<import("./agent-sdk-types.js").SteerResult> {
+      if (options.expectedTurnId !== `active-turn-${this.startCount}`) {
+        return { status: "unavailable" };
+      }
+      return { status: "accepted", providerMessageId: `provider-${options.clientMessageId}` };
+    }
+
+    override async revertFiles({ messageId }: { messageId: string }): Promise<void> {
+      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<AgentSession> {
+          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<void>();
   const release = deferred<void>();

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<void> {
     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);
   }

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;

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", "<task-notification><task-id>t1</task-id></task-notification>"),
+      { 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" }] } };

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"]);

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" }]);
+  });
+});

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" });

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<void> {
     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<ClaudeConversationRewindTarget> {
     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: (

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 });
+    }
+  },
+);

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<string | null> {
+  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;