fix(runtime): honor durable authority and retired lease handoffs

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

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

@@ -156,3 +156,17 @@ 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.

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

@@ -3,7 +3,7 @@ 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, environmentLeases, 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";
@@ -13,6 +13,7 @@ import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from
 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";
@@ -22,10 +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_queued", "stale_overlap", "stale_success", "stale_busy", "stale_durable",
+  "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"])("offline native controller migration (%s)", scenario => {
+  "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) {
@@ -269,17 +272,47 @@ 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") {
+    if (scenario.startsWith("stale_durable")) {
-      const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, { runId: f.runId,
+      const state = JSON.parse(f.source);
-        runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId },
+      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_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);
@@ -360,6 +393,27 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
         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));
@@ -376,7 +430,7 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
     }
     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"].includes(scenario)) {
+    } 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") {

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

@@ -201,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>>;

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

@@ -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" };

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

@@ -16,7 +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 } from "../remote-execution-termination.js";
+import { hasRemoteTerminationReceipt, resolveRemoteTerminationLease } from "../remote-execution-termination.js";
 import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
 
 const SCHEMA = "paperclip.native-harness-object.v1";
@@ -121,7 +121,8 @@ async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner:
           || 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");
-        if (!leases.length || leases.some(lease => {
+        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)