diff --git a/doc/plans/2026-11-07-db-only-failed-controller-preservation.md b/doc/plans/2026-11-07-db-only-failed-controller-preservation.md new file mode 100644 index 00000000000..4538695f101 --- /dev/null +++ b/doc/plans/2026-11-07-db-only-failed-controller-preservation.md @@ -0,0 +1,26 @@ +# Preserve database-only failed controller evidence + +A retained failed run can already have a complete encrypted controller checkpoint +and its original inline session snapshot while both legacy controller files are +absent. This is preserved evidence. It does not become successful or executable +recovery authority merely because the instance changes storage mode. + +The migration owner accepts this case only after the canonical checkpoint decoder +validates every record and its company/issue/run encryption binding. The selected +run, admitted execution and inline snapshot must match company, agent, issue, run +and normalized session. The decoded checkpoint and protocol header must also match +the run, Runner instance and normalized session. The exact controller and supervisor +paths must both be absent. A remaining file, symbolic link, unsafe path, invalid +binding or failed read keeps the obligation blocked. + +The earlier failed checkpoint case with no inline snapshot keeps its existing +predicate. The new case does not replace an inline snapshot, create recovery +metadata, change a failed status or coordinator, settle an action, or move a task +pointer. Present legacy payloads still need the existing preservation path. + +The focused regression uses the existing encrypted PostgreSQL checkpoint fixture. +It first reproduces the startup blocker. It then verifies the absent-files case, +remaining controller and supervisor files, foreign inline identities and foreign +checkpoint Runner/session identities. Existing null-snapshot and online/offline +archive controls run with it. The original queued turn, run, coordinator and +checkpoint rows stay unchanged, and failed evidence stays non-recoverable. diff --git a/server/src/services/native-runtime/native-controller-migration.integration.test.ts b/server/src/services/native-runtime/native-controller-migration.integration.test.ts index 43535febb1d..a58ee298adb 100644 --- a/server/src/services/native-runtime/native-controller-migration.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-migration.integration.test.ts @@ -23,7 +23,7 @@ import { legacyNativeControllerArchive } from "./native-controller-archive.js"; import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archive_online_changed", "archive_online_commit", "archive_online_active", "archive_online_foreign", - "online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed", + "online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed", "durable_inline_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", @@ -331,6 +331,60 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv expect(await readFile(f.file, "utf8")).toBe(f.source); } finally { f.db.$client.options.debug = previousDebug; } }, 60_000); + if (scenario === "durable_inline_failed") it("preserves exact DB-only failed inline evidence while retaining every local payload and binding guard", async () => { + const f = await fixture(), binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId }; + const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "unused-durable-objects"))); + const put = vi.spyOn(storage, "putFile"); + const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, { runId: f.runId, + runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId }, + { version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` }); + await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, checkpoint)); + await f.db.update(nativeRunFinalizations).set({ recoveryState: "blocked", failureCode: "original_failure" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true); + await rm(f.file); + await mkdir(dirname(f.paths.supervisorPath), { recursive: true }); + await writeFile(f.paths.supervisorPath, "retained newer supervisor\n"); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true); + expect(await readFile(f.paths.supervisorPath, "utf8")).toBe("retained newer supervisor\n"); + await rm(f.paths.supervisorPath); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false); + for (const field of ["companyId", "agentId", "issueId", "runId", "sessionId"] as const) { + const profile = { ...f.profile, sessionCheckpoint: { ...f.snapshot, identity: { ...f.snapshot.identity, [field]: randomUUID() } } }; + await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true); + } + await f.db.update(heartbeatRuns).set({ runnerProfileJson: f.profile }).where(eq(heartbeatRuns.id, f.runId)); + for (const field of ["runnerInstanceId", "normalizedSessionId"] as const) { + const state = JSON.parse(f.source); state.identity[field] = randomUUID(); + const foreign = 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.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, foreign)); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true); + } + await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, checkpoint)); + 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 rows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + expect(await migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000), storage })) + .toMatchObject({ preservedFailed: 1, imported: 0, archivedFailed: 0 }); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false); + 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(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(rows); + 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(heartbeatRuns).where(eq(heartbeatRuns.id, f.queuedId)))[0].status).toBe("queued"); + expect(put).not.toHaveBeenCalled(); + expect((await loadNativeRunnerCheckpoint(f.db, binding))?.records).toEqual(checkpoint.records); + await f.db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId }) + .where(eq(heartbeatRuns.id, f.queuedId)); + const recovery = await resolveNativeControllerRecovery(f.db, { companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, normalizedSessionId: f.native.normalizedSessionId, currentRunId: f.queuedId }); + expect(recovery).toEqual({ kind: "unsupported", reason: "metadata_format" }); + }); if (scenario === "durable_failed") it("preserves a failed durable checkpoint without inventing a missing session snapshot", async () => { const f = await fixture(); const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId }; diff --git a/server/src/services/native-runtime/native-controller-migration.ts b/server/src/services/native-runtime/native-controller-migration.ts index 87481b9d683..c4997488c63 100644 --- a/server/src/services/native-runtime/native-controller-migration.ts +++ b/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, 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 { captureLegacyDurablePrpCheckpoint, inspectDurablePrpCheckpointHeader, 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"; @@ -150,6 +150,31 @@ function preservedFailedCheckpoint(source: NonNullable>>, + checkpoint: Awaited>) { + const snapshot = record(source.runnerProfileJson.sessionCheckpoint), identity = record(snapshot.identity); + if (source.status !== "failed" || source.runtimeMode !== "native" || !checkpoint + || source.companyId !== selected.companyId || source.agentId !== selected.agentId + || source.nativeIssueId !== selected.issueId || source.nativeSessionId !== selected.sessionId + || identity.companyId !== selected.companyId || identity.agentId !== selected.agentId + || identity.issueId !== selected.issueId || identity.runId !== source.id || identity.sessionId !== selected.sessionId + || typeof snapshot.sessionId !== "string" || !snapshot.sessionId.trim() + || checkpoint.runId !== source.id || checkpoint.recoveryEvidence.runnerInstanceId !== source.runnerInstanceId + || checkpoint.recoveryEvidence.normalizedSessionId !== selected.sessionId) return false; + const header = inspectDurablePrpCheckpointHeader(checkpoint.records.get("header") ?? ""); + if (header?.runId !== source.id || header.runnerInstanceId !== source.runnerInstanceId + || header.normalizedSessionId !== selected.sessionId) return false; + const paths = await localPaths(source); + for (const file of [paths.controlFile, paths.supervisorPath]) { + const present = await lstat(file).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return undefined; throw error; + }); + if (present) return false; + } + return true; +} + export async function hasLocalNativeControllerObligations(db: Db, storage?: StorageService | null): Promise { const roots = await readdir(nativeHarnessStateBase()).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return [] as string[]; throw error; @@ -165,6 +190,8 @@ export async function hasLocalNativeControllerObligations(db: Db, storage?: Stor const present = await lstat(paths.controlFile).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return undefined; throw error; }); + if (!present && source.status === "failed" && source.runnerProfileJson.sessionCheckpoint != null + && await preservedDatabaseOnlyFailedCheckpoint(selected, source, await loadNativeRunnerCheckpoint(db, binding))) continue; if (present && source.status === "failed") { const archiveStorage = storage === undefined ? durableNativeHarnessStorage() : storage; const controller = await readPrivateSource(paths.controlFile, 256 * 1024 * 1024); @@ -234,6 +261,7 @@ 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; } + if (await preservedDatabaseOnlyFailedCheckpoint(selected, source, checkpoint)) { counts.preservedFailed++; continue; } 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; }