Bound parallel reads during sandbox folder hydration

Open up to sixteen response streams while keeping queued bodies paused and preserving ordered publication and cleanup. This addresses the storage-response waits measured on cold staging runs.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/sandbox-work-folders.md

@@ -237,8 +237,9 @@ saving outgoing changes. Imported bytes therefore cannot overwrite a newer
 shared version by being mistaken for an agent edit; actual subsequent edits
 still synchronize normally. Failed explicit refreshes retain the same intent.
 Other providers keep the small-argument transport. Incoming storage responses
-are prefetched four at a time and closed if transfer fails. Repository restores
-use the same path, then recreate confined repository links.
+are prefetched sixteen at a time without consuming queued response bodies, and
+closed if transfer fails. The transport keeps its separate bounded batch buffer.
+Repository restores use the same path, then recreate confined repository links.
 
 Read-only sandbox commands retry transient connection failures and HTTP
 502/503/504 responses up to three attempts within one 120-second deadline.

server/src/__tests__/work-folder-transfer.test.ts

@@ -4,7 +4,7 @@ import { prefetchWorkFiles } from "../services/work-folder-transfer.js";
 
 const entry = { path: "file", kind: "file" as const, byteSize: 0, sha256: null, executable: false };
 describe("bounded work file prefetch", () => {
-  it("preserves order and keeps at most four streams open", async () => {
+  it("preserves order and keeps at most sixteen streams open", async () => {
     let opened = 0, peak = 0;
     const seen: string[] = [];
     for await (const transfer of prefetchWorkFiles(Array.from({ length: 30 }, (_, i) => i), async (i) => {
@@ -13,9 +13,28 @@ describe("bounded work file prefetch", () => {
       return { entry: { ...entry, path: String(i) }, body };
     })) seen.push(transfer.entry.path);
     expect(seen).toEqual(Array.from({ length: 30 }, (_, i) => String(i)));
-    expect(peak).toBe(4);
+    expect(peak).toBe(16);
     expect(opened).toBe(0);
   });
+  it("opens a bounded response window without reading queued bodies", async () => {
+    const bodies: Readable[] = [];
+    let reads = 0;
+    const transfers = prefetchWorkFiles(Array.from({ length: 40 }, (_, i) => i), async (i) => {
+      const body = new Readable({ read() { reads++; this.push(Buffer.alloc(1024)); } });
+      bodies.push(body);
+      return { entry: { ...entry, path: String(i), byteSize: 1024 }, body };
+    });
+    try {
+      const first = await transfers.next();
+      expect(first.value?.entry.path).toBe("0");
+      expect(bodies).toHaveLength(16);
+      expect(reads).toBe(0);
+    } finally {
+      await transfers.return();
+    }
+    expect(bodies.every((body) => body.destroyed)).toBe(true);
+    expect(reads).toBe(0);
+  });
   it("handles a prefetched stream failure before the consumer reaches it", async () => {
     const source = new Readable({ read() {} });
     await expect((async () => {
@@ -39,12 +58,12 @@ describe("bounded work file prefetch", () => {
       const body = Readable.from([]); streams.push(body);
       return { entry, body };
     };
-    for await (const _transfer of prefetchWorkFiles([0, 1, 2, 3, 4], open)) break;
-    expect(streams).toHaveLength(3);
+    for await (const _transfer of prefetchWorkFiles(Array.from({ length: 40 }, (_, i) => i), open)) break;
+    expect(streams).toHaveLength(15);
     expect(streams.every((stream) => stream.destroyed)).toBe(true);
     streams.length = 0;
     await expect((async () => {
-      for await (const _transfer of prefetchWorkFiles([0, 1, 2, 3, 4], open)) { /* consume */ }
+      for await (const _transfer of prefetchWorkFiles(Array.from({ length: 40 }, (_, i) => i), open)) { /* consume */ }
     })()).rejects.toThrow("storage unavailable");
     expect(streams.every((stream) => stream.destroyed)).toBe(true);
   });

server/src/services/sandbox-work-folders.ts

@@ -1,5 +1,5 @@
 import { measureSandboxOperation, measureSandboxStream, captureSandboxPerformanceContext } from "./sandbox-performance.js";
-import { prefetchWorkFiles } from "./work-folder-transfer.js";
+import { prefetchWorkFiles, WORK_FOLDER_PREFETCH_CONCURRENCY } from "./work-folder-transfer.js";
 import { createHash, randomUUID } from "node:crypto";
 import path from "node:path";
 import { managedAgentFiles } from "./work-folder-agent-import.js";
@@ -242,7 +242,7 @@ export async function prepareSandboxWorkFolders(input: {
         }
         const changed = saved.sort((a, b) => a.path.length - b.path.length)
           .filter((entry) => signature(current.get(entry.path)) !== signature(entry));
-        await measureSandboxOperation("work_folder.scope.hydrate", { scope, files: changed.length, parallelism: 4 }, async () => (transport.writeMany(paths[scope]!, staging, prefetchWorkFiles(changed, async (entry, fileIndex) => {
+        await measureSandboxOperation("work_folder.scope.hydrate", { scope, files: changed.length, parallelism: WORK_FOLDER_PREFETCH_CONCURRENCY }, async () => (transport.writeMany(paths[scope]!, staging, prefetchWorkFiles(changed, async (entry, fileIndex) => {
           if (entry.kind === "directory") return { entry };
           const result = await svc.content(folder, entry.path, fileIndex);
           // A shared file can change after listing. Validate and baseline the

server/src/services/work-folder-transfer.ts

@@ -1,7 +1,11 @@
 import { measureSandboxOperation } from "./sandbox-performance.js";
 import type { WorkFileTransfer } from "./work-folder-transport.js";
 
-// Open a few storage responses ahead without buffering their bodies. Drain all
+// Bound open responses independently of file size. Bodies stay paused until
+// consumed, and the transport retains its separate batch byte limit.
+export const WORK_FOLDER_PREFETCH_CONCURRENCY = 16;
+
+// Open storage responses ahead without buffering their bodies. Drain all
 // pending opens on failure so abandoned HTTP response streams are closed too.
 export async function* prefetchWorkFiles<T>(
   entries: Iterable<T>,
@@ -15,7 +19,7 @@ export async function* prefetchWorkFiles<T>(
     const next = iterator.next();
     if (!next.done) {
       const fileIndex = opened++;
-      pending.push(Promise.resolve().then(() => measureSandboxOperation("work_folder.prefetch.open", { fileIndex, parallelism: 4 }, async () => open(next.value, fileIndex)))
+      pending.push(Promise.resolve().then(() => measureSandboxOperation("work_folder.prefetch.open", { fileIndex, parallelism: WORK_FOLDER_PREFETCH_CONCURRENCY }, async () => open(next.value, fileIndex)))
       .then((value): Result => {
         // A response can fail while queued, before its async iterator exists.
         // Keep that error handled; consuming the stream still throws it.
@@ -25,16 +29,16 @@ export async function* prefetchWorkFiles<T>(
     }
   }
   try {
-    for (let i = 0; i < 4; i++) enqueue();
+    for (let i = 0; i < WORK_FOLDER_PREFETCH_CONCURRENCY; i++) enqueue();
     while (pending.length) {
       const next = pending.shift()!;
-      const result = await measureSandboxOperation("work_folder.prefetch.wait", { fileIndex: consumed++, parallelism: 4 }, async () => next);
+      const result = await measureSandboxOperation("work_folder.prefetch.wait", { fileIndex: consumed++, parallelism: WORK_FOLDER_PREFETCH_CONCURRENCY }, async () => next);
       if ("error" in result) throw result.error;
       try { yield result.value; } finally { result.value.body?.destroy(); }
       enqueue();
     }
   } finally {
-    for (const result of await measureSandboxOperation("work_folder.prefetch.drain", { files: pending.length, parallelism: 4 }, async () => Promise.all(pending))) {
+    for (const result of await measureSandboxOperation("work_folder.prefetch.drain", { files: pending.length, parallelism: WORK_FOLDER_PREFETCH_CONCURRENCY }, async () => Promise.all(pending))) {
       if ("value" in result) result.value.body?.destroy();
     }
     iterator.return?.();