diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e9a7df8bdc4..fe9ee0c9615 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -535,6 +535,15 @@ Warm sandbox reuse must match the current host Git tip and branch as well as the file snapshot and saved stamp, including managed nested repositories. A history or branch mismatch restages the host before the next run begins. +Disk-backed native sandbox workspace snapshots run in at most two worker threads per +controller process. Each capture retains the same file hashes, modes, symlinks, +exclusions and ignored-path manifest as an inline capture. A worker seals its +private manifest before handing it to the controller; ordinary snapshot disposal +removes that worker's temporary directory. A failed capture removes its temporary +state before releasing its worker slot. This scheduling boundary does not replace +source authorization, warm-stamp validation, Git identity checks or restore +conflict detection. Legacy adapter snapshots remain inline. + ### Remote Codex model compatibility Fresh remote Codex Runner runs check the selected image CLI before saving the diff --git a/doc/plans/2026-10-15-workspace-snapshot-isolation.md b/doc/plans/2026-10-15-workspace-snapshot-isolation.md new file mode 100644 index 00000000000..46d4699416b --- /dev/null +++ b/doc/plans/2026-10-15-workspace-snapshot-isolation.md @@ -0,0 +1,37 @@ +# Isolate workspace snapshot work from the controller + +## Measured problem + +A warm native Codex run spent 12.2 seconds capturing an 11,380-entry workspace +baseline. A separate process on the same host produced the exact saved snapshot +hash in 4.3–4.6 seconds. These were separate observation windows; their difference +alone does not establish a controlled performance result. + +A local full tracked-tree fixture reproduces controller contention with a fixed +60% main-thread busy load. Alternating inline and worker captures preserves every +entry and hash. The compiled JavaScript implementation reduced the median from +1,511ms to 947ms across three samples of each path, exceeding the declared 30% +reduction criterion. This fixture measures snapshot work, not complete agent +response latency. + +## Change boundary + +- Move disk-backed native sandbox snapshot capture into at most two concurrent workers. +- Keep legacy adapter snapshot capture inline; worker startup must not enter their + existing lifecycle deadlines. +- Keep the existing snapshot algorithm, privacy decisions and restore semantics. +- Give each worker a private temporary directory; transfer sealed manifest + ownership only after successful completion and clean failed captures before + releasing capacity. +- Preserve in-memory captures and filesystem error codes. +- Keep source-worker loading available in development without requiring a + development loader for compiled JavaScript workers. + +## Finish conditions + +Correctness, confinement, cleanup, checkpoint, restore and native recovery tests +must pass, as must full repository validation and current-head PR checks/review. +Then repeat a real warm Codex greeting on the intended staging workspace using +the same model, effort, Daytona provider and isolated Git worktree. Snapshot +benchmarks, passing checks and deployment success do not establish the separate +under-30-second reply objective. diff --git a/packages/adapter-utils/package.json b/packages/adapter-utils/package.json index 5c928747a03..8f424019ab8 100644 --- a/packages/adapter-utils/package.json +++ b/packages/adapter-utils/package.json @@ -49,6 +49,7 @@ }, "devDependencies": { "@types/node": "^24.0.0", + "tsx": "^4.23.15", "typescript": "^7.0.2" }, "engines": { diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index 0fad2165327..50265a7bfd0 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -545,6 +545,7 @@ export async function prepareCommandManagedRuntime(input: { runtimeRootDir?: string; syncWorkspace?: boolean; workspaceInboundMode?: WorkspaceInboundMode; + workspaceSnapshotIsolation?: "worker"; workspaceDurableSeed?: WorkspaceDurableSeedPaths; workspaceCheckpoint?: boolean; onWorkspaceCheckpoint?: (metrics: WorkspaceCheckpointMetrics) => Promise; @@ -619,6 +620,7 @@ export async function prepareCommandManagedRuntime(input: { runtimeRootDir: input.runtimeRootDir, syncWorkspace: input.syncWorkspace, workspaceInboundMode: input.workspaceInboundMode, + workspaceSnapshotIsolation: input.workspaceSnapshotIsolation, workspaceDurableSeed: input.workspaceDurableSeed, workspaceCheckpoint: input.workspaceCheckpoint, onWorkspaceCheckpoint: input.onWorkspaceCheckpoint, @@ -670,6 +672,7 @@ export async function prepareCommandManagedRuntime(input: { runtimeRootDir: input.runtimeRootDir, syncWorkspace: input.syncWorkspace, workspaceInboundMode: input.workspaceInboundMode, + workspaceSnapshotIsolation: input.workspaceSnapshotIsolation, workspaceDurableSeed: input.workspaceDurableSeed, workspaceCheckpoint: input.workspaceCheckpoint, onWorkspaceCheckpoint: input.onWorkspaceCheckpoint, diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 5a5b9d79b78..79c0a4495f3 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -1439,6 +1439,7 @@ export async function prepareAdapterExecutionTargetRuntime(input: { runtimeRootDir?: string; syncWorkspace?: boolean; workspaceInboundMode?: WorkspaceInboundMode; + workspaceSnapshotIsolation?: "worker"; workspaceDurableSeed?: WorkspaceDurableSeedPaths; workspaceCheckpoint?: boolean; onWorkspaceCheckpoint?: (metrics: WorkspaceCheckpointMetrics) => Promise; @@ -1533,6 +1534,7 @@ export async function prepareAdapterExecutionTargetRuntime(input: { runtimeRootDir: input.runtimeRootDir, syncWorkspace: input.syncWorkspace, workspaceInboundMode: input.workspaceInboundMode, + workspaceSnapshotIsolation: input.workspaceSnapshotIsolation, workspaceDurableSeed: input.workspaceDurableSeed, workspaceCheckpoint: input.workspaceCheckpoint, onWorkspaceCheckpoint: input.onWorkspaceCheckpoint, diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.test.ts b/packages/adapter-utils/src/sandbox-managed-runtime.test.ts index 6aba93e702c..d9331d310d2 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.test.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.test.ts @@ -416,6 +416,36 @@ function recordingFilesystemClient( describe("sandbox managed runtime", () => { const cleanupDirs: string[] = []; + it.each([ + { isolation: undefined, detect: false }, + { isolation: undefined, detect: true }, + { isolation: "worker" as const, detect: false }, + { isolation: "worker" as const, detect: true }, + ])("keeps snapshot workers opt-in through command preparation: %j", async ({ isolation, detect }) => { + const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-snapshot-mode-")); + cleanupDirs.push(root); + const local = path.join(root, "local"), remote = path.join(root, "remote"); + await mkdir(local); + await writeFile(path.join(local, "file.txt"), "before"); + const capture = vi.spyOn(restoreMerge, "captureDirectorySnapshotOffThread"); + try { + const prepared = await prepareCommandManagedRuntime({ + runner: makeInlineSpawnRunner(), + spec: { remoteCwd: remote, timeoutMs: 30_000 }, + adapterKey: "test", + workspaceLocalDir: local, + workspaceSnapshotIsolation: isolation, + workspaceCheckpoint: true, + ...(detect ? { installCommand: "exit 99", detectCommand: "sh" } : {}), + }); + expect(capture.mock.calls.length > 0).toBe(isolation === "worker"); + expect(await readFile(path.join(remote, "file.txt"), "utf8")).toBe("before"); + await writeFile(path.join(remote, "file.txt"), "after"); + await prepared.restoreWorkspace(); + expect(await readFile(path.join(local, "file.txt"), "utf8")).toBe("after"); + } finally { capture.mockRestore(); } + }); + it.each(["host_current", "adopt_remote", "durable_seed"] as const)("stages and restores both project repositories with independent Git histories (%s)", async (mode) => { const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-multi-repo-")); cleanupDirs.push(root); diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.ts b/packages/adapter-utils/src/sandbox-managed-runtime.ts index 21a6a5f5e26..3e591d7dffe 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.ts @@ -34,6 +34,7 @@ import { } from "./git-workspace-sync.js"; import { captureDirectorySnapshot, + captureDirectorySnapshotOffThread, directorySnapshotSha256, classifyWorkspaceRestoreFailure, disposeDirectorySnapshot, @@ -1173,6 +1174,7 @@ export async function prepareSandboxManagedRuntime(input: { syncWorkspace?: boolean; /** Selects authoritative host staging, exact durable-seed replay, or no-overwrite adoption. */ workspaceInboundMode?: WorkspaceInboundMode; + workspaceSnapshotIsolation?: "worker"; workspaceDurableSeed?: WorkspaceDurableSeedPaths; workspaceCheckpoint?: boolean; onWorkspaceCheckpoint?: (metrics: WorkspaceCheckpointMetrics) => Promise; @@ -1205,6 +1207,9 @@ export async function prepareSandboxManagedRuntime(input: { /** Backoff before each automatic workspace copy-back retry. Tests shorten it. */ workspaceRestoreRetryDelaysMs?: readonly number[]; }): Promise { + const captureWorkspaceSnapshot = input.workspaceSnapshotIsolation === "worker" + ? captureDirectorySnapshotOffThread + : captureDirectorySnapshot; const workspaceRemoteDir = input.workspaceRemoteDir ?? input.spec.remoteCwd; const runtimeRootDir = input.runtimeRootDir ?? path.posix.join(workspaceRemoteDir, ".paperclip-runtime", input.adapterKey); if (input.runtimeRootDir !== undefined) { @@ -1338,7 +1343,7 @@ export async function prepareSandboxManagedRuntime(input: { baselineSnapshot = syncWorkspace ? (input.workspaceBaseline ?? (await runStepSpan("snapshot.baseline", () => - captureDirectorySnapshot(input.workspaceLocalDir, { + captureWorkspaceSnapshot(input.workspaceLocalDir, { exclude: restoreExclude, ignoredPaths, diskBacked: true, @@ -1652,7 +1657,7 @@ export async function prepareSandboxManagedRuntime(input: { } await fs.rm(path.join(verifiedRoot, relative), { recursive: true, force: true }); } - const verified = await captureDirectorySnapshot(verifiedRoot, { exclude: baselineSnapshot!.exclude, ignoredPaths: baselineSnapshot!.ignoredPaths, diskBacked: true }); + const verified = await captureWorkspaceSnapshot(verifiedRoot, { exclude: baselineSnapshot!.exclude, ignoredPaths: baselineSnapshot!.ignoredPaths, diskBacked: true }); try { return directorySnapshotSha256(verified) === directorySnapshotSha256(baselineSnapshot!); } finally { await disposeDirectorySnapshot(verified); } } finally { await fs.rm(verifiedRoot, { recursive: true, force: true }); } @@ -1905,6 +1910,7 @@ export async function prepareSandboxManagedRuntime(input: { adapterKey: input.adapterKey, workspaceLocalDir: localDir, workspaceInboundMode: "adopt_remote", + workspaceSnapshotIsolation: input.workspaceSnapshotIsolation, workspaceCheckpoint: input.workspaceCheckpoint, onWorkspaceCheckpoint: input.onWorkspaceCheckpoint, workspaceGitSnapshot: repository.snapshot, @@ -2141,6 +2147,7 @@ export async function prepareSandboxManagedRuntime(input: { }); try { await mergeDirectoryWithBaseline({ + workspaceSnapshotIsolation: input.workspaceSnapshotIsolation, baseline: mergeBaseline, sourceDir: extractedDir, ...(checkpointSnapshot ? { snapshots: { source: checkpointSnapshot } } : {}), diff --git a/packages/adapter-utils/src/workspace-restore-merge.ts b/packages/adapter-utils/src/workspace-restore-merge.ts index 85fd25155c4..e35923638ae 100644 --- a/packages/adapter-utils/src/workspace-restore-merge.ts +++ b/packages/adapter-utils/src/workspace-restore-merge.ts @@ -6,6 +6,7 @@ import { DatabaseSync } from "node:sqlite"; import { createWorkspaceManifest, WorkspaceManifestMap, workspacePathMatcher, type PathManifest, type WorkspacePaths, type WorkspaceManifestWriter } from "./workspace-manifest.js"; import { shouldExcludePath } from "./exclude-patterns.js"; import { resolvePaperclipInstanceRootForAdapter } from "./server-utils.js"; +import { runSnapshotThread } from "./workspace-snapshot-thread.js"; import type { WorkspaceRestoreDiagnostic } from "./workspace-restore-diagnostics.js"; export type SnapshotEntry = @@ -640,6 +641,22 @@ export async function captureDirectorySnapshot( } finally { ignored.close(); } } +export async function captureDirectorySnapshotOffThread( + rootDir: string, + options: { exclude?: string[]; ignoredPaths?: WorkspacePaths; diskBacked?: boolean } = {}, +): Promise { + if (!options.diskBacked) return captureDirectorySnapshot(rootDir, options); + const result = await runSnapshotThread({ rootDir, exclude: options.exclude, ignoredPaths: options.ignoredPaths }); + try { + const snapshot = openDirectorySnapshot(result.snapshot); + ownedDirectorySnapshots.set(snapshot, result.directory); + return snapshot; + } catch (error) { + await fs.rm(result.directory, { recursive: true, force: true }); + throw error; + } +} + /** A disk-backed subset for independent nested-repository merges. */ export async function selectDirectorySnapshot(snapshot: DirectorySnapshot, options: { prefix?: string; omit?: string[]; exclude: string[]; ignoredPaths?: WorkspacePaths; @@ -716,14 +733,16 @@ export async function mergeDirectoryWithBaseline(input: { sourceDir: string; targetDir: string; conflictPolicy?: "reject"; + workspaceSnapshotIsolation?: "worker"; beforeApply?: () => Promise; afterApply?: () => Promise; /** Caller holds the target's writer lock and validated an immutable sparse * source. Unchanged entries need no payload and are never copied. */ snapshots?: { source: DirectorySnapshot; current?: DirectorySnapshot }; }): Promise { + const capture = input.workspaceSnapshotIsolation === "worker" ? captureDirectorySnapshotOffThread : captureDirectorySnapshot; const options = { exclude: input.baseline.exclude, ignoredPaths: input.baseline.ignoredPaths, diskBacked: true }; - const source = input.snapshots?.source ?? await captureDirectorySnapshot(input.sourceDir, options); + const source = input.snapshots?.source ?? await capture(input.sourceDir, options); try { await withDirectoryMergeLock(input.targetDir, async (canonicalTargetDir) => { await input.beforeApply?.(); @@ -733,7 +752,7 @@ export async function mergeDirectoryWithBaseline(input: { } // Strict preflight must see excluded children before a directory is // replaced. The merge still applies only the filtered source/baseline. - const current = input.snapshots?.current ?? await captureDirectorySnapshot(canonicalTargetDir, + const current = input.snapshots?.current ?? await capture(canonicalTargetDir, input.conflictPolicy === "reject" ? { exclude: [], diskBacked: true } : options); try { if (input.conflictPolicy === "reject") { diff --git a/packages/adapter-utils/src/workspace-snapshot-thread-limit.test.ts b/packages/adapter-utils/src/workspace-snapshot-thread-limit.test.ts new file mode 100644 index 00000000000..7b519474cc1 --- /dev/null +++ b/packages/adapter-utils/src/workspace-snapshot-thread-limit.test.ts @@ -0,0 +1,86 @@ +import { promises as fs } from "node:fs"; +import { afterEach, expect, it, vi } from "vitest"; + +const control = vi.hoisted(() => ({ closing: false, onStart: undefined as undefined | (() => void), workers: [] as Array<{ + directory: string; + requestId: string; + emit(event: string, value: unknown): boolean; +}> })); + +vi.mock("node:worker_threads", async () => { + const { EventEmitter } = await import("node:events"); + return { Worker: class extends EventEmitter { + directory: string; + requestId: string; + constructor(_source: unknown, options: { env: { TMPDIR: string }; workerData: { rootDir: string } }) { + super(); + this.directory = options.env.TMPDIR; + this.requestId = options.workerData.rootDir; + control.workers.push(this); + control.onStart?.(); + if (control.closing) queueMicrotask(() => { + this.emit("message", { error: { name: "Error", message: "cleanup" } }); + this.emit("exit", 0); + }); + } + async terminate() { return 0; } + } }; +}); + +import { runSnapshotThread } from "./workspace-snapshot-thread.js"; + +afterEach(() => { vi.restoreAllMocks(); control.workers.length = 0; control.closing = false; control.onStart = undefined; }); + +it.each([false, true])("holds capacity through cleanup with reversed directory completion: %s", async reverse => { + let allowFirst!: () => void; + const firstGate = new Promise(resolve => { allowFirst = resolve; }); + const mkdtemp = fs.mkdtemp.bind(fs); + let directories = 0; + vi.spyOn(fs, "mkdtemp").mockImplementation(async (...args: Parameters) => { + if (reverse && ++directories === 1) await firstGate; + return await mkdtemp(...args); + }); + control.onStart = allowFirst; + const rm = fs.rm.bind(fs); + let releaseCleanup!: () => void; + const cleanupGate = new Promise(resolve => { releaseCleanup = resolve; }); + let cleaning = false; + vi.spyOn(fs, "rm").mockImplementation(async (...args: Parameters) => { + if (args[0] === control.workers[0]?.directory) { + cleaning = true; + await cleanupGate; + } + return await rm(...args); + }); + const jobs = new Map(["first", "second", "third"].map(requestId => [requestId, + runSnapshotThread({ rootDir: requestId }).then(() => null, error => error)])); + const fail = (worker: typeof control.workers[number]) => { + worker.emit("message", { error: { name: "Error", message: "failed", code: "EACCES" } }); + worker.emit("exit", 0); + }; + try { + await vi.waitFor(() => expect(control.workers).toHaveLength(2)); + if (reverse) expect(control.workers[0]!.requestId).toBe("second"); + const failed = control.workers[0]!; + fail(failed); + await vi.waitFor(() => expect(cleaning).toBe(true)); + expect(control.workers).toHaveLength(2); + expect((await fs.stat(control.workers[0]!.directory)).isDirectory()).toBe(true); + releaseCleanup(); + expect(await jobs.get(failed.requestId)).toMatchObject({ code: "EACCES" }); + await vi.waitFor(() => expect(control.workers).toHaveLength(3)); + await expect(fs.stat(control.workers[0]!.directory)).rejects.toMatchObject({ code: "ENOENT" }); + fail(control.workers[1]!); + fail(control.workers[2]!); + for (const job of jobs.values()) expect(await job).toMatchObject({ code: "EACCES" }); + for (const worker of control.workers) { + await expect(fs.stat(worker.directory)).rejects.toMatchObject({ code: "ENOENT" }); + } + } finally { + control.closing = true; + allowFirst(); + releaseCleanup(); + for (const worker of control.workers) fail(worker); + await Promise.all(jobs.values()); + } +}); diff --git a/packages/adapter-utils/src/workspace-snapshot-thread-worker.ts b/packages/adapter-utils/src/workspace-snapshot-thread-worker.ts new file mode 100644 index 00000000000..336a562ca92 --- /dev/null +++ b/packages/adapter-utils/src/workspace-snapshot-thread-worker.ts @@ -0,0 +1,29 @@ +import { parentPort, workerData } from "node:worker_threads"; +import { captureDirectorySnapshot, serializeDirectorySnapshot } from "./workspace-restore-merge.js"; +import type { SnapshotThreadRequest, SnapshotThreadResponse } from "./workspace-snapshot-thread.js"; + +if (!parentPort) throw new Error("Workspace snapshot worker requires a parent"); +const port = parentPort; +const request = workerData as SnapshotThreadRequest; + +async function run() { + let response: SnapshotThreadResponse; + try { + const snapshot = await captureDirectorySnapshot(request.rootDir, { + exclude: request.exclude, + ignoredPaths: request.ignoredPaths, + diskBacked: true, + }); + response = { snapshot: serializeDirectorySnapshot(snapshot) }; + } catch (value) { + const error = value as NodeJS.ErrnoException; + response = { error: { + name: error.name ?? "Error", message: error.message ?? String(value), stack: error.stack, + code: error.code, errno: error.errno, syscall: error.syscall, path: error.path, + } }; + } + port.postMessage(response); + port.close(); +} + +void run(); diff --git a/packages/adapter-utils/src/workspace-snapshot-thread.test.ts b/packages/adapter-utils/src/workspace-snapshot-thread.test.ts new file mode 100644 index 00000000000..5830b7039aa --- /dev/null +++ b/packages/adapter-utils/src/workspace-snapshot-thread.test.ts @@ -0,0 +1,142 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createWorkspaceManifest } from "./workspace-manifest.js"; +import { + captureDirectorySnapshot, captureDirectorySnapshotOffThread, + directorySnapshotSha256, disposeDirectorySnapshot, mergeDirectoryWithBaseline, + type DirectorySnapshot, +} from "./workspace-restore-merge.js"; + +const roots: string[] = []; +const snapshots: DirectorySnapshot[] = []; + +async function fixture() { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-snapshot-thread-test-")); + roots.push(root); + return root; +} + +afterEach(async () => { + vi.restoreAllMocks(); + await Promise.all(snapshots.splice(0).map(disposeDirectorySnapshot)); + await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true }))); +}); + +describe("workspace snapshot threads", () => { + it("preserves snapshot entries, digest, exclusions, modes and links across the thread boundary", async () => { + const root = await fixture(), outside = await fixture(); + await fs.mkdir(path.join(root, "nested")); + for (let index = 0; index < 39; index++) { + const name = path.join(root, index < 20 ? "" : "nested", `name ${index}\nbytes`); + await fs.writeFile(name, Buffer.from(`entry ${index}\0content`)); + await fs.chmod(name, index % 2 ? 0o600 : 0o755); + } + await fs.writeFile(path.join(outside, "private"), "outside"); + await fs.symlink(outside, path.join(root, "outside-link")); + await fs.writeFile(path.join(root, "excluded"), "excluded"); + await fs.writeFile(path.join(root, "ignored"), "ignored"); + const options = { exclude: ["excluded"], ignoredPaths: ["ignored"], diskBacked: true }; + const original = await captureDirectorySnapshot(root, options); + snapshots.push(original); + const isolated = await captureDirectorySnapshotOffThread(root, options); + snapshots.push(isolated); + expect(new Map(isolated.entries)).toEqual(new Map(original.entries)); + expect(directorySnapshotSha256(isolated)).toBe(directorySnapshotSha256(original)); + expect(isolated.entries.has("outside-link/private")).toBe(false); + expect(await fs.readFile(path.join(outside, "private"), "utf8")).toBe("outside"); + }); + + it("reads an existing ignored-path manifest without taking its cleanup ownership", async () => { + const root = await fixture(); + await fs.writeFile(path.join(root, "kept"), "kept"); + await fs.mkdir(path.join(root, "ignored")); + await fs.writeFile(path.join(root, "ignored", "private"), "ignored"); + const writer = await createWorkspaceManifest(); + roots.push(path.dirname(writer.filePath)); + writer.add("ignored", "ignored"); + const ignoredPaths = writer.paths("ignored"); + writer.close(); + const isolated = await captureDirectorySnapshotOffThread(root, { ignoredPaths, diskBacked: true }); + expect([...new Map(isolated.entries).keys()]).toEqual(["kept"]); + await disposeDirectorySnapshot(isolated); + expect((await fs.stat(ignoredPaths.filePath)).isFile()).toBe(true); + }); + + it("removes the worker's complete temporary directory when its snapshot is disposed", async () => { + const root = await fixture(); + await fs.writeFile(path.join(root, "file"), "bytes"); + const created: string[] = []; + const mkdtemp = fs.mkdtemp.bind(fs); + vi.spyOn(fs, "mkdtemp").mockImplementation(async (...args: Parameters) => { + const result = await mkdtemp(...args); + if (typeof result === "string" && path.basename(result).startsWith("paperclip-snapshot-thread-")) created.push(result); + return result; + }); + const isolated = await captureDirectorySnapshotOffThread(root, { diskBacked: true }); + expect(created).toHaveLength(1); + expect((await fs.stat(created[0]!)).isDirectory()).toBe(true); + await disposeDirectorySnapshot(isolated); + await expect(fs.stat(created[0]!)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("preserves filesystem errors, cleans failed workers and releases queued jobs", async () => { + const root = await fixture(); + const file = path.join(root, "file"); + await fs.writeFile(file, "bytes"); + const created: string[] = []; + const mkdtemp = fs.mkdtemp.bind(fs); + vi.spyOn(fs, "mkdtemp").mockImplementation(async (...args: Parameters) => { + const result = await mkdtemp(...args); + if (typeof result === "string" && path.basename(result).startsWith("paperclip-snapshot-thread-")) created.push(result); + return result; + }); + const results = await Promise.allSettled([ + captureDirectorySnapshotOffThread(path.join(root, "missing"), { diskBacked: true }), + captureDirectorySnapshotOffThread(file, { diskBacked: true }), + captureDirectorySnapshotOffThread(root, { diskBacked: true }), + ]); + expect(results[0]).toMatchObject({ status: "rejected", reason: { code: "ENOENT" } }); + expect(results[1]).toMatchObject({ status: "rejected", reason: { code: "ENOTDIR" } }); + expect(results[2].status).toBe("fulfilled"); + if (results[2].status === "fulfilled") await disposeDirectorySnapshot(results[2].value); + expect(created).toHaveLength(3); + for (const directory of created) await expect(fs.stat(directory)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it.each(["unchanged", "changed", "sparse"] as const)("isolates restore captures and cleans their manifests: %s", async mode => { + const target = await fixture(), source = await fixture(); + await fs.writeFile(path.join(target, "file"), "before"); + const baseline = await captureDirectorySnapshot(target, { diskBacked: true }); + snapshots.push(baseline); + await fs.writeFile(path.join(target, "concurrent"), "host bytes"); + await fs.writeFile(path.join(source, "file"), mode === "unchanged" ? "before" : "after"); + const supplied = mode === "sparse" ? await captureDirectorySnapshot(source, { diskBacked: true }) : undefined; + if (supplied) snapshots.push(supplied); + const created: string[] = []; + const mkdtemp = fs.mkdtemp.bind(fs); + vi.spyOn(fs, "mkdtemp").mockImplementation(async (...args: Parameters) => { + const result = await mkdtemp(...args); + if (typeof result === "string" && path.basename(result).startsWith("paperclip-snapshot-thread-")) created.push(result); + return result; + }); + await mergeDirectoryWithBaseline({ + baseline, sourceDir: source, targetDir: target, workspaceSnapshotIsolation: "worker", + ...(supplied ? { snapshots: { source: supplied } } : {}), + }); + expect(created).toHaveLength(mode === "changed" ? 2 : 1); + expect(await fs.readFile(path.join(target, "file"), "utf8")).toBe(mode === "unchanged" ? "before" : "after"); + expect(await fs.readFile(path.join(target, "concurrent"), "utf8")).toBe("host bytes"); + for (const directory of created) await expect(fs.stat(directory)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("retains the in-memory snapshot contract when disk backing is not requested", async () => { + const root = await fixture(); + await fs.writeFile(path.join(root, "file"), "bytes"); + const isolated = await captureDirectorySnapshotOffThread(root); + snapshots.push(isolated); + expect(isolated.entries).toBeInstanceOf(Map); + expect(isolated.entries.size).toBe(1); + }); +}); diff --git a/packages/adapter-utils/src/workspace-snapshot-thread.ts b/packages/adapter-utils/src/workspace-snapshot-thread.ts new file mode 100644 index 00000000000..91c3f479202 --- /dev/null +++ b/packages/adapter-utils/src/workspace-snapshot-thread.ts @@ -0,0 +1,80 @@ +import { Worker } from "node:worker_threads"; +import { createRequire } from "node:module"; +import { pathToFileURL } from "node:url"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import type { SerializedDirectorySnapshot } from "./workspace-restore-merge.js"; +import type { WorkspacePaths } from "./workspace-manifest.js"; + +export interface SnapshotThreadRequest { + rootDir: string; + exclude?: string[]; + ignoredPaths?: WorkspacePaths; +} + +export type SnapshotThreadResponse = + | { snapshot: SerializedDirectorySnapshot } + | { error: { name: string; message: string; stack?: string; code?: string; errno?: number; syscall?: string; path?: string } }; + +let active = 0; +const waiting: Array<() => void> = []; + +async function acquire() { + if (active < 2) { active++; return; } + await new Promise(resolve => waiting.push(resolve)); +} + +function release() { + const next = waiting.shift(); + if (next) next(); + else active--; +} + +export async function runSnapshotThread(request: SnapshotThreadRequest): Promise<{ + snapshot: Extract; + directory: string; +}> { + await acquire(); + let directory: string | undefined; + let worker: Worker | undefined; + let published = false; + try { + directory = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-snapshot-thread-")); + const source = new URL(import.meta.url).pathname.endsWith(".ts"); + const url = new URL(source ? "./workspace-snapshot-thread-worker.ts" : "./workspace-snapshot-thread-worker.js", import.meta.url); + const options = { workerData: request, env: { ...process.env, TMPDIR: directory, TMP: directory, TEMP: directory } }; + worker = source + ? new Worker(`import(${JSON.stringify(pathToFileURL(createRequire(import.meta.url).resolve("tsx/esm/api")).href)}).then(({tsImport}) => tsImport(${JSON.stringify(url.href)}, ${JSON.stringify(import.meta.url)}));`, { ...options, eval: true }) + : new Worker(url, options); + const child = worker; + const response = await new Promise((resolve, reject) => { + let result: SnapshotThreadResponse | undefined; + child.once("message", value => { result = value; }); + child.once("error", reject); + child.once("exit", code => { + if (code === 0 && result) resolve(result); + else reject(new Error(`Workspace snapshot worker exited without a result (${code})`)); + }); + }); + if ("error" in response) { + const error = Object.assign(new Error(response.error.message), response.error); + throw error; + } + const snapshot = response.snapshot; + if (snapshot.version !== 2) throw new Error("Workspace snapshot worker returned an in-memory snapshot"); + const relative = path.relative(directory, snapshot.entries.filePath); + if (!relative || path.isAbsolute(relative) || relative === ".." || relative.startsWith(`..${path.sep}`)) { + throw new Error("Workspace snapshot worker returned an unowned manifest"); + } + const stat = await fs.lstat(snapshot.entries.filePath); + if (!stat.isFile() || stat.isSymbolicLink()) throw new Error("Workspace snapshot worker manifest is not a regular file"); + published = true; + return { snapshot, directory }; + } finally { + try { + if (worker) await worker.terminate(); + if (directory && !published) await fs.rm(directory, { recursive: true, force: true }); + } finally { release(); } + } +} diff --git a/server/src/__tests__/native-workspace-sync-history.test.ts b/server/src/__tests__/native-workspace-sync-history.test.ts index 5e68c8527c8..a1f2e0edcdb 100644 --- a/server/src/__tests__/native-workspace-sync-history.test.ts +++ b/server/src/__tests__/native-workspace-sync-history.test.ts @@ -7,6 +7,7 @@ import { heartbeatRuns, type Db } from "@paperclipai/db"; import type { EnvironmentLease } from "@paperclipai/shared"; import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import * as restoreMerge from "@paperclipai/adapter-utils/workspace-restore-merge"; import * as workspaceSeedCache from "@paperclipai/adapter-utils/workspace-seed-cache"; import * as gitWorkspaceSync from "@paperclipai/adapter-utils/git-workspace-sync"; import { runLocalGit } from "@paperclipai/adapter-utils/git-workspace-sync"; @@ -51,6 +52,7 @@ describe("native warm workspace Git history", () => { "checks Git identity before reusing a warm sandbox: %s", async (change) => { const invalidRunSeed = change === "remote_commit_missing_run_seed" || change === "remote_commit_corrupt_run_seed"; const cacheRefused = change === "remote_commit_cache_refused" || change === "host_bytes_cache_refused" || invalidRunSeed; + const snapshotWorker = vi.spyOn(restoreMerge, "captureDirectorySnapshotOffThread"); const packClone = vi.spyOn(gitWorkspaceSync, "withShallowGitWorkspaceClone"); const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-warm-history-")); roots.push(root); @@ -101,6 +103,7 @@ describe("native warm workspace Git history", () => { const input = { db: db as unknown as Db, companyId: "company", workspaceId: "workspace", workspaceLocalDir: host, lease, target }; const first = (await prepareNativeWorkspaceSync({ ...input, runId: "first" }))!; expect(first.mode).toBe("host_current"); + expect(snapshotWorker).toHaveBeenCalled(); expect(first.reference.schema).toBe("paperclip.native-workspace-sync/v3"); if (change === "plain_nested_commit") { const recovered = await nativeWorkspaceSyncInternals.readDescriptor({ runId: "first", reference: first.reference }); diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index 3a11f3c30bb..cb8ea5be2e2 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -801,6 +801,7 @@ async function prepareRuntime(input: { workspaceLocalDir: input.workspaceLocalDir, workspaceRemoteDir: input.target.remoteCwd, workspaceInboundMode: input.mode, + workspaceSnapshotIsolation: "worker", workspaceDurableSeed: input.durableSeed, workspaceBaseline: input.baseline, workspaceGitSnapshot: input.gitSnapshot, @@ -828,7 +829,7 @@ async function finalizePreparedRuntime(input: { await input.assertOwnership?.(); const finalSnapshot = await import("@paperclipai/adapter-utils/workspace-restore-merge").then( - ({ captureDirectorySnapshot }) => + ({ captureDirectorySnapshotOffThread: captureDirectorySnapshot }) => captureDirectorySnapshot(input.descriptor.binding.localCwd, { exclude: input.runtime.workspaceSyncSnapshot?.baseline.exclude ?? [], ignoredPaths: input.runtime.workspaceSyncSnapshot?.baseline.ignoredPaths,