perf(workspaces): overlap bounded snapshot file hashing (#19458)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - The server saves portable workspace archives before it moves file
authority to object storage.
> - Each capture checks and hashes every eligible file before and after
archive creation.
> - Those stable reads run in series and leave file I/O idle on
workspaces with many files.
> - This pull request overlaps at most four stable reads across the
whole directory walk.
> - It preserves entry order, source checks, Git checks, and archive
bytes.
> - The benefit is shorter archive capture with bounded extra I/O
concurrency.

## Linked Issues or Issue Description

**What existing behavior does this improve?**

Portable remote workspace archive capture in the server. Related: #19396
preserves saved-upload recovery. #17132 applies a similar technique to
the separate Runner file-diff snapshot path; this PR changes the server
archive owner only.

**Current behavior**

The archive walker waits for each file's stable read and hash before
visiting the next file. A local packed-Git fixture with 6,000 files and
52.4 MB of contents took a median 4.640 seconds in the final two serial
samples.

**Proposed behavior**

Keep up to four stable reads active across all nested directories.
Reserve each entry's position before reading it. Await all reads before
returning success, failure, or cancellation. The final two treatment
samples took a median 3.439 seconds, a 25.9% reduction in capture time.

**Reason and benefit**

Captures run twice during a fresh workspace migration. Bounded I/O
overlap reduces time spent in this owner without changing the archive
contract. This measurement does not establish an end-to-end migration
speedup.

**Breaking changes**

None intended. The file format, manifest order, source validation, Git
validation, API, and schema stay the same.

## What Changed

- Share one four-read limit across the recursive walk. Preserve
deterministic entry positions when reads finish out of order.
- Drain active reads on every exit before callers can remove scratch
files or return control to a source writer.
- Check cancellation before creating a file read stream.
- Exercise real file handles across nested directories, out-of-order
completion, source replacement, cancellation, and serialized archive
capture.

## Verification

- Focused archive, archive-environment, and snapshot-storage suites: 113
tests passed. The first sandbox run could not bind local test sockets;
the same suites passed with local listener access.
- Full workspace typecheck passed before rebasing. The rebase changed
neither touched file.
- Full local `pnpm test:run` stopped at an unchanged chat integration
test: `resolves ambiguous provider effects explicitly and never lets
terminal failures starve reconciliation`, with `socket hang up`. The
same case passed once in isolation on the rebased head. Its original
cause remains unknown; this is not a full-suite pass.
- Full `pnpm build` passed on the rebased head. Current-head CI is
pending.
- Benchmark: actual owning sources, Node 26.4 on macOS arm64, one
importer, the same packed-Git fixture. All 12 trial archives had
identical manifest digest, SHA-256, and 72,908,826 bytes. Final paired
medians were 4.640 seconds serial and 3.439 seconds with four reads.
Total process plus Git CPU increased about 12%.
- An initial pair was slower with concurrency (23.352 seconds versus
11.228 seconds); the cause remains unknown. Later paired reductions
ranged from 24% to 37%. The benchmark host was shared. Linux Node 24 and
eight-importer throughput are not yet qualified.
- No user commands, configuration, or document contracts changed. Code
comments explain ordering and cleanup.

## Risks

- Four simultaneous reads can increase I/O and CPU contention when
several captures run together. The bound is per capture. Hosted
throughput needs measurement after the ordinary release.
- A shared-host local benchmark has limited predictive value. This PR
claims a measured capture improvement on that fixture only.
- Rollback is a code revert. No migration or stored-data rewrite is
required.

## Model Used

OpenAI Codex, GPT-6 family, with tool use, code execution, and delegated
review. The runtime did not expose an exact deployment variant or
context-window size.

## 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
- [ ] 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
- [ ] All Paperclip CI gates are green
- [ ] 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-snapshot-archive.test.ts

@@ -11,13 +11,21 @@ import { captureRemoteWorkspaceDirectory, readRemoteWorkspaceArchiveDirectories,
 const archiveIo = vi.hoisted(() => ({
   afterChmod: undefined as ((target: string, mode: number) => void) | undefined,
   beforeOpen: undefined as ((target: string) => Promise<void>) | undefined,
+  afterOpen: undefined as ((target: string) => Promise<void>) | undefined,
+  afterClose: undefined as ((target: string) => void) | undefined,
   captures: [] as string[],
 }));
 vi.mock("node:fs/promises", async importOriginal => {
   const actual = await importOriginal<typeof import("node:fs/promises")>();
   return { ...actual, async open(...args: Parameters<typeof actual.open>) {
     await archiveIo.beforeOpen?.(String(args[0]));
-    return actual.open(...args);
+    const handle = await actual.open(...args), close = handle.close.bind(handle);
+    let closed = false;
+    handle.close = async () => {
+      try { await close(); } finally { if (!closed) { closed = true; archiveIo.afterClose?.(String(args[0])); } }
+    };
+    await archiveIo.afterOpen?.(String(args[0]));
+    return handle;
   }, async mkdtemp(...args: Parameters<typeof actual.mkdtemp>) {
     const directory = await actual.mkdtemp(...args);
     if (String(args[0]).includes("pc-workspace-capture-")) archiveIo.captures.push(String(directory));
@@ -35,7 +43,8 @@ describe("portable remote workspace archives", () => {
   let root: string;
   beforeEach(async () => { root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-archive-test-")); });
   afterEach(async () => {
-    archiveIo.afterChmod = undefined; archiveIo.beforeOpen = undefined; archiveIo.captures = [];
+    archiveIo.afterChmod = undefined; archiveIo.beforeOpen = undefined; archiveIo.afterOpen = undefined;
+    archiveIo.afterClose = undefined; archiveIo.captures = [];
     await fs.rm(root, { recursive: true, force: true });
   });
   async function source() { const cwd = path.join(root, "source"); await fs.mkdir(cwd); return cwd; }
@@ -54,6 +63,87 @@ describe("portable remote workspace archives", () => {
     }
     return JSON.stringify(entries);
   }
+  function deferred() {
+    let resolve!: () => void;
+    const promise = new Promise<void>(done => { resolve = done; });
+    return { promise, resolve };
+  }
+  async function parallelSource() {
+    const cwd = await source();
+    for (const name of ["a", "b", "c", "d", "e", "f"]) {
+      await fs.mkdir(path.join(cwd, name));
+      await fs.writeFile(path.join(cwd, name, "file"), name.repeat(1024));
+    }
+    return fs.realpath(cwd);
+  }
+  it("bounds stable reads across directories and preserves bytes after out-of-order completion", async () => {
+    const cwd = await parallelSource(), blocked: Array<ReturnType<typeof deferred>> = [];
+    let gating = true, active = 0, maximum = 0;
+    const closed: string[] = [];
+    archiveIo.afterOpen = async target => {
+      if (!target.startsWith(`${cwd}${path.sep}`)) return;
+      maximum = Math.max(maximum, ++active);
+      if (gating) { const gate = deferred(); blocked.push(gate); await gate.promise; }
+    };
+    archiveIo.afterClose = target => {
+      if (target.startsWith(`${cwd}${path.sep}`)) { active--; closed.push(target); }
+    };
+    const capturing = capture(cwd);
+    try {
+      await expect.poll(() => blocked.length).toBe(4);
+      expect(active).toBe(4);
+      blocked[3]!.resolve();
+      await expect.poll(() => blocked.length).toBe(5);
+      expect(closed[0]).toBe(path.join(cwd, "d", "file"));
+      expect(maximum).toBe(4);
+    } finally {
+      gating = false;
+      for (const gate of blocked) gate.resolve();
+      await capturing;
+    }
+    const snapshot = await capturing;
+    expect(active).toBe(0); expect(maximum).toBe(4);
+    archiveIo.afterOpen = undefined; archiveIo.afterClose = undefined;
+    const helper = path.join(root, "helper.cjs"), archivePath = path.join(root, "serialized.pcws");
+    await fs.writeFile(helper, remoteWorkspaceArchiveScript());
+    const serialized = JSON.parse((await execute(process.execPath, [helper, "capture", cwd, archivePath])).stdout);
+    expect(serialized).toMatchObject({ manifestDigest: snapshot.manifestDigest, sha256: snapshot.sha256 });
+    expect(await fs.readFile(archivePath)).toEqual(await fs.readFile(snapshot.archivePath));
+  });
+  it.each(["cancel", "replace"])("drains every parallel stable read after %s", async kind => {
+    const cwd = await parallelSource(), blocked: Array<ReturnType<typeof deferred>> = [];
+    const controller = new AbortController(), reason = new Error("capture cancelled");
+    let gating = true, active = 0, settled = false;
+    archiveIo.afterOpen = async target => {
+      if (!target.startsWith(`${cwd}${path.sep}`)) return;
+      active++;
+      if (gating) { const gate = deferred(); blocked.push(gate); await gate.promise; }
+    };
+    archiveIo.afterClose = target => { if (target.startsWith(`${cwd}${path.sep}`)) active--; };
+    const capturing = captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, "snapshot.pcws"), signal: controller.signal })
+      .then(() => { settled = true; return null; }, error => { settled = true; return error as Error; });
+    try {
+      await expect.poll(() => blocked.length).toBe(4);
+      if (kind === "cancel") controller.abort(reason);
+      else {
+        await fs.unlink(path.join(cwd, "a", "file"));
+        await fs.writeFile(path.join(cwd, "a", "file"), "replacement");
+      }
+      for (const gate of blocked.slice(0, 3)) gate.resolve();
+      await expect.poll(() => active).toBe(1);
+      expect(settled).toBe(false);
+    } finally {
+      gating = false;
+      for (const gate of blocked) gate.resolve();
+      await capturing;
+    }
+    const error = await capturing;
+    if (kind === "cancel") expect(error === reason || error?.cause === reason).toBe(true);
+    else expect(error?.message).toBe("remote_workspace_archive_changed");
+    expect(active).toBe(0);
+    await expect(fs.access(path.join(root, "snapshot.pcws"))).rejects.toMatchObject({ code: "ENOENT" });
+    for (const directory of archiveIo.captures) await expect(fs.access(directory)).rejects.toMatchObject({ code: "ENOENT" });
+  });
   it("reads captured directory coverage and rejects a different manifest digest", async () => {
     const cwd = await source();
     await fs.mkdir(path.join(cwd, "checkout"));

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

@@ -252,6 +252,7 @@ async function workspaceArchive(input: {
       const before = await handle.stat({ bigint: true });
       if (!before.isFile() || original.dev !== before.dev || original.ino !== before.ino || await fs.realpath(source) !== source) fail("changed");
       const digest = crypto.createHash("sha256"); let size = 0;
+      check();
       for await (const bytes of handle.createReadStream({ autoClose: false, signal: input.signal })) {
         check(); digest.update(bytes); size += bytes.length;
         await consume?.(bytes);
@@ -309,6 +310,11 @@ async function workspaceArchive(input: {
     || next === ".paperclip-runtime" || next.startsWith(".paperclip-runtime/");
   const walk = async (root: string, skipGit = false) => {
     const entries: Entry[] = [];
+    // One bound for the whole walk, including nested directories. Reserve each
+    // entry's position before hashing so completion order cannot change identity.
+    const pending = new Set<Promise<void>>();
+    let failed = false, failure: unknown;
+    const checkWalk = () => { if (failed) throw failure; check(); };
     const visit = async (relative: string, parentIsRepository = false) => {
       const directory = relative ? path.join(root, relative) : root;
       const directoryStat = await fs.lstat(directory);
@@ -322,7 +328,7 @@ async function workspaceArchive(input: {
         await assertRepositoryRoot(directory);
       }
       for (const name of names) {
-        check();
+        checkWalk();
         const next = relative ? `${relative}/${name}` : name;
         if (!safeRelative(next)) fail("unsupported_filename");
         if (excluded(next)) {
@@ -356,12 +362,27 @@ async function workspaceArchive(input: {
           if (!safeLink(next, link)) fail("unsafe_symlink");
           entries.push({ path: next, type: "symlink", target: link });
         } else if (stat.isFile()) {
-          entries.push({ path: next, type: "file", ...await readStableFile(source) });
+          const entry: Entry = { path: next, type: "file" };
+          entries.push(entry);
+          const read = readStableFile(source).then(value => { Object.assign(entry, value); }, error => {
+            if (!failed) { failed = true; failure = error; }
+          });
+          pending.add(read);
+          void read.then(() => pending.delete(read));
+          if (pending.size >= 4) { await Promise.race(pending); checkWalk(); }
         } else fail("unsupported_entry");
       }
     };
-    await visit("");
-    return entries;
+    try {
+      await visit("");
+      await Promise.all(pending);
+      checkWalk();
+      return entries;
+    } finally {
+      // A rejected/cancelled walk must close every active stable read before its
+      // caller removes scratch or returns control to a mutating source owner.
+      await Promise.all(pending);
+    }
   };
   const manifestHash = (manifest: { schema: string; entries: Entry[] }) => crypto.createHash("sha256").update(JSON.stringify(manifest)).digest("hex");
   if (input.kind === "capture") {