From 7fc2dc8b3236e1149ae32756b6153a503623404b Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 15 Oct 2026 06:31:59 -0500 Subject: [PATCH] fix: isolate workspace snapshots from controller work (#16414) Move native sandbox snapshot captures off the controller callback queue while preserving the existing capture and restore contract. Keep legacy adapters inline, bound captures to two workers, and verify cleanup and reordered worker completion. Validation: 461 affected cases, full typecheck/build, exact-head CI/E2E/Superagent and Apex 5/5. The full local canonical suite remains running; no full local pass or under-30-second staging response is claimed. Co-Authored-By: Paperclip --- doc/execution-semantics.md | 9 ++ ...2026-10-15-workspace-snapshot-isolation.md | 37 +++++ packages/adapter-utils/package.json | 1 + .../src/command-managed-runtime.ts | 3 + .../adapter-utils/src/execution-target.ts | 2 + .../src/sandbox-managed-runtime.test.ts | 30 ++++ .../src/sandbox-managed-runtime.ts | 11 +- .../src/workspace-restore-merge.ts | 23 ++- .../workspace-snapshot-thread-limit.test.ts | 86 +++++++++++ .../src/workspace-snapshot-thread-worker.ts | 29 ++++ .../src/workspace-snapshot-thread.test.ts | 142 ++++++++++++++++++ .../src/workspace-snapshot-thread.ts | 80 ++++++++++ .../native-workspace-sync-history.test.ts | 3 + .../native-runtime/native-workspace-sync.ts | 3 +- 14 files changed, 454 insertions(+), 5 deletions(-) create mode 100644 doc/plans/2026-10-15-workspace-snapshot-isolation.md create mode 100644 packages/adapter-utils/src/workspace-snapshot-thread-limit.test.ts create mode 100644 packages/adapter-utils/src/workspace-snapshot-thread-worker.ts create mode 100644 packages/adapter-utils/src/workspace-snapshot-thread.test.ts create mode 100644 packages/adapter-utils/src/workspace-snapshot-thread.ts 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,