mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-01 14:12:51 +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 { 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 () => {
|
||||
|
||||
@@ -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})`)));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 { 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 });
|
||||
}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user