fix(runtime): bound parallel offline harness backup transfers (#19108)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - An ephemeral web server must retain provider history outside its
local disk.
> - Offline enrollment moves verified local harness backups to object
storage.
> - A cold rehearsal reached 699 current backup rows while processing
1,953 sources before its 25-minute deadline.
> - The serial transfer loop prevents this existing dataset from
enrolling.
> - This pull request overlaps at most four transfers within the
existing total scratch limit.
> - The benefit is faster preservation with the same ownership and
publication checks.

## Linked Issues or Issue Description

Refs #19038, #18943, #18932.

**What happened?**

Offline enrollment processed 1,953 local provider backups one at a time.
The normal Cloud image reached 699 current backup rows (including nine
pre-existing rows) before maintenance expired. The serving instance
remained unchanged during the isolated test.

**Expected behavior**

Enrollment must preserve every source within bounded time and disk use.
It must stop all in-flight transfers before it reports failure.

**Steps to reproduce**

1. Prepare a settled instance with multiple verified native harness
backups.
2. Run offline enrollment against shared object storage.
3. Observe that the serial owner waits for each archive, upload,
readback and publication before starting the next source.

## What Changed

- Process groups of up to four verified offline harness backups.
- Reserve space for the archive and encrypted copy before each group
starts. Keep the total scratch limit equal to one maximum-size serial
transfer.
- Derive a smaller archive limit from a closed manifest directory set.
Preserve unlisted files with an exclusive maximum-size reservation.
- Enforce the reservation while streaming. Keep the existing maximum for
ordinary runtime backups.
- Cancel and await peer transfers after failure. Keep completed
publications valid for retry.
- Preserve the original archive-stream failure when the tar writer also
fails after destruction.
- Document the migration bound and add integration regressions for
overlap, restoration, cancellation, retry, extra files and archive
limits.

## Verification

- Serial baseline: the new overlap regression fails with one active
transfer when four are required.
- Candidate: 59 targeted tests pass with real PostgreSQL, S3, filesystem
archives and restored provider bytes. The bundled-entry test is skipped
locally because this source worktree has no runtime bundle. The full
normal image rehearsal remains required.
- Follow-up: 48 archive/storage checks pass after correcting the
PAX-header allowance. The real encoder produces a 972,801,024-byte
full-entry projection, above the old 805,306,368-byte allowance and
below the new 1,024,001,024-byte bound. The existing global object limit
remains in force. The old formula fails the new regression. Exact-size
restoration and one-byte-over-limit rejection pass.
- The previous revision passed full CI after one setup-route timeout
retry. That failure remains recorded; the corrected current revision
f7c2be1289 passed fresh full CI (41937700826) and Greptile 5/5. Its
review thread is resolved.
- Local server typecheck found four test-only Promise library errors;
those are fixed. The remaining 26 diagnostics come from stale linked
dependency declarations outside the changed files. Clean full CI is
required before merge.
- The resumed isolated staging-copy experiment also exceeded its
25-minute deadline. It reached 1,389 current backups. This improves
transfer throughput but does not by itself qualify enrollment; online
preparation is tracked in #19137.
- Full fresh normal-image migration, empty-home restoration and hosted
release handoffs remain deployment acceptance gates.

## Risks

- Parallel work increases concurrent object-storage and database
requests to at most four transfers. The scratch reservation still bounds
archive and ciphertext files together.
- A malformed or changed source can fail its streaming limit. The
migration retains its source and stops before export.
- Partial completed publications survive failure. Retry verifies them
under the existing hash and ownership rules.
- No serving configuration, deployment pin, database contract or
provider lifecycle policy changes in this pull request.

## Model Used

OpenAI GPT-6.1 Codex, exact model ID `gpt-6.1-sol`. Used for reasoning,
code editing, shell tools and test execution. The runtime does not
expose its 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>

doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md

@@ -170,3 +170,30 @@ keeps its supervisor and ownership requirements.
 For a retired reusable sandbox lease, the existing remote-termination resolver
 may prove destruction through its exact successor handoff chain. A stopped
 successor, broken chain or competing tip does not grant archival authority.
+
+### Bounded offline harness transfers
+
+The full bundled cold rehearsal reached only 699 of 1,953 local harness backups
+before the 25-minute maintenance deadline. Keep every source and its existing
+ownership and hash checks. Migrate up to four sources together. Reserve space for
+both each tar archive and its encrypted copy before starting a group, with the
+same total scratch limit as one maximum-size serial transfer. Use the verified
+manifest sizes for a closed directory set; give sources with extra files an
+exclusive maximum-size reservation so those files are preserved. Enforce the
+reservation while streaming the archive. Ordinary runtime backups keep their
+existing maximum archive limit.
+
+Derive tar overhead from the maximum entry count and both maximum path lengths.
+PAX can encode a path and symlink target for every entry. The difference between
+the global object and content limits is not a safe per-source overhead estimate.
+Check the bound against real maximum-path PAX encoding at the full entry budget,
+and check streaming at an exact byte limit and one byte beyond it.
+
+On failure, cancel and await every transfer in the group before returning. A
+completed publication remains valid for the next attempt. Tests must exercise
+real PostgreSQL publication and byte restoration, the four-transfer bound,
+failure while another storage call is unresolved, retry, extra-file preservation,
+and rejection at the archive limit before any object or row is published. This
+finite offline operation adds no idle background work. A resumed rehearsal can
+measure the fix but cannot replace fresh full-image migration and empty-home
+recovery proof.

server/src/services/native-runtime/native-harness-backup-archive.test.ts

@@ -5,7 +5,9 @@ import { tmpdir } from "node:os";
 import { pipeline } from "node:stream/promises";
 import tar from "tar-stream";
 import { afterEach, expect, it } from "vitest";
-import { captureNativeHarnessArchive, restoreNativeHarnessArchive, MAX_NATIVE_HARNESS_BYTES } from "./native-harness-backup-archive.js";
+import { captureNativeHarnessArchive, restoreNativeHarnessArchive, MAX_NATIVE_HARNESS_BYTES,
+  MAX_NATIVE_HARNESS_ENTRIES, MAX_NATIVE_HARNESS_OBJECT_BYTES, MAX_NATIVE_HARNESS_PATH_BYTES,
+  nativeHarnessArchiveSizeLimit } from "./native-harness-backup-archive.js";
 
 const roots: string[] = [];
 afterEach(async () => { for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true }); });
@@ -16,6 +18,32 @@ async function maliciousArchive(root: string, headers: Array<tar.Headers>) {
   for (const header of headers) archive.entry(header, Buffer.alloc(0));
   archive.finalize(); await written; return file;
 }
+it("reserves the full entry budget using real maximum-path PAX encoding", async () => {
+  const root = await fixture(), name = "a/".repeat(MAX_NATIVE_HARNESS_PATH_BYTES / 2 - 1) + "aa";
+  const file = await maliciousArchive(root, [{ name, type: "symlink", linkname: "b".repeat(MAX_NATIVE_HARNESS_PATH_BYTES) }]);
+  // One real encoded record gives the repeated-entry size without creating a
+  // gigabyte fixture. Each hypothetical entry includes its own PAX records.
+  const entryBytes = (await stat(file)).size - 1024;
+  const denseArchiveBytes = entryBytes * MAX_NATIVE_HARNESS_ENTRIES + 1024;
+  const bound = nativeHarnessArchiveSizeLimit(0);
+  expect(denseArchiveBytes).toBeGreaterThan(MAX_NATIVE_HARNESS_OBJECT_BYTES - MAX_NATIVE_HARNESS_BYTES);
+  expect(denseArchiveBytes).toBeLessThanOrEqual(bound);
+  expect(denseArchiveBytes).toBeGreaterThan(bound * 0.9);
+  expect(nativeHarnessArchiveSizeLimit(MAX_NATIVE_HARNESS_BYTES)).toBe(MAX_NATIVE_HARNESS_OBJECT_BYTES);
+});
+it("captures at the exact archive byte limit and refuses one byte beyond it", async () => {
+  const root = await fixture(), source = join(root, "source"); await mkdir(source);
+  await writeFile(join(source, "manifest.json"), "{}"); await writeFile(join(source, "state"), "retained provider bytes");
+  const original = join(root, "original"); await captureNativeHarnessArchive(source, original);
+  const bytes = (await stat(original)).size, exact = join(root, "exact");
+  await captureNativeHarnessArchive(source, exact, undefined, bytes);
+  expect((await stat(exact)).size).toBe(bytes);
+  await restoreNativeHarnessArchive(exact, join(root, "restored-exact"));
+  expect(await readFile(join(root, "restored-exact", "state"), "utf8")).toBe("retained provider bytes");
+  const exceeded = join(root, "exceeded");
+  await expect(captureNativeHarnessArchive(source, exceeded, undefined, bytes - 1)).rejects.toThrow("native_harness_archive_invalid");
+  expect((await stat(exceeded)).size).toBeLessThan(bytes);
+});
 it.each([
   { name: "../escaped", type: "file" as const },
   { name: "/absolute", type: "file" as const },

server/src/services/native-runtime/native-harness-backup-archive.ts

@@ -8,12 +8,22 @@ import tar from "tar-stream";
 export const MAX_NATIVE_HARNESS_BYTES = 4 * 1024 * 1024 * 1024;
 export const MAX_NATIVE_HARNESS_OBJECT_BYTES = MAX_NATIVE_HARNESS_BYTES + 768 * 1024 * 1024;
 export const MAX_NATIVE_HARNESS_ENTRIES = 100_000;
+export const MAX_NATIVE_HARNESS_PATH_BYTES = 4096;
+// PAX can store both a full path and a full symlink target for each entry.
+// Two KiB also cover PAX framing, both headers and payload padding. The final
+// two zero blocks terminate the archive. Keep this separate from content size.
+export const MAX_NATIVE_HARNESS_TAR_OVERHEAD_BYTES = MAX_NATIVE_HARNESS_ENTRIES
+  * (2 * MAX_NATIVE_HARNESS_PATH_BYTES + 2048) + 1024;
 const fail = () => new Error("native_harness_archive_invalid");
-const safeName = (name: string) => name.length > 0 && Buffer.byteLength(name) <= 4096 && !name.includes("\\")
+export function nativeHarnessArchiveSizeLimit(contentBytes: number) {
+  if (!Number.isSafeInteger(contentBytes) || contentBytes < 0 || contentBytes > MAX_NATIVE_HARNESS_BYTES) throw fail();
+  return Math.min(MAX_NATIVE_HARNESS_OBJECT_BYTES, contentBytes + MAX_NATIVE_HARNESS_TAR_OVERHEAD_BYTES);
+}
+const safeName = (name: string) => name.length > 0 && Buffer.byteLength(name) <= MAX_NATIVE_HARNESS_PATH_BYTES && !name.includes("\\")
   && !/[\u0000-\u001f\u007f]/u.test(name) && !posix.isAbsolute(name)
   && name.split("/").every(part => part !== ".." && part !== "." && part !== "");
 
-const safeLink = (name: string, target: string) => target.length > 0 && Buffer.byteLength(target) <= 4096
+const safeLink = (name: string, target: string) => target.length > 0 && Buffer.byteLength(target) <= MAX_NATIVE_HARNESS_PATH_BYTES
   && !target.includes("\\") && !/[\u0000-\u001f\u007f]/u.test(target) && !posix.isAbsolute(target)
   && !["..", "."].includes(posix.normalize(posix.join(posix.dirname(name), target)))
   && !posix.normalize(posix.join(posix.dirname(name), target)).startsWith("../");
@@ -25,19 +35,21 @@ const validateLinks = (links: Map<string, string>) => {
     }
   }
 };
-const boundedArchive = () => {
+const boundedArchive = (limit = MAX_NATIVE_HARNESS_OBJECT_BYTES) => {
   let bytes = 0;
   return new Transform({ transform(chunk, _encoding, callback) {
     bytes += chunk.length;
-    callback(bytes > MAX_NATIVE_HARNESS_OBJECT_BYTES ? fail() : null, chunk);
+    callback(bytes > limit ? fail() : null, chunk);
   } });
 };
 
-export async function captureNativeHarnessArchive(root: string, target: string, signal?: AbortSignal) {
+export async function captureNativeHarnessArchive(root: string, target: string, signal?: AbortSignal, maxArchiveBytes = MAX_NATIVE_HARNESS_OBJECT_BYTES) {
+  if (!Number.isSafeInteger(maxArchiveBytes) || maxArchiveBytes <= 0 || maxArchiveBytes > MAX_NATIVE_HARNESS_OBJECT_BYTES) throw fail();
   if (!(await lstat(root)).isDirectory()) throw fail();
   const archive = tar.pack();
-  const output = pipeline(archive, boundedArchive(), createWriteStream(target, { flags: "wx", mode: 0o600 }), { signal });
-  void output.catch(() => undefined);
+  let outputFailure: unknown;
+  const output = pipeline(archive, boundedArchive(maxArchiveBytes), createWriteStream(target, { flags: "wx", mode: 0o600 }), { signal });
+  void output.catch(error => { outputFailure = error; });
   let count = 0, bytes = 0;
   const links = new Map<string, string>();
   const visit = async (relative: string): Promise<void> => {
@@ -72,7 +84,11 @@ export async function captureNativeHarnessArchive(root: string, target: string,
     }
   };
   try { await visit(""); validateLinks(links); archive.finalize(); await output; }
-  catch (error) { archive.destroy(error as Error); await output.catch(() => undefined); throw error; }
+  catch (error) {
+    archive.destroy((outputFailure ?? error) as Error);
+    await output.catch(() => undefined);
+    throw outputFailure ?? error;
+  }
 }
 
 export async function restoreNativeHarnessArchive(source: string, target: string, signal?: AbortSignal) {

server/src/services/native-runtime/native-harness-backup-migration.ts

@@ -6,6 +6,7 @@ import { environmentLeases, heartbeatRuns, nativeHarnessBackups, type Db } from
 import type { StorageService } from "../../storage/types.js";
 import { nativeHarnessStateBase, readNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js";
 import { persistNativeHarnessBackup, verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
+import { MAX_NATIVE_HARNESS_BYTES, MAX_NATIVE_HARNESS_OBJECT_BYTES, nativeHarnessArchiveSizeLimit } from "./native-harness-backup-archive.js";
 import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
 
 export async function hasLocalNativeHarnessObligations(db: Db): Promise<boolean> {
@@ -21,6 +22,79 @@ export async function hasLocalNativeHarnessObligations(db: Db): Promise<boolean>
   return rows.length > 0;
 }
 
+const MAX_PARALLEL_BACKUPS = 4;
+// An archive and its encrypted copy coexist. Keep the same worst-case scratch
+// bound as one maximum-size serial transfer, including tar header overhead.
+const MAX_SCRATCH_BYTES = 2 * MAX_NATIVE_HARNESS_OBJECT_BYTES;
+
+async function prepareSource(base: string, scopeDigest: string) {
+  const root = join(base, scopeDigest, "failover-backups", "current");
+  const stat = await lstat(root).catch((error: NodeJS.ErrnoException) => {
+    if (error.code === "ENOENT") return null; throw error;
+  });
+  if (!stat) return null;
+  if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("native_harness_migration_unsafe_source");
+  const manifestBytes = await readFile(join(root, "manifest.json"));
+  const manifest = JSON.parse(manifestBytes.toString("utf8")) as Record<string, unknown>;
+  const state = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8"));
+  if (state.lifecycle !== "suspended" || typeof state.runId !== "string"
+    || manifest.schema !== "paperclip.native-harness-backup.v1") throw new Error("native_harness_migration_source_unsettled");
+  const names = new Set<string>();
+  let bytes = manifestBytes.length, closed = Array.isArray(manifest.directories) && manifest.directories.length > 0;
+  for (const value of Array.isArray(manifest.directories) ? manifest.directories : []) {
+    const entry = value as { name?: unknown; bytes?: unknown } | null;
+    if (!entry || typeof entry.name !== "string" || !/^[A-Za-z0-9._-]+$/.test(entry.name)
+      || [".", ".."].includes(entry.name) || !Number.isSafeInteger(entry.bytes) || Number(entry.bytes) < 0) { closed = false; break; }
+    names.add(entry.name); bytes += Number(entry.bytes);
+  }
+  const entries = await readdir(root, { withFileTypes: true });
+  closed &&= entries.length === names.size + 1 && entries.every(entry => entry.name === "manifest.json"
+    ? entry.isFile() : names.has(entry.name) && entry.isDirectory() && !entry.isSymbolicLink());
+  // Unknown extra files are preserved with an exclusive full-size reservation.
+  const archiveLimit = closed && Number.isSafeInteger(bytes) && bytes <= MAX_NATIVE_HARNESS_BYTES
+    ? nativeHarnessArchiveSizeLimit(bytes)
+    : MAX_NATIVE_HARNESS_OBJECT_BYTES;
+  return { root, scopeDigest, runId: state.runId as string, archiveLimit,
+    sessionId: manifest.normalizedSessionId, runnerId: manifest.runnerInstanceId,
+    manifestSha256: `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}` };
+}
+
+type Source = NonNullable<Awaited<ReturnType<typeof prepareSource>>>;
+
+async function migrateSource(db: Db, input: { storage: StorageService; signal: AbortSignal }, source: Source) {
+  input.signal.throwIfAborted();
+  const { root, scopeDigest, manifestSha256 } = source;
+  const [run] = await db.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId,
+    sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns)
+    .where(eq(heartbeatRuns.id, source.runId));
+  if (!run) return 0;
+  if (!run.issueId || run.sessionId !== source.sessionId || run.runnerId !== source.runnerId) {
+    throw new Error("native_harness_migration_binding_mismatch");
+  }
+  const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId),
+    eq(nativeHarnessBackups.scopeDigest, scopeDigest), eq(nativeHarnessBackups.status, "current")));
+  if (current) {
+    if (current.manifestSha256 !== manifestSha256) throw new Error("native_harness_migration_durable_source_conflict");
+    await verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: run.companyId,
+      scopeDigest, manifestSha256: current.manifestSha256, signal: input.signal });
+    return 1;
+  }
+  const leases = await db.select({ stamp: sql<unknown>`${environmentLeases.metadata}->'nativeHarnessBackup'`,
+    providerLeaseId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and(
+    eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.issueId, run.issueId),
+    sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${manifestSha256}`));
+  let valid = false;
+  for (const lease of leases) {
+    const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? "");
+    if (stamp?.sessionScopeSha256 === scopeDigest && await verifyNativeHarnessBackupStampAsync(stamp, lease.providerLeaseId!, input.signal)) { valid = true; break; }
+  }
+  if (!valid) throw new Error("native_harness_migration_backup_unverified");
+  await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest, signal: input.signal,
+    maxArchiveBytes: source.archiveLimit,
+    owner: { kind: "offline_migration", writersStopped: true, companyId: run.companyId, issueId: run.issueId, runId: source.runId } });
+  return 1;
+}
+
 export async function migrateNativeHarnessBackups(db: Db, input: {
   storage: StorageService; writersStopped: true; signal: AbortSignal;
 }): Promise<number> {
@@ -31,49 +105,31 @@ export async function migrateNativeHarnessBackups(db: Db, input: {
   const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => {
     if (error.code === "ENOENT") return [] as string[]; throw error;
   });
-  let count = 0;
+  let count = 0, reserved = 0, wave: Source[] = [];
+  const flush = async () => {
+    const cancellation = new AbortController();
+    const signal = AbortSignal.any([input.signal, cancellation.signal]);
+    const settled = await Promise.allSettled(wave.map(async source => {
+      try { return await migrateSource(db, { ...input, signal }, source); }
+      catch (error) { cancellation.abort(error); throw error; }
+    }));
+    // Never return failure while a peer can still upload or publish. Partial
+    // completed transfers remain valid for the next offline attempt.
+    if (cancellation.signal.aborted) throw cancellation.signal.reason;
+    for (const result of settled) {
+      if (result.status === "rejected") throw result.reason;
+      count += result.value;
+    }
+    wave = []; reserved = 0;
+  };
   for (const scopeDigest of directories.filter(name => /^[a-f0-9]{64}$/.test(name)).sort()) {
     input.signal.throwIfAborted();
-    const root = join(base, scopeDigest, "failover-backups", "current");
-    const stat = await lstat(root).catch((error: NodeJS.ErrnoException) => {
-      if (error.code === "ENOENT") return null; throw error;
-    });
-    if (!stat) continue;
-    if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("native_harness_migration_unsafe_source");
-    const manifestBytes = await readFile(join(root, "manifest.json"));
-    const manifest = JSON.parse(manifestBytes.toString("utf8"));
-    const state = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8"));
-    if (state.lifecycle !== "suspended" || typeof state.runId !== "string"
-      || manifest.schema !== "paperclip.native-harness-backup.v1") throw new Error("native_harness_migration_source_unsettled");
-    const [run] = await db.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId,
-      sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns)
-      .where(eq(heartbeatRuns.id, state.runId));
-    if (!run) continue;
-    if (!run.issueId || run.sessionId !== manifest.normalizedSessionId || run.runnerId !== manifest.runnerInstanceId) {
-      throw new Error("native_harness_migration_binding_mismatch");
-    }
-    const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId),
-      eq(nativeHarnessBackups.scopeDigest, scopeDigest), eq(nativeHarnessBackups.status, "current")));
-    const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`;
-    if (current) {
-      if (current.manifestSha256 !== manifestSha256) throw new Error("native_harness_migration_durable_source_conflict");
-      await verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: run.companyId,
-        scopeDigest, manifestSha256: current.manifestSha256, signal: input.signal });
-      count++; continue;
-    }
-    const leases = await db.select({ stamp: sql<unknown>`${environmentLeases.metadata}->'nativeHarnessBackup'`,
-      providerLeaseId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and(
-      eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.issueId, run.issueId),
-      sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${manifestSha256}`));
-    let valid = false;
-    for (const lease of leases) {
-      const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? "");
-      if (stamp?.sessionScopeSha256 === scopeDigest && await verifyNativeHarnessBackupStampAsync(stamp, lease.providerLeaseId!, input.signal)) { valid = true; break; }
-    }
-    if (!valid) throw new Error("native_harness_migration_backup_unverified");
-    await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest, signal: input.signal,
-      owner: { kind: "offline_migration", writersStopped: true, companyId: run.companyId, issueId: run.issueId, runId: state.runId } });
-    count++;
+    const source = await prepareSource(base, scopeDigest);
+    if (!source) continue;
+    const reservation = 2 * source.archiveLimit;
+    if (wave.length && (wave.length === MAX_PARALLEL_BACKUPS || reserved + reservation > MAX_SCRATCH_BYTES)) await flush();
+    wave.push(source); reserved += reservation;
   }
+  if (wave.length) await flush();
   return count;
 }

server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts

@@ -19,6 +19,11 @@ import { legacyControllerBootId } from "../legacy-controller-lease.js";
 describeEmbeddedPostgres("durable native provider backups", () => {
   const environment = nativeRestartFixture();
   const ownedHomes: string[] = [];
+  function deferred() {
+    let resolve!: () => void;
+    const promise = new Promise<void>(done => { resolve = done; });
+    return { promise, resolve };
+  }
   afterEach(async () => {
     for (const home of ownedHomes.splice(0)) await removeNativeHarnessBackup(home);
     vi.unstubAllEnvs();
@@ -45,6 +50,99 @@ describeEmbeddedPostgres("durable native provider backups", () => {
     return { ...f, home, root, companyId, owner, scopeDigest, sessionScopeId, storage, identity,
       manifestSha256: `sha256:${createHash("sha256").update(manifest).digest("hex")}` };
   }
+  async function offlineFixture(base: string) {
+    const f = await fixture("codex-home"), home = join(base, f.scopeDigest);
+    await rename(f.home, home);
+    const root = join(home, "failover-backups", "current"), runnerInstanceId = randomUUID(), providerLeaseId = randomUUID();
+    const execution = { provider: { kind: "codex" }, binding: { companyId: f.companyId, issueId: f.issueId,
+      runId: f.runId, agentId: environment.current.agentId, executionWorkspaceId: randomUUID() },
+      workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
+      session: { normalizedSessionId: f.identity.normalizedSessionId, driverKind: "codex_app_server",
+        protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } as never;
+    const manifest = buildNativeHarnessBackupManifest({ backupRoot: root, execution, runnerInstanceId,
+      providerSessionIdentity: { providerSessionId: f.identity.providerSessionId }, sourceProviderLeaseId: providerLeaseId });
+    await writeFile(join(root, "manifest.json"), JSON.stringify(manifest));
+    const stamp = createNativeHarnessBackupStamp({ manifestPath: join(root, "manifest.json"), sessionScopeId: f.sessionScopeId,
+      authorizedProviderLeaseId: providerLeaseId, runnerInstanceId, normalizedSessionId: f.identity.normalizedSessionId, completedAt: manifest.completedAt });
+    await f.db.update(heartbeatRuns).set({ nativeSessionId: f.identity.normalizedSessionId, runnerInstanceId }).where(eq(heartbeatRuns.id, f.runId));
+    await f.db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId,
+      providerLeaseId, status: "expired", metadata: { nativeHarnessBackup: stamp } });
+    return { ...f, home, root };
+  }
+  async function offlineFixtures(count: number) {
+    const base = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(base); await mkdir(base);
+    vi.stubEnv("PAPERCLIP_RUNNER_STATE_DIR", base);
+    const fixtures = [];
+    for (let i = 0; i < count; i++) fixtures.push(await offlineFixture(base));
+    const runs = await fixtures[0].db.select({ id: heartbeatRuns.id, issueId: heartbeatRuns.nativeIssueId }).from(heartbeatRuns)
+      .where(eq(heartbeatRuns.companyId, fixtures[0].companyId));
+    for (const run of runs) await settle({ ...fixtures[0], runId: run.id, issueId: run.issueId! }, new Date());
+    return fixtures.sort((a, b) => a.scopeDigest.localeCompare(b.scopeDigest));
+  }
+  it("overlaps verified offline backups within a four-transfer bound and restores their bytes", async () => {
+    const fixtures = await offlineFixtures(6), f = fixtures[0], gate = deferred();
+    const put = f.storage.putFile.bind(f.storage); let active = 0, peak = 0, entered = 0;
+    vi.spyOn(f.storage, "putFile").mockImplementation(async input => {
+      active++; entered++; peak = Math.max(peak, active);
+      try { await gate.promise; return await put(input); } finally { active--; }
+    });
+    const migration = migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) });
+    try { await vi.waitFor(() => expect(active).toBe(4), { timeout: 3000 }); }
+    finally { gate.resolve(); await migration.catch(() => undefined); }
+    expect(await migration).toBe(6); expect(peak).toBe(4); expect(entered).toBe(6); expect(active).toBe(0);
+    for (const source of fixtures) {
+      await removeNativeHarnessBackup(source.home);
+      const replacement = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(replacement);
+      expect(await restoreDurableNativeHarnessBackup({ ...source, storage: f.storage, backupRoot: replacement })).toBe(true);
+      expect(JSON.parse(await readFile(join(replacement, "current", "codex-home", "sessions", "history.json"), "utf8"))).toEqual(source.identity);
+    }
+  });
+  it("waits for cancelled peer transfers before returning failure and permits a complete retry", async () => {
+    const fixtures = await offlineFixtures(2), f = fixtures[0], peerEntered = deferred(), release = deferred();
+    const failed = deferred(), put = f.storage.putFile.bind(f.storage); let entered = 0, active = 0, settled = false;
+    const spy = vi.spyOn(f.storage, "putFile").mockImplementation(async input => {
+      const ordinal = ++entered; active++;
+      try {
+        if (ordinal === 1) { await peerEntered.promise; failed.resolve(); throw new Error("storage unavailable"); }
+        peerEntered.resolve(); await release.promise; return await put(input);
+      } finally { active--; }
+    });
+    const migration = migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) });
+    const completed = migration.then(() => { settled = true; }, () => { settled = true; });
+    try {
+      await vi.waitFor(() => expect(entered).toBe(2), { timeout: 3000 }); await failed.promise;
+      await new Promise<void>(resolve => setImmediate(resolve)); expect(settled).toBe(false);
+    } finally { peerEntered.resolve(); release.resolve(); await completed; }
+    await expect(migration).rejects.toThrow("storage unavailable"); expect(active).toBe(0); spy.mockRestore();
+    for (const source of fixtures) expect(await readFile(join(source.root, "manifest.json"), "utf8")).toContain("paperclip.native-harness-backup.v1");
+    expect(await migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) })).toBe(2);
+    for (const source of fixtures) {
+      const rows = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, source.scopeDigest));
+      expect(rows.filter(row => row.status === "current")).toHaveLength(1);
+    }
+  });
+  it("preserves unlisted files with an exclusive scratch reservation", async () => {
+    const fixtures = await offlineFixtures(3), f = fixtures[0], extra = fixtures[1];
+    await writeFile(join(extra.root, "retained-extra.txt"), "unlisted surviving bytes");
+    const put = f.storage.putFile.bind(f.storage); let active = 0, peak = 0;
+    vi.spyOn(f.storage, "putFile").mockImplementation(async input => {
+      active++; peak = Math.max(peak, active);
+      try { await new Promise<void>(resolve => setImmediate(resolve)); return await put(input); } finally { active--; }
+    });
+    expect(await migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) })).toBe(3);
+    expect(peak).toBe(1);
+    await removeNativeHarnessBackup(extra.home);
+    const replacement = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(replacement);
+    expect(await restoreDurableNativeHarnessBackup({ ...extra, storage: f.storage, backupRoot: replacement })).toBe(true);
+    expect(await readFile(join(replacement, "current", "retained-extra.txt"), "utf8")).toBe("unlisted surviving bytes");
+  });
+  it("enforces the reserved archive size while streaming before publishing any object", async () => {
+    const f = await fixture(), put = vi.spyOn(f.storage, "putFile");
+    await expect(persistNativeHarnessBackup({ ...f, maxArchiveBytes: 512 })).rejects.toThrow("native_harness_archive_invalid");
+    expect(put).not.toHaveBeenCalled();
+    expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toHaveLength(0);
+    expect(JSON.parse(await readFile(join(f.root, "codex", "sessions", "history.json"), "utf8"))).toEqual(f.identity);
+  });
   it.each(["codex", "claude", "opencode", "grok", "new-provider"])("restores %s session bytes after deleting the entire controller home", async provider => {
     const f = await fixture(provider);
     await persistNativeHarnessBackup(f);

server/src/services/native-runtime/native-harness-backup-storage.ts

@@ -142,7 +142,7 @@ async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner:
 
 export async function persistNativeHarnessBackup(input: {
   db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner;
-  scopeDigest: string; root: string; signal?: AbortSignal;
+  scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number;
 }): Promise<void> {
   if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid");
   const manifestBytes = await readFile(join(input.root, "manifest.json"));
@@ -152,7 +152,7 @@ export async function persistNativeHarnessBackup(input: {
   const key = randomBytes(32);
   try {
     const archive = join(scratch, "archive"), encrypted = join(scratch, "encrypted");
-    await captureNativeHarnessArchive(input.root, archive, input.signal);
+    await captureNativeHarnessArchive(input.root, archive, input.signal, input.maxArchiveBytes);
     const plaintext = await digest(archive, input.signal), iv = randomBytes(12);
     const cipher = createCipheriv("aes-256-gcm", key, iv); cipher.setAAD(aad(binding));
     await pipeline(createReadStream(archive), cipher, createWriteStream(encrypted, { flags: "wx", mode: 0o600 }), { signal: input.signal });