feat(runtime): pre-copy settled harness history while serving

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

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

@@ -214,3 +214,39 @@ source, export-failure cause or unknown recovery kind retains the existing
 blocker unless the importer already records that exact legacy recovery source.
 Live migration keeps its recovery exclusion. Recheck recovery evidence before
 publishing the saved workspace.
+
+### Online preparation for large provider histories
+
+Parallel offline transfer alone did not fit the copied staging dataset in the
+maintenance window. A resumed attempt reached 1,389 of 1,953 sources. Keep the
+offline deadline and move successful released history before draining the server.
+The inventory contains 1,802 such backups, approximately 48.5 GB of the 54.8 GB
+declared source bytes. This is preparatory publication under recorded runtime
+authority, not permission to detach a volume.
+
+Run the finite operator entry from a normal released image:
+
+```sh
+PAPERCLIP_NATIVE_HARNESS_PRESTAGE=true node --max-old-space-size=512 server/dist-runtime/native-harness-prestage.mjs
+```
+
+The entry loads the stored Cloud configuration and requires shared object storage.
+It processes one source at a time, has a 90-minute deadline, and stops on shutdown.
+It starts no scheduler, does not drain tasks, and always reports
+`enrollmentReady: false`. Check serving memory headroom before running a separate
+process on a tenant. Keep the deployment on normal releases.
+
+Only a successful, finished source with a released committed coordinator and an
+accepted result can publish through this path. At both database fences, hold the
+task share lock and recheck its checkout, related queued/running executions,
+recorded session and Runner, exact backup-stamp lease, and unchanged local
+manifest and suspended state. A concurrent turn, moved source, revoked stamp or
+changed owner defers that source. Unrelated executions continue unchanged.
+No lease, result, task or lifecycle state is rewritten to make a source eligible.
+
+The final offline migration still stops all writers, verifies every durable
+object, imports deferred sources and all other data, and exports only its normal
+complete eligibility receipt. Qualification starts from the saved pristine
+database, records the separate preparation phase, and then proves the normal
+bundled offline entry and empty-home restoration. Preparation alone cannot pass
+enrollment or active-agent release acceptance.

server/scripts/bundle-runtime.mjs

@@ -71,11 +71,12 @@ function externalizeThirdParty(isBundled, runtimeDir) {
   };
 }
 
-export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) {
+export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, nativeHarnessPrestageEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) {
   await rm(outdir, { recursive: true, force: true });
   const result = await build({
     entryPoints: { index: "paperclip-runtime-entry", server: entry, ...(maintenanceEntry ? { "agent-files": maintenanceEntry } : {}),
-      ...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}) },
+      ...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}),
+      ...(nativeHarnessPrestageEntry ? { "native-harness-prestage": nativeHarnessPrestageEntry } : {}) },
     outdir,
     outExtension: { ".js": ".mjs" },
     chunkNames: "chunk-[hash]",
@@ -105,7 +106,8 @@ async function main() {
   const started = performance.now();
   const summary = await bundleRuntime({ entry: join(serverDir, "dist", "index.js"), outdir: join(serverDir, "dist-runtime"),
     maintenanceEntry: join(serverDir, "dist", "maintenance", "agent-files.js"),
-    ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js") });
+    ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js"),
+    nativeHarnessPrestageEntry: join(serverDir, "dist", "maintenance", "native-harness-prestage.js") });
   console.log(`[bundle-runtime] ${summary.inputs} inputs -> ${summary.chunks} chunks in ${Math.round(performance.now() - started)} ms: ${summary.outfile}`);
 }
 

server/scripts/bundle-runtime.test.mjs

@@ -89,6 +89,15 @@ describe("bundleRuntime", () => {
     expect(JSON.parse(stdout)).toEqual({ migrated: 42 });
   });
 
