fix(server): restore durable agent folders with bounded concurrency (#18531)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - A replacement web server restores agent files from shared storage.
> - The old restore reads each object in sequence.
> - Modest network latency makes a 1,024-file folder exceed its
deadline.
> - This pull request uses at most eight concurrent restore streams.
> - Every file must pass its size and hash checks before the folder is
published.

## Linked Issues or Issue Description

Refs #18518 and #18526. Local deployment qualification found that a 64
MiB folder could not be restored after the old server filesystem was
removed.

## What Changed

- Restore at most eight files concurrently within the existing file
authority service.
- Cancel sibling transfers on failure. Await all workers before removing
the temporary folder.
- Keep the existing deadline, file-transfer database pool, authorization
and publication rules.
- Add regression coverage for bounded concurrency, corruption, settled
cancellation, retained previous bytes and retry.
- Document the behavior and update the deployment plan with local
qualification evidence.

## Verification

- Real PostgreSQL and S3-compatible storage: 1,024 files of 64 KiB each,
plus instructions, with 25 ms added to each object GET. Delete the
entire predecessor home and disable local adoption before reading
through the successor.
- Before: the read failed after 30.065 seconds. After: the same case
passed in 4.180 seconds and fetched 1,025 objects. All owned test
resources were cleaned.
- The durable-file suite passed eight tests against PostgreSQL. Server
TypeScript compilation passed. All 56 latest-head CI checks are complete
and green or intentionally skipped, including full typecheck, build,
unit and browser suites. Greptile is 5/5 with no review findings or open
threads.
- The plan separately records a released-container overlap proof: 130
reads, 26 saves, zero failures, two deleted predecessor containers and a
retained authenticated session. That image predates this restore change.

## Risks

- A cold folder can use eight simultaneous streams instead of one. Two
file-transfer database connections bound the simultaneous restore
callers. Streams verify bytes without loading entire objects into
memory.
- The 30-second deadline is unchanged. The measured size and latency do
not qualify every folder at the maximum storage limits.
- No schema, API or UI changes. Paid-model and hosted deployment
acceptance remain separate work.

## Model Used

OpenAI Codex, GPT-6. Tool use, code execution and repository review. The
runtime does not expose a more specific model ID 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
- [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/__tests__/agent-file-durability.test.ts

@@ -3,6 +3,7 @@ import os from "node:os";
 import path from "node:path";
 import { randomUUID } from "node:crypto";
 import { Readable } from "node:stream";
+import { setTimeout as delay } from "node:timers/promises";
 import { beforeAll, beforeEach, afterEach, afterAll, describe, expect, it, vi } from "vitest";
 import { eq } from "drizzle-orm";
 import { agents, agentFileRoots, authUsers, companies, companyMemberships, principalPermissionGrants, heartbeatRuns, createDb, closeFileStorageDb } from "@paperclipai/db";
@@ -14,7 +15,7 @@ import { resolveManagedInstructionsRoot } from "../services/agent-instructions.j
 import { AGENT_FILE_OBJECT_GRACE_MS, collectAgentFileObjects, migrateAgentFiles } from "../services/agent-file-maintenance.js";
 import { getStorageService } from "../storage/index.js";
 
-const objects = vi.hoisted(() => ({ bytes: new Map<string, Buffer>(), rejectWrites: false, corruptReads: false }));
+const objects = vi.hoisted(() => ({ bytes: new Map<string, Buffer>(), rejectWrites: false, corruptReads: false, readDelayMs: 0, activeReads: 0, peakReads: 0 }));
 vi.mock("../storage/index.js", () => ({ getStorageService: () => ({
   provider: "s3",
   async putFile(input: { companyId: string; namespace: string; body: Readable; byteSize: number; sha256: string }) {
@@ -28,11 +29,18 @@ vi.mock("../storage/index.js", () => ({ getStorageService: () => ({
     objects.bytes.set(objectKey, body);
     return { objectKey };
   },
-  async getObject(companyId: string, objectKey: string) {
+  async getObject(companyId: string, objectKey: string, options: { signal?: AbortSignal } = {}) {
     expect(objectKey.startsWith(`${companyId}/`)).toBe(true);
     const body = objects.bytes.get(objectKey);
     if (!body) throw new Error("fixture object missing");
-    return { stream: Readable.from([objects.corruptReads ? Buffer.from("wrong bytes") : body]) };
+    return { stream: Readable.from((async function* () {
+      objects.activeReads++;
+      objects.peakReads = Math.max(objects.peakReads, objects.activeReads);
+      try {
+        if (objects.readDelayMs) await delay(objects.readDelayMs, undefined, { signal: options.signal });
+        yield objects.corruptReads ? Buffer.from("wrong bytes") : body;
+      } finally { objects.activeReads--; }
+    })()) };
   },
   async headObject(_companyId: string, key: string) { return { exists: objects.bytes.has(key), lastModified: new Date(0) }; },
   async deleteObject(companyId: string, key: string, options: { allVersions: boolean }) {
@@ -60,6 +68,7 @@ describe("durable agent file authority", () => {
     home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "durable-agent-files-")));
     vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_DURABLE_AGENT_FILES", "true");
     objects.bytes.clear(); objects.rejectWrites = false; objects.corruptReads = false;
+    objects.readDelayMs = 0; objects.activeReads = 0; objects.peakReads = 0;
     companyId = randomUUID(); agentId = randomUUID(); userId = randomUUID();
     await db.insert(companies).values({ id: companyId, name: "File authority", issuePrefix: randomUUID().slice(0, 8) });
     await db.insert(authUsers).values({ id: userId, name: "Editor", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
@@ -98,6 +107,34 @@ describe("durable agent file authority", () => {
     expect((await fs.stat(path.join(root, "empty-folder"))).isDirectory()).toBe(true);
   });
 
+  it("restores a cold folder with bounded parallel streams and settles all failed work before cleanup", async () => {
+    const contents = Buffer.alloc(128 * 1024, 0x81);
+    for (let index = 0; index < 40; index++) {
+      await materializeInstructionBytes(root, `nested/file-${index}.bin`, contents);
+    }
+    expect(await migrateAgentFiles(db, { companyId, apply: true, writersStopped: true })).toMatchObject({ pageReady: true });
+    await replaceFilesystem();
+    objects.readDelayMs = 25;
+    expect(await store().read(companyId, agentId, "nested/file-39.bin", actor())).toEqual(contents);
+    expect(objects.peakReads).toBeGreaterThan(1);
+    expect(objects.peakReads).toBeLessThanOrEqual(8);
+    expect(objects.activeReads).toBe(0);
+    for (let index = 0; index < 40; index++) {
+      expect(await fs.readFile(path.join(root, `nested/file-${index}.bin`))).toEqual(contents);
+    }
+
+    // A corrupt replacement must preserve the old projection and await every
+    // cancelled sibling stream before deleting its temporary directory.
+    await invalidateAgentFileProjection(root);
+    objects.corruptReads = true;
+    await expect(store().read(companyId, agentId, "nested/file-39.bin", actor())).rejects.toThrow(/committed (size|digest)/);
+    expect(objects.activeReads).toBe(0);
+    expect((await fs.readdir(path.dirname(root))).filter(name => name.startsWith(".agent-files-"))).toEqual([]);
+    expect(await fs.readFile(path.join(root, "nested/file-39.bin"))).toEqual(contents);
+    objects.corruptReads = false;
+    expect(await store().read(companyId, agentId, "nested/file-39.bin", actor())).toEqual(contents);
+  });
+
   it("does not promote a rejected first save through subsequent local adoption", async () => {
     objects.rejectWrites = true;
     await expect(store().write({ companyId, agentId, path: "failed.txt", bytes: Buffer.from("uncommitted"), baseHash: null }, actor()))

server/src/services/agent-file-durability.ts

@@ -84,15 +84,19 @@ export async function restoreDurableAgentFiles(tx: Tx, owner: Owner, root: strin
   if (generation === head.generation && await fs.stat(root).then(s => s.isDirectory()).catch(() => false)) return true;
   const temporary = path.join(path.dirname(root), `.agent-files-${randomUUID()}`);
   await fs.mkdir(temporary, { recursive: true, mode: 0o700 });
-  const signal = AbortSignal.timeout(30_000);
+  const failed = new AbortController();
+  const signal = AbortSignal.any([AbortSignal.timeout(30_000), failed.signal]);
   try {
-    // Complete and verify the replacement before discarding an older projection.
-    for (const entry of head.entries) {
+    // Bound open streams and file handles while avoiding one network round trip
+    // per file in series. Await every worker before removing a failed projection.
+    const entries = head.entries;
+    let next = 0;
+    const restoreEntry = async (entry: DurableAgentFileEntry) => {
       signal.throwIfAborted();
       const relative = instructionPath(entry.path);
       if (entry.kind === "dir") {
         await assertInstructionPathSafe(temporary, `${relative}/.path-check`);
-        await fs.mkdir(path.join(temporary, relative), { recursive: true, mode: 0o700 }); continue;
+        await fs.mkdir(path.join(temporary, relative), { recursive: true, mode: 0o700 }); return;
       }
       const filename = await assertInstructionPathSafe(temporary, relative);
       await fs.mkdir(path.dirname(filename), { recursive: true, mode: 0o700 });
@@ -110,7 +114,20 @@ export async function restoreDurableAgentFiles(tx: Tx, owner: Owner, root: strin
         }
         if (size !== entry.size || hash.digest("hex") !== entry.sha256) throw new Error("Agent file does not match its committed digest");
       } finally { await handle.close(); }
-    }
+    };
+    const workers = Array.from({ length: Math.min(8, entries.length) }, async () => {
+      try {
+        while (next < entries.length) {
+          signal.throwIfAborted();
+          await restoreEntry(entries[next++]!);
+        }
+      } catch (error) {
+        failed.abort(error);
+        throw error;
+      }
+    });
+    await Promise.allSettled(workers);
+    signal.throwIfAborted();
     await invalidateAgentFileProjection(root);
     await fs.rm(root, { recursive: true, force: true });
     await fs.rename(temporary, root);