fix(runtime): preserve failed sources and durable checkpoint authority (#19038)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A volume-free web server must preserve retained native session data. > - An old failed follow-up can keep its own snapshot while its controller file still belongs to the preceding successful turn. > - The Runner correctly refuses to import that file as the failed turn's execution authority. > - That refusal also blocks migration when no warm supervisor is present. > - This pull request preserves the file after it proves the older run's lineage, and uses existing typed database authority when present. > - Failed history and files survive, while normal execution authority stays unchanged. ## Linked Issues or Issue Description Refs #19024, which preserves the related older-supervisor case. The duplicate search found no PR for this older-controller case. **What happened?** Offline native-controller migration stops with `native_runner_checkpoint_binding_mismatch` for a failed retained follow-up. The local controller still contains the preceding successful run. A per-turn source can have no supervisor file, so the older-supervisor archive path does not apply. Read-only checks matched the database snapshots and controller-file hashes between the original source and the isolated rehearsal copy. **Expected behavior** Preserve the available file as evidence through the existing encrypted archive. Require a valid older controller, a completed successful prior run, and the same company, agent, task, native session, Runner and provider thread. Keep successful sources and active owners outside this archival fallback. When a valid typed database checkpoint already exists, derive only its missing recovery metadata and ignore obsolete local projections. **Steps to reproduce** Create a successful native turn and a later failed follow-up in the same session. Retain the successful controller file. Keep the failed snapshot and admitted input in the database. Leave the per-turn source without a supervisor. Run offline migration. The new positive regression fails with the binding mismatch before this change. ## What Changed - Share the scoped older-successful-run proof with the existing supervisor preservation path. - Validate the older controller through the Runner parser and artifact checks before archive publication. - Archive this failed legacy source without changing its snapshot, status, ownership, queued work or session pointer. - Preserve the normal refusal for foreign, changed, active, successful, invalid and already-durable cases. - Permit evidence archival for a stale observed coordinator only with exact host-owned stop receipts for every remote lease, no current task ownership, no result and no retry deadline. Recheck the proof before publication. - Use execution start for lifecycle ordering. A queued follow-up can predate its prior turn completion, and a remote stop can predate a later recovery timeout. - Resolve destruction receipts through the existing reusable-lease handoff verifier. Reject stopped tips, broken chains and competing tips. - Prefer a valid typed database checkpoint over a stale or absent local warm supervisor. Add only missing recovery metadata after current admitted-input and checkpoint binding checks. - Document the evidence and execution-authority boundary. ## Verification - Positive regression: red before the owner change and green afterward. - `pnpm exec vitest run server/src/services/native-runtime/native-controller-migration.integration.test.ts server/src/services/native-runtime/native-controller-recovery.integration.test.ts server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts`: 81 checks passed across migration, recovery and durable storage. - The checks cover storage failure, repeated migration, exact file restoration after old-home deletion, unchanged history and ownership, and refusal to treat the archive as executable recovery authority. - A second regression reproduced the archive fence refusal for an observed failed source with a valid stopped-remote receipt. Negative controls cover missing, foreign, old and revoked receipts, live leases, pending results, retry deadlines and task ownership. Revoking the receipt during upload prevents publication. - Two additional regressions cover a follow-up queued before the prior turn finished and a remote stop before a later logical failure. Both fail with the earlier timestamp comparisons and pass with execution-start ordering. Genuine execution overlap and old receipts stay blocked. - Two historical source snapshots and controller-file hashes matched in a read-only live-versus-copy check. The first candidate reached the archive fence and exposed the observed-coordinator case. That failure is retained. A fresh candidate replay is required for the expanded proof. It does not qualify the full packaged migration or hosted Railway handoff. - Existing typed-authority checks cover stale and missing supervisors, repeat migration, byte-identical checkpoint rows, unchanged snapshots/owners/history, old-home deletion and ordinary queued-turn recovery. Foreign Runner, session and snapshot controls remain blocked. - Handoff checks cover an exact destroyed successor, a stopped successor, a broken proof and a competing tip. - Isolated replay of the two retained typed-checkpoint sources imported only derived recovery metadata. Checks confirmed unchanged checkpoint rows (12 and 9 parts), history, snapshots, finalization owners and task-session pointers. The replay used three source-module overlays with a blocked provider network; it does not qualify the released bundle. - Final-head CI attempt 1 passed typecheck, build and the migration suites but failed one unrelated budget-concurrency assertion in `decision-models.test.ts` (one provider call observed instead of two). The original log is retained. The isolated local comparison cannot classify the timing failure because this checkout has stale linked budget schema exports. The one retry of that failed CI job passed, and final-head CI attempt 2 is green. The original failure remains retained; its cause is not established. - Full local typecheck, repository tests and build are not claimed. The server typecheck reports pre-existing linked declaration mismatches outside the changed files; the full clean CI typecheck has passed. This checkout uses linked generated declarations; clean CI is required before merge. Prior head d71dc913 had passing full CI and a 5/5 Greptile review. Head 8ff3a22109 adds the handoff and typed-authority fixes. Its fresh full CI is green and Greptile is 5/5, with no unresolved review threads. Earlier head7729 hit unrelated 5-second CLI ZIP and 15-second setup-token test timeouts; those failures remain retained. ## Risks This path preserves evidence for a source that has already failed. It does not make that source resumable. The Runner must accept the older controller under its own identity and artifact binding. Scoped successful history must match its session and provider thread. The archive owner still requires stopped writers, a finished failed run and no live lease. An observed coordinator also needs exact remote-stop receipts after execution started, no accepted result or retry deadline, and no current task ownership. The coordinator stays unchanged. Existing durable authority stays authoritative. ## Model Used OpenAI Codex, exact model ID `gpt-6.1-sol`, with reasoning, tool use and code execution. The host does not expose 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 described the issue in-PR following the bug template - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal ticket id or instance-derived details - [x] I have run the relevant tests locally; verification limits are listed above - [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 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>
2026-11-08 22:13:59 +01:00 · 2026-11-04 15:37:54 -06:00
1 parent 4740d251b7
commit 61b2455319
@@ -137,3 +137,36 @@ test resumes that same queued run after deleting the old disk, verifies saved
 workspace bytes and instructions at provider dispatch, and checks one result and
 one wake consumption even after a second queue sweep. Its sandbox transport and
 provider session are scripted; database, S3, admission and restoration are real.
+
+A failed follow-up can also have no warm supervisor and retain the preceding
+successful run in its controller file. Archive this source only after the Runner
+parser validates that older authority and scoped history confirms the same
+Runner, session, provider thread and an earlier successful finish. Keep current
+snapshots, ownership and typed checkpoints unchanged. The archive cannot become
+executable recovery authority. Foreign, invalid and active sources remain blocked.
+
+For a finished failed source with a stale observed coordinator, evidence archival
+also requires an expired controller lease, no accepted result or retry deadline,
+and no current task ownership. Every remote lease must carry the exact host-owned
+termination receipt, confirmed after execution started. Recheck those receipts
+before publication. Preserve the coordinator as recorded; this proof permits
+evidence storage, not execution or a claim that the coordinator has settled.
+
+Compare retained turn completion to the follow-up execution start, not its queue
+creation time. A queued follow-up can predate the prior turn completion. A remote
+stop can also predate a later logical failure caused by recovery timeout. These
+proofs use the recorded execution start and never rewrite lifecycle timestamps.
+
+### Existing typed checkpoint authority
+
+When a retained source already has a valid typed database checkpoint, derive its
+small recovery metadata from that authority and the current admitted input. Do
+not consult a stale or absent local warm supervisor. Validate the same company,
+agent, task, normalized session and Runner bindings and retain the checkpoint,
+snapshot, owner and session pointer unchanged. Offline warm import without a
+supervisor still requires this existing typed authority; normal runtime import
+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.
@@ -3,16 +3,18 @@ import { mkdir, readFile, rm, 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, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db";
+import { agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, 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 { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
-import { resolveNativeControllerRecovery } from "./native-controller-recovery.js";
+import { createNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js";
 import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
 import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
+import { createReusableLeaseHandoff } from "../environment-lease-handoff.js";
+import { remoteTerminationReceipt } from "../remote-execution-termination.js";
 import { nativeSha256 } from "./canonical.js";
 import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js";
 import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js";
@@ -21,7 +23,12 @@ import { legacyNativeControllerArchive } from "./native-controller-archive.js";
 import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";

 describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable_failed",
-  "stale_confirmed", "stale_foreign", "stale_changed", "stale_success", "stale_busy", "stale_durable"])("offline native controller migration (%s)", scenario => {
+  "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",
+  "controller_observed_stopped", "controller_observed_stop_before_finish", "controller_observed_missing", "controller_observed_foreign",
+  "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());
  async function fixture(commandCount = 1, warm = false) {
@@ -68,7 +75,7 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
        digest: `sha256:${createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex")}` }),
      runnerArtifact: () => ({ version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` }) });
    pack.verified();
-    const snapshot = { backendKind: "mock", sessionId: f.native.normalizedSessionId,
+    const snapshot = { backendKind: "mock" as const, sessionId: f.native.normalizedSessionId,
      identity: { companyId, agentId, issueId: f.issueId, runId: f.runId, sessionId: f.native.normalizedSessionId } };
    const profile = { ...f.preparedProfile, nativeExecutionInput: execution, nativeToolContractFingerprint: fingerprint, sessionCheckpoint: snapshot,
      nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v1", state: "prepared", descriptorSha256: "a".repeat(64),
@@ -249,6 +256,12 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
      nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, runnerInstanceId: f.native.runnerInstanceId,
      runtimeMode: "native", status: "succeeded", createdAt: new Date(Date.now() - 60_000),
      finishedAt: new Date(Date.now() - 50_000), runnerProfileJson: priorProfile });
+    if (scenario === "stale_queued" || scenario === "stale_overlap") {
+      const now = Date.now();
+      await f.db.update(heartbeatRuns).set({ createdAt: new Date(now - 45_000),
+        startedAt: new Date(now - (scenario === "stale_queued" ? 30_000 : 43_000)) }).where(eq(heartbeatRuns.id, f.runId));
+      await f.db.update(heartbeatRuns).set({ finishedAt: new Date(now - 40_000) }).where(eq(heartbeatRuns.id, priorId));
+    }
    const supervisor = JSON.stringify({ schema: "paperclip.native-session-supervisor.v1",
      configDigest: f.paths.configDigest, snapshot: priorSnapshot });
    await mkdir(dirname(f.paths.supervisorPath), { recursive: true });
@@ -259,18 +272,48 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
    if (scenario === "stale_success") await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
    if (scenario === "stale_busy") await f.db.update(nativeRunFinalizations).set({ leaseOwner: "live-owner",
      leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
-    if (scenario === "stale_durable") {
-      const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, { runId: f.runId,
-        runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId },
+    if (scenario.startsWith("stale_durable")) {
+      const state = JSON.parse(f.source);
+      if (scenario === "stale_durable_runner") state.identity.runnerInstanceId = randomUUID();
+      if (scenario === "stale_durable_session") state.identity.normalizedSessionId = randomUUID();
+      const checkpoint = captureLegacyDurablePrpCheckpoint(JSON.stringify(state), { runId: f.runId,
+        runnerInstanceId: state.identity.runnerInstanceId, normalizedSessionId: state.identity.normalizedSessionId },
      { version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` });
      await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx,
        { companyId: f.companyId, issueId: f.issueId, runId: f.runId }, checkpoint));
+      if (scenario === "stale_durable_absent") await rm(f.paths.supervisorPath);
+      if (scenario === "stale_durable_snapshot") await f.db.update(heartbeatRuns).set({ runnerProfileJson: {
+        ...f.profile, sessionCheckpoint: { ...f.snapshot, identity: { ...f.snapshot.identity, agentId: randomUUID() } },
+      } }).where(eq(heartbeatRuns.id, f.runId));
    }
    const [originalRun] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
    const [originalOwner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
    const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "durable-objects")));
    const options = { storage, writersStopped: true as const, signal: AbortSignal.timeout(30_000) };
