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;