+  it("packages online harness preparation as a standalone plain-Node entry", async () => {
+    const nativeHarnessPrestageEntry = path.join(root, "server", "dist", "prestage.js");
+    write(nativeHarnessPrestageEntry, 'import { values } from "@paperclipai/fixture"; console.log(JSON.stringify({ copied: values.doubled, enrollmentReady: false }));');
+    const outdir = path.join(root, "server", "dist-runtime");
+    await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir, nativeHarnessPrestageEntry });
+    const { stdout } = await execFileAsync(process.execPath, [path.join(outdir, "native-harness-prestage.mjs")], { timeout: 30_000 });
+    expect(JSON.parse(stdout)).toEqual({ copied: 42, enrollmentReady: false });
+  });
+
   it("bundles workspace TypeScript and runs with plain node", async () => {
     const outdir = path.join(root, "server", "dist-runtime");
     const summary = await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir });

server/src/maintenance/native-harness-prestage.ts

@@ -0,0 +1,36 @@
+import { closeRegisteredClients, createDb } from "@paperclipai/db";
+import { cloudRuntimeConfigService } from "../services/cloud-runtime-config.js";
+import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js";
+import { isCloudManagedInstance } from "../services/cloud-instance.js";
+import { getStorageService } from "../storage/index.js";
+import { prestageNativeHarnessBackups } from "../services/native-runtime/native-harness-backup-migration.js";
+
+// Operator-only finite preparation. No task drain, scheduler, app routes,
+// enrollment export, or assertion that other executions have stopped.
+if (process.env.PAPERCLIP_NATIVE_HARNESS_PRESTAGE !== "true") throw new Error("Explicit harness pre-copy mode is required");
+const url = process.env.DATABASE_URL;
+if (!url) throw new Error("Harness pre-copy requires an explicit database");
+const cancellation = new AbortController();
+const stop = () => cancellation.abort(new Error("native_harness_prestage_stopped"));
+const deadline = setTimeout(() => cancellation.abort(new Error("native_harness_prestage_deadline")), 90 * 60_000);
+process.once("SIGINT", stop); process.once("SIGTERM", stop);
+const db = createDb(url, { maxConnections: 2 });
+try {
+  if (isCloudManagedInstance()) {
+    const config = await cloudRuntimeConfigService(db).inspectStored();
+    if (!config) throw new Error("cloud_runtime_configuration_missing");
+    setCloudRuntimeConfigSnapshot(config);
+  }
+  const storage = getStorageService();
+  if (storage.provider !== "s3") throw new Error("shared_object_storage_required");
+  const counts = await prestageNativeHarnessBackups(db, { storage, signal: cancellation.signal });
+  console.log(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", counts, enrollmentReady: false }));
+} catch (error) {
+  const message = error instanceof Error ? error.message : "native_harness_prestage_failed";
+  const failure = /^[a-z_]{1,80}$/.test(message) ? message : "native_harness_prestage_failed";
+  console.error(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", state: "blocked", failure, enrollmentReady: false }));
+  process.exitCode = 1;
+} finally {
+  clearTimeout(deadline); process.off("SIGINT", stop); process.off("SIGTERM", stop);
+  await closeRegisteredClients(url, { scope: "database" });
+}

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

@@ -1,11 +1,11 @@
 import { createHash } from "node:crypto";
 import { lstat, readFile, readdir } from "node:fs/promises";
 import { join } from "node:path";
-import { and, eq, sql } from "drizzle-orm";
-import { environmentLeases, heartbeatRuns, nativeHarnessBackups, type Db } from "@paperclipai/db";
+import { and, eq, inArray, or, sql } from "drizzle-orm";
+import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db";
 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 { nativeHarnessBackupOwner, 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";
 
@@ -61,16 +61,34 @@ async function prepareSource(base: string, scopeDigest: string) {
 
 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;
+async function sourceRun(db: Db, source: Source) {
   const [run] = await db.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId,
-    sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns)
+    sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId,
+    status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt }).from(heartbeatRuns)
     .where(eq(heartbeatRuns.id, source.runId));
-  if (!run) return 0;
-  if (!run.issueId || run.sessionId !== source.sessionId || run.runnerId !== source.runnerId) {
+  if (!run) return null;
+  if (!run.issueId || !run.sessionId || !run.runnerId || run.sessionId !== source.sessionId || run.runnerId !== source.runnerId) {
     throw new Error("native_harness_migration_binding_mismatch");
   }
+  return { ...run, issueId: run.issueId, sessionId: run.sessionId, runnerId: run.runnerId };
+}
+
+async function verifiedSourceLease(db: Db, source: Source, run: NonNullable<Awaited<ReturnType<typeof sourceRun>>>, signal: AbortSignal) {
+  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' = ${source.manifestSha256}`));
+  for (const lease of leases) {
+    const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? "");
+    if (stamp?.sessionScopeSha256 === source.scopeDigest && await verifyNativeHarnessBackupStampAsync(stamp, lease.providerLeaseId!, signal)) return lease.providerLeaseId!;
+  }
+  return null;
+}
+
+async function migrateSource(db: Db, input: { storage: StorageService; signal: AbortSignal }, source: Source) {
+  input.signal.throwIfAborted();
+  const { root, scopeDigest, manifestSha256 } = source, run = await sourceRun(db, source);
+  if (!run) return 0;
   const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId),
     eq(nativeHarnessBackups.scopeDigest, scopeDigest), eq(nativeHarnessBackups.status, "current")));
   if (current) {
@@ -79,57 +97,120 @@ async function migrateSource(db: Db, input: { storage: StorageService; signal: A
       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");
+  if (!await verifiedSourceLease(db, source, run, input.signal)) 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;
 }
 
+async function mapSources<T>(input: { signal: AbortSignal; concurrency?: number;
+  unavailableSource?: (error: unknown) => T | undefined }, visit: (source: Source, signal: AbortSignal) => Promise<T>) {
+  const base = nativeHarnessStateBase();
+  const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => {
+    if (error.code === "ENOENT") return [] as string[]; throw error;
+  });
+  let reserved = 0, wave: Source[] = [];
+  const results: T[] = [];
+  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 visit(source, signal); }
+      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;
+      results.push(result.value);
+    }
+    wave = []; reserved = 0;
+  };
+  for (const scopeDigest of directories.filter(name => /^[a-f0-9]{64}$/.test(name)).sort()) {
+    input.signal.throwIfAborted();
+    let source;
+    try { source = await prepareSource(base, scopeDigest); }
+    catch (error) {
+      input.signal.throwIfAborted();
+      const deferred = input.unavailableSource?.(error);
+      if (deferred === undefined) throw error;
+      results.push(deferred); continue;
+    }
+    if (!source) continue;
+    const reservation = 2 * source.archiveLimit;
+    if (wave.length && (wave.length === (input.concurrency ?? MAX_PARALLEL_BACKUPS) || reserved + reservation > MAX_SCRATCH_BYTES)) await flush();
+    wave.push(source); reserved += reservation;
+  }
+  if (wave.length) await flush();
+  return results;
+}
+
 export async function migrateNativeHarnessBackups(db: Db, input: {
   storage: StorageService; writersStopped: true; signal: AbortSignal;
 }): Promise<number> {
   if (!input.writersStopped) throw new Error("native_harness_migration_requires_offline_writers");
   const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1);
   if (active) throw new Error("native_harness_migration_requires_offline_writers");