-    if (scenario !== "stale_confirmed") {
+    if (scenario === "stale_durable" || scenario === "stale_durable_absent") {
+      const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId };
+      const before = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
+      expect(await migrateNativeControllers(f.db, options)).toMatchObject({ imported: 1, archivedFailed: 0 });
+      expect(await migrateNativeControllers(f.db, options)).toMatchObject({ unchanged: 1, archivedFailed: 0 });
+      expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false);
+      expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(before);
+      expect(await f.db.select().from(nativeHarnessBackups)).toHaveLength(0);
+      const [afterRun] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+      expect(afterRun.runnerProfileJson?.sessionCheckpoint).toEqual(originalRun.runnerProfileJson?.sessionCheckpoint);
+      expect(afterRun).toEqual({ ...originalRun, runnerProfileJson: afterRun.runnerProfileJson, updatedAt: afterRun.updatedAt });
+      expect(afterRun.runnerProfileJson).toEqual({ ...originalRun.runnerProfileJson,
+        nativeControllerRecovery: expect.objectContaining({ snapshotSha256: nativeSha256(f.snapshot),
+          checkpoint: { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId } }) });
+      expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(originalOwner);
+      await rm(f.paperclipHome, { recursive: true, force: true });
+      expect((await loadNativeRunnerCheckpoint(f.db, binding))?.records.size).toBe(before.length);
+      await f.db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId,
+        status: "running", runtimeMode: "native" }).where(eq(heartbeatRuns.id, f.queuedId));
+      expect((await resolveNativeControllerRecovery(f.db, { companyId: f.companyId, agentId: f.agentId,
+        issueId: f.issueId, normalizedSessionId: f.native.normalizedSessionId, currentRunId: f.queuedId })).kind).toBe("recoverable");
+      return;
+    }
+    if (!["stale_confirmed", "stale_queued"].includes(scenario)) {
      await expect(migrateNativeControllers(f.db, options)).rejects.toThrow();
      expect(await f.db.select().from(nativeHarnessBackups)).toHaveLength(0);
      expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true);
