fix(runtime): preserve failed sources with older controller files

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

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

@@ -137,3 +137,10 @@ 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.

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

@@ -10,7 +10,7 @@ import { nativeToolContractFingerprintForTarget } from "./native-session-resume.
 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 { nativeSha256 } from "./canonical.js";
@@ -21,7 +21,8 @@ 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_success", "stale_busy", "stale_durable",
+  "controller_confirmed", "controller_foreign", "controller_thread", "controller_runner", "controller_success", "controller_busy", "controller_invalid", "controller_durable"])("offline native controller migration (%s)", scenario => {
   const environment = nativeRestartFixture();
   afterEach(() => vi.unstubAllEnvs());
   async function fixture(commandCount = 1, warm = false) {
@@ -299,6 +300,68 @@ 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));
+    }
+    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_durable") {
+      expect(await migrateNativeControllers(f.db, options)).toMatchObject({ unchanged: 1, archivedFailed: 0 });
+    } else if (scenario !== "controller_confirmed") {
+      await expect(migrateNativeControllers(f.db, options)).rejects.toThrow();
+      expect(await f.db.select().from(nativeHarnessBackups)).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(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"));

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

@@ -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")));
+  const priorSnapshot = record(prior?.snapshot), priorIdentity = record(priorSnapshot.identity);
   if (!prior?.finishedAt || prior.createdAt >= source.createdAt || prior.finishedAt > source.createdAt
     || !source.runnerInstanceId || prior.runnerInstanceId !== source.runnerInstanceId
-    || nativeSha256(prior.snapshot) !== nativeSha256(redactRunnerApiPayloads(olderSnapshot))) return false;
+    || 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 }).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 ? 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) {
@@ -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++;