-  const base = nativeHarnessStateBase();
-  const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => {
-    if (error.code === "ENOENT") return [] as string[]; throw error;
+  const counts = await mapSources(input, (source, signal) => migrateSource(db, { storage: input.storage, signal }, source));
+  return counts.reduce<number>((total, count) => total + count, 0);
+}
+
+/** Copy settled successful history using its released runtime owner. This is
+ * preparatory work, never an offline-writer assertion or enrollment receipt. */
+export async function prestageNativeHarnessBackups(db: Db, input: { storage: StorageService; signal: AbortSignal }) {
+  // Keep preparatory work below offline throughput so the serving process can
+  // continue handling real work. This finite operation has no idle schedule.
+  const outcomes = await mapSources({ ...input, concurrency: 1, unavailableSource: error =>
+    (error as NodeJS.ErrnoException)?.code === "ENOENT" || error instanceof Error && error.message === "native_harness_migration_source_unsettled"
+      ? "source_unavailable" : undefined }, async (source, signal) => {
+    signal.throwIfAborted();
+    const run = await sourceRun(db, source).catch(error => {
+      if (error instanceof Error && error.message === "native_harness_migration_binding_mismatch") return null;
+      throw error;
+    });
+    if (!run) return "missing_run";
+    const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId),
+      eq(nativeHarnessBackups.scopeDigest, source.scopeDigest), eq(nativeHarnessBackups.status, "current")));
+    if (current) {
+      if (current.manifestSha256 !== source.manifestSha256) return "durable_source_changed";
+      await verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: run.companyId,
+        scopeDigest: source.scopeDigest, manifestSha256: source.manifestSha256, signal });
+      return "current";
+    }
+    if (run.status !== "succeeded" || !run.finishedAt) return "source_not_successful";
+    const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
+      eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.issueId),
+      eq(nativeRunFinalizations.runId, source.runId)));
+    if (coordinator?.phase !== "committed" || coordinator.leaseOwner || !coordinator.resultId || !coordinator.controllerBootId) return "owner_not_released";
+    const [task] = await db.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues)
+      .where(and(eq(issues.companyId, run.companyId), eq(issues.id, run.issueId)));
+    const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+      eq(heartbeatRuns.companyId, run.companyId), inArray(heartbeatRuns.status, ["queued", "running"]),
+      or(eq(heartbeatRuns.nativeIssueId, run.issueId), eq(heartbeatRuns.issueId, run.issueId), eq(heartbeatRuns.nativeSessionId, run.sessionId)))).limit(1);
+    if (!task || task.executionRunId || task.checkoutRunId || active) return "source_active";
+    const providerLeaseId = await verifiedSourceLease(db, source, run, signal);
+    if (!providerLeaseId) return "source_unverified";
+    try {
+      const owner = await nativeHarnessBackupOwner(db, { companyId: run.companyId, issueId: run.issueId,
+        runId: source.runId, bootId: coordinator.controllerBootId });
+      await persistNativeHarnessBackup({ db, storage: input.storage, root: source.root, scopeDigest: source.scopeDigest,
+        owner, signal, maxArchiveBytes: source.archiveLimit,
+        inactiveSource: { normalizedSessionId: run.sessionId, runnerInstanceId: run.runnerId, providerLeaseId } });
+      return "published";
+    } catch (error) {
+      signal.throwIfAborted();
+      if ((error as NodeJS.ErrnoException)?.code === "ENOENT" || error instanceof Error && ["native_runner_checkpoint_owner_lost", "native_harness_storage_owner_missing",
+        "native_harness_storage_prestage_source_active", "native_harness_storage_prestage_source_changed",
+        "native_harness_storage_owner_superseded", "native_harness_archive_invalid"].includes(error.message)) return "source_changed";
+      throw error;
+    }
   });
