mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-01 22:23:07 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f2d8cbf5b9 |
@@ -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"
|
||||||
|
);
|
||||||
|
}
|
||||||
@@ -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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,4 +1,5 @@
|
|||||||
import { randomUUID } from "node:crypto";
|
import { randomUUID } from "node:crypto";
|
||||||
|
import { subscribe, unsubscribe } from "node:diagnostics_channel";
|
||||||
import {
|
import {
|
||||||
mkdirSync,
|
mkdirSync,
|
||||||
mkdtempSync,
|
mkdtempSync,
|
||||||
@@ -10,9 +11,11 @@ import {
|
|||||||
} from "node:fs";
|
} from "node:fs";
|
||||||
import { tmpdir } from "node:os";
|
import { tmpdir } from "node:os";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
|
import { Worker } from "node:worker_threads";
|
||||||
import { afterEach, describe, expect, test } from "vitest";
|
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 {
|
import {
|
||||||
findSteeredMessageParent,
|
findSteeredMessageParent,
|
||||||
revertClaudeConversation,
|
revertClaudeConversation,
|
||||||
@@ -20,6 +23,7 @@ import {
|
|||||||
revertClaudeFiles,
|
revertClaudeFiles,
|
||||||
realClaudeRewindSdk,
|
realClaudeRewindSdk,
|
||||||
} from "./rewind.js";
|
} from "./rewind.js";
|
||||||
|
import { projectFolderSessionStore } from "./rewind-session-store.js";
|
||||||
import { FakeClaudeSdk } from "./test-rewind-claude-sdk.js";
|
import { FakeClaudeSdk } from "./test-rewind-claude-sdk.js";
|
||||||
|
|
||||||
describe("Claude rewind", () => {
|
describe("Claude rewind", () => {
|
||||||
@@ -278,8 +282,196 @@ describe("Claude rewind against the Claude SDK", () => {
|
|||||||
"tool output",
|
"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")(
|
test.runIf(process.platform === "linux")(
|
||||||
"releases the transcript descriptor after an early steer match",
|
"releases the transcript descriptor after an early steer match",
|
||||||
async () => {
|
async () => {
|
||||||
|
|||||||
@@ -1,13 +1,10 @@
|
|||||||
import { createReadStream } from "node:fs";
|
import { createReadStream } from "node:fs";
|
||||||
import { appendFile, readFile } from "node:fs/promises";
|
|
||||||
import path from "node:path";
|
|
||||||
import { createInterface } from "node:readline";
|
import { createInterface } from "node:readline";
|
||||||
import {
|
import { Worker } from "node:worker_threads";
|
||||||
forkSession as claudeForkSession,
|
import type { Query } from "@anthropic-ai/claude-agent-sdk";
|
||||||
type Query,
|
import pLimit from "p-limit";
|
||||||
type SessionStore,
|
import { resolveWorkerEntrypoint } from "../../../../utils/worker-entrypoint.js";
|
||||||
type SessionStoreEntry,
|
import type { ClaudeForkReply, ClaudeForkRequest } from "./rewind-worker.js";
|
||||||
} from "@anthropic-ai/claude-agent-sdk";
|
|
||||||
|
|
||||||
export interface ClaudeRewindSdk {
|
export interface ClaudeRewindSdk {
|
||||||
forkSession(
|
forkSession(
|
||||||
@@ -16,60 +13,42 @@ export interface ClaudeRewindSdk {
|
|||||||
): Promise<{ sessionId: string }>;
|
): Promise<{ sessionId: string }>;
|
||||||
}
|
}
|
||||||
|
|
||||||
// On its own, the SDK looks for the session under the daemon's CLAUDE_CONFIG_DIR, while a
|
const source = import.meta.url.endsWith(".ts");
|
||||||
// provider can point Claude Code at another one. Forking through a store rooted at the
|
const workerUrl = new URL(source ? "./rewind-worker.ts" : "./rewind-worker.js", import.meta.url);
|
||||||
// transcript's folder reads the session where Claude Code wrote it and writes the fork beside it.
|
const rewindWorker = resolveWorkerEntrypoint(workerUrl);
|
||||||
|
const forks = pLimit(1);
|
||||||
|
|
||||||
export const realClaudeRewindSdk: ClaudeRewindSdk = {
|
export const realClaudeRewindSdk: ClaudeRewindSdk = {
|
||||||
forkSession: (sessionId, options) =>
|
forkSession: (sessionId, options) =>
|
||||||
claudeForkSession(sessionId, {
|
forks(() =>
|
||||||
upToMessageId: options.upToMessageId,
|
forkInWorker({
|
||||||
sessionStore: projectFolderSessionStore(path.dirname(options.transcriptPath)),
|
sessionId,
|
||||||
}),
|
upToMessageId: options.upToMessageId,
|
||||||
|
transcriptPath: options.transcriptPath,
|
||||||
|
}),
|
||||||
|
),
|
||||||
};
|
};
|
||||||
|
|
||||||
function projectFolderSessionStore(projectFolder: string): SessionStore {
|
function forkInWorker(request: ClaudeForkRequest): Promise<{ sessionId: string }> {
|
||||||
const transcriptPath = (sessionId: string) => path.join(projectFolder, `${sessionId}.jsonl`);
|
return new Promise((resolve, reject) => {
|
||||||
return {
|
const worker = new Worker(rewindWorker.entrypoint, {
|
||||||
async load({ sessionId }) {
|
eval: rewindWorker.eval,
|
||||||
let text: string;
|
execArgv: [],
|
||||||
try {
|
workerData: request,
|
||||||
text = await readFile(transcriptPath(sessionId), "utf8");
|
});
|
||||||
} catch (error) {
|
worker.once("message", (reply: ClaudeForkReply) => {
|
||||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
|
void worker.terminate();
|
||||||
throw error;
|
if (reply.kind === "forked") {
|
||||||
|
resolve({ sessionId: reply.sessionId });
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
return parseTranscriptEntries(text);
|
const error = new Error(reply.message);
|
||||||
},
|
error.stack = reply.stack;
|
||||||
async append({ sessionId }, entries) {
|
reject(Object.assign(error, reply.properties));
|
||||||
const lines = entries.map((entry) => `${JSON.stringify(entry)}\n`).join("");
|
});
|
||||||
await appendFile(transcriptPath(sessionId), lines, "utf8");
|
worker.once("error", reject);
|
||||||
},
|
worker.once("exit", (code) => reject(new Error(`Claude rewind worker exited (code ${code})`)));
|
||||||
};
|
});
|
||||||
}
|
|
||||||
|
|
||||||
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"
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -1,8 +1,7 @@
|
|||||||
import { createRequire } from "node:module";
|
|
||||||
import { pathToFileURL } from "node:url";
|
|
||||||
import { Worker } from "node:worker_threads";
|
import { Worker } from "node:worker_threads";
|
||||||
import pLimit from "p-limit";
|
import pLimit from "p-limit";
|
||||||
import { ZodError } from "zod";
|
import { ZodError } from "zod";
|
||||||
|
import { resolveWorkerEntrypoint } from "../../utils/worker-entrypoint.js";
|
||||||
import type * as build from "./compiler-build.js";
|
import type * as build from "./compiler-build.js";
|
||||||
import type { PluginManifest } from "./manifest.js";
|
import type { PluginManifest } from "./manifest.js";
|
||||||
import type { CompilerReply } from "./compiler-worker.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
|
// Resolve the development loader once at module initialization, never during a
|
||||||
// compile request. Packaged workers execute ordinary JavaScript without tsx.
|
// compile request. Packaged workers execute ordinary JavaScript without tsx.
|
||||||
const workerEntrypoint = source
|
const compilerWorker = resolveWorkerEntrypoint(workerUrl);
|
||||||
? `import(${JSON.stringify(pathToFileURL(createRequire(import.meta.url).resolve("tsx/esm/api")).href)}).then(({ register }) => { register(); return import(${JSON.stringify(workerUrl.href)}); });`
|
|
||||||
: workerUrl;
|
|
||||||
|
|
||||||
// Reuse the isolate during a build batch (notably daemon bootstrap), then release
|
// 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.
|
// TypeScript's heap when idle. Only this owner admits work or stops the worker.
|
||||||
@@ -74,7 +71,10 @@ class PluginCompiler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private startWorker(): Worker {
|
private startWorker(): Worker {
|
||||||
const worker = new Worker(workerEntrypoint, { eval: source, execArgv: [] });
|
const worker = new Worker(compilerWorker.entrypoint, {
|
||||||
|
eval: compilerWorker.eval,
|
||||||
|
execArgv: [],
|
||||||
|
});
|
||||||
this.worker = worker;
|
this.worker = worker;
|
||||||
// Keep error/exit ownership while idle too. The next caller gets a fresh
|
// Keep error/exit ownership while idle too. The next caller gets a fresh
|
||||||
// isolate after a crash; author build commands are never retried.
|
// isolate after a crash; author build commands are never retried.
|
||||||
|
|||||||
@@ -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));
|
||||||
@@ -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"));
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -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,
|
||||||
|
};
|
||||||
|
}
|
||||||
@@ -1,6 +1,7 @@
|
|||||||
import assert from "node:assert/strict";
|
import assert from "node:assert/strict";
|
||||||
import { execFile } from "node:child_process";
|
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 os from "node:os";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import test from "node:test";
|
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 repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..");
|
||||||
const bridgeModule = "packages/server/dist/server/server/agent/providers/opencode/bridge.js";
|
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.
|
// Mirrors nix/package.nix's installPhase: copy every traced path into $out.
|
||||||
async function installTracedDaemon(outRoot) {
|
async function installTracedDaemon(outRoot) {
|
||||||
@@ -17,12 +24,14 @@ async function installTracedDaemon(outRoot) {
|
|||||||
[path.join(repoRoot, "scripts/trace-daemon.mjs")],
|
[path.join(repoRoot, "scripts/trace-daemon.mjs")],
|
||||||
{ cwd: repoRoot, maxBuffer: 64 * 1024 * 1024 },
|
{ 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), {
|
await cp(path.join(repoRoot, file), path.join(outRoot, file), {
|
||||||
recursive: true,
|
recursive: true,
|
||||||
verbatimSymlinks: true,
|
verbatimSymlinks: true,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
return files;
|
||||||
}
|
}
|
||||||
|
|
||||||
test("traced daemon closure ships the OpenCode bridge plugin for every OpenCode version", async () => {
|
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 });
|
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 });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user