@@ -299,6 +342,126 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
    expect((await resolveNativeControllerRecovery(f.db, { companyId: f.companyId, agentId: f.agentId,
      issueId: f.issueId, normalizedSessionId: f.native.normalizedSessionId, currentRunId: f.queuedId })).kind).not.toBe("recoverable");
  });
+  if (scenario.startsWith("controller_")) it("preserves failed legacy files only with proven older controller lineage", async () => {
+    const f = await fixture(), priorId = randomUUID();
+    const priorSnapshot = { ...f.snapshot, identity: { ...f.snapshot.identity, runId: priorId } };
+    if (scenario === "controller_foreign") priorSnapshot.identity.agentId = randomUUID();
+    const priorProfile = { ...f.profile, nativeExecutionInput: { ...f.profile.nativeExecutionInput,
+      binding: { ...f.profile.nativeExecutionInput.binding, runId: priorId } }, sessionCheckpoint: priorSnapshot };
+    await f.db.insert(heartbeatRuns).values({ id: priorId, companyId: f.companyId,
+      agentId: f.agentId,
+      nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId,
+      runnerInstanceId: scenario === "controller_runner" ? randomUUID() : f.native.runnerInstanceId,
+      runtimeMode: "native", status: "succeeded", createdAt: new Date(Date.now() - 60_000),
+      finishedAt: new Date(Date.now() - 50_000), runnerProfileJson: scenario === "controller_thread"
+        ? { ...priorProfile, sessionCheckpoint: { ...priorSnapshot, sessionId: "foreign-provider-thread" } } : priorProfile });
+    const state = JSON.parse(f.source);
+    state.identity = { ...state.identity, runId: priorId, turnId: `turn-${priorId}`, itemId: `item-${priorId}` };
+    if (scenario === "controller_invalid") state.schema = "unrecognized";
+    const source = JSON.stringify(state);
+    await writeFile(f.file, source, { mode: 0o600 });
+    if (scenario === "controller_success") await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
+    if (scenario === "controller_busy") await f.db.update(nativeRunFinalizations).set({ leaseOwner: "live-owner",
+      leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
+    if (scenario === "controller_durable") {
+      const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, f.identity,
+        { version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` });
+      await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx,
+        { companyId: f.companyId, issueId: f.issueId, runId: f.runId }, checkpoint));
+      await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile, nativeControllerRecovery:
+        createNativeControllerRecovery(f.snapshot, { configDigest: f.paths.configDigest, processConfigDigest: null },
+          { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId }) } })
+        .where(eq(heartbeatRuns.id, f.runId));
+    }
+    if (scenario.startsWith("controller_observed_")) {
+      await f.db.update(nativeRunFinalizations).set({ phase: scenario === "controller_observed_result" ? "result_accepted" : "observed",
+        leaseOwner: "retired-controller", leaseExpiresAt: new Date(Date.now() + (scenario === "controller_observed_live" ? 60_000 : -60_000)),
+        nextAttemptAt: scenario === "controller_observed_retry" ? new Date(Date.now() + 60_000) : null,
+      }).where(eq(nativeRunFinalizations.runId, f.runId));
+      if (scenario === "controller_observed_task") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
+      if (scenario !== "controller_observed_missing") {
+        const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.runId, issueId: f.issueId,
+          provider: "daytona", providerLeaseId: randomUUID(), status: "released", cleanupStatus: "success", releasedAt: new Date() };
+        const receipt = remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" })!;
+        if (scenario === "controller_observed_stop_before_finish") {
+          const now = Date.now();
+          await f.db.update(heartbeatRuns).set({ createdAt: new Date(now - 40_000), startedAt: new Date(now - 30_000) })
+            .where(eq(heartbeatRuns.id, f.runId));
+          receipt.confirmedAt = new Date(now - 10_000).toISOString();
+          lease.releasedAt = new Date(now - 10_000);
+        }
+        if (scenario === "controller_observed_foreign") receipt.runId = randomUUID();
+        if (scenario === "controller_observed_old_receipt") receipt.confirmedAt = new Date(0).toISOString();
+        await f.db.insert(environmentLeases).values({ ...lease, metadata: { remoteExecutionTermination: receipt } });
+        if (scenario.startsWith("controller_observed_handoff")) {
+          const environmentId = randomUUID(), successorRunId = randomUUID(), handoffAt = new Date(Date.now() - 1000);
+          await f.db.insert(environments).values({ id: environmentId, name: environmentId, driver: "sandbox" });
+          await f.db.insert(heartbeatRuns).values({ id: successorRunId, companyId: f.companyId, agentId: f.agentId,
+            nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, runtimeMode: "native", status: "succeeded", finishedAt: new Date() });
+          const [previous] = await f.db.update(environmentLeases).set({ environmentId, leasePolicy: "reuse_by_environment",
+            status: "expired", acquiredAt: new Date(handoffAt.getTime() - 1000), releasedAt: handoffAt,
+            metadata: { agentId: f.agentId } }).where(eq(environmentLeases.id, lease.id)).returning();
+          const successor = { ...previous, id: randomUUID(), heartbeatRunId: successorRunId, acquiredAt: handoffAt,
+            status: "released", releasedAt: new Date(), metadata: { agentId: f.agentId } };
+          const handoff = createReusableLeaseHandoff(previous, successor);
+          const termination = remoteTerminationReceipt(successor, { providerLeaseId: successor.providerLeaseId,
+            state: scenario === "controller_observed_handoff_stopped" ? "stopped" : "destroyed" });
+          await f.db.update(environmentLeases).set({ metadata: { agentId: f.agentId, reusableLeaseSuccessor: handoff } })
+            .where(eq(environmentLeases.id, previous.id));
+          await f.db.insert(environmentLeases).values({ ...successor, metadata: { agentId: f.agentId,
+            reusableLeasePredecessor: scenario === "controller_observed_handoff_broken" ? { ...handoff, handoffId: randomUUID() } : handoff,
+            remoteExecutionTermination: termination } });
+          if (scenario === "controller_observed_handoff_fork") await f.db.insert(environmentLeases).values({ ...previous,
+            id: randomUUID(), heartbeatRunId: f.queuedId, status: "active", releasedAt: null, metadata: { agentId: f.agentId } });
+        }
+      }
+    }
+    const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+    const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
+    const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "durable-objects")));
+    const options = { storage, writersStopped: true as const, signal: AbortSignal.timeout(30_000) };
+    if (scenario === "controller_observed_revoked") {
+      const put = storage.putFile.bind(storage);
+      vi.spyOn(storage, "putFile").mockImplementation(async args => {
+        const result = await put(args);
+        await f.db.update(environmentLeases).set({ metadata: {} }).where(eq(environmentLeases.heartbeatRunId, f.runId));
+        return result;
+      });
+    }
+    if (scenario === "controller_durable") {
+      expect(await migrateNativeControllers(f.db, options)).toMatchObject({ unchanged: 1, archivedFailed: 0 });
+    } else if (!["controller_confirmed", "controller_observed_stopped", "controller_observed_stop_before_finish", "controller_observed_handoff"].includes(scenario)) {
+      await expect(migrateNativeControllers(f.db, options)).rejects.toThrow();
+      const archives = await f.db.select().from(nativeHarnessBackups);
+      if (scenario === "controller_observed_revoked") {
+        expect(storage.putFile).toHaveBeenCalledOnce();
+        expect(archives).toHaveLength(1);
+        expect(archives[0].status).toBe("pending");
+      }
+      else expect(archives).toHaveLength(0);
+    } else {
+      const put = vi.spyOn(storage, "putFile").mockRejectedValue(new Error("storage unavailable"));
+      await expect(migrateNativeControllers(f.db, options)).rejects.toThrow("storage unavailable");
+      put.mockRestore();
+      expect(await migrateNativeControllers(f.db, options)).toMatchObject({ archivedFailed: 1, imported: 0 });
+      expect(await migrateNativeControllers(f.db, options)).toMatchObject({ archivedFailed: 1, imported: 0 });
+      expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false);
+      expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toHaveLength(0);
+      await rm(f.paperclipHome, { recursive: true, force: true });
+      const descriptor = legacyNativeControllerArchive({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, controller: source });
+      const replacement = join(f.runtimeRoot, "archived-older-controller");
+      expect(await restoreDurableNativeHarnessBackup({ db: f.db, storage, companyId: f.companyId,
+        scopeDigest: descriptor.scopeDigest, backupRoot: replacement })).toBe(true);
+      expect(await readFile(join(replacement, "current", "control-plane-state.json"), "utf8")).toBe(source);
+      expect((await resolveNativeControllerRecovery(f.db, { companyId: f.companyId, agentId: f.agentId,
+        issueId: f.issueId, normalizedSessionId: f.native.normalizedSessionId, currentRunId: f.queuedId })).kind).not.toBe("recoverable");
+    }
+    expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(run);
+    expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(owner);
+    expect((await f.db.select().from(agentTaskSessions).where(eq(agentTaskSessions.id, f.session.id)))[0].lastRunId).toBe(f.runId);
+    expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId)
+      .toBe(scenario === "controller_observed_task" ? f.runId : f.queuedId);
+  });
  if (scenario === "success") it("does not substitute a failure archive for a successful retained session", async () => {
    const f = await fixture();
    await rm(join(f.paths.root, "provider-pack-authority.json"));
@@ -3,7 +3,7 @@ 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 { parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
+import { captureLegacyDurablePrpCheckpoint, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
 import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
 import { findNativeSessionResumeRun } from "./native-session-resume.js";
 import { nativeControllerMigrationDescriptor } from "./native-session-executor.js";
@@ -24,34 +24,54 @@ function record(value: unknown): Record<string, unknown> {
  return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
 }

-async function hasConfirmedOlderSupervisor(db: Db, selected: Selected,
+async function confirmedOlderSuccessfulSnapshot(db: Db, selected: Selected,
  source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>,
-  paths: Awaited<ReturnType<typeof localPaths>>, supervisor: unknown) {
+  paths: Awaited<ReturnType<typeof localPaths>>, olderRunId: unknown) {
  const snapshot = record(source.runnerProfileJson.sessionCheckpoint), identity = record(snapshot.identity);
-  const olderSnapshot = record(record(supervisor).snapshot), olderIdentity = record(olderSnapshot.identity);
  const scope = { companyId: selected.companyId, agentId: selected.agentId, issueId: selected.issueId, sessionId: selected.sessionId };
  if (source.status !== "failed" || identity.runId !== source.id
-    || Object.entries(scope).some(([key, value]) => identity[key] !== value || olderIdentity[key] !== value)
+    || Object.entries(scope).some(([key, value]) => identity[key] !== value)
    || paths.execution.binding.runId !== source.id || paths.execution.binding.companyId !== selected.companyId
    || paths.execution.binding.agentId !== selected.agentId || paths.execution.binding.issueId !== selected.issueId
    || paths.execution.session.normalizedSessionId !== selected.sessionId
    || typeof snapshot.sessionId !== "string" || !snapshot.sessionId.trim()
-    || snapshot.sessionId !== olderSnapshot.sessionId || snapshot.backendKind !== olderSnapshot.backendKind
-    || typeof olderIdentity.runId !== "string" || olderIdentity.runId === source.id
-    || !/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(olderIdentity.runId)) return false;
+    || typeof olderRunId !== "string" || olderRunId === source.id
+    || !/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(olderRunId)) return null;
  const [prior] = await db.select({ createdAt: heartbeatRuns.createdAt, finishedAt: heartbeatRuns.finishedAt,
    runnerInstanceId: heartbeatRuns.runnerInstanceId, snapshot: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'sessionCheckpoint'` })
-    .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, olderIdentity.runId), eq(heartbeatRuns.companyId, selected.companyId),
+    .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, olderRunId), eq(heartbeatRuns.companyId, selected.companyId),
      eq(heartbeatRuns.agentId, selected.agentId), eq(heartbeatRuns.nativeIssueId, selected.issueId),
      eq(heartbeatRuns.nativeSessionId, selected.sessionId), eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "succeeded")));