-  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 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;
+  const counts = { inspected: outcomes.length, published: 0, current: 0, deferred: {} as Record<string, number> };
+  for (const outcome of outcomes) {
+    if (outcome === "published" || outcome === "current") counts[outcome]++;
+    else counts.deferred[outcome] = (counts.deferred[outcome] ?? 0) + 1;
   }
-  if (wave.length) await flush();
-  return count;
+  return counts;
 }

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

@@ -12,7 +12,7 @@ import { nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNat
 import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
 import { buildNativeHarnessBackupManifest } from "./native-session-executor.js";
 import { createNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js";
-import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups } from "./native-harness-backup-migration.js";
+import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups, prestageNativeHarnessBackups } from "./native-harness-backup-migration.js";
 import { nativeHarnessBackupExpirationService } from "./native-harness-backup-expiration.js";
 import { legacyControllerBootId } from "../legacy-controller-lease.js";
 
@@ -67,7 +67,7 @@ describeEmbeddedPostgres("durable native provider backups", () => {
     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 };
+    return { ...f, home, root, runnerInstanceId, providerLeaseId };
   }
   async function offlineFixtures(count: number) {
     const base = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(base); await mkdir(base);
@@ -179,6 +179,73 @@ describeEmbeddedPostgres("durable native provider backups", () => {
       leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId));
     return result;
   }
+  async function inactiveFixture() {
+    const [f] = await offlineFixtures(1);
+    await commitWarmRun(f);
+    await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId));
+    const owner = await nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId,
+      runId: f.runId, bootId: environment.current.successor.bootId });
+    return { ...f, owner, inactiveSource: { normalizedSessionId: f.identity.normalizedSessionId,
+      runnerInstanceId: f.runnerInstanceId, providerLeaseId: f.providerLeaseId } };
+  }
+  it("pre-copies released history and restores its bytes while an unrelated execution stays active", async () => {
+    const f = await inactiveFixture(), active = await environment.seedRun(randomUUID());
+    const input = { storage: f.storage, signal: AbortSignal.timeout(15_000) };
+    expect(await prestageNativeHarnessBackups(f.db, input)).toEqual({ inspected: 1, published: 1, current: 0, deferred: {} });
+    expect(await prestageNativeHarnessBackups(f.db, input)).toEqual({ inspected: 1, published: 0, current: 1, deferred: {} });
+    await expect(migrateNativeHarnessBackups(f.db, { ...input, writersStopped: true })).rejects.toThrow("requires_offline_writers");
+    expect((await active.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, active.runId)))[0].status).toBe("running");
+    await removeNativeHarnessBackup(f.home);
+    const replacement = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(replacement);
+    expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement })).toBe(true);
+    expect(JSON.parse(await readFile(join(replacement, "current", "codex-home", "sessions", "history.json"), "utf8"))).toEqual(f.identity);
+  });
+  it.each(["queued", "running"] as const)("defers pre-copy when the same task has a %s execution", async status => {
+    const f = await inactiveFixture(), put = vi.spyOn(f.storage, "putFile");
+    await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId, agentId: environment.current.agentId,
+      issueId: f.issueId, nativeIssueId: f.issueId, nativeSessionId: f.identity.normalizedSessionId, status, runtimeMode: "native" });
+    expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
+      .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_active: 1 } });
+    await expect(persistNativeHarnessBackup(f)).rejects.toThrow("native_harness_storage_prestage_source_active");
+    expect(put).not.toHaveBeenCalled();
+    expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]);
+  });
+  it("defers failed history without treating pre-copy as offline authority", async () => {
+    const [f] = await offlineFixtures(1);
+    expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
+      .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_not_successful: 1 } });
+    expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]);
+  });
+  it("defers a source being replaced online while offline migration still refuses it", async () => {
+    const f = await inactiveFixture();
+    await writeFile(join(f.root, "runner", "runner-state.json"), JSON.stringify({ runId: f.runId, lifecycle: "running" }));
+    expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
+      .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_unavailable: 1 } });
+    await expect(migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) }))
+      .rejects.toThrow("native_harness_migration_source_unsettled");
+  });
+  it.each(["source_advanced", "queued_turn", "lease_revoked", "owner_changed"])("keeps the published source when %s happens during pre-copy upload", async change => {
+    const f = await inactiveFixture(); await persistNativeHarnessBackup(f);
+    const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
+    const put = f.storage.putFile.bind(f.storage);
+    vi.spyOn(f.storage, "putFile").mockImplementation(async input => {
+      const result = await put(input);
+      if (change === "source_advanced") await writeFile(join(f.root, "manifest.json"), (await readFile(join(f.root, "manifest.json"), "utf8")) + "\n");
+      if (change === "queued_turn") await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId,
+        agentId: environment.current.agentId, issueId: f.issueId, status: "queued" });
+      if (change === "lease_revoked") await f.db.update(environmentLeases).set({ metadata: {} })
+        .where(eq(environmentLeases.providerLeaseId, f.providerLeaseId));
+      if (change === "owner_changed") await f.db.update(nativeRunFinalizations).set({ controllerGeneration: 2 })
+        .where(eq(nativeRunFinalizations.runId, f.runId));
+      return result;
+    });
+    await expect(persistNativeHarnessBackup(f)).rejects.toThrow(change === "owner_changed" ? "native_runner_checkpoint_owner_lost"
+      : change === "queued_turn" ? "native_harness_storage_prestage_source_active" : "native_harness_storage_prestage_source_changed");
+    const rows = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
+    expect(rows.filter(row => row.status === "current").map(row => row.id)).toEqual([before.id]);
+    expect(rows.filter(row => row.status === "pending")).toHaveLength(1);
+    expect(await readFile(join(f.root, "manifest.json"), "utf8")).toContain("paperclip.native-harness-backup.v1");
+  });
   it.each(["idle", "preparing", "native"])("preserves a released committed warm session with a %s successor", async state => {
     const f = await fixture();
     const result = await commitWarmRun(f);

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

@@ -6,7 +6,7 @@ import { join } from "node:path";
 import os from "node:os";
 import { pipeline } from "node:stream/promises";
 import { Transform } from "node:stream";
-import { and, eq, inArray, sql } from "drizzle-orm";
+import { and, eq, inArray, or, sql } from "drizzle-orm";
 import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js";
 import type { StorageService } from "../../storage/types.js";
@@ -28,6 +28,8 @@ type Reference = Binding & {
 type Row = typeof nativeHarnessBackups.$inferSelect;
 type OfflineOwner = { kind: "offline_migration"; companyId: string; issueId: string; runId: string;
   writersStopped: true; terminalFailureOnly?: true };
+type InactiveSource = { root: string; manifestSha256: string; scopeDigest: string;
+  normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string };
 const fail = (reason: string): never => { throw new Error(`native_harness_storage_${reason}`); };
 const aad = (binding: Binding) => Buffer.from(JSON.stringify([SCHEMA, binding.companyId, binding.scopeDigest, binding.id]));
 const scope = (binding: Pick<Binding, "companyId" | "scopeDigest">) => and(
@@ -89,9 +91,11 @@ export async function nativeHarnessBackupOwner(db: Db, input: { companyId: strin
   if (!row || !row.leaseOwner && (row.phase !== "committed" || !row.resultId)) fail("owner_missing");
   return { ...input, authorityRunId: input.runId, leaseOwner: row!.leaseOwner, generation: row!.controllerGeneration } satisfies NativeRunnerCheckpointFenceOwner;
 }
-async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner) {
+async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner,
+  inactiveSource?: InactiveSource) {
+  if (inactiveSource && "kind" in owner) fail("prestage_requires_runtime_owner");
   const checkpointOwner = "kind" in owner ? null : await lockNativeRunnerCheckpointOwner(tx, owner);
-  const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues)
+  const [task] = await tx.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues)
     .where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("share");
   const authorityRunId = "kind" in owner ? owner.runId : owner.authorityRunId;
   if (!task) fail("owner_superseded");
@@ -138,15 +142,44 @@ async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner:
   if (owner.leaseOwner === null && !committed) fail("owner_missing");
   if (task.executionRunId && task.executionRunId !== authorityRunId && !committed) fail("owner_superseded");
   await validateNativeRunnerCheckpointOwner(tx, owner, owned, false);
+  if (inactiveSource) {
+    if (!committed || task.executionRunId || task.checkoutRunId) fail("prestage_source_active");
+    const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+      eq(heartbeatRuns.companyId, owner.companyId), inArray(heartbeatRuns.status, ["queued", "running"]),
+      or(eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.issueId, owner.issueId),
+        eq(heartbeatRuns.nativeSessionId, inactiveSource.normalizedSessionId)))).limit(1);
+    if (active) fail("prestage_source_active");
+    const [binding] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+      eq(heartbeatRuns.id, owner.authorityRunId), eq(heartbeatRuns.companyId, owner.companyId),
+      eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, inactiveSource.normalizedSessionId),
+      eq(heartbeatRuns.runnerInstanceId, inactiveSource.runnerInstanceId)));
+    const [lease] = await tx.select({ stamp: sql<unknown>`${environmentLeases.metadata}->'nativeHarnessBackup'` })
+      .from(environmentLeases).where(and(eq(environmentLeases.companyId, owner.companyId),
+        eq(environmentLeases.issueId, owner.issueId), eq(environmentLeases.providerLeaseId, inactiveSource.providerLeaseId),
+        sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${inactiveSource.manifestSha256}`)).limit(1);
+    const stamp = readNativeHarnessBackupStamp(lease?.stamp, inactiveSource.providerLeaseId);
+    if (!binding || stamp?.sessionScopeSha256 !== inactiveSource.scopeDigest
+      || stamp.normalizedSessionId !== inactiveSource.normalizedSessionId || stamp.runnerInstanceId !== inactiveSource.runnerInstanceId) {
+      fail("prestage_source_changed");
+    }
+    // The task share lock serializes publication with execution checkout. A
+    // turn that ran during upload also changes the immutable backup manifest.
+    const manifest = await readFile(join(inactiveSource.root, "manifest.json"));
+    const state = JSON.parse(await readFile(join(inactiveSource.root, "runner", "runner-state.json"), "utf8"));
+    if (`sha256:${createHash("sha256").update(manifest).digest("hex")}` !== inactiveSource.manifestSha256
+      || state.runId !== owner.authorityRunId || state.lifecycle !== "suspended") fail("prestage_source_changed");
+  }
 }
 
 export async function persistNativeHarnessBackup(input: {
   db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner;
   scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number;
+  inactiveSource?: { normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string };
 }): Promise<void> {
   if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid");
   const manifestBytes = await readFile(join(input.root, "manifest.json"));
   const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`;
+  const inactiveSource = input.inactiveSource ? { ...input.inactiveSource, root: input.root, manifestSha256, scopeDigest: input.scopeDigest } : undefined;
   const binding = { companyId: input.owner.companyId, scopeDigest: input.scopeDigest, id: randomUUID() };
   const scratch = await mkdtemp(join(os.tmpdir(), "pc-harness-upload-"));
   const key = randomBytes(32);
@@ -163,7 +196,7 @@ export async function persistNativeHarnessBackup(input: {
         key: key.toString("base64"), iv: iv.toString("base64"), tag: cipher.getAuthTag().toString("base64"), plaintext })) };
     if (ref.byteSize > MAX_NATIVE_HARNESS_OBJECT_BYTES) fail("size_limit_exceeded");
     await input.db.transaction(async tx => {
-      await fence(tx, input.owner);
+      await fence(tx, input.owner, inactiveSource);
       await tx.insert(nativeHarnessBackups).values({ ...binding, issueId: input.owner.issueId, runId: input.owner.runId,
         manifestSha256, reference: ref });
     });
@@ -172,7 +205,7 @@ export async function persistNativeHarnessBackup(input: {
       byteSize: ref.byteSize, sha256: ref.sha256, signal: input.signal });
     await receive(input.storage, ref, undefined, input.signal);
     await input.db.transaction(async tx => {
-      await fence(tx, input.owner);
+      await fence(tx, input.owner, inactiveSource);
       await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`${ref.companyId}:${ref.scopeDigest}`}, 0))`);
       const [pending] = await tx.select({ status: nativeHarnessBackups.status }).from(nativeHarnessBackups)
         .where(eq(nativeHarnessBackups.id, ref.id)).for("update");
@@ -182,7 +215,7 @@ export async function persistNativeHarnessBackup(input: {
           eq(nativeHarnessBackups.status, "current"), sql`${nativeHarnessBackups.runId} <> ${input.owner.runId}::uuid`,
           sql`${heartbeatRuns.createdAt} >= (select created_at from heartbeat_runs where id = ${input.owner.runId}::uuid)`)).limit(1);
       if (successor) fail("owner_superseded");
-      await fence(tx, input.owner);
+      await fence(tx, input.owner, inactiveSource);
       await tx.update(nativeHarnessBackups).set({ status: "superseded", updatedAt: new Date() })
         .where(and(scope(ref), eq(nativeHarnessBackups.status, "current")));
       await tx.update(nativeHarnessBackups).set({ status: "current", updatedAt: new Date() })