Migrate settled controller checkpoints while unrelated work runs (#19631)

## Thinking Path

> - Paperclip manages work performed by AI agents.
> - Server replacement must preserve retained Runner sessions.
> - Older sessions can retain controller checkpoints on the server
volume.
> - The existing importer requires all execution writers to stop,
including unrelated runs.
> - This change adds a separate online path for successful, settled
sources.
> - It locks the source task and finalizer while it reads and publishes
that source.
> - Unrelated work continues. The offline importer keeps its existing
authority.

## Linked Issues or Issue Description

**What happened?**

A retained controller checkpoint cannot be converted by the existing
maintenance importer while another task runs. Requiring a busy server to
become globally idle prevents online migration.

**Expected behavior**

An operator can convert a successful retained source after its finalizer
releases ownership, while unrelated tasks continue. An expired owner
remains an owner. Related active work and uncertain evidence remain
blocked.

**Steps to reproduce**

1. Retain a valid legacy controller for a successful run with an
accepted result and a released committed finalizer.
2. Start a different task.
3. Run controller preparation. The old importer refuses because
execution writers are active.

Related preservation contracts: #19038 and #19024. This change does not
replace workspace migration or authorize volume removal.

## What Changed

- Add `prestageNativeControllers` with task, finalizer, and source-run
row locks. Recheck related runs and exact checkpoint identity under
those locks.
- Allow legacy checkpoint material through normal settled-owner import
checks. Keep offline authority separate.
- Compare captured local controller and supervisor bytes before commit.
A changed source rolls back publication.
- Add `--controllers-only` to the existing finite native preparation
entry. This avoids re-reading provider archives during controller
conversion.
- Add PostgreSQL cases for unrelated work, expired owners, related
active work, source-lock contention, and source changes.
- Document the finite command and its limits.

## Verification

- Real PostgreSQL affected suites: 56 passed across controller migration
and recovery. A first per-server run was interrupted because it was
slow; the completed run used the repository's shared PostgreSQL test
wrapper.
- Initial online cases: 3 passed, covering unrelated work, an expired
owner, and related active work.
- Source lock and changed-file cases: 2 passed. Publication rolled back
on a source change.
- Final source-evidence regression: 2 passed. A first unproven source
deferred and a later valid source imported; changed-file rollback
remained intact.
- Server TypeScript check passed with the SDK and Runner declarations
built.
- Server package build passed, including the Runner vendor checks and
compiled command.
- Runtime bundling suite: 10 passed.
- Full repository CI and exact-head automated review remain required
before merge. This draft is not staging acceptance.


- PASS: 7 CLI cases after adding the controller-only dependency fixture
and explicit-mode coverage. The first CI run exposed missing mock/export
coverage; the fixture repair passed after rebase.
- The unrelated model-default failures were fixed on master; this branch
now includes that fix.

## Risks

- A source task can wait briefly on its own migration lock. A one-second
lock acquisition budget defers a contended source. No global admission
hold is acquired.
- Only successful sources with a released committed finalizer are
eligible. Failed or unsettled sources stay deferred and require their
preservation owners.
- The operator must run the command on the host that owns the original
files. The command does not certify enrollment readiness or detach any
volume.
- The existing offline path remains unchanged and still requires stopped
writers.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning, tool use, and code
execution. The exact serving model ID and context window are not exposed
in this session.

## 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
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

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

@@ -289,6 +289,24 @@ 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.
 
+Run the controller-only phase separately so it does not re-read provider
+archives that were already verified:
+
+```sh
+PAPERCLIP_NATIVE_HARNESS_PRESTAGE=true node --max-old-space-size=512 server/dist-runtime/native-harness-prestage.mjs --controllers-only
+```
+
+This finite entry adopts legacy controller checkpoints for successful
+retained sources whose finalizer is committed, has an accepted result, and has
+released its ownership. It locks only the source task and coordinator while
+reading and publishing the original controller state. Related queued/running
+work, checkout, failed sources, recovery state, or any remaining coordinator
+owner (including an expired owner) defers that source. Local controller and warm
+supervisor bytes are rechecked before commit; a changed source rolls back.
+Unrelated runs remain active. Existing controller checkpoint identity, artifact
+provenance, configuration and snapshot checks still apply. The `paperclip.native-controller-prestage.v1` counts describe this phase; neither successful prestaging
+nor zero deferred controller sources authorizes volume detachment.
+
 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

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

@@ -1,9 +1,9 @@
-import { afterEach, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, expect, it, vi } from "vitest";
 import { spawnSync } from "node:child_process";
 import { fileURLToPath } from "node:url";
 import { build } from "esbuild";
 
-const mocks = vi.hoisted(() => ({ close: vi.fn(async () => {}), prestage: vi.fn(), key: vi.fn(async () => {}), obligations: vi.fn(async () => false) }));
+const mocks = vi.hoisted(() => ({ close: vi.fn(async () => {}), prestage: vi.fn(), controllers: vi.fn(), key: vi.fn(async () => {}), obligations: vi.fn(async () => false) }));
 vi.mock("@paperclipai/db", () => ({ createDb: () => ({}), closeRegisteredClients: mocks.close }));
 vi.mock("../config.js", () => ({ loadConfig: () => ({ secretsMasterKeyFilePath: "/fixture/key" }) }));
 vi.mock("../services/cloud-runtime-config.js", () => ({ cloudRuntimeConfigService: () => ({}) }));
@@ -13,12 +13,15 @@ vi.mock("../storage/index.js", () => ({ getStorageService: () => ({ provider: "s
 vi.mock("../services/native-runtime/native-harness-backup-migration.js", () => ({
   prestageNativeHarnessBackups: mocks.prestage, hasLocalNativeHarnessObligations: mocks.obligations,
 }));
+vi.mock("../services/native-runtime/native-controller-migration.js", () => ({ prestageNativeControllers: mocks.controllers }));
 vi.mock("./native-harness-prestage-runtime.js", async importOriginal => ({
   ...await importOriginal<typeof import("./native-harness-prestage-runtime.js")>(), initializeNativeHarnessPrestageKey: mocks.key,
 }));
 
 const initialExitCode = process.exitCode;
-afterEach(() => { process.exitCode = initialExitCode; vi.useRealTimers(); vi.unstubAllEnvs(); vi.restoreAllMocks(); vi.resetModules(); mocks.close.mockClear(); mocks.obligations.mockReset().mockResolvedValue(false); });
+const initialArgv = process.argv;
+beforeEach(() => { process.argv = [process.execPath, "native-harness-prestage"]; });
+afterEach(() => { process.argv = initialArgv; mocks.controllers.mockReset(); process.exitCode = initialExitCode; vi.useRealTimers(); vi.unstubAllEnvs(); vi.restoreAllMocks(); vi.resetModules(); mocks.close.mockClear(); mocks.obligations.mockReset().mockResolvedValue(false); });
 
 it.each([true, false])("checks unfulfilled database obligations for an empty source scan (pending: %s)", async pending => {
   vi.stubEnv("PAPERCLIP_NATIVE_HARNESS_PRESTAGE", "true");
@@ -69,6 +72,7 @@ it("reports a failed deadline with real timers when the dependency holds no even
         export const isCloudManagedInstance = () => false, getStorageService = () => ({ provider: 's3' });
         export const initializeNativeHarnessPrestageKey = async () => {}, nativeHarnessPrestageDatabaseUrl = url => url;
         export const prestageNativeHarnessBackups = () => new Promise(() => {});
+        export const prestageNativeControllers = () => new Promise(() => {});
         export const hasLocalNativeHarnessObligations = async () => false;
       ` }));
     } }],
@@ -111,3 +115,20 @@ it.each(["SIGTERM", "deadline"])("bounds shutdown if a dependency does not settl
   expect(process.exitCode).toBe(1);
   expect(vi.getTimerCount()).toBe(0);
 });
+
+it("copies only settled controllers when the explicit mode is selected", async () => {
+  vi.stubEnv("PAPERCLIP_NATIVE_HARNESS_PRESTAGE", "true");
+  vi.stubEnv("DATABASE_URL", "postgres://fixture:fixture@localhost/fixture");
+  process.argv.push("--controllers-only");
+  const counts = { inspected: 2, published: 1, current: 0, deferred: { source_active: 1 } };
+  mocks.controllers.mockResolvedValue(counts);
+  mocks.prestage.mockClear();
+  const success = vi.spyOn(console, "log").mockImplementation(() => {});
+  await import("./native-harness-prestage.js");
+  expect(mocks.controllers).toHaveBeenCalledOnce();
+  expect(mocks.controllers.mock.calls[0][1].signal.aborted).toBe(false);
+  expect(mocks.prestage).not.toHaveBeenCalled();
+  expect(mocks.obligations).not.toHaveBeenCalled();
+  expect(success).toHaveBeenCalledWith(JSON.stringify({ schema: "paperclip.native-controller-prestage.v1", counts, enrollmentReady: false }));
+  expect(mocks.close).toHaveBeenCalled();
+});

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

@@ -5,9 +5,14 @@ import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-
 import { isCloudManagedInstance } from "../services/cloud-instance.js";
 import { getStorageService } from "../storage/index.js";
 import { hasLocalNativeHarnessObligations, prestageNativeHarnessBackups } from "../services/native-runtime/native-harness-backup-migration.js";
+import { prestageNativeControllers } from "../services/native-runtime/native-controller-migration.js";
 import { initializeNativeHarnessPrestageKey, nativeHarnessPrestageDatabaseUrl } from "./native-harness-prestage-runtime.js";
 
 if (process.env.PAPERCLIP_NATIVE_HARNESS_PRESTAGE !== "true") throw new Error("Explicit harness pre-copy mode is required");
+const args = process.argv.slice(2);
+if (args.length > 1 || args.some(arg => arg !== "--controllers-only")) throw new Error("Unknown native pre-copy mode");
+const controllersOnly = args[0] === "--controllers-only";
+const schema = controllersOnly ? "paperclip.native-controller-prestage.v1" : "paperclip.native-harness-prestage.v1";
 const sourceUrl = process.env.DATABASE_URL;
 if (!sourceUrl) throw new Error("Harness pre-copy requires an explicit database");
 const url = nativeHarnessPrestageDatabaseUrl(sourceUrl);
@@ -20,7 +25,7 @@ const reportBlocked = (error: unknown) => {
   blocked = true;
   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 }));
+  console.error(JSON.stringify({ schema, state: "blocked", failure, enrollmentReady: false }));
 };
 const abort = (reason: string) => {
   if (cancellation.signal.aborted) return;
@@ -42,14 +47,20 @@ try {
   await initializeNativeHarnessPrestageKey(loadConfig().secretsMasterKeyFilePath);
   const storage = getStorageService();
   if (storage.provider !== "s3") throw new Error("shared_object_storage_required");
-  const counts = await prestageNativeHarnessBackups(db, { storage, signal: cancellation.signal });
-  cancellation.signal.throwIfAborted();
-  if (counts.inspected === 0) {
-    const obligations = await hasLocalNativeHarnessObligations(db);
+  if (controllersOnly) {
+    const counts = await prestageNativeControllers(db, { signal: cancellation.signal });
     cancellation.signal.throwIfAborted();
-    if (obligations) throw new Error("native_harness_prestage_source_missing");
+    console.log(JSON.stringify({ schema, counts, enrollmentReady: false }));
+  } else {
+    const counts = await prestageNativeHarnessBackups(db, { storage, signal: cancellation.signal });
+    cancellation.signal.throwIfAborted();
+    if (counts.inspected === 0) {
+      const obligations = await hasLocalNativeHarnessObligations(db);
+      cancellation.signal.throwIfAborted();
+      if (obligations) throw new Error("native_harness_prestage_source_missing");
+    }
+    console.log(JSON.stringify({ schema, counts, enrollmentReady: false }));
   }
-  console.log(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", counts, enrollmentReady: false }));
 } catch (error) {
   reportBlocked(error);
 } finally {

server/src/services/native-runtime/native-controller-migration.integration.test.ts

@@ -1,14 +1,14 @@
 import { createHash, randomUUID } from "node:crypto";
-import { mkdir, readFile, rm, writeFile } from "node:fs/promises";
+import { mkdir, open, readFile, rm, stat, writeFile } from "node:fs/promises";
 import { dirname, join } from "node:path";
 import { eq } from "drizzle-orm";
 import { afterEach, expect, it, vi } from "vitest";
-import { agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db";
+import { agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db";
 import { captureLegacyDurablePrpCheckpoint, DurablePrpControlPlane, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
 import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
 import { nativeToolContractFingerprintForTarget } from "./native-session-resume.js";
 import { nativeControllerMigrationDescriptor } from "./native-session-executor.js";
-import { hasLocalNativeControllerObligations, migrateNativeControllers } from "./native-controller-migration.js";
+import { hasLocalNativeControllerObligations, migrateNativeControllers, prestageNativeControllers } from "./native-controller-migration.js";
 import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
 import { createNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js";
 import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
@@ -22,7 +22,7 @@ import { createStorageService } from "../../storage/service.js";
 import { legacyNativeControllerArchive } from "./native-controller-archive.js";
 import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
 
-describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable_failed",
+describeEmbeddedPostgres.each(["online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed",
   "stale_confirmed", "stale_foreign", "stale_changed", "stale_queued", "stale_overlap", "stale_success", "stale_busy", "stale_durable", "stale_durable_absent",
   "stale_durable_runner", "stale_durable_session", "stale_durable_snapshot",
   "controller_confirmed", "controller_foreign", "controller_thread", "controller_runner", "controller_success", "controller_busy", "controller_invalid", "controller_durable",
@@ -30,7 +30,7 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
   "controller_observed_live", "controller_observed_result", "controller_observed_retry", "controller_observed_task", "controller_observed_old_receipt", "controller_observed_revoked",
   "controller_observed_handoff", "controller_observed_handoff_stopped", "controller_observed_handoff_broken", "controller_observed_handoff_fork"])("offline native controller migration (%s)", scenario => {
   const environment = nativeRestartFixture();
-  afterEach(() => vi.unstubAllEnvs());
+  afterEach(() => { vi.unstubAllEnvs(); vi.restoreAllMocks(); });
   async function fixture(commandCount = 1, warm = false) {
     vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "73".repeat(32));
     const f = await environment.seedRun(randomUUID());
@@ -90,6 +90,91 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
       taskKey: f.issueId, lastRunId: f.runId }).returning();
     return { ...f, companyId, agentId, paperclipHome, runtimeRoot, paths, file, source, profile, session, queuedId, identity, snapshot };
   }
+  if (scenario.startsWith("online_")) it("prestages only a settled legacy controller while unrelated work continues", async () => {
+    const f = await fixture();
+    const settle = async (f: Awaited<ReturnType<typeof fixture>>) => {
+      const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+      const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId,
+        completionContractId: run.completionContractId!, serverFingerprint: "online-migration", schemaStatus: "accepted",
+        resultJson: {}, canonicalSha256: "online-migration" }).returning();
+      await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
+      await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId));
+      await f.db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, f.queuedId));
+      await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null,
+        leaseExpiresAt: null, recoveryState: null }).where(eq(nativeRunFinalizations.runId, f.runId));
+    };
+    await settle(f);
+    const unrelated = await environment.seedRun(randomUUID());
+    const [unrelatedBefore] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId));
+    if (scenario === "online_owned") await f.db.update(nativeRunFinalizations).set({
+      leaseOwner: "expired-owner-is-still-an-owner", leaseExpiresAt: new Date(0),
+    }).where(eq(nativeRunFinalizations.runId, f.runId));
+    if (scenario === "online_same_source") await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId,
+      agentId: f.agentId, issueId: f.issueId, nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId,
+      status: "running", runtimeMode: "native" });
+    const before = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
+    if (scenario === "online_evidence") {
+      await rm(join(f.paths.root, "provider-pack-authority.json"));
+      await f.db.update(agentTaskSessions).set({ id: "00000000-0000-4000-8000-000000000001" }).where(eq(agentTaskSessions.id, f.session.id));
+      const next = await fixture();
+      await settle(next);
+      const actual = await prestageNativeControllers(f.db, { signal: AbortSignal.timeout(30_000) });
+      expect(actual).toMatchObject({ inspected: 2, imported: 1, deferred: { artifact_unproven: 1 } });
+      expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).toBeNull();
+      expect(await loadNativeRunnerCheckpoint(f.db, { companyId: next.companyId, issueId: next.issueId, runId: next.runId })).not.toBeNull();
+      expect(await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(before);
+      expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0]).toEqual(unrelatedBefore);
+      return;
+    }
+    if (scenario === "online_changed") {
+      const inode = (await stat(f.file)).ino;
+      const probe = await open(f.file, "r");
+      const prototype = Object.getPrototypeOf(probe);
+      const originalRead = prototype.readFile;
+      await probe.close();
+      let changed = false;
+      vi.spyOn(prototype, "readFile").mockImplementation(async function(this: { stat: () => Promise<{ ino: number }> }, ...args: unknown[]) {
+        const bytes = await originalRead.apply(this, args);
+        if (!changed && (await this.stat()).ino === inode) {
+          changed = true;
+          await writeFile(f.file, f.source + "\n", { mode: 0o600 });
+        }
+        return bytes;
+      });
+      expect(await prestageNativeControllers(f.db, { signal: AbortSignal.timeout(30_000) }))
+        .toMatchObject({ imported: 0, deferred: { source_changed: 1 } });
+      expect(changed).toBe(true);
+      expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).toBeNull();
+      expect(await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(before);
+      expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0]).toEqual(unrelatedBefore);
+      return;
+    }
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    let acquired!: () => void;
+    const locked = new Promise<void>(resolve => { acquired = resolve; });
+    const writer = scenario === "online_busy" ? f.db.transaction(async tx => {
+      await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
+      acquired();
+      await gate;
+    }) : null;
+    if (writer) await locked;
+    let actual: Awaited<ReturnType<typeof prestageNativeControllers>>;
+    try { actual = await prestageNativeControllers(f.db, { signal: AbortSignal.timeout(30_000) }); }
+    finally { release(); await writer; }
+    if (scenario === "online_busy") expect(actual.deferred.source_busy).toBe(1);
+    if (scenario === "online_success") {
+      expect(actual.imported).toBe(1);
+      expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).not.toBeNull();
+      expect((await prestageNativeControllers(f.db, { signal: AbortSignal.timeout(30_000) })).unchanged).toBeGreaterThanOrEqual(1);
+    } else {
+      expect(actual.imported).toBe(0);
+      expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).toBeNull();
+    }
+    expect(await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(before);
+    expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0]).toEqual(unrelatedBefore);
+    expect(await readFile(f.file, "utf8")).toBe(f.source);
+  });
   if (scenario === "import") it("imports an exact settled authority once, preserves queued work, and resumes from an empty home", async () => {
     const f = await fixture();
     const { companyId, agentId, paperclipHome, runtimeRoot, profile, file, source, session, queuedId, identity, snapshot } = f;

server/src/services/native-runtime/native-controller-migration.ts

@@ -1,8 +1,8 @@
 import { constants } from "node:fs";
 import { lstat, open, readdir } from "node:fs/promises";
 import { join } from "node:path";
-import { and, eq, sql } from "drizzle-orm";
-import { heartbeatRuns, type Db } from "@paperclipai/db";
+import { and, eq, inArray, or, sql } from "drizzle-orm";
+import { heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import { captureLegacyDurablePrpCheckpoint, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
 import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
 import { findNativeSessionResumeRun } from "./native-session-resume.js";
@@ -157,6 +157,54 @@ export async function hasLocalNativeControllerObligations(db: Db, storage?: Stor
   return false;
 }
 
+function hasMatchingControllerRecovery(selected: Selected,
+  source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>,
+  checkpoint: Awaited<ReturnType<typeof loadNativeRunnerCheckpoint>>) {
+  const metadata = readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint);
+  const snapshot = source.runnerProfileJson.sessionCheckpoint as { identity?: Record<string, unknown> } | null;
+  return Boolean(checkpoint && metadata?.checkpoint?.runId === source.id
+      && metadata.checkpoint.runnerInstanceId === source.runnerInstanceId
+      && metadata.checkpoint.normalizedSessionId === selected.sessionId
+      && snapshot?.identity?.runId === source.id && snapshot.identity.companyId === selected.companyId
+      && snapshot.identity.agentId === selected.agentId && snapshot.identity.issueId === selected.issueId
+      && snapshot.identity.sessionId === selected.sessionId
+      && checkpoint.recoveryEvidence.runnerInstanceId === source.runnerInstanceId
+      && checkpoint.recoveryEvidence.normalizedSessionId === selected.sessionId);
+}
+
+async function readMigrationMaterial(db: Db, selected: Selected,
+  source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>,
+  checkpoint: Awaited<ReturnType<typeof loadNativeRunnerCheckpoint>>) {
+  const paths = await localPaths(source);
+  const legacyControllerState = checkpoint ? undefined : await readPrivateSource(paths.controlFile, 256 * 1024 * 1024);
+  let legacyRunnerArtifact: { version: string; digest: string } | undefined;
+  if (!checkpoint && legacyControllerState === undefined) return { paths, legacyControllerState, legacyRunnerArtifact, supervisor: undefined };
+  if (legacyControllerState !== undefined && await readPrivateSource(join(paths.root, "provider-pack-authority.json"), 512 * 1024)) {
+    const identity = JSON.parse(legacyControllerState).identity;
+    const execution = paths.execution;
+    const [admission] = await db.select({ workspace: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'` })
+      .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, selected.companyId),
+        eq(heartbeatRuns.agentId, selected.agentId)));
+    const workspace = readNativeWorkspaceSyncReference(admission?.workspace);
+    const remoteCwd = workspace?.remoteCwd ?? execution.workspace.cwd;
+    if (typeof identity?.environmentLeaseId !== "string" || !remoteCwd
+      || (workspace && workspace.workspaceId !== execution.binding.executionWorkspaceId)) {
+      throw new Error("native_controller_migration_artifact_binding");
+    }
+    const unavailable = (): never => { throw new Error("native_controller_migration_artifact_unavailable"); };
+    legacyRunnerArtifact = resolveRemoteProviderPackAuthority({ stateDirectory: paths.root, reattach: true,
+      binding: { companyId: selected.companyId, agentId: selected.agentId,
+        workspaceId: providerPackWorkspaceId(execution.binding.executionWorkspaceId, source.id, identity.environmentLeaseId),
+        sessionId: selected.sessionId, runnerInstanceId: source.runnerInstanceId!,
+        environmentLeaseId: identity.environmentLeaseId, remoteCwd,
+        providerDigest: nativeSha256(execution.provider) },
+      current: unavailable, runnerArtifact: unavailable }).runnerArtifact;
+  }
+  const supervisor = !checkpoint && paths.execution.session.lifecyclePolicy.mode === "warm"
+    ? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined;
+  return { paths, legacyControllerState, legacyRunnerArtifact, supervisor };
+}
+
 export async function migrateNativeControllers(db: Db, input: { writersStopped: true; signal: AbortSignal; storage?: StorageService }) {
   const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1);
   if (!input.writersStopped || active) throw new Error("native_controller_migration_requires_offline_writers");
@@ -166,43 +214,9 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped:
     const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
     const checkpoint = await loadNativeRunnerCheckpoint(db, binding);
     if (preservedFailedCheckpoint(source, selected.sessionId, checkpoint)) { counts.preservedFailed++; continue; }
-    const metadata = readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint);
-    const snapshot = source.runnerProfileJson.sessionCheckpoint as { identity?: Record<string, unknown> } | null;
-    if (checkpoint && metadata?.checkpoint?.runId === source.id
-      && metadata.checkpoint.runnerInstanceId === source.runnerInstanceId
-      && metadata.checkpoint.normalizedSessionId === selected.sessionId
-      && snapshot?.identity?.runId === source.id && snapshot.identity.companyId === selected.companyId
-      && snapshot.identity.agentId === selected.agentId && snapshot.identity.issueId === selected.issueId
-      && snapshot.identity.sessionId === selected.sessionId
-      && checkpoint.recoveryEvidence.runnerInstanceId === source.runnerInstanceId
-      && checkpoint.recoveryEvidence.normalizedSessionId === selected.sessionId) { counts.unchanged++; continue; }
-    const paths = await localPaths(source);
-    const legacyControllerState = checkpoint ? undefined : await readPrivateSource(paths.controlFile, 256 * 1024 * 1024);
+    if (hasMatchingControllerRecovery(selected, source, checkpoint)) { counts.unchanged++; continue; }
+    const { paths, legacyControllerState, legacyRunnerArtifact, supervisor } = await readMigrationMaterial(db, selected, source, checkpoint);
     if (!checkpoint && legacyControllerState === undefined) { counts.sourceAlreadyAbsent++; continue; }
-    let legacyRunnerArtifact: { version: string; digest: string } | undefined;
-    if (legacyControllerState !== undefined && await readPrivateSource(join(paths.root, "provider-pack-authority.json"), 512 * 1024)) {
-      const identity = JSON.parse(legacyControllerState).identity;
-      const execution = paths.execution;
-      const [admission] = await db.select({ workspace: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'` })
-        .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, selected.companyId),
-          eq(heartbeatRuns.agentId, selected.agentId)));
-      const workspace = readNativeWorkspaceSyncReference(admission?.workspace);
-      const remoteCwd = workspace?.remoteCwd ?? execution.workspace.cwd;
-      if (typeof identity?.environmentLeaseId !== "string" || !remoteCwd
-        || (workspace && workspace.workspaceId !== execution.binding.executionWorkspaceId)) {
-        throw new Error("native_controller_migration_artifact_binding");
-      }
-      const unavailable = (): never => { throw new Error("native_controller_migration_artifact_unavailable"); };
-      legacyRunnerArtifact = resolveRemoteProviderPackAuthority({ stateDirectory: paths.root, reattach: true,
-        binding: { companyId: selected.companyId, agentId: selected.agentId,
-          workspaceId: providerPackWorkspaceId(execution.binding.executionWorkspaceId, source.id, identity.environmentLeaseId),
-          sessionId: selected.sessionId, runnerInstanceId: source.runnerInstanceId!,
-          environmentLeaseId: identity.environmentLeaseId, remoteCwd,
-          providerDigest: nativeSha256(execution.provider) },
-        current: unavailable, runnerArtifact: unavailable }).runnerArtifact;
-    }
-    const supervisor = !checkpoint && paths.execution.session.lifecyclePolicy.mode === "warm"
-      ? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined;
     input.signal.throwIfAborted();
     let result: Awaited<ReturnType<typeof importNativeControllerRecovery>>;
     try {
@@ -240,3 +254,86 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped:
   }
   return counts;
 }
+
+function onlineControllerSourceDeferral(error: unknown): string | null {
+  if (!(error instanceof Error)) return null;
+  const reasons: Record<string, string> = {
+    native_runner_checkpoint_artifact_unproven: "artifact_unproven",
+    native_runner_checkpoint_binding_mismatch: "checkpoint_binding",
+    native_controller_migration_source_changed: "source_changed",
+    native_controller_migration_unsafe_source: "unsafe_source",
+    native_controller_migration_artifact_binding: "artifact_binding",
+    native_controller_migration_artifact_unavailable: "artifact_unavailable",
+  };
+  if (reasons[error.message]) return reasons[error.message];
+  if (error.message.startsWith("runner_remote_provider_artifact_incompatible:")) return "artifact_incompatible";
+  if (error instanceof SyntaxError) return "invalid_source_json";
+  if (["ENOENT", "ENOTDIR", "EISDIR", "ELOOP", "EACCES", "EPERM"].includes((error as NodeJS.ErrnoException).code ?? "")) return "source_unavailable";
+  return null;
+}
+
+export async function prestageNativeControllers(db: Db, input: { signal: AbortSignal }) {
+  const counts = { inspected: 0, imported: 0, unchanged: 0, sourceAlreadyAbsent: 0, deferred: {} as Record<string, number> };
+  const defer = (reason: string) => { counts.deferred[reason] = (counts.deferred[reason] ?? 0) + 1; };
+  for await (const { selected, source } of retainedSources(db, input.signal)) {
+    counts.inspected++;
+    input.signal.throwIfAborted();
+    const outcome = await db.transaction(async tx => {
+      await tx.execute(sql`set local lock_timeout = '1s'`);
+      const [task] = await tx.select().from(issues).where(and(eq(issues.companyId, selected.companyId),
+        eq(issues.id, selected.issueId))).for("update");
+      if (!task || task.executionRunId || task.checkoutRunId) return "source_active";
+      const [owner] = await tx.select().from(nativeRunFinalizations).where(and(
+        eq(nativeRunFinalizations.companyId, selected.companyId), eq(nativeRunFinalizations.issueId, selected.issueId),
+        eq(nativeRunFinalizations.runId, source.id))).for("update");
+      if (!owner || owner.leaseOwner || owner.phase !== "committed" || !owner.resultId || owner.recoveryState) return "owner_not_released";
+      const [current] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id),
+        eq(heartbeatRuns.companyId, selected.companyId), eq(heartbeatRuns.agentId, selected.agentId),
+        eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.nativeSessionId, selected.sessionId))).for("update");
+      if (current?.status !== "succeeded" || !current.finishedAt) return "source_not_successful";
+      const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+        eq(heartbeatRuns.companyId, selected.companyId), inArray(heartbeatRuns.status, ["queued", "running"]),
+        or(eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.issueId, selected.issueId),
+          eq(heartbeatRuns.nativeSessionId, selected.sessionId)))).limit(1);
+      if (active) return "source_active";
+      const scoped = tx as unknown as Db;
+      const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
+      const checkpoint = await loadNativeRunnerCheckpoint(scoped, binding);
+      const profile = record(current.runnerProfileJson);
+      const freshSource = { ...source, status: current.status, runnerInstanceId: current.runnerInstanceId,
+        runnerProfileJson: { nativeToolContractFingerprint: profile.nativeToolContractFingerprint,
+          nativeExecutionInput: profile.nativeExecutionInput, sessionCheckpoint: profile.sessionCheckpoint,
+          nativeControllerRecovery: profile.nativeControllerRecovery } };
+      if (hasMatchingControllerRecovery(selected, freshSource, checkpoint)) return "unchanged";
+      const material = await readMigrationMaterial(scoped, selected, freshSource, checkpoint);
+      if (!checkpoint && material.legacyControllerState === undefined) return "source_already_absent";
+      input.signal.throwIfAborted();
+      const result = await importNativeControllerRecovery(scoped, { ...binding, agentId: selected.agentId,
+        normalizedSessionId: selected.sessionId, expectedConfigDigest: material.paths.configDigest,
+        supervisor: material.supervisor === undefined ? undefined : JSON.parse(material.supervisor),
+        legacyCheckpoint: { legacyControllerState: material.legacyControllerState,
+          legacyRunnerArtifact: material.legacyRunnerArtifact } });
+      if (result.kind !== "imported" && result.kind !== "unchanged") return "source_unproven";
+      if (material.legacyControllerState !== undefined
+        && await readPrivateSource(material.paths.controlFile, 256 * 1024 * 1024) !== material.legacyControllerState) {
+        throw new Error("native_controller_migration_source_changed");
+      }
+      if (material.supervisor !== undefined
+        && await readPrivateSource(material.paths.supervisorPath, 4 * 1024 * 1024) !== material.supervisor) {
+        throw new Error("native_controller_migration_source_changed");
+      }
+      input.signal.throwIfAborted();
+      return result.kind;
+    }).catch(error => {
+      input.signal.throwIfAborted();
+      const reason = onlineControllerSourceDeferral(error);
+      if (reason) return reason;
+      if (error?.code === "55P03" || error?.cause?.code === "55P03") return "source_busy";
+      throw error;
+    });
+    if (outcome === "imported" || outcome === "unchanged") counts[outcome]++;
+    else if (outcome === "source_already_absent") counts.sourceAlreadyAbsent++;
+    else defer(outcome);
+  }
+  return counts;
+}

server/src/services/native-runtime/native-controller-recovery.ts

@@ -1,4 +1,4 @@
-import { and, eq, ne, sql } from "drizzle-orm";
+import { and, eq, inArray, ne, or, sql } from "drizzle-orm";
 import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db";
 import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
 import { captureLegacyDurablePrpCheckpoint, parseNativeExecutionInput, type DurablePrpCheckpoint, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
@@ -226,13 +226,16 @@ export async function importNativeControllerRecovery(db: Db, input: {
   expectedConfigDigest: string;
   supervisor?: unknown;
   offline?: { writersStopped: true; legacyControllerState?: string; legacyRunnerArtifact?: { version: string; digest: string } };
+  legacyCheckpoint?: { legacyControllerState?: string; legacyRunnerArtifact?: { version: string; digest: string } };
 }): Promise<{ kind: "imported" | "unchanged" } | Exclude<NativeControllerRecoveryOutcome, { kind: "recoverable" } | { kind: "no_prior_execution" }>> {
+  if (input.offline && input.legacyCheckpoint) throw new Error("Conflicting native controller migration authority");
+  const legacy = input.offline ?? input.legacyCheckpoint;
   const supervisor = record(input.supervisor);
   if (input.supervisor !== undefined && (supervisor.schema !== "paperclip.native-session-supervisor.v1" || supervisor.configDigest !== input.expectedConfigDigest)) {
     return { kind: "unsupported", reason: "supervisor_compatibility" };
   }
   return db.transaction(async tx => {
-    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, input.companyId), eq(issues.id, input.issueId))).for("share");
     const [owner] = await tx.select().from(nativeRunFinalizations).where(and(
       eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId),
@@ -252,6 +255,13 @@ export async function importNativeControllerRecovery(db: Db, input: {
     if (!input.offline && (!(owner.phase === "committed" && owner.resultId) || owner.recoveryState)) {
       return { kind: "missing_evidence" as const, reason: "pending_finalization" };
     }
+    if (input.legacyCheckpoint) {
+      const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+        eq(heartbeatRuns.companyId, input.companyId), inArray(heartbeatRuns.status, ["queued", "running"]),
+        or(eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.issueId, input.issueId),
+          eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId)))).limit(1);
+      if (task.checkoutRunId || active) return { kind: "missing_evidence" as const, reason: "legacy_source_active" };
+    }
     const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId),
       eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId),
       eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId))).for("update");
@@ -266,7 +276,7 @@ export async function importNativeControllerRecovery(db: Db, input: {
     if (input.supervisor === undefined) {
       try {
         const admitted = parseNativeExecutionInput(profile.nativeExecutionInput);
-        if (!input.offline || (!checkpoint && admitted.session.lifecyclePolicy.mode !== "per_turn")
+        if (!legacy || (!checkpoint && admitted.session.lifecyclePolicy.mode !== "per_turn")
           || admitted.session.normalizedSessionId !== input.normalizedSessionId
           || admitted.binding.runId !== input.runId || admitted.binding.companyId !== input.companyId
           || admitted.binding.agentId !== input.agentId || admitted.binding.issueId !== input.issueId) {
@@ -274,12 +284,12 @@ export async function importNativeControllerRecovery(db: Db, input: {
         }
       } catch { return { kind: "unsupported" as const, reason: "admitted_input_format" }; }
     }
-    const importing = !checkpoint && input.offline?.legacyControllerState !== undefined;
+    const importing = !checkpoint && legacy?.legacyControllerState !== undefined;
     if (importing) {
       if (!run.runnerInstanceId) return { kind: "missing_evidence" as const, reason: "checkpoint_binding" };
-      checkpoint = captureLegacyDurablePrpCheckpoint(input.offline!.legacyControllerState!, {
+      checkpoint = captureLegacyDurablePrpCheckpoint(legacy!.legacyControllerState!, {
         runId: input.runId, runnerInstanceId: run.runnerInstanceId, normalizedSessionId: input.normalizedSessionId,
-      }, input.offline!.legacyRunnerArtifact);
+      }, legacy!.legacyRunnerArtifact);
     }
     if (!checkpoint || checkpoint.recoveryEvidence.normalizedSessionId !== input.normalizedSessionId
       || checkpoint.recoveryEvidence.runnerInstanceId !== run.runnerInstanceId) {