fix(claude): keep the daemon responsive during conversation rewind

The SDK parses and copies the complete transcript while forking a Claude
conversation. Running that work in the daemon's event loop blocks all
connected clients, even though only one conversation is being rewound.

Run the fork in a worker, stream transcript records, and append in bounded
chunks. Preserve the SDK's fork content, transcript location, and error
fields. Share the existing compiler worker loader and verify that the
installed daemon closure contains and starts the rewind worker.

Retain Dominic Anaeto's implementation and regressions from #8444, rebased
onto main to include the correction for the moved worktrees documentation
page in the website test.

Fixes #8209

Co-authored-by: Dominic Anaeto <dominic.anaeto@gmail.com>

packages/server/src/server/agent/providers/claude/rewind-session-store.ts

@@ -0,0 +1,63 @@
+import { createReadStream } from "node:fs";
+import { appendFile } from "node:fs/promises";
+import path from "node:path";
+import { createInterface } from "node:readline";
+import type { SessionStore, SessionStoreEntry } from "@anthropic-ai/claude-agent-sdk";
+
+const APPEND_CHUNK_LENGTH = 1 << 20;
+
+export function projectFolderSessionStore(projectFolder: string): SessionStore {
+  const transcriptPath = (sessionId: string) => path.join(projectFolder, `${sessionId}.jsonl`);
+  return {
+    async load({ sessionId }) {
+      const input = createReadStream(transcriptPath(sessionId), { encoding: "utf8" });
+      const lines = createInterface({ input, crlfDelay: Infinity });
+      const entries: SessionStoreEntry[] = [];
+      try {
+        for await (const line of lines) {
+          const entry = parseTranscriptEntry(line);
+          if (entry) entries.push(entry);
+        }
+      } catch (error) {
+        if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
+        throw error;
+      } finally {
+        lines.close();
+        input.destroy();
+      }
+      return entries;
+    },
+    async append({ sessionId }, entries) {
+      let chunk = "";
+      for (const entry of entries) {
+        chunk += `${JSON.stringify(entry)}\n`;
+        if (chunk.length >= APPEND_CHUNK_LENGTH) {
+          await appendFile(transcriptPath(sessionId), chunk, "utf8");
+          chunk = "";
+        }
+      }
+      await appendFile(transcriptPath(sessionId), chunk, "utf8");
+    },
+  };
+}
+
+function parseTranscriptEntry(line: string): SessionStoreEntry | null {
+  if (!line.trim()) return null;
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(line);
+  } catch {
+    // Skip a line that is not JSON, as the SDK's own transcript reader does.
+    return null;
+  }
+  return isSessionStoreEntry(parsed) ? parsed : null;
+}
+
+function isSessionStoreEntry(value: unknown): value is SessionStoreEntry {
+  return (
+    typeof value === "object" &&
+    value !== null &&
+    !Array.isArray(value) &&
+    typeof (value as { type?: unknown }).type === "string"
+  );
+}

packages/server/src/server/agent/providers/claude/rewind-worker.ts

@@ -0,0 +1,43 @@
+import path from "node:path";
+import { parentPort, workerData } from "node:worker_threads";
+import { forkSession } from "@anthropic-ai/claude-agent-sdk";
+import { projectFolderSessionStore } from "./rewind-session-store.js";
+
+export interface ClaudeForkRequest {
+  sessionId: string;
+  upToMessageId: string;
+  transcriptPath: string;
+}
+
+export type ClaudeForkReply =
+  | { kind: "forked"; sessionId: string }
+  | {
+      kind: "failed";
+      message: string;
+      stack: string | undefined;
+      properties: Record<string, unknown>;
+    };
+
+const port = parentPort!;
+const request: ClaudeForkRequest = workerData;
+try {
+  // On its own, the SDK looks for the session under the daemon's CLAUDE_CONFIG_DIR, while a
+  // provider can point Claude Code at another one. Forking through a store rooted at the
+  // transcript's folder reads the session where Claude Code wrote it and writes the fork beside it.
+  const fork = await forkSession(request.sessionId, {
+    upToMessageId: request.upToMessageId,
+    sessionStore: projectFolderSessionStore(path.dirname(request.transcriptPath)),
+  });
+  port.postMessage({ kind: "forked", sessionId: fork.sessionId } satisfies ClaudeForkReply);
+} catch (cause) {
+  const error = cause instanceof Error ? cause : new Error(String(cause));
+  const failure = { kind: "failed", message: error.message, stack: error.stack } as const;
+  try {
+    port.postMessage({ ...failure, properties: { ...error } } satisfies ClaudeForkReply);
+  } catch (cloneError) {
+    if (!(cloneError instanceof DOMException) || cloneError.name !== "DataCloneError") {
+      throw cloneError;
+    }
+    port.postMessage({ ...failure, properties: {} } satisfies ClaudeForkReply);
+  }
+}

packages/server/src/server/agent/providers/claude/rewind.test.ts

@@ -1,4 +1,5 @@
 import { randomUUID } from "node:crypto";
+import { subscribe, unsubscribe } from "node:diagnostics_channel";
 import {
   mkdirSync,
   mkdtempSync,
@@ -10,9 +11,11 @@ import {
 } from "node:fs";
 import { tmpdir } from "node:os";
 import path from "node:path";
+import { Worker } from "node:worker_threads";
 import { afterEach, describe, expect, test } from "vitest";
-import type { Query } from "@anthropic-ai/claude-agent-sdk";
+import { forkSession, type Query } from "@anthropic-ai/claude-agent-sdk";
 
+import { measureMainThreadWork } from "../../../../test-utils/main-thread-work.js";
 import {
   findSteeredMessageParent,
   revertClaudeConversation,
@@ -20,6 +23,7 @@ import {
   revertClaudeFiles,
   realClaudeRewindSdk,
 } from "./rewind.js";
+import { projectFolderSessionStore } from "./rewind-session-store.js";
 import { FakeClaudeSdk } from "./test-rewind-claude-sdk.js";
 
 describe("Claude rewind", () => {
@@ -278,8 +282,196 @@ describe("Claude rewind against the Claude SDK", () => {
       "tool output",
     ]);
   });
+
+  test("writes the fork beside the original and leaves the original untouched", async () => {
+    tempRoot = mkdtempSync(path.join(tmpdir(), "claude-rewind-fork-"));
+    const projectDir = path.join(tempRoot, "projects", "-repo");
+    mkdirSync(projectDir, { recursive: true });
+    const sessionId = randomUUID();
+    const firstUser = randomUUID();
+    const firstReply = randomUUID();
+    const secondUser = randomUUID();
+    const transcriptPath = path.join(projectDir, `${sessionId}.jsonl`);
+    writeTranscript(transcriptPath, sessionId, [
+      { type: "user", uuid: firstUser, parentUuid: null, text: "first" },
+      { type: "assistant", uuid: firstReply, parentUuid: firstUser, text: "first reply" },
+      { type: "user", uuid: secondUser, parentUuid: firstReply, text: "second" },
+      { type: "assistant", uuid: randomUUID(), parentUuid: secondUser, text: "second reply" },
+    ]);
+    const original = readFileSync(transcriptPath, "utf8");
+
+    const fork = await realClaudeRewindSdk.forkSession(sessionId, {
+      upToMessageId: secondUser,
+      transcriptPath,
+    });
+
+    const forkPath = path.join(projectDir, `${fork.sessionId}.jsonl`);
+    expect(readdirSync(projectDir).sort()).toEqual(
+      [`${sessionId}.jsonl`, `${fork.sessionId}.jsonl`].sort(),
+    );
+    expect(readMessageTexts(forkPath)).toEqual(["first", "first reply", "second"]);
+    const forkSessionIds = readFileSync(forkPath, "utf8")
+      .trim()
+      .split("\n")
+      .map((line) => (JSON.parse(line) as { sessionId: string }).sessionId);
+    expect(forkSessionIds).toEqual([
+      fork.sessionId,
+      fork.sessionId,
+      fork.sessionId,
+      fork.sessionId,
+    ]);
+    expect(readFileSync(transcriptPath, "utf8")).toBe(original);
+  });
+
+  test("rejects with the SDK's own error when the transcript is missing", async () => {
+    tempRoot = mkdtempSync(path.join(tmpdir(), "claude-rewind-missing-"));
+    const sessionId = randomUUID();
+
+    await expect(
+      realClaudeRewindSdk.forkSession(sessionId, {
+        upToMessageId: randomUUID(),
+        transcriptPath: path.join(tempRoot, `${sessionId}.jsonl`),
+      }),
+    ).rejects.toThrow(new Error(`Session ${sessionId} not found`));
+  });
+
+  test("rejects with the name and message of the error the SDK raises", async () => {
+    tempRoot = mkdtempSync(path.join(tmpdir(), "claude-rewind-invalid-"));
+    const sessionId = randomUUID();
+    const sdkError = await forkSession(sessionId, { upToMessageId: "not-a-uuid" }).then(
+      () => null,
+      (error: Error) => error,
+    );
+    expect(sdkError).toBeInstanceOf(Error);
+
+    await expect(
+      realClaudeRewindSdk.forkSession(sessionId, {
+        upToMessageId: "not-a-uuid",
+        transcriptPath: path.join(tempRoot, `${sessionId}.jsonl`),
+      }),
+    ).rejects.toMatchObject({ name: sdkError?.name, message: sdkError?.message });
+  });
+
+  test("keeps a filesystem error's code when the fork fails in the worker", async () => {
+    tempRoot = mkdtempSync(path.join(tmpdir(), "claude-rewind-eisdir-"));
+    const sessionId = randomUUID();
+    const transcriptPath = path.join(tempRoot, `${sessionId}.jsonl`);
+    mkdirSync(transcriptPath);
+
+    await expect(
+      realClaudeRewindSdk.forkSession(sessionId, { upToMessageId: randomUUID(), transcriptPath }),
+    ).rejects.toMatchObject({ name: "Error", code: "EISDIR" });
+  });
+
+  test("forks one conversation at a time", async () => {
+    tempRoot = mkdtempSync(path.join(tmpdir(), "claude-rewind-queue-"));
+    const projectDir = path.join(tempRoot, "projects", "-repo");
+    mkdirSync(projectDir, { recursive: true });
+    const forks = [randomUUID(), randomUUID()].map((sessionId) => {
+      const user = randomUUID();
+      const transcriptPath = path.join(projectDir, `${sessionId}.jsonl`);
+      writeTranscript(transcriptPath, sessionId, [
+        { type: "user", uuid: user, parentUuid: null, text: "first" },
+        { type: "assistant", uuid: randomUUID(), parentUuid: user, text: "first reply" },
+      ]);
+      return { sessionId, upToMessageId: user, transcriptPath };
+    });
+    const events: string[] = [];
+    let started = 0;
+    function recordWorker(message: unknown): void {
+      if (typeof message !== "object" || message === null || !("worker" in message)) return;
+      if (!(message.worker instanceof Worker)) return;
+      const fork = ++started;
+      events.push(`start ${fork}`);
+      message.worker.once("message", () => events.push(`reply ${fork}`));
+    }
+
+    subscribe("worker_threads", recordWorker);
+    try {
+      await Promise.all(
+        forks.map(({ sessionId, upToMessageId, transcriptPath }) =>
+          realClaudeRewindSdk.forkSession(sessionId, { upToMessageId, transcriptPath }),
+        ),
+      );
+    } finally {
+      unsubscribe("worker_threads", recordWorker);
+    }
+
+    expect(events).toEqual(["start 1", "reply 1", "start 2", "reply 2"]);
+  });
+
+  test("forks a long transcript without blocking the daemon's event loop", async () => {
+    tempRoot = mkdtempSync(path.join(tmpdir(), "claude-rewind-long-"));
+    const projectDir = path.join(tempRoot, "projects", "-repo");
+    mkdirSync(projectDir, { recursive: true });
+    const sessionId = randomUUID();
+    const transcriptPath = path.join(projectDir, `${sessionId}.jsonl`);
+    const turns = 43_000;
+    const lastUser = writeLongTranscript(transcriptPath, sessionId, turns);
+
+    const { result: fork, blockingMs } = await measureMainThreadWork(
+      "claude fork main-thread work",
+      () => realClaudeRewindSdk.forkSession(sessionId, { upToMessageId: lastUser, transcriptPath }),
+    );
+
+    expect(blockingMs).toBeLessThan(50);
+    const forkLines = readFileSync(path.join(projectDir, `${fork.sessionId}.jsonl`), "utf8")
+      .trim()
+      .split("\n");
+    expect(forkLines).toHaveLength(turns * 2);
+    expect(JSON.parse(forkLines[turns * 2 - 2]!).forkedFrom).toEqual({
+      sessionId,
+      messageUuid: lastUser,
+    });
+  });
 });
 
+describe("Claude project folder session store", () => {
+  let root: string | null = null;
+
+  afterEach(() => {
+    if (root) rmSync(root, { recursive: true, force: true });
+    root = null;
+  });
+
+  test("appends entries after existing ones, byte for byte, and loads them back", async () => {
+    root = mkdtempSync(path.join(tmpdir(), "claude-rewind-store-"));
+    const store = projectFolderSessionStore(root);
+    const key = { projectKey: "-repo", sessionId: randomUUID() };
+    const transcriptPath = path.join(root, `${key.sessionId}.jsonl`);
+    const existing = { type: "user", uuid: randomUUID(), message: { content: "already there" } };
+    writeFileSync(transcriptPath, `${JSON.stringify(existing)}\n`);
+    const entries = Array.from({ length: 4_000 }, (_, index) => ({
+      type: "assistant",
+      uuid: randomUUID(),
+      message: { content: `reply ${index} ✓ ${"x".repeat(1_000)}` },
+    }));
+
+    await store.append(key, entries);
+
+    expect(readFileSync(transcriptPath, "utf8")).toBe(
+      [existing, ...entries].map((entry) => `${JSON.stringify(entry)}\n`).join(""),
+    );
+    expect(await store.load(key)).toEqual([existing, ...entries]);
+  });
+});
+
+function writeLongTranscript(transcriptPath: string, sessionId: string, turns: number): string {
+  const toolOutput = 'line 1: "quoted" output\twith\\escapes\n'.repeat(4);
+  const lines: TranscriptLine[] = [];
+  let parentUuid: string | null = null;
+  let lastUser = "";
+  for (let turn = 0; turn < turns; turn++) {
+    lastUser = randomUUID();
+    const reply = randomUUID();
+    lines.push({ type: "user", uuid: lastUser, parentUuid, text: `question ${turn}` });
+    lines.push({ type: "assistant", uuid: reply, parentUuid: lastUser, text: toolOutput });
+    parentUuid = reply;
+  }
+  writeTranscript(transcriptPath, sessionId, lines);
+  return lastUser;
+}
+
 test.runIf(process.platform === "linux")(
   "releases the transcript descriptor after an early steer match",
   async () => {

packages/server/src/server/agent/providers/claude/rewind.ts

@@ -1,13 +1,10 @@
 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,
-  type SessionStore,
-  type SessionStoreEntry,
-} from "@anthropic-ai/claude-agent-sdk";
+import { Worker } from "node:worker_threads";
+import type { Query } from "@anthropic-ai/claude-agent-sdk";
+import pLimit from "p-limit";
+import { resolveWorkerEntrypoint } from "../../../../utils/worker-entrypoint.js";
+import type { ClaudeForkReply, ClaudeForkRequest } from "./rewind-worker.js";
 
 export interface ClaudeRewindSdk {
   forkSession(
@@ -16,60 +13,42 @@ export interface ClaudeRewindSdk {
   ): Promise<{ sessionId: string }>;
 }
 
-// On its own, the SDK looks for the session under the daemon's CLAUDE_CONFIG_DIR, while a
-// provider can point Claude Code at another one. Forking through a store rooted at the
-// transcript's folder reads the session where Claude Code wrote it and writes the fork beside it.
+const source = import.meta.url.endsWith(".ts");
+const workerUrl = new URL(source ? "./rewind-worker.ts" : "./rewind-worker.js", import.meta.url);
+const rewindWorker = resolveWorkerEntrypoint(workerUrl);
+const forks = pLimit(1);
+
 export const realClaudeRewindSdk: ClaudeRewindSdk = {
   forkSession: (sessionId, options) =>
-    claudeForkSession(sessionId, {
-      upToMessageId: options.upToMessageId,
-      sessionStore: projectFolderSessionStore(path.dirname(options.transcriptPath)),
-    }),
+    forks(() =>
+      forkInWorker({
+        sessionId,
+        upToMessageId: options.upToMessageId,
+        transcriptPath: options.transcriptPath,
+      }),
+    ),
 };
 
-function projectFolderSessionStore(projectFolder: string): SessionStore {
-  const transcriptPath = (sessionId: string) => path.join(projectFolder, `${sessionId}.jsonl`);
-  return {
-    async load({ sessionId }) {
-      let text: string;
-      try {
-        text = await readFile(transcriptPath(sessionId), "utf8");
-      } catch (error) {
-        if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
-        throw error;
+function forkInWorker(request: ClaudeForkRequest): Promise<{ sessionId: string }> {
+  return new Promise((resolve, reject) => {
+    const worker = new Worker(rewindWorker.entrypoint, {
+      eval: rewindWorker.eval,
+      execArgv: [],
+      workerData: request,
+    });
+    worker.once("message", (reply: ClaudeForkReply) => {
+      void worker.terminate();
+      if (reply.kind === "forked") {
+        resolve({ sessionId: reply.sessionId });
+        return;
       }
-      return parseTranscriptEntries(text);
-    },
-    async append({ sessionId }, entries) {
-      const lines = entries.map((entry) => `${JSON.stringify(entry)}\n`).join("");
-      await appendFile(transcriptPath(sessionId), lines, "utf8");
-    },
-  };
-}
-
-function parseTranscriptEntries(text: string): SessionStoreEntry[] {
-  const entries: SessionStoreEntry[] = [];
-  for (const line of text.split("\n")) {
-    if (!line.trim()) continue;
-    let parsed: unknown;
-    try {
-      parsed = JSON.parse(line);
-    } catch {
-      // Skip a line that is not JSON, as the SDK's own transcript reader does.
-      continue;
-    }
-    if (isSessionStoreEntry(parsed)) entries.push(parsed);
-  }
-  return entries;
-}
-
-function isSessionStoreEntry(value: unknown): value is SessionStoreEntry {
-  return (
-    typeof value === "object" &&
-    value !== null &&
-    !Array.isArray(value) &&
-    typeof (value as { type?: unknown }).type === "string"
-  );
+      const error = new Error(reply.message);
+      error.stack = reply.stack;
+      reject(Object.assign(error, reply.properties));
+    });
+    worker.once("error", reject);
+    worker.once("exit", (code) => reject(new Error(`Claude rewind worker exited (code ${code})`)));
+  });
 }
 
 /**

packages/server/src/server/plugins/compiler.ts

@@ -1,8 +1,7 @@
-import { createRequire } from "node:module";
-import { pathToFileURL } from "node:url";
 import { Worker } from "node:worker_threads";
 import pLimit from "p-limit";
 import { ZodError } from "zod";
+import { resolveWorkerEntrypoint } from "../../utils/worker-entrypoint.js";
 import type * as build from "./compiler-build.js";
 import type { PluginManifest } from "./manifest.js";
 import type { CompilerReply } from "./compiler-worker.js";
@@ -37,9 +36,7 @@ const workerUrl = new URL(
 );
 // Resolve the development loader once at module initialization, never during a
 // compile request. Packaged workers execute ordinary JavaScript without tsx.
-const workerEntrypoint = source
-  ? `import(${JSON.stringify(pathToFileURL(createRequire(import.meta.url).resolve("tsx/esm/api")).href)}).then(({ register }) => { register(); return import(${JSON.stringify(workerUrl.href)}); });`
-  : workerUrl;
+const compilerWorker = resolveWorkerEntrypoint(workerUrl);
 
 // Reuse the isolate during a build batch (notably daemon bootstrap), then release
 // TypeScript's heap when idle. Only this owner admits work or stops the worker.
@@ -74,7 +71,10 @@ class PluginCompiler {
   }
 
   private startWorker(): Worker {
-    const worker = new Worker(workerEntrypoint, { eval: source, execArgv: [] });
+    const worker = new Worker(compilerWorker.entrypoint, {
+      eval: compilerWorker.eval,
+      execArgv: [],
+    });
     this.worker = worker;
     // Keep error/exit ownership while idle too. The next caller gets a fresh
     // isolate after a crash; author build commands are never retried.

packages/server/src/utils/test-fixtures/expand-tilde-worker.ts

@@ -0,0 +1,6 @@
+import { parentPort, workerData } from "node:worker_threads";
+import { expandTilde } from "../path.js";
+
+const port = parentPort!;
+const input: string = workerData;
+port.postMessage(expandTilde(input));

packages/server/src/utils/worker-entrypoint.test.ts

@@ -0,0 +1,34 @@
+import { once } from "node:events";
+import { Worker } from "node:worker_threads";
+import { describe, expect, test } from "vitest";
+
+import { expandTilde } from "./path.js";
+import { resolveWorkerEntrypoint } from "./worker-entrypoint.js";
+
+describe("worker entrypoint", () => {
+  test("starts a compiled worker from its own URL", () => {
+    const entry = resolveWorkerEntrypoint(
+      new URL("file:///app/dist/server/server/plugins/compiler-worker.js"),
+    );
+
+    expect({ ...entry, entrypoint: String(entry.entrypoint) }).toEqual({
+      entrypoint: "file:///app/dist/server/server/plugins/compiler-worker.js",
+      eval: false,
+    });
+  });
+
+  test("runs a TypeScript worker, with its .js imports, through tsx", async () => {
+    const entry = resolveWorkerEntrypoint(
+      new URL("./test-fixtures/expand-tilde-worker.ts", import.meta.url),
+    );
+    const worker = new Worker(entry.entrypoint, {
+      eval: entry.eval,
+      execArgv: [],
+      workerData: "~/project",
+    });
+
+    const [expanded] = await once(worker, "message");
+
+    expect(expanded).toBe(expandTilde("~/project"));
+  });
+});

packages/server/src/utils/worker-entrypoint.ts

@@ -0,0 +1,16 @@
+import { createRequire } from "node:module";
+import { pathToFileURL } from "node:url";
+
+export interface WorkerEntrypoint {
+  entrypoint: string | URL;
+  eval: boolean;
+}
+
+export function resolveWorkerEntrypoint(workerUrl: URL): WorkerEntrypoint {
+  if (!workerUrl.pathname.endsWith(".ts")) return { entrypoint: workerUrl, eval: false };
+  const tsxUrl = pathToFileURL(createRequire(import.meta.url).resolve("tsx/esm/api")).href;
+  return {
+    entrypoint: `import(${JSON.stringify(tsxUrl)}).then(({ register }) => { register(); return import(${JSON.stringify(workerUrl.href)}); });`,
+    eval: true,
+  };
+}