From 72f7b42fe398f87885514c5ee1df3c21fd935db9 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 18:38:50 -0600 Subject: [PATCH] Preserve settled failed controller state during online migration (#19678) ## Thinking Path > - Paperclip controls agent work across web-server releases. > - Volume-free servers need retained controller state in shared storage. > - Successful settled sources already support online migration. > - Some retained failed sources have no accepted result and must stay available for recovery inspection. > - This change copies those sources under task, coordinator and publication fences while unrelated work continues. > - The original files, failed status and recovery warnings remain unchanged. ## Linked Issues or Issue Description Refs #19038. Refs #19631. **What are you trying to improve?** Preserve retained failed controller files without waiting for global idle. **What problem does this solve?** The online producer skips terminal failed controllers. Their files can still be needed for recovery inspection even when no accepted provider artifact proves executable continuation. **What would the improvement look like?** A settled exact source can enter the existing encrypted archive store. An accepted committed failed source can use the existing checkpoint import. Busy, changed, unowned or foreign sources remain blocked. ## What Changed - Add a source owner bound to the selected task session, run, coordinator generation, admitted input and file digests. - Check the task, physical finalizer lock, released coordinator, canonical resume source and active related turns before upload and before publication. - Preserve failed sources through the existing encrypted archive format. Do not treat archived bytes as execution authority or old-writer stop proof. - Allow the existing accepted committed checkpoint import for a finished failed run without changing its status. - Validate admitted input bindings before looking up local controller files. - Document the online preservation path and the final static writer fence. ## Verification - Real PostgreSQL: 125 cases passed across migration, controller recovery and durable backup storage. One added assertion inspected an unadmitted queued fixture; it was corrected while retaining the queued-turn preservation and stale-publication assertions. The final migration rerun passes 51/51; the other three affected suites pass 75/75. - Genuine prepatch RED: online failed-source preservation and accepted committed failed-source import both fail on the previous producer and recovery owners. - New cases cover independent restore after deleting the original source, an expired but non-null coordinator owner, source mutation after upload, a new related queued turn during upload, foreign input binding and an accepted committed failed result. - The full rebased server source and test-tree typecheck passes after restoring its missing ignored workspace SDK link and building that existing dependency. - Module boundaries and whitespace checks pass. - Rebased onto current master; both provider-artifact imports and the new archive owner are preserved. Rebased type and module-boundary checks pass. All six archive cases pass on the exact rebased head (39.68 seconds). All CI gates pass on exact head 81fc974cd939377136b34efd3c0d7939f1dc4562. Fresh Greptile review is 5/5 with no unresolved comments. No serving migration, deletion or deployment is performed by this PR. - Review memory fix: 64 KiB streamed hashes avoid retaining another controller string at each publication fence. A 256 MiB source is hashed three times while the original string remains live under a 512 MiB heap. The old full-string comparison reproduces an OOM; the fixed two-case suite passes. Full source/test-tree typecheck and module boundaries pass on the new code. All six real-PostgreSQL archive cases pass again with streaming enabled (17.31 seconds); ownership, queued-turn and post-upload mutation guards remain covered. ## Risks - This is a copy-only preservation operation. It does not prove the original process has stopped. Final enrollment must arm static ephemeral mode on the original volume, observe the old deployment stopped and repeat data eligibility. - A source change during upload leaves an unpublished pending object. The normal retention owner handles obsolete attempts; the source remains required until a verified current archive exists. - An archive is forensic evidence. It does not make an unaccepted failed provider result resumable. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The host does not expose the exact serving model ID or 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 either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [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 --- ...-existing-workspace-ephemeral-migration.md | 22 +++++ .../native-controller-archive-source.ts | 89 +++++++++++++++++++ .../native-controller-archive.ts | 11 ++- ...e-controller-migration.integration.test.ts | 74 ++++++++++++++- .../native-controller-migration.ts | 46 ++++++++-- .../native-controller-recovery.ts | 3 +- .../native-controller-source-digest.test.ts | 42 +++++++++ .../native-controller-source-digest.ts | 23 +++++ .../native-harness-backup-storage.ts | 9 +- 9 files changed, 306 insertions(+), 13 deletions(-) create mode 100644 server/src/services/native-runtime/native-controller-archive-source.ts create mode 100644 server/src/services/native-runtime/native-controller-source-digest.test.ts create mode 100644 server/src/services/native-runtime/native-controller-source-digest.ts diff --git a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md index 6da3228cbfd..63a525cfb1b 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -343,6 +343,28 @@ Unrelated runs remain active. Existing controller checkpoint identity, artifact provenance, configuration and snapshot checks still apply. The `paperclip.native-controller-prestage.v1` counts describe this phase; neither successful prestaging nor zero deferred controller sources authorizes volume detachment. +The same finite controller-only entry also preserves settled failed sources. +A committed failed result may import its exact checkpoint under the normal +owner checks without changing the failed outcome. A terminal failed controller +that cannot earn executable recovery authority instead gets an encrypted +evidence archive through the existing native harness object owner. + +Online archival verifies the exact retained session pointer, company, agent, +task, admitted source binding, terminal run and released coordinator. It rejects +related queued/running work, checkout, expired-but-retained ownership and a busy +physical finalizer. Source/profile/coordinator fingerprints and original file +bytes are checked at publication, including after the upload. Pending object +records commit before upload; a raced source cannot publish a current archive. +The operation leaves original files, failed status, leases and recovery history +unchanged. The `archivedFailed` count records preservation, never successful +execution or permission to take over an agent. + +The live Cloud path uses online preparation and static ephemeral arming on the +retained volume. Replacement startup must recheck the original bytes after the +old deployment stops. That final check, not an archive copy or process-local +busy map, establishes the cutover boundary. The offline fallback below is for +an independently quiet instance; it is not the live staging delivery path. + The final offline migration still stops all writers, verifies every durable object, imports deferred sources and all other data, and exports only its normal complete eligibility receipt. Qualification starts from the saved pristine diff --git a/server/src/services/native-runtime/native-controller-archive-source.ts b/server/src/services/native-runtime/native-controller-archive-source.ts new file mode 100644 index 00000000000..5101dbe71b5 --- /dev/null +++ b/server/src/services/native-runtime/native-controller-archive-source.ts @@ -0,0 +1,89 @@ +import { and, eq, inArray, or, sql } from "drizzle-orm"; +import { createHash } from "node:crypto"; +import { agentTaskSessions, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { nativeSha256 } from "./canonical.js"; +import { findNativeSessionResumeRun } from "./native-session-resume.js"; +import { readLegacyNativeControllerArchiveDigests, readLegacyNativeControllerArchiveSource } from "./native-controller-migration.js"; +import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; +import { archiveFailedLegacyNativeController } from "./native-controller-archive.js"; +import type { StorageService } from "../../storage/types.js"; + +export type SettledControllerArchiveOwner = { + kind: "settled_controller_archive"; + companyId: string; issueId: string; agentId: string; runId: string; sessionId: string; + selectedSessionId: string; selectedLastRunId: string; + profileDigest: string; coordinatorDigest: string; controllerDigest: string; supervisorDigest: string | null; +}; +const fileDigest = (value: string) => createHash("sha256").update(value).digest("hex"); + +export function nativeControllerArchiveProfileDigest(value: unknown) { + const profile = value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + return nativeSha256({ nativeExecutionInput: profile.nativeExecutionInput ?? null, nativeToolContractFingerprint: profile.nativeToolContractFingerprint ?? null, + sessionCheckpoint: profile.sessionCheckpoint ?? null, nativeControllerRecovery: profile.nativeControllerRecovery ?? null }); +} + +export function nativeControllerArchiveCoordinatorDigest(value: { + phase: string; resultId: string | null; controllerBootId: string | null; controllerGeneration: number; recoveryState: unknown; +}) { + return nativeSha256({ phase: value.phase, resultId: value.resultId, controllerBootId: value.controllerBootId, + controllerGeneration: value.controllerGeneration, recoveryState: value.recoveryState }); +} + +export async function assertSettledControllerArchiveSource(db: Db, owner: SettledControllerArchiveOwner) { + function fail(): never { throw new Error("native_controller_archive_source_unsettled"); } + await db.execute(sql`set local lock_timeout = '1s'`); + if (!await tryLockNativeWorkspaceFinalization(db, owner)) fail(); + const [task] = await db.select().from(issues).where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("update"); + if (!task || task.executionRunId || task.checkoutRunId) fail(); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, owner.companyId), eq(nativeRunFinalizations.issueId, owner.issueId), + eq(nativeRunFinalizations.runId, owner.runId))).for("update"); + if (!coordinator || coordinator.leaseOwner || coordinator.nextAttemptAt + || !(coordinator.phase === "terminal_failure" || coordinator.phase === "committed" && coordinator.resultId) + || nativeControllerArchiveCoordinatorDigest(coordinator) !== owner.coordinatorDigest) fail(); + const [selection] = await db.select().from(agentTaskSessions).where(and(eq(agentTaskSessions.id, owner.selectedSessionId), + eq(agentTaskSessions.companyId, owner.companyId), eq(agentTaskSessions.agentId, owner.agentId), + eq(agentTaskSessions.lastRunId, owner.selectedLastRunId))).for("share"); + const [selectedRun] = await db.select({ createdAt: heartbeatRuns.createdAt }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, owner.selectedLastRunId), eq(heartbeatRuns.companyId, owner.companyId), eq(heartbeatRuns.agentId, owner.agentId), + eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, owner.sessionId))); + const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, owner.runId), + eq(heartbeatRuns.companyId, owner.companyId), eq(heartbeatRuns.agentId, owner.agentId), eq(heartbeatRuns.nativeIssueId, owner.issueId), + eq(heartbeatRuns.nativeSessionId, owner.sessionId), eq(heartbeatRuns.runtimeMode, "native"))).for("share"); + if (!selection || !selectedRun || !run?.finishedAt || run.status !== "failed" + || nativeControllerArchiveProfileDigest(run.runnerProfileJson) !== owner.profileDigest) fail(); + const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId), + inArray(heartbeatRuns.status, ["queued", "running"]), or(eq(heartbeatRuns.nativeIssueId, owner.issueId), + eq(heartbeatRuns.issueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, owner.sessionId)))).limit(1); + if (active) fail(); + const selected = await findNativeSessionResumeRun(db, { companyId: owner.companyId, agentId: owner.agentId, issueId: owner.issueId, + normalizedSessionId: owner.sessionId, beforeCreatedAt: selectedRun.createdAt, includeProgressBarrier: true }); + if (!selected || selected.id !== owner.runId || nativeControllerArchiveProfileDigest(selected.runnerProfileJson) !== owner.profileDigest) fail(); + const material = await readLegacyNativeControllerArchiveDigests(selected); + if (material.controllerDigest !== owner.controllerDigest + || (material.supervisorDigest ?? null) !== owner.supervisorDigest) fail(); +} + +export async function archiveSettledFailedNativeController(db: Db, input: { + selected: { id: string; lastRunId: string; companyId: string; agentId: string; issueId: string; sessionId: string }; + source: NonNullable>>; + storage: StorageService; signal: AbortSignal; +}) { + input.signal.throwIfAborted(); + const { selected, source } = input; + const material = await readLegacyNativeControllerArchiveSource(source); + if (material.controller === undefined) return "source_already_absent" as const; + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, selected.companyId), + eq(nativeRunFinalizations.issueId, selected.issueId), eq(nativeRunFinalizations.runId, source.id))); + if (!coordinator) throw new Error("native_controller_archive_source_unsettled"); + const owner: SettledControllerArchiveOwner = { kind: "settled_controller_archive", companyId: selected.companyId, + issueId: selected.issueId, agentId: selected.agentId, runId: source.id, sessionId: selected.sessionId, + selectedSessionId: selected.id, selectedLastRunId: selected.lastRunId, + profileDigest: nativeControllerArchiveProfileDigest(source.runnerProfileJson), coordinatorDigest: nativeControllerArchiveCoordinatorDigest(coordinator), + controllerDigest: fileDigest(material.controller), supervisorDigest: material.supervisor === undefined ? null : fileDigest(material.supervisor) }; + await db.transaction(tx => assertSettledControllerArchiveSource(tx as unknown as Db, owner)); + input.signal.throwIfAborted(); + await archiveFailedLegacyNativeController(db, { companyId: selected.companyId, issueId: selected.issueId, runId: source.id, + controller: material.controller, supervisor: material.supervisor, storage: input.storage, onlineOwner: owner, signal: input.signal }); + return "archived_failed" as const; +} diff --git a/server/src/services/native-runtime/native-controller-archive.ts b/server/src/services/native-runtime/native-controller-archive.ts index d0602ce24f1..d47eda47aba 100644 --- a/server/src/services/native-runtime/native-controller-archive.ts +++ b/server/src/services/native-runtime/native-controller-archive.ts @@ -6,6 +6,7 @@ import type { Db } from "@paperclipai/db"; import type { StorageService } from "../../storage/types.js"; import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; import { persistNativeHarnessBackup, verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; +import type { SettledControllerArchiveOwner } from "./native-controller-archive-source.js"; type Source = { companyId: string; issueId: string; runId: string; controller: string; supervisor?: string }; const sha256 = (value: string) => createHash("sha256").update(value).digest("hex"); @@ -24,8 +25,14 @@ export async function hasArchivedLegacyNativeController(db: Db, input: Source & } export async function archiveFailedLegacyNativeController(db: Db, input: Source & { - storage: StorageService; writersStopped: true; signal: AbortSignal; + storage: StorageService; writersStopped?: true; onlineOwner?: SettledControllerArchiveOwner; signal: AbortSignal; }) { + if (input.onlineOwner ? input.writersStopped || input.onlineOwner.kind !== "settled_controller_archive" + || input.onlineOwner.companyId !== input.companyId || input.onlineOwner.issueId !== input.issueId + || input.onlineOwner.runId !== input.runId || input.onlineOwner.controllerDigest !== sha256(input.controller) + || input.onlineOwner.supervisorDigest !== (input.supervisor === undefined ? null : sha256(input.supervisor)) : !input.writersStopped) { + throw new Error("native_controller_archive_authority_invalid"); + } if (await hasArchivedLegacyNativeController(db, input)) return; const descriptor = legacyNativeControllerArchive(input); const root = await mkdtemp(join(tmpdir(), "pc-legacy-controller-archive-")); @@ -34,7 +41,7 @@ export async function archiveFailedLegacyNativeController(db: Db, input: Source await writeFile(join(root, "control-plane-state.json"), input.controller, { mode: 0o600 }); if (input.supervisor !== undefined) await writeFile(join(root, "supervisor.json"), input.supervisor, { mode: 0o600 }); await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest: descriptor.scopeDigest, signal: input.signal, - owner: { kind: "offline_migration", writersStopped: input.writersStopped, terminalFailureOnly: true, + owner: input.onlineOwner ?? { kind: "offline_migration", writersStopped: true, terminalFailureOnly: true, companyId: input.companyId, issueId: input.issueId, runId: input.runId } }); } finally { await removeNativeHarnessBackup(root); } } 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 4caf3e39967..43535febb1d 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 @@ -22,7 +22,8 @@ import { createStorageService } from "../../storage/service.js"; import { legacyNativeControllerArchive } from "./native-controller-archive.js"; import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; -describeEmbeddedPostgres.each(["online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed", +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", "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", @@ -90,6 +91,77 @@ describeEmbeddedPostgres.each(["online_success", "online_owned", "online_same_so taskKey: f.issueId, lastRunId: f.runId }).returning(); return { ...f, companyId, agentId, paperclipHome, runtimeRoot, paths, file, source, profile, session, queuedId, identity, snapshot }; } + if (scenario.startsWith("archive_online")) it("preserves a failed source online without changing its execution or unrelated work", async () => { + const f = await fixture(); + await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + await f.db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, f.queuedId)); + if (scenario === "archive_online_commit") { + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, + completionContractId: run.completionContractId!, serverFingerprint: "failed-committed-migration", schemaStatus: "accepted", + resultJson: {}, canonicalSha256: "failed-committed-migration" }).returning(); + await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, + leaseExpiresAt: null, recoveryState: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + } else { + await rm(join(f.paths.root, "provider-pack-authority.json")); + await f.db.update(nativeRunFinalizations).set({ recoveryState: "blocked", failureCode: "native_provider_terminal_failed" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + } + if (scenario === "archive_online_owned") await f.db.update(nativeRunFinalizations).set({ + leaseOwner: "expired-is-still-owned", leaseExpiresAt: new Date(0), + }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (scenario === "archive_online_foreign") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile, + nativeExecutionInput: { ...f.profile.nativeExecutionInput, binding: { ...f.profile.nativeExecutionInput.binding, companyId: randomUUID() } } } }) + .where(eq(heartbeatRuns.id, f.runId)); + const unrelated = await environment.seedRun(randomUUID()); + const [unrelatedBefore] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)); + const [runBefore] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [ownerBefore] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "durable-objects"))); + const originalPut = storage.putFile.bind(storage); + const put = vi.spyOn(storage, "putFile").mockImplementation(async input => { + const result = await originalPut(input); + if (scenario === "archive_online_changed") await writeFile(f.file, f.source + "\n", { mode: 0o600 }); + if (scenario === "archive_online_active") await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId, + agentId: f.agentId, issueId: f.issueId, nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, + status: "queued", runtimeMode: "native" }); + return result; + }); + const counts = await prestageNativeControllers(f.db, { storage, signal: AbortSignal.timeout(30_000) }); + if (["archive_online_owned", "archive_online_changed", "archive_online_active", "archive_online_foreign"].includes(scenario)) { + expect(counts.archivedFailed).toBe(0); + expect(counts.deferred[scenario === "archive_online_foreign" ? "artifact_binding" : "archive_source_unsettled"]).toBe(1); + if (["archive_online_owned", "archive_online_changed"].includes(scenario)) expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true); + const backups = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.runId, f.runId)); + expect(backups.some(backup => backup.status === "current")).toBe(false); + if (scenario === "archive_online_owned") expect(put).not.toHaveBeenCalled(); + if (scenario === "archive_online_changed") expect(put).toHaveBeenCalledOnce(); + if (scenario === "archive_online_active") { + expect(put).toHaveBeenCalledOnce(); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.issueId, f.issueId))).some(run => run.status === "queued")).toBe(true); + } + if (scenario === "archive_online_foreign") expect(put).not.toHaveBeenCalled(); + if (scenario === "archive_online_foreign") await expect(hasLocalNativeControllerObligations(f.db, storage)) + .rejects.toThrow("native_controller_migration_artifact_binding"); + } else if (scenario === "archive_online_commit") { + expect(counts.imported).toBe(1); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed"); + expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).not.toBeNull(); + } else { + expect(counts.archivedFailed).toBe(1); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false); + expect((await prestageNativeControllers(f.db, { storage, signal: AbortSignal.timeout(30_000) })).archivedFailed).toBe(1); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(runBefore); + const descriptor = legacyNativeControllerArchive({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, controller: f.source }); + const replacement = join(f.runtimeRoot, "online-archive-inspection"); + await rm(f.paperclipHome, { recursive: true, force: true }); + 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(f.source); + } + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(ownerBefore); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0]).toEqual(unrelatedBefore); + }); if (scenario.startsWith("online_")) it("prestages only a settled legacy controller while unrelated work continues", async () => { const f = await fixture(); const settle = async (f: Awaited>) => { diff --git a/server/src/services/native-runtime/native-controller-migration.ts b/server/src/services/native-runtime/native-controller-migration.ts index 77d2af449c3..87481b9d683 100644 --- a/server/src/services/native-runtime/native-controller-migration.ts +++ b/server/src/services/native-runtime/native-controller-migration.ts @@ -17,8 +17,10 @@ import type { StorageService } from "../../storage/types.js"; import { durableNativeHarnessStorage } from "./native-harness-backup-storage.js"; import { archiveFailedLegacyNativeController, hasArchivedLegacyNativeController } from "./native-controller-archive.js"; import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js"; +import { archiveSettledFailedNativeController } from "./native-controller-archive-source.js"; +import { nativeControllerSourceDigest } from "./native-controller-source-digest.js"; -type Selected = { id: string; companyId: string; agentId: string; issueId: string; sessionId: string; createdAt: Date }; +type Selected = { id: string; companyId: string; agentId: string; issueId: string; sessionId: string; createdAt: Date; lastRunId: string }; function record(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; @@ -79,7 +81,7 @@ async function* retainedSources(db: Db, signal?: AbortSignal) { for (;;) { signal?.throwIfAborted(); const rows = await db.execute(sql`select s.id, s.company_id as "companyId", s.agent_id as "agentId", - r.native_issue_id as "issueId", r.native_session_id as "sessionId", r.created_at as "createdAt" + r.native_issue_id as "issueId", r.native_session_id as "sessionId", r.created_at as "createdAt", r.id as "lastRunId" from agent_task_sessions s join heartbeat_runs r on r.id = s.last_run_id and r.company_id = s.company_id and r.agent_id = s.agent_id where s.adapter_type = 'paperclip_runner' and r.native_issue_id is not null and r.native_session_id is not null @@ -110,6 +112,9 @@ async function readPrivateSource(file: string, maxBytes: number): Promise>>) { const execution = parseNativeExecutionInput(source.runnerProfileJson.nativeExecutionInput); + if (execution.binding.companyId !== source.companyId || execution.binding.agentId !== source.agentId + || execution.binding.issueId !== source.nativeIssueId || execution.binding.runId !== source.id + || execution.session.normalizedSessionId !== source.nativeSessionId) throw new Error("native_controller_migration_artifact_binding"); const descriptor = nativeControllerMigrationDescriptor(execution, source.runnerProfileJson.nativeToolContractFingerprint); const root = await lstat(descriptor.root).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return undefined; throw error; @@ -123,6 +128,21 @@ async function localPaths(source: NonNullable>>) { + const paths = await localPaths(source); + const controller = await readPrivateSource(paths.controlFile, 256 * 1024 * 1024); + const supervisor = paths.execution.session.lifecyclePolicy.mode === "warm" + ? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined; + return { paths, controller, supervisor }; +} + +export async function readLegacyNativeControllerArchiveDigests(source: NonNullable>>) { + const paths = await localPaths(source); + return { controllerDigest: await nativeControllerSourceDigest(paths.controlFile, 256 * 1024 * 1024), + supervisorDigest: paths.execution.session.lifecyclePolicy.mode === "warm" + ? await nativeControllerSourceDigest(paths.supervisorPath, 4 * 1024 * 1024) : undefined }; +} + function preservedFailedCheckpoint(source: NonNullable>>, sessionId: string, checkpoint: Awaited>) { return source.status === "failed" && source.runnerProfileJson.sessionCheckpoint == null @@ -264,6 +284,7 @@ function onlineControllerSourceDeferral(error: unknown): string | null { native_controller_migration_unsafe_source: "unsafe_source", native_controller_migration_artifact_binding: "artifact_binding", native_controller_migration_artifact_unavailable: "artifact_unavailable", + native_controller_archive_source_unsettled: "archive_source_unsettled", }; if (reasons[error.message]) return reasons[error.message]; if (error.message.startsWith("runner_remote_provider_artifact_incompatible:")) return "artifact_incompatible"; @@ -272,13 +293,13 @@ function onlineControllerSourceDeferral(error: unknown): string | null { return null; } -export async function prestageNativeControllers(db: Db, input: { signal: AbortSignal }) { - const counts = { inspected: 0, imported: 0, unchanged: 0, sourceAlreadyAbsent: 0, deferred: {} as Record }; +export async function prestageNativeControllers(db: Db, input: { signal: AbortSignal; storage?: StorageService }) { + const counts = { inspected: 0, imported: 0, unchanged: 0, archivedFailed: 0, sourceAlreadyAbsent: 0, deferred: {} as Record }; const defer = (reason: string) => { counts.deferred[reason] = (counts.deferred[reason] ?? 0) + 1; }; for await (const { selected, source } of retainedSources(db, input.signal)) { counts.inspected++; input.signal.throwIfAborted(); - const outcome = await db.transaction(async tx => { + let outcome = await db.transaction(async tx => { await tx.execute(sql`set local lock_timeout = '1s'`); const [task] = await tx.select().from(issues).where(and(eq(issues.companyId, selected.companyId), eq(issues.id, selected.issueId))).for("update"); @@ -290,7 +311,7 @@ export async function prestageNativeControllers(db: Db, input: { signal: AbortSi const [current] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, selected.companyId), eq(heartbeatRuns.agentId, selected.agentId), eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.nativeSessionId, selected.sessionId))).for("update"); - if (current?.status !== "succeeded" || !current.finishedAt) return "source_not_successful"; + if (!current || !["succeeded", "failed"].includes(current.status) || !current.finishedAt) return "source_not_successful"; const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, selected.companyId), inArray(heartbeatRuns.status, ["queued", "running"]), or(eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.issueId, selected.issueId), @@ -331,7 +352,18 @@ export async function prestageNativeControllers(db: Db, input: { signal: AbortSi if (error?.code === "55P03" || error?.cause?.code === "55P03") return "source_busy"; throw error; }); - if (outcome === "imported" || outcome === "unchanged") counts[outcome]++; + if (source.status === "failed" && ["owner_not_released", "artifact_unproven", "checkpoint_binding", "source_unproven", "artifact_binding"].includes(outcome)) { + const storage = input.storage ?? durableNativeHarnessStorage(); + if (storage) outcome = await archiveSettledFailedNativeController(db, { selected, source, storage, signal: input.signal }).catch(error => { + input.signal.throwIfAborted(); + const reason = onlineControllerSourceDeferral(error); + if (reason) return reason; + if (error?.code === "55P03" || error?.cause?.code === "55P03") return "source_busy"; + throw error; + }); + } + if (outcome === "archived_failed") counts.archivedFailed++; + else if (outcome === "imported" || outcome === "unchanged") counts[outcome]++; else if (outcome === "source_already_absent") counts.sourceAlreadyAbsent++; else defer(outcome); } diff --git a/server/src/services/native-runtime/native-controller-recovery.ts b/server/src/services/native-runtime/native-controller-recovery.ts index e4fbad449e0..1b27be232a5 100644 --- a/server/src/services/native-runtime/native-controller-recovery.ts +++ b/server/src/services/native-runtime/native-controller-recovery.ts @@ -266,7 +266,8 @@ export async function importNativeControllerRecovery(db: Db, input: { eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId))).for("update"); const profile = record(run?.runnerProfileJson), snapshot = record(profile.sessionCheckpoint), identity = record(snapshot.identity); - if (run?.runtimeMode !== "native" || !(run.status === "succeeded" || input.offline && ["failed", "cancelled", "interrupted", "timed_out"].includes(run.status)) || !run.finishedAt + if (run?.runtimeMode !== "native" || !(run.status === "succeeded" || input.legacyCheckpoint && run.status === "failed" + || input.offline && ["failed", "cancelled", "interrupted", "timed_out"].includes(run.status)) || !run.finishedAt || identity.runId !== input.runId || identity.companyId !== input.companyId || identity.issueId !== input.issueId || identity.agentId !== input.agentId || identity.sessionId !== input.normalizedSessionId || input.supervisor !== undefined && nativeSha256(snapshot) !== nativeSha256(redactRunnerApiPayloads(supervisor.snapshot))) { diff --git a/server/src/services/native-runtime/native-controller-source-digest.test.ts b/server/src/services/native-runtime/native-controller-source-digest.test.ts new file mode 100644 index 00000000000..6f0c8fde069 --- /dev/null +++ b/server/src/services/native-runtime/native-controller-source-digest.test.ts @@ -0,0 +1,42 @@ +import { execFileSync } from "node:child_process"; +import { createHash } from "node:crypto"; +import { link, mkdtemp, open, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, it } from "vitest"; +import { nativeControllerSourceDigest } from "./native-controller-source-digest.js"; + +it("hashes a maximum-size source repeatedly with an original copy retained on a 512 MiB heap", async () => { + const root = await mkdtemp(join(tmpdir(), "pc-controller-digest-")); + try { + const file = join(root, "controller.json"), chunk = Buffer.alloc(1024 * 1024, "x"); + const expected = createHash("sha256"), handle = await open(file, "wx", 0o600); + try { for (let index = 0; index < 256; index++) { await handle.write(chunk); expected.update(chunk); } } + finally { await handle.close(); } + const source = new URL("./native-controller-source-digest.ts", import.meta.url).href; + const code = `import { readFile } from "node:fs/promises"; + import { nativeControllerSourceDigest } from ${JSON.stringify(source)}; + globalThis.retainedController = await readFile(${JSON.stringify(file)}, "utf8"); + const hashes = []; + for (let index = 0; index < 3; index++) hashes.push(await nativeControllerSourceDigest(${JSON.stringify(file)}, 256 * 1024 * 1024)); + console.log(JSON.stringify({ hashes, retainedBytes: globalThis.retainedController.length }));`; + const result = JSON.parse(execFileSync(process.execPath, ["--max-old-space-size=512", "--import", import.meta.resolve("tsx/esm"), + "--input-type=module", "--eval", code], { encoding: "utf8", timeout: 60_000, maxBuffer: 256 * 1024 })); + expect(result).toEqual({ hashes: Array(3).fill(expected.digest("hex")), retainedBytes: 256 * 1024 * 1024 }); + } finally { await rm(root, { recursive: true, force: true }); } +}, 60_000); + +it("keeps private-file and size guards while hashing", async () => { + const root = await mkdtemp(join(tmpdir(), "pc-controller-digest-guards-")); + try { + const file = join(root, "controller"); await writeFile(file, "source", { mode: 0o600 }); + expect(await nativeControllerSourceDigest(join(root, "missing"), 10)).toBeUndefined(); + expect(await nativeControllerSourceDigest(file, 10)).toBe(createHash("sha256").update("source").digest("hex")); + await expect(nativeControllerSourceDigest(file, 5)).rejects.toThrow("native_controller_migration_unsafe_source"); + await expect(nativeControllerSourceDigest(root, 10)).rejects.toThrow("native_controller_migration_unsafe_source"); + const alias = join(root, "alias"); await symlink(file, alias); + await expect(nativeControllerSourceDigest(alias, 10)).rejects.toMatchObject({ code: "ELOOP" }); + await link(file, join(root, "hardlink")); + await expect(nativeControllerSourceDigest(file, 10)).rejects.toThrow("native_controller_migration_unsafe_source"); + } finally { await rm(root, { recursive: true, force: true }); } +}); diff --git a/server/src/services/native-runtime/native-controller-source-digest.ts b/server/src/services/native-runtime/native-controller-source-digest.ts new file mode 100644 index 00000000000..546c84b8e47 --- /dev/null +++ b/server/src/services/native-runtime/native-controller-source-digest.ts @@ -0,0 +1,23 @@ +import { constants } from "node:fs"; +import { open } from "node:fs/promises"; +import { createHash } from "node:crypto"; + +export async function nativeControllerSourceDigest(file: string, maxBytes: number): Promise { + const handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return undefined; + throw error; + }); + if (!handle) return undefined; + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.nlink !== 1 || stat.size > maxBytes) throw new Error("native_controller_migration_unsafe_source"); + const hash = createHash("sha256"); + let bytes = 0; + for await (const chunk of handle.createReadStream({ autoClose: false, highWaterMark: 64 * 1024 })) { + bytes += chunk.length; + if (bytes > maxBytes) throw new Error("native_controller_migration_unsafe_source"); + hash.update(chunk); + } + return hash.digest("hex"); + } finally { await handle.close(); } +} diff --git a/server/src/services/native-runtime/native-harness-backup-storage.ts b/server/src/services/native-runtime/native-harness-backup-storage.ts index 74bdd963e1c..711c72af9a8 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -22,6 +22,7 @@ import { readNativeHarnessBackupSource } from "./native-harness-backup-source.js import { pinNativeProviderArtifactsForBackup, releaseRetiredNativeProviderArtifacts, restoreNativeProviderArtifacts, retainNativeProviderArtifacts, type NativeProviderArtifactReceipt, type NativeProviderArtifactSource } from "./native-provider-artifacts.js"; import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js"; +import type { SettledControllerArchiveOwner } from "./native-controller-archive-source.js"; const SCHEMA = "paperclip.native-harness-object.v1"; type Binding = { companyId: string; scopeDigest: string; id: string }; @@ -96,9 +97,13 @@ export async function nativeHarnessBackupOwner(db: Db, input: { companyId: strin if (!row || !row.leaseOwner && (row.phase !== "committed" || !row.resultId)) fail("owner_missing"); return { ...input, authorityRunId: input.runId, leaseOwner: row!.leaseOwner, generation: row!.controllerGeneration } satisfies NativeRunnerCheckpointFenceOwner; } -async function fence(tx: Parameters[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner, +async function fence(tx: Parameters[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner, inactiveSource?: InactiveSource) { if (inactiveSource && "kind" in owner) fail("prestage_requires_runtime_owner"); + if ("kind" in owner && owner.kind === "settled_controller_archive") { + await (await import("./native-controller-archive-source.js")).assertSettledControllerArchiveSource(tx as unknown as Db, owner); + return; + } const checkpointOwner = "kind" in owner ? null : await lockNativeRunnerCheckpointOwner(tx, owner); const [task] = await tx.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues) .where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("share"); @@ -183,7 +188,7 @@ export async function assertNativeHarnessBackupOwner(db: Db, owner: NativeRunner } export async function persistNativeHarnessBackup(input: { - db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner; + db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner; scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number; inactiveSource?: { normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string }; providerArtifacts?: NativeProviderArtifactSource;