diff --git a/packages/server/src/server/agent/providers/claude/rewind-session-store.ts b/packages/server/src/server/agent/providers/claude/rewind-session-store.ts new file mode 100644 index 0000000000..0b028cc927 --- /dev/null +++ b/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" + ); +} diff --git a/packages/server/src/server/agent/providers/claude/rewind-worker.ts b/packages/server/src/server/agent/providers/claude/rewind-worker.ts new file mode 100644 index 0000000000..73c8d2b87a --- /dev/null +++ b/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; + }; + +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); + } +} 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 226f41d39c..81d4ddcdb8 100644 --- a/packages/server/src/server/agent/providers/claude/rewind.test.ts +++ b/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 () => { diff --git a/packages/server/src/server/agent/providers/claude/rewind.ts b/packages/server/src/server/agent/providers/claude/rewind.ts index 75607c184b..ebe192048a 100644 --- a/packages/server/src/server/agent/providers/claude/rewind.ts +++ b/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})`))); + }); } /** diff --git a/packages/server/src/server/plugins/compiler.ts b/packages/server/src/server/plugins/compiler.ts index b96c8ecd4e..183e5e754f 100644 --- a/packages/server/src/server/plugins/compiler.ts +++ b/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. diff --git a/packages/server/src/utils/test-fixtures/expand-tilde-worker.ts b/packages/server/src/utils/test-fixtures/expand-tilde-worker.ts new file mode 100644 index 0000000000..2464dd4ac4 --- /dev/null +++ b/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)); diff --git a/packages/server/src/utils/worker-entrypoint.test.ts b/packages/server/src/utils/worker-entrypoint.test.ts new file mode 100644 index 0000000000..42e28b491a --- /dev/null +++ b/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")); + }); +}); diff --git a/packages/server/src/utils/worker-entrypoint.ts b/packages/server/src/utils/worker-entrypoint.ts new file mode 100644 index 0000000000..a52343bc62 --- /dev/null +++ b/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, + }; +} diff --git a/scripts/trace-daemon-dist.test.mjs b/scripts/trace-daemon-dist.test.mjs index a2c2149daa..5d086ae9e6 100644 --- a/scripts/trace-daemon-dist.test.mjs +++ b/scripts/trace-daemon-dist.test.mjs @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import { execFile } from "node:child_process"; -import { cp, mkdtemp, rm } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; +import { cp, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import test from "node:test"; @@ -9,6 +10,12 @@ import { promisify } from "node:util"; const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); const bridgeModule = "packages/server/dist/server/server/agent/providers/opencode/bridge.js"; +const claudeDist = "packages/server/dist/server/server/agent/providers/claude"; +const workerThreadEntrypoints = [ + "packages/server/dist/server/server/plugins/compiler-worker.js", + `${claudeDist}/rewind-worker.js`, + `${claudeDist}/rewind-session-store.js`, +]; // Mirrors nix/package.nix's installPhase: copy every traced path into $out. async function installTracedDaemon(outRoot) { @@ -17,12 +24,14 @@ async function installTracedDaemon(outRoot) { [path.join(repoRoot, "scripts/trace-daemon.mjs")], { cwd: repoRoot, maxBuffer: 64 * 1024 * 1024 }, ); - for (const file of stdout.split("\n").filter(Boolean)) { + const files = stdout.split("\n").filter(Boolean); + for (const file of files) { await cp(path.join(repoRoot, file), path.join(outRoot, file), { recursive: true, verbatimSymlinks: true, }); } + return files; } test("traced daemon closure ships the OpenCode bridge plugin for every OpenCode version", async () => { @@ -44,3 +53,32 @@ test("traced daemon closure ships the OpenCode bridge plugin for every OpenCode await rm(outRoot, { recursive: true, force: true }); } }); + +test("traced daemon closure ships the worker-thread entrypoints and what they load", async () => { + const outRoot = await mkdtemp(path.join(os.tmpdir(), "trace-daemon-dist-")); + try { + const files = await installTracedDaemon(outRoot); + for (const file of workerThreadEntrypoints) { + assert.ok(files.includes(file), `${file} is missing from the traced closure`); + } + + const { realClaudeRewindSdk } = await import( + pathToFileURL(path.join(outRoot, claudeDist, "rewind.js")).href + ); + const sessionId = randomUUID(); + const user = randomUUID(); + const transcriptPath = path.join(outRoot, `${sessionId}.jsonl`); + await writeFile( + transcriptPath, + `${JSON.stringify({ type: "user", uuid: user, parentUuid: null, sessionId, message: { role: "user", content: "first" } })}\n`, + ); + const fork = await realClaudeRewindSdk.forkSession(sessionId, { + upToMessageId: user, + transcriptPath, + }); + const forked = await readFile(path.join(outRoot, `${fork.sessionId}.jsonl`), "utf8"); + assert.equal(JSON.parse(forked.split("\n")[0]).forkedFrom.messageUuid, user); + } finally { + await rm(outRoot, { recursive: true, force: true }); + } +});