-  if (!prior?.finishedAt || prior.createdAt >= source.createdAt || prior.finishedAt > source.createdAt
+  const priorSnapshot = record(prior?.snapshot), priorIdentity = record(priorSnapshot.identity);
+  if (!prior?.finishedAt || prior.createdAt >= source.createdAt
    || !source.runnerInstanceId || prior.runnerInstanceId !== source.runnerInstanceId
-    || nativeSha256(prior.snapshot) !== nativeSha256(redactRunnerApiPayloads(olderSnapshot))) return false;
-  const [failed] = await db.select({ finishedAt: heartbeatRuns.finishedAt }).from(heartbeatRuns).where(and(
+    || priorIdentity.runId !== olderRunId || Object.entries(scope).some(([key, value]) => priorIdentity[key] !== value)
+    || priorSnapshot.sessionId !== snapshot.sessionId || priorSnapshot.backendKind !== snapshot.backendKind) return null;
+  const [failed] = await db.select({ finishedAt: heartbeatRuns.finishedAt, startedAt: heartbeatRuns.startedAt }).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),
    eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "failed")));
-  return Boolean(failed?.finishedAt);
+  return failed?.finishedAt && prior.finishedAt <= (failed.startedAt ?? source.createdAt) ? priorSnapshot : null;
+}
+
+async function hasConfirmedOlderSupervisor(db: Db, selected: Selected,
+  source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>,
+  paths: Awaited<ReturnType<typeof localPaths>>, supervisor: unknown) {
+  const olderSnapshot = record(record(supervisor).snapshot);
+  const prior = await confirmedOlderSuccessfulSnapshot(db, selected, source, paths, record(olderSnapshot.identity).runId);
+  return prior !== null && nativeSha256(prior) === nativeSha256(redactRunnerApiPayloads(olderSnapshot));
+}
+
+async function hasConfirmedOlderController(db: Db, selected: Selected,
+  source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>,
+  paths: Awaited<ReturnType<typeof localPaths>>, controller: string, artifact?: { version: string; digest: string }) {
+  const identity = record(record(JSON.parse(controller)).identity);
+  if (!await confirmedOlderSuccessfulSnapshot(db, selected, source, paths, identity.runId)) return false;
+  try {
+    captureLegacyDurablePrpCheckpoint(controller, { runId: String(identity.runId),
+      runnerInstanceId: source.runnerInstanceId!, normalizedSessionId: selected.sessionId }, artifact);
+    return true;
+  } catch { return false; }
 }

 async function* retainedSources(db: Db, signal?: AbortSignal) {
@@ -181,7 +201,7 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped:
          providerDigest: nativeSha256(execution.provider) },
        current: unavailable, runnerArtifact: unavailable }).runnerArtifact;
    }
-    const supervisor = paths.execution.session.lifecyclePolicy.mode === "warm"
+    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>>;
@@ -192,8 +212,11 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped:
        offline: { writersStopped: true, legacyControllerState, legacyRunnerArtifact } });
    } catch (error) {
      const storage = input.storage ?? durableNativeHarnessStorage();
-      if (!(error instanceof Error) || error.message !== "native_runner_checkpoint_artifact_unproven"
-        || source.status !== "failed" || legacyControllerState === undefined || !storage) throw error;
+      if (!(error instanceof Error) || source.status !== "failed" || legacyControllerState === undefined || !storage) throw error;
+      const archivable = error.message === "native_runner_checkpoint_artifact_unproven"
+        || error.message === "native_runner_checkpoint_binding_mismatch"
+          && await hasConfirmedOlderController(db, selected, source, paths, legacyControllerState, legacyRunnerArtifact);
+      if (!archivable) throw error;
      await archiveFailedLegacyNativeController(db, { ...binding, storage, controller: legacyControllerState, supervisor,
        writersStopped: true, signal: input.signal });
      counts.archivedFailed++;
@@ -262,17 +262,18 @@ export async function importNativeControllerRecovery(db: Db, input: {
      || input.supervisor !== undefined && nativeSha256(snapshot) !== nativeSha256(redactRunnerApiPayloads(supervisor.snapshot))) {
      return { kind: "missing_evidence" as const, reason: "stale_supervisor_snapshot" };
    }
+    let checkpoint = await loadNativeRunnerCheckpoint(tx, input);
    if (input.supervisor === undefined) {
      try {
        const admitted = parseNativeExecutionInput(profile.nativeExecutionInput);
-        if (!input.offline || admitted.session.lifecyclePolicy.mode !== "per_turn"
+        if (!input.offline || (!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) {
          return { kind: "missing_evidence" as const, reason: "admitted_per_turn_binding" };
        }
      } catch { return { kind: "unsupported" as const, reason: "admitted_input_format" }; }
    }
-    let checkpoint = await loadNativeRunnerCheckpoint(tx, input);
    const importing = !checkpoint && input.offline?.legacyControllerState !== undefined;
    if (importing) {
      if (!run.runnerInstanceId) return { kind: "missing_evidence" as const, reason: "checkpoint_binding" };
@@ -7,7 +7,7 @@ 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 { heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db";
+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";
 import { lockNativeRunnerCheckpointOwner, validateNativeRunnerCheckpointOwner, type NativeRunnerCheckpointFenceOwner } from "./native-runner-checkpoint.js";
@@ -16,6 +16,7 @@ import { getStorageService } from "../../storage/index.js";
 import { readNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp } from "./native-harness-backup-stamp.js";
 import { captureNativeHarnessArchive, restoreNativeHarnessArchive, MAX_NATIVE_HARNESS_OBJECT_BYTES } from "./native-harness-backup-archive.js";
 import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
+import { hasRemoteTerminationReceipt, resolveRemoteTerminationLease } from "../remote-execution-termination.js";
 import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";

 const SCHEMA = "paperclip.native-harness-object.v1";
@@ -98,18 +99,37 @@ async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner:
    if (!owner.writersStopped) fail("offline_writers_required");
    const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1);
    if (active) fail("offline_writers_required");
-    const [run] = await tx.select({ finishedAt: heartbeatRuns.finishedAt, status: heartbeatRuns.status }).from(heartbeatRuns).where(and(
+    const [run] = await tx.select({ finishedAt: heartbeatRuns.finishedAt, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, status: heartbeatRuns.status }).from(heartbeatRuns).where(and(
      eq(heartbeatRuns.id, owner.runId), eq(heartbeatRuns.companyId, owner.companyId),
      eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.runtimeMode, "native"))).for("share");
    const [coordinator] = await tx.select({ leaseOwner: nativeRunFinalizations.leaseOwner,
-      leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, phase: nativeRunFinalizations.phase })
+      leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, phase: nativeRunFinalizations.phase,
+      resultId: nativeRunFinalizations.resultId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt })
      .from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, owner.runId)).for("share");
    const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`);
    if (!run?.finishedAt || !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(run.status)
-      || !coordinator || coordinator.leaseOwner && (!coordinator.leaseExpiresAt || coordinator.leaseExpiresAt > new Date(clock.now))
-      || owner.terminalFailureOnly && (run.status !== "failed" || coordinator.phase !== "terminal_failure")) {
+      || !coordinator || coordinator.leaseOwner && (!coordinator.leaseExpiresAt || coordinator.leaseExpiresAt > new Date(clock.now))) {
      fail("offline_owner_unsettled");
    }
+    if (owner.terminalFailureOnly) {
+      if (run.status !== "failed") fail("offline_owner_unsettled");
+      if (coordinator.phase !== "terminal_failure") {
+        // Evidence archival does not settle or resume this stale coordinator.
+        // Lease expiry alone is insufficient: every remote writer needs its
+        // exact host-owned stop receipt, recorded after execution began.
+        if (coordinator.phase !== "observed" || coordinator.resultId || coordinator.nextAttemptAt
+          || task.executionRunId === owner.runId) fail("offline_owner_unsettled");
+        const leases = await tx.select().from(environmentLeases).where(and(
+          eq(environmentLeases.companyId, owner.companyId), eq(environmentLeases.heartbeatRunId, owner.runId))).for("share");
+        const stopped = await Promise.all(leases.map(lease => resolveRemoteTerminationLease(tx as unknown as Db, lease)));
+        if (!stopped.length || stopped.some(lease => {
+          const receipt = lease.metadata?.remoteExecutionTermination as { confirmedAt?: unknown } | undefined;
+          const confirmedAt = Date.parse(String(receipt?.confirmedAt));
+          return lease.issueId !== owner.issueId || !hasRemoteTerminationReceipt(lease)
+            || !Number.isFinite(confirmedAt) || confirmedAt < (run.startedAt ?? run.createdAt).getTime() || confirmedAt > new Date(clock.now).getTime();
+        })) fail("offline_owner_unsettled");
+      }
+    }
    return;
  }
  const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`);