perf(workspace): run portable archives in bounded CPU workers (#20404)

## Thinking Path

> - Paperclip preserves agent work when its control plane changes.
> - Workspace imports capture and verify portable filesystem archives.
> - The archive function uses asynchronous I/O but hashes and handles
manifests on the caller's JavaScript thread.
> - Concurrent imports therefore share one CPU execution context for
that work.
> - This pull request runs the same archive function in bounded native
Node workers.
> - The parent keeps all database, storage, source locks and publication
decisions.

## Linked Issues or Issue Description

**What happened?**

Concurrent administrative workspace imports can spend many minutes
capturing archives while the owning Node process uses about one CPU
core. Promise concurrency does not move hashing and manifest work off
that process's JavaScript thread.

**Expected behavior**

Independent archive calls use separate CPU execution contexts. At most
eight archive workers run in one process. Cancellation must wait for
file cleanup and actual worker exit before the caller releases its
locks.

**Steps to reproduce**

Run several workspace captures in one process. The existing wrappers
call the same archive function on the parent thread. The regression
holds eight real worker threads at a CPU barrier while the parent
remains responsive, then cancels a queued ninth call without spawning
it.

**Deployment mode**

All server callers of the portable capture and restore wrappers. Related
import work: https://github.com/paperclipai/paperclip/pull/20311 and
https://github.com/paperclipai/paperclip/pull/20394.

## What Changed

- Run the existing self-contained archive function in native workers.
Its algorithm and format stay byte-identical.
- Limit active workers to eight per process, with a 1 GiB old-generation
heap limit each. Workers stop when their owned call finishes.
- Forward cancellation to the worker's AbortController. Wait for actual
exit before returning, throwing or releasing an active slot. Cancel
queued calls before they spawn.
- Own capture scratch in a private parent-created worker temporary root.
Remove it after actual exit, including abnormal exit.
- Keep direct archive algorithm fault tests on the canonical
implementation. Add real-worker byte, manifest, restore, CPU barrier and
cancellation tests.

## Verification

- Four real-worker regressions passed: forced-exit capture scratch
cleanup; exact capture bytes/manifest and restored files/modes/links;
eight CPU barriers plus queued abort; abort cleanup joined through
actual worker exit.
- Canonical workspaceArchive declaration/body hash is unchanged from the
base:
`d6cffeab112b817fab8f65a3c0d8ecdc8d23589d37bf44ae30fe9e53213d54f4`.
- The combined archive run passed 114 cases. Three Unix-socket cases
failed because the sandbox denied listen with EPERM; a focused retry
with socket permission passed all three. The initial separate worker run
passed all three cases; the new head passed all four, including
forced-exit cleanup. All server layered typechecks and module boundary
checks passed.
- The initial ac4bc5760d source passed a synthetic smoke on staging’s
Linux Node v24.22.2 through the installed tsx loader: direct/worker byte
and digest parity, worker restore of three files plus a symlink, and
private fixture cleanup. It made no database, object-storage or provider
calls.
- Greptile reviewed final head
`3a51fe67ff17a56b2ddd7877c89734ba60afd18e` at 5/5 with no actionable
issues. All applicable final-head CI passed, including aggregate verify,
typecheck, build, workspace consumer suites and all server shards. The
second review remains queued. No measured migration speedup claim is
made yet.

## Risks

Each active worker adds a bounded V8 heap and native overhead. Eight
workers can use more memory and I/O bandwidth than the former single
JavaScript thread. The existing 64 MiB manifest limit and streaming file
buffers stay in place. Worker failures refuse the call and cannot
publish authority. The canonical restore staging path stays unchanged;
this PR does not claim cleanup after every arbitrary process crash.
Cooperative cancellation retains the existing cleanup path; the parent
waits for actual worker exit. No database, key, object-storage or
provider behavior changes.

## Model Used

OpenAI Codex, based on GPT-6, with tool use and code execution. The
exact deployment identifier and context-window size are not exposed in
this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` /
`Closes: #` / `Refs: #` OR (b) described the issue in-PR following the
relevant issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

server/src/services/remote-workspace-archive-worker.test.ts

@@ -0,0 +1,121 @@
+import { createHash } from "node:crypto";
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const state = vi.hoisted(() => ({ control: undefined as SharedArrayBuffer | undefined,
+  mode: "" as "" | "cpu" | "abort" | "crash", spawned: 0, exited: 0, roots: [] as string[] }));
+vi.mock("node:worker_threads", async importOriginal => {
+  const actual = await importOriginal<typeof import("node:worker_threads")>();
+  return { ...actual, Worker: class extends actual.Worker {
+    constructor(source: string | URL, options?: import("node:worker_threads").WorkerOptions) {
+      if (state.control && typeof source === "string") {
+        const setup = `const testControl=new Int32Array(workerData.testControl);`;
+        source = source.replace("const archive=", `${setup}const archive=`);
+        if (state.mode === "cpu") source = source.replace("const before = await walk(root, true);", `
+          Atomics.add(testControl,0,1);
+          while(!Atomics.load(testControl,1)) Atomics.add(testControl,2,1);
+          const before = await walk(root, true);`);
+        if (state.mode === "abort") {
+          source = source.replace("const before = await walk(root, true);", `
+            Atomics.store(testControl,0,1);
+            await new Promise(resolve=>{input.signal.addEventListener('abort',resolve,{once:true});if(input.signal.aborted)resolve();});
+            check();
+            const before = await walk(root, true);`);
+          source = source.replace(/await fs\.rm\(scratch,\s*\{\s*recursive:\s*true,\s*force:\s*true\s*\}\);?/, `
+            Atomics.store(testControl,3,1);
+            while(!Atomics.load(testControl,4)) await new Promise(resolve=>setTimeout(resolve,5));
+            await fs.rm(scratch, { recursive: true, force: true });`);
+        }
+        if (state.mode === "crash") source = source.replace("const before = await walk(root, true);", `
+          await fs.mkdir(path.join(scratch,'copied-repository'));
+          await fs.writeFile(path.join(scratch,'copied-repository','retained'),'synthetic repository copy');
+          process.exit(73);
+          const before = await walk(root, true);`);
+        options = { ...options, workerData: { ...options?.workerData, testControl: state.control } };
+      }
+      super(source, options);
+      if (options?.env && typeof options.env === "object" && typeof options.env.TMPDIR === "string") state.roots.push(options.env.TMPDIR);
+      state.spawned++;
+      this.once("exit", () => { state.exited++; });
+    }
+  } };
+});
+import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive, workspaceArchive } from "./remote-workspace-snapshot-archive.js";
+
+describe("workspace archive worker boundary", () => {
+  let root: string, cwd: string;
+  beforeEach(async () => {
+    state.control = undefined; state.mode = ""; state.spawned = 0; state.exited = 0; state.roots = [];
+    root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-archive-worker-test-"));
+    cwd = path.join(root, "source"); await fs.mkdir(cwd);
+    await fs.writeFile(path.join(cwd, "saved.txt"), "retained work\n");
+    await fs.mkdir(path.join(cwd, "nested"));
+    await fs.writeFile(path.join(cwd, "nested", "bytes"), Buffer.alloc(1024 * 1024, 42));
+    await fs.chmod(path.join(cwd, "saved.txt"), 0o751);
+    await fs.symlink("saved.txt", path.join(cwd, "link"));
+  });
+  afterEach(async () => {
+    if (state.control) { const control = new Int32Array(state.control); Atomics.store(control,1,1); Atomics.store(control,4,1); }
+    await fs.rm(root, { recursive: true, force: true });
+  });
+  it("preserves exact archive bytes, manifest and restored files across real workers", async () => {
+    const directPath = path.join(root, "direct.pcws"), archivePath = path.join(root, "worker.pcws");
+    const direct = await workspaceArchive({ kind: "capture", cwd, archivePath: directPath });
+    const captured = await captureRemoteWorkspaceDirectory({ cwd, archivePath });
+    expect(captured).toEqual(direct);
+    expect(await fs.readFile(archivePath)).toEqual(await fs.readFile(directPath));
+    const targetDir = path.join(root, "restored");
+    await expect(restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: direct.manifestDigest })).resolves.toEqual(direct);
+    expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe("retained work\n");
+    expect((await fs.stat(path.join(targetDir, "saved.txt"))).mode & 0o777).toBe(0o751);
+    expect(await fs.readlink(path.join(targetDir, "link"))).toBe("saved.txt");
+    expect(createHash("sha256").update(await fs.readFile(path.join(targetDir, "nested", "bytes"))).digest("hex"))
+      .toBe(createHash("sha256").update(await fs.readFile(path.join(cwd, "nested", "bytes"))).digest("hex"));
+    expect(state.spawned).toBe(2); expect(state.exited).toBe(2);
+  });
+  it("runs CPU barriers on eight real workers while an aborted queued capture never spawns", async () => {
+    state.control = new SharedArrayBuffer(5 * 4); state.mode = "cpu";
+    const control = new Int32Array(state.control), controller = new AbortController(), reason = new Error("queued capture cancelled");
+    const captures = Array.from({ length: 8 }, (_, i) => captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, `${i}.pcws`) }));
+    const queued = captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, "queued.pcws"), signal: controller.signal }).catch(error => error);
+    try {
+      await expect.poll(() => Atomics.load(control,0), { timeout: 10_000 }).toBe(8);
+      expect(state.spawned).toBe(8); expect(state.exited).toBe(0);
+      expect(Atomics.load(control,2)).toBeGreaterThan(0);
+      controller.abort(reason);
+      expect(await queued).toBe(reason); expect(state.spawned).toBe(8);
+      Atomics.store(control,1,1);
+      await Promise.all(captures);
+      expect(state.exited).toBe(8);
+      await expect(fs.access(path.join(root, "queued.pcws"))).rejects.toMatchObject({ code: "ENOENT" });
+    } finally { Atomics.store(control,1,1); await Promise.allSettled(captures); }
+  });
+  it("removes parent-owned capture scratch after an abnormal worker exit", async () => {
+    state.control = new SharedArrayBuffer(5 * 4); state.mode = "crash";
+    await expect(captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, "crashed.pcws") }))
+      .rejects.toThrow("remote_workspace_archive_worker_failed");
+    expect(state.exited).toBe(1); expect(state.roots).toHaveLength(1);
+    await expect(fs.access(state.roots[0]!)).rejects.toMatchObject({ code: "ENOENT" });
+    expect(await fs.readFile(path.join(cwd,"saved.txt"),"utf8")).toBe("retained work\n");
+  });
+  it("waits for abort cleanup and actual worker exit before rejecting or releasing its slot", async () => {
+    state.control = new SharedArrayBuffer(5 * 4); state.mode = "abort";
+    const control = new Int32Array(state.control), controller = new AbortController(), reason = new Error("owned capture cancelled");
+    let settled = false;
+    const archivePath = path.join(root, "cancelled.pcws");
+    const capture = captureRemoteWorkspaceDirectory({ cwd, archivePath, signal: controller.signal }).then(
+      () => { settled = true; return null; }, error => { settled = true; return error; });
+    try {
+      await expect.poll(() => Atomics.load(control,0), { timeout: 10_000 }).toBe(1);
+      controller.abort(reason);
+      await expect.poll(() => Atomics.load(control,3), { timeout: 10_000 }).toBe(1);
+      expect(settled).toBe(false); expect(state.exited).toBe(0);
+      Atomics.store(control,4,1);
+      expect(await capture).toBe(reason);
+      expect(state.exited).toBe(1);
+      await expect(fs.access(archivePath)).rejects.toMatchObject({ code: "ENOENT" });
+    } finally { Atomics.store(control,4,1); controller.abort(reason); await capture; }
+  });
+});

server/src/services/remote-workspace-snapshot-archive.test.ts

@@ -6,7 +6,7 @@ import os from "node:os";
 import path from "node:path";
 import { promisify } from "node:util";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { captureRemoteWorkspaceDirectory, readRemoteWorkspaceArchiveDirectories, remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js";
+import { workspaceArchive, readRemoteWorkspaceArchiveDirectories, remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage } from "./remote-workspace-snapshot-archive.js";
 
 const archiveIo = vi.hoisted(() => ({
   afterChmod: undefined as ((target: string, mode: number) => void) | undefined,
@@ -35,6 +35,9 @@ vi.mock("node:fs/promises", async importOriginal => {
   } };
 });
 
+const captureRemoteWorkspaceDirectory = (input: Omit<Parameters<typeof workspaceArchive>[0], "kind">) => workspaceArchive({ ...input, kind: "capture" });
+const restoreRemoteWorkspaceArchive = (input: Omit<Parameters<typeof workspaceArchive>[0], "kind">) => workspaceArchive({ ...input, kind: "restore" });
+
 const execute = promisify(execFile);
 const env = { ...process.env, GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: os.devNull,
   GIT_AUTHOR_NAME: "Snapshot", GIT_COMMITTER_NAME: "Snapshot", GIT_AUTHOR_EMAIL: "snapshot@example.test", GIT_COMMITTER_EMAIL: "snapshot@example.test" };

server/src/services/remote-workspace-snapshot-archive.ts

@@ -1,8 +1,10 @@
+import { Worker } from "node:worker_threads";
+
 /**
  * A portable, streaming workspace archive. This function is intentionally
  * self-contained: the same implementation runs locally and in a sandbox.
  */
-async function workspaceArchive(input: {
+export async function workspaceArchive(input: {
   kind: "capture" | "restore";
   cwd?: string;
   archivePath: string;
@@ -662,8 +664,81 @@ async function workspaceArchive(input: {
   }
 }
 
+type ArchiveInput = Parameters<typeof workspaceArchive>[0];
+let activeArchiveWorkers = 0;
+const archiveWaiters: Array<() => void> = [];
+
+async function archiveWorkerSlot(signal?: AbortSignal): Promise<() => void> {
+  signal?.throwIfAborted();
+  await new Promise<void>((resolve, reject) => {
+    const start = () => {
+      signal?.removeEventListener("abort", abort);
+      activeArchiveWorkers++;
+      resolve();
+    };
+    const abort = () => {
+      const index = archiveWaiters.indexOf(start);
+      if (index >= 0) archiveWaiters.splice(index, 1);
+      reject(signal!.reason);
+    };
+    if (activeArchiveWorkers < 8) start();
+    else {
+      archiveWaiters.push(start);
+      signal?.addEventListener("abort", abort, { once: true });
+    }
+  });
+  return () => { activeArchiveWorkers--; archiveWaiters.shift()?.(); };
+}
+
+async function runArchiveWorker(input: ArchiveInput): ReturnType<typeof workspaceArchive> {
+  const release = await archiveWorkerSlot(input.signal);
+  let workerScratch: string | undefined;
+  const fs = await import("node:fs/promises");
+  try {
+    input.signal?.throwIfAborted();
+    const path = await import("node:path"), os = await import("node:os");
+    workerScratch = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-archive-worker-"));
+    await fs.chmod(workerScratch, 0o700);
+    const { signal, ...parameters } = input;
+    const source = workspaceArchive.toString().replaceAll("__vite_ssr_dynamic_import__", "import");
+    const worker = new Worker(`const {parentPort,workerData}=require("node:worker_threads");
+const __name=value=>value;
+const controller=new AbortController();
+parentPort.on("message",()=>controller.abort());
+const archive=${source};
+archive({...workerData,signal:controller.signal}).then(result=>parentPort.postMessage({result}),error=>parentPort.postMessage({error:{...error,name:error.name,message:error.message}})).finally(()=>parentPort.close());`, {
+      eval: true, workerData: parameters, execArgv: [], stdout: true, stderr: true,
+      env: { ...process.env, TMPDIR: workerScratch, TMP: workerScratch, TEMP: workerScratch },
+      resourceLimits: { maxOldGenerationSizeMb: 1024, maxYoungGenerationSizeMb: 64, stackSizeMb: 4 },
+    });
+    worker.stdout.resume(); worker.stderr.resume();
+    const abort = () => worker.postMessage("abort");
+    signal?.addEventListener("abort", abort, { once: true });
+    if (signal?.aborted) abort();
+    try {
+      return await new Promise<Awaited<ReturnType<typeof workspaceArchive>>>((resolve, reject) => {
+        let result: Awaited<ReturnType<typeof workspaceArchive>> | undefined, failure: Error | undefined;
+        worker.once("message", (message: { result?: typeof result; error?: Record<string, unknown> }) => {
+          result = message.result;
+          if (message.error) failure = Object.assign(new Error(String(message.error.message)), message.error);
+        });
+        worker.once("error", error => { failure = error; });
+        worker.once("exit", code => {
+          if (signal?.aborted) reject(signal.reason);
+          else if (failure) reject(failure);
+          else if (code === 0 && result) resolve(result);
+          else reject(new Error("remote_workspace_archive_worker_failed"));
+        });
+      });
+    } finally { signal?.removeEventListener("abort", abort); }
+  } finally {
+    try { if (workerScratch) await fs.rm(workerScratch, { recursive: true, force: true }); }
+    finally { release(); }
+  }
+}
+
 export function captureRemoteWorkspaceDirectory(input: { cwd: string; archivePath: string; manifestPath?: string; exclude?: string[]; requireNoGit?: boolean; gitSourceRoots?: string[]; signal?: AbortSignal }) {
-  return workspaceArchive({ ...input, kind: "capture" });
+  return runArchiveWorker({ ...input, kind: "capture" });
 }
 
 export async function readRemoteWorkspaceArchiveDirectories(archivePath: string, expectedManifestDigest: string) {
@@ -691,7 +766,7 @@ export async function readRemoteWorkspaceArchiveDirectories(archivePath: string,
 }
 
 export function restoreRemoteWorkspaceArchive(input: { archivePath: string; targetDir: string; expectedManifestDigest: string; signal?: AbortSignal; maxEntries?: number; maxManifestBytes?: number }) {
-  return workspaceArchive({ ...input, kind: "restore" });
+  return runArchiveWorker({ ...input, kind: "restore" });
 }
 
 export function remoteWorkspaceCaptureFailureMessage(error: unknown): string | null {