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 63a525cfb1b..0a5ed7492cc 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -317,14 +317,26 @@ It starts no scheduler, does not drain tasks, and always reports `enrollmentReady: false`. Check serving memory headroom before running a separate process on a tenant. Keep the deployment on normal releases. -Only a successful, finished source with a released committed coordinator and an -accepted result can publish through this path. At both database fences, hold the +A successful, finished source with a released committed coordinator and an +accepted result publishes through its normal runtime owner. At both database fences, hold the task share lock and recheck its checkout, related queued/running executions, recorded session and Runner, exact backup-stamp lease, and unchanged local manifest and suspended state. A concurrent turn, moved source, revoked stamp or changed owner defers that source. Unrelated executions continue unchanged. No lease, result, task or lifecycle state is rewritten to make a source eligible. +The same entry can preserve sealed failed or cancelled backup files through a +separate source owner. The source must retain its exact selected task/session, +released committed result or terminal-failure coordinator, and recorded backup +stamp. Every publication checks the physical finalization lock, task checkout, +source run and coordinator, source lease, manifest, and payload hashes. A remaining +coordinator or physical owner blocks the copy even after its deadline expires. +Historical PID fields are preserved and are not termination proof. A valid +`previous` tree cannot authenticate a changed `current` tree. Failed publication +retains the pending encrypted object receipt and every original file. The copy +does not change the failed outcome, warnings, result, allocation or session. +It does not prove that an absent backup was recovered or authorize detachment. + Run the controller-only phase separately so it does not re-read provider archives that were already verified: diff --git a/server/src/services/native-runtime/native-harness-backup-migration.ts b/server/src/services/native-runtime/native-harness-backup-migration.ts index 12d55d49757..eb4260c3766 100644 --- a/server/src/services/native-runtime/native-harness-backup-migration.ts +++ b/server/src/services/native-runtime/native-harness-backup-migration.ts @@ -9,6 +9,7 @@ import { nativeHarnessBackupOwner, persistNativeHarnessBackup, verifyDurableNati import { MAX_NATIVE_HARNESS_BYTES, MAX_NATIVE_HARNESS_OBJECT_BYTES, nativeHarnessArchiveSizeLimit } from "./native-harness-backup-archive.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; import { NativeHarnessBackupSourceError, readNativeHarnessBackupSource } from "./native-harness-backup-source.js"; +import { settledNativeHarnessBackupOwner } from "./native-harness-preservation-source.js"; export async function hasLocalNativeHarnessObligations(db: Db): Promise { const rows = await db.execute(sql`select 1 from agent_task_sessions s @@ -181,7 +182,29 @@ export async function prestageNativeHarnessBackups(db: Db, input: { storage: Sto scopeDigest: source.scopeDigest, manifestSha256: source.manifestSha256, signal }); return "current"; } - if (run.status !== "succeeded" || !run.finishedAt) return "source_not_successful"; + if (run.status !== "succeeded" || !run.finishedAt) { + if (!run.finishedAt || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status)) return "source_not_successful"; + const providerLeaseId = await verifiedSourceLease(db, source, run, signal); + if (!providerLeaseId) return "source_unverified"; + try { + const owner = await settledNativeHarnessBackupOwner(db, { companyId: run.companyId, issueId: run.issueId, + runId: source.runId, sessionId: run.sessionId, runnerInstanceId: run.runnerId, scopeDigest: source.scopeDigest, + root: source.root, manifestSha256: source.manifestSha256, providerLeaseId, signal }); + if (!owner) return "source_not_successful"; + await persistNativeHarnessBackup({ db, storage: input.storage, owner, root: source.root, + scopeDigest: source.scopeDigest, signal, maxArchiveBytes: source.archiveLimit }); + return "published"; + } catch (error) { + signal.throwIfAborted(); + if (error instanceof NativeHarnessBackupSourceError || error instanceof Error && ["native_harness_preservation_source_unsettled", + "native_harness_archive_invalid"].includes(error.message) + || ((error as { code?: string; cause?: { code?: string } })?.code + ?? (error as { cause?: { code?: string } })?.cause?.code) === "55P03") { + return deferSource(source.scopeDigest, error instanceof Error ? error.message : "native_harness_source_busy", "source_changed"); + } + throw error; + } + } const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.issueId), eq(nativeRunFinalizations.runId, source.runId))); diff --git a/server/src/services/native-runtime/native-harness-backup-stamp.ts b/server/src/services/native-runtime/native-harness-backup-stamp.ts index 11b84d80492..8168009fba7 100644 --- a/server/src/services/native-runtime/native-harness-backup-stamp.ts +++ b/server/src/services/native-runtime/native-harness-backup-stamp.ts @@ -211,10 +211,14 @@ export function verifyNativeHarnessBackupStamp(value: unknown, expectedProviderL return false; } -export async function verifyNativeHarnessBackupStampAsync(value: unknown, expectedProviderLeaseId: string, signal: AbortSignal): Promise { +export async function verifyNativeHarnessBackupStampAsync(value: unknown, expectedProviderLeaseId: string, signal: AbortSignal, + exactRoot?: string): Promise { const stamp = readNativeHarnessBackupStamp(value, expectedProviderLeaseId); if (!stamp) return false; for (const candidate of backupCandidates(stamp)) { + // A preservation publisher must authenticate the tree it actually copied; + // an intact previous backup cannot authorize a changed current payload. + if (exactRoot !== undefined && resolve(candidate) !== resolve(exactRoot)) continue; signal.throwIfAborted(); try { const directories = matchingBackupDirectories(candidate, stamp); 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 711c72af9a8..6f0612efc18 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -23,6 +23,8 @@ import { pinNativeProviderArtifactsForBackup, releaseRetiredNativeProviderArtifa 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"; +import type { SettledNativeHarnessBackupOwner } from "./native-harness-preservation-source.js"; +import { logActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; const SCHEMA = "paperclip.native-harness-object.v1"; type Binding = { companyId: string; scopeDigest: string; id: string }; @@ -97,9 +99,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 | SettledControllerArchiveOwner, - inactiveSource?: InactiveSource) { +async function fence(tx: Parameters[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner | SettledNativeHarnessBackupOwner, + inactiveSource?: InactiveSource, signal?: AbortSignal) { if (inactiveSource && "kind" in owner) fail("prestage_requires_runtime_owner"); + if ("kind" in owner && owner.kind === "settled_native_harness_backup") { + await (await import("./native-harness-preservation-source.js")).assertSettledNativeHarnessBackupSource(tx as unknown as Db, owner, signal); + return; + } 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; @@ -188,12 +194,14 @@ export async function assertNativeHarnessBackupOwner(db: Db, owner: NativeRunner } export async function persistNativeHarnessBackup(input: { - db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner; + db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner | SettledNativeHarnessBackupOwner; scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number; inactiveSource?: { normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string }; providerArtifacts?: NativeProviderArtifactSource; }): Promise { if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid"); + if ("kind" in input.owner && input.owner.kind === "settled_native_harness_backup" + && (input.scopeDigest !== input.owner.scopeDigest || input.root !== input.owner.root)) fail("preservation_source_changed"); if (input.providerArtifacts) { if ("kind" in input.owner) fail("provider_artifacts_require_runtime_owner"); const receipt = await retainNativeProviderArtifacts({ ...input, owner: input.owner, source: input.providerArtifacts }); @@ -206,10 +214,13 @@ export async function persistNativeHarnessBackup(input: { const manifestBytes = source?.manifestBytes ?? await readFile(join(input.root, "manifest.json")); const providerArtifacts = JSON.parse(manifestBytes.toString("utf8")).providerArtifacts as NativeProviderArtifactReceipt | undefined; const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`; + if ("kind" in input.owner && input.owner.kind === "settled_native_harness_backup" + && manifestSha256 !== input.owner.manifestSha256) fail("preservation_source_changed"); const inactiveSource = input.inactiveSource ? { ...input.inactiveSource, root: input.root, manifestSha256, scopeDigest: input.scopeDigest } : undefined; const binding = { companyId: input.owner.companyId, scopeDigest: input.scopeDigest, id: randomUUID() }; const scratch = await mkdtemp(join(os.tmpdir(), "pc-harness-upload-")); const key = randomBytes(32); + const publications: ActivityPublication[] = []; try { const archive = join(scratch, "archive"), encrypted = join(scratch, "encrypted"); await captureNativeHarnessArchive(input.root, archive, input.signal, input.maxArchiveBytes, providerArtifacts ? ["provider-artifacts"] : []); @@ -223,7 +234,7 @@ export async function persistNativeHarnessBackup(input: { key: key.toString("base64"), iv: iv.toString("base64"), tag: cipher.getAuthTag().toString("base64"), plaintext })) }; if (ref.byteSize > MAX_NATIVE_HARNESS_OBJECT_BYTES) fail("size_limit_exceeded"); await input.db.transaction(async tx => { - await fence(tx, input.owner, inactiveSource); + await fence(tx, input.owner, inactiveSource, input.signal); await tx.insert(nativeHarnessBackups).values({ ...binding, issueId: input.owner.issueId, runId: input.owner.runId, manifestSha256, reference: ref }); if (providerArtifacts) await pinNativeProviderArtifactsForBackup(tx, { companyId: input.owner.companyId, @@ -234,7 +245,7 @@ export async function persistNativeHarnessBackup(input: { byteSize: ref.byteSize, sha256: ref.sha256, signal: input.signal }); await receive(input.storage, ref, undefined, input.signal); await input.db.transaction(async tx => { - await fence(tx, input.owner, inactiveSource); + await fence(tx, input.owner, inactiveSource, input.signal); await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`${ref.companyId}:${ref.scopeDigest}`}, 0))`); const [pending] = await tx.select({ status: nativeHarnessBackups.status }).from(nativeHarnessBackups) .where(eq(nativeHarnessBackups.id, ref.id)).for("update"); @@ -244,12 +255,19 @@ export async function persistNativeHarnessBackup(input: { eq(nativeHarnessBackups.status, "current"), sql`${nativeHarnessBackups.runId} <> ${input.owner.runId}::uuid`, sql`${heartbeatRuns.createdAt} >= (select created_at from heartbeat_runs where id = ${input.owner.runId}::uuid)`)).limit(1); if (successor) fail("owner_superseded"); - await fence(tx, input.owner, inactiveSource); + await fence(tx, input.owner, inactiveSource, input.signal); await tx.update(nativeHarnessBackups).set({ status: "superseded", updatedAt: new Date() }) .where(and(scope(ref), eq(nativeHarnessBackups.status, "current"))); await tx.update(nativeHarnessBackups).set({ status: "current", updatedAt: new Date() }) .where(and(eq(nativeHarnessBackups.id, ref.id), eq(nativeHarnessBackups.status, "pending"))); + if ("kind" in input.owner && input.owner.kind === "settled_native_harness_backup") { + await logActivity(tx as unknown as Db, { companyId: input.owner.companyId, actorType: "system", actorId: "native-harness-preservation", + action: "native_harness_backup_preserved", entityType: "heartbeat_run", entityId: input.owner.runId, + agentId: input.owner.agentId, runId: input.owner.runId, + details: { scopeDigest: ref.scopeDigest, manifestSha256, backupId: ref.id, originalFilesRetained: true } }, publications); + } }); + for (const publication of publications) publishActivity(publication); } finally { key.fill(0); await removeNativeHarnessBackup(scratch); } } diff --git a/server/src/services/native-runtime/native-harness-online-preservation.integration.test.ts b/server/src/services/native-runtime/native-harness-online-preservation.integration.test.ts new file mode 100644 index 00000000000..b334c96f5a6 --- /dev/null +++ b/server/src/services/native-runtime/native-harness-online-preservation.integration.test.ts @@ -0,0 +1,176 @@ +import { createHash, randomUUID } from "node:crypto"; +import { cp, mkdir, readFile, symlink, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { eq } from "drizzle-orm"; +import { agentTaskSessions, createDb, environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { afterEach, expect, it, vi } from "vitest"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js"; +import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js"; +import { createStorageService } from "../../storage/service.js"; +import { buildNativeHarnessBackupManifest } from "./native-session-executor.js"; +import { createNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js"; +import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; +import { prestageNativeHarnessBackups } from "./native-harness-backup-migration.js"; +import { persistNativeHarnessBackup, restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; +import { settledNativeHarnessBackupOwner } from "./native-harness-preservation-source.js"; +import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; +import * as backupStamp from "./native-harness-backup-stamp.js"; +import { nativeHarnessPrestageDatabaseUrl } from "../../maintenance/native-harness-prestage-runtime.js"; + +describeEmbeddedPostgres("online preservation of settled legacy harness backups", () => { + const environment = nativeRestartFixture(); + const homes: string[] = []; + afterEach(async () => { for (const home of homes.splice(0)) await removeNativeHarnessBackup(home); vi.unstubAllEnvs(); }); + async function fixture(options: { base?: string; scopeId?: string; unsafeLink?: boolean } = {}) { + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "51".repeat(32)); + const f = await environment.seedRun(randomUUID()); + const { companyId, agentId, runtimeRoot } = environment.current; + const base = options.base ?? join(runtimeRoot, randomUUID()); + if (!options.base) { homes.push(base); await mkdir(base); } + vi.stubEnv("PAPERCLIP_RUNNER_STATE_DIR", base); + const sessionScopeId = options.scopeId ?? randomUUID(), scopeDigest = createHash("sha256").update(sessionScopeId).digest("hex"); + const root = join(base, scopeDigest, "failover-backups", "current"); + await mkdir(join(root, "codex-home", "sessions"), { recursive: true }); await mkdir(join(root, "runner")); + const saved = { providerSessionId: randomUUID(), history: "surviving failed provider history" }; + await writeFile(join(root, "codex-home", "sessions", "history.json"), JSON.stringify(saved), { mode: 0o600 }); + if (options.unsafeLink) await symlink("/outside-owned-backup", join(root, "codex-home", "outside-link")); + await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ runId: f.runId, lifecycle: "suspended" })); + const normalizedSessionId = randomUUID(), runnerInstanceId = randomUUID(), providerLeaseId = randomUUID(); + const execution = { provider: { kind: "codex" }, binding: { companyId, issueId: f.issueId, runId: f.runId, + agentId, executionWorkspaceId: randomUUID() }, workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } as never; + const manifest = buildNativeHarnessBackupManifest({ backupRoot: root, execution, runnerInstanceId, + providerSessionIdentity: { providerSessionId: saved.providerSessionId }, sourceProviderLeaseId: providerLeaseId }); + await writeFile(join(root, "manifest.json"), JSON.stringify(manifest)); + const stamp = createNativeHarnessBackupStamp({ manifestPath: join(root, "manifest.json"), sessionScopeId, + authorizedProviderLeaseId: providerLeaseId, runnerInstanceId, normalizedSessionId, completedAt: manifest.completedAt }); + await f.db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), nativeSessionId: normalizedSessionId, + runnerInstanceId, processPid: 123456, processStartedAt: new Date(1) }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, + recoveryState: "original_failure_retained" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(issues).set({ status: "done", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + await f.db.insert(agentTaskSessions).values({ companyId, agentId, adapterType: "paperclip_runner", taskKey: f.issueId, lastRunId: f.runId }); + await f.db.insert(environmentLeases).values({ companyId, issueId: f.issueId, heartbeatRunId: f.runId, providerLeaseId, + status: "expired", metadata: { nativeHarnessBackup: stamp } }); + const storage = createStorageService(createLocalDiskStorageProvider(join(runtimeRoot, "durable-objects"))); + return { ...f, companyId, base, root, scopeDigest, storage, saved, stamp, normalizedSessionId, runnerInstanceId, providerLeaseId }; + } + it("preserves failed backup bytes online while retaining failure and unrelated active work", async () => { + const f = await fixture(), active = await environment.seedRun(randomUUID()); + const [before] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [coordinator] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 1, published: 1, current: 0, deferred: {} }); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(before); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(coordinator); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, active.runId)))[0].status).toBe("running"); + expect(await readFile(join(f.root, "codex-home", "sessions", "history.json"), "utf8")).toBe(JSON.stringify(f.saved)); + const restored = join(environment.current.runtimeRoot, randomUUID()); homes.push(restored); + expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: restored })).toBe(true); + expect(JSON.parse(await readFile(join(restored, "current", "codex-home", "sessions", "history.json"), "utf8"))).toEqual(f.saved); + expect((await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)))[0].status).toBe("current"); + }); + it.each([["failed", "committed"], ["cancelled", "terminal_failure"], ["cancelled", "committed"]])("preserves %s source bytes with %s ownership without changing its status", async (status, phase) => { + const f = await fixture(); + await f.db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, f.runId)); + if (phase === "committed") { + 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: "settled-historical-result", schemaStatus: "accepted", + resultJson: { summary: "Original settled result" }, canonicalSha256: "settled-historical-result" }).returning(); + await f.db.update(nativeRunFinalizations).set({ phase, resultId: result.id }).where(eq(nativeRunFinalizations.runId, f.runId)); + } + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toMatchObject({ published: 1, deferred: {} }); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe(status); + }); + it("never takes a nonnull expired coordinator or durable physical owner", async () => { + const f = await fixture(), put = vi.spyOn(f.storage, "putFile"); + await f.db.update(nativeRunFinalizations).set({ leaseOwner: "original-owner", leaseExpiresAt: new Date(0) }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toMatchObject({ published: 0, deferred: { source_not_successful: 1 } }); + await f.db.update(nativeRunFinalizations).set({ leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, nativeWorkspaceFinalizationOwner: { token: "physical-owner" } } }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toMatchObject({ published: 0, deferred: { source_changed: 1 } }); + expect(put).not.toHaveBeenCalled(); + }); + it("respects the physical advisory lock while unrelated work stays active", async () => { + const f = await fixture(), put = vi.spyOn(f.storage, "putFile"); + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + const holder = f.db.transaction(async tx => { expect(await tryLockNativeWorkspaceFinalization(tx as never, f)).toBe(true); entered(); await gate; }); + try { + await ready; + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toMatchObject({ published: 0, deferred: { source_changed: 1 } }); + expect(put).not.toHaveBeenCalled(); + } finally { release(); await holder; } + }); + it.each(["owner", "payload"])("retains the uploaded pending receipt when %s changes before publication", async change => { + const f = await fixture(), put = f.storage.putFile.bind(f.storage); + vi.spyOn(f.storage, "putFile").mockImplementation(async input => { + const result = await put(input); + if (change === "owner") await f.db.update(nativeRunFinalizations).set({ leaseOwner: "new-owner", leaseExpiresAt: new Date(0) }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + else await writeFile(join(f.root, "codex-home", "sessions", "history.json"), "newer surviving files"); + return result; + }); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toMatchObject({ published: 0, deferred: { source_changed: 1 } }); + expect(await f.db.select({ status: nativeHarnessBackups.status }).from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))) + .toEqual([{ status: "pending" }]); + if (change === "owner") expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].leaseOwner).toBe("new-owner"); + else expect(await readFile(join(f.root, "codex-home", "sessions", "history.json"), "utf8")).toBe("newer surviving files"); + }); + it("does not authenticate a changed current payload from an intact previous backup", async () => { + const f = await fixture(), previous = join(f.root, "..", "previous"); + await cp(f.root, previous, { recursive: true }); + await writeFile(join(f.root, "codex-home", "sessions", "history.json"), "changed current payload"); + const signal = AbortSignal.timeout(15_000); + expect(await verifyNativeHarnessBackupStampAsync(f.stamp, f.providerLeaseId, signal)).toBe(true); + expect(await verifyNativeHarnessBackupStampAsync(f.stamp, f.providerLeaseId, signal, f.root)).toBe(false); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal })).toMatchObject({ published: 0 }); + expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]); + }); + it("binds the archive payload and scope to the checked preservation owner", async () => { + const f = await fixture(), signal = AbortSignal.timeout(15_000); + const owner = await settledNativeHarnessBackupOwner(f.db, { ...f, sessionId: f.normalizedSessionId, + manifestSha256: f.stamp.manifestSha256, signal }); + expect(owner).not.toBeNull(); + const put = vi.spyOn(f.storage, "putFile"); + await expect(persistNativeHarnessBackup({ ...f, owner: owner!, root: join(f.root, "..", "previous"), signal })) + .rejects.toThrow("preservation_source_changed"); + await expect(persistNativeHarnessBackup({ ...f, owner: owner!, scopeDigest: "f".repeat(64), signal })) + .rejects.toThrow("preservation_source_changed"); + expect(put).not.toHaveBeenCalled(); + }); + it("keeps the command transaction alive while exact payload verification exceeds thirty seconds", async () => { + const f = await fixture(), db = createDb(nativeHarnessPrestageDatabaseUrl(environment.current.temporary.connectionString)); + const original = backupStamp.verifyNativeHarnessBackupStampAsync; + let slowed = false; + const spy = vi.spyOn(backupStamp, "verifyNativeHarnessBackupStampAsync").mockImplementation(async (...args) => { + if (args[3] !== undefined && !slowed) { slowed = true; await new Promise(resolve => setTimeout(resolve, 32_000)); } + return original(...args); + }); + try { + expect((await db.$client.unsafe("show idle_in_transaction_session_timeout"))[0].idle_in_transaction_session_timeout).toBe("30s"); + expect(await prestageNativeHarnessBackups(db, { storage: f.storage, signal: AbortSignal.timeout(60_000) })) + .toMatchObject({ published: 1, deferred: {} }); + expect(slowed).toBe(true); + } finally { spy.mockRestore(); await db.$client.end({ timeout: 1 }); } + }, 60_000); + it("defers a stamped archive-invalid source and preserves the next eligible source", async () => { + const bad = await fixture({ scopeId: "b", unsafeLink: true }); + const good = await fixture({ base: bad.base, scopeId: "a" }); + expect(bad.scopeDigest < good.scopeDigest).toBe(true); + expect(await prestageNativeHarnessBackups(bad.db, { storage: bad.storage, signal: AbortSignal.timeout(15_000) })) + .toMatchObject({ inspected: 2, published: 1, deferred: { source_changed: 1 } }); + expect(await bad.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, bad.scopeDigest))).toEqual([]); + expect((await good.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, good.scopeDigest)))[0].status).toBe("current"); + }); +}); diff --git a/server/src/services/native-runtime/native-harness-preservation-source.ts b/server/src/services/native-runtime/native-harness-preservation-source.ts new file mode 100644 index 00000000000..de6c36d5dd8 --- /dev/null +++ b/server/src/services/native-runtime/native-harness-preservation-source.ts @@ -0,0 +1,112 @@ +import { createHash } from "node:crypto"; +import { resolve } from "node:path"; +import { and, eq, inArray, or, sql } from "drizzle-orm"; +import { agentTaskSessions, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { adapterExecutionControls } from "../adapter-execution-control.js"; +import { nativeSha256 } from "./canonical.js"; +import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; +import { nativeHarnessStateBase, readNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js"; +import { readNativeHarnessBackupSource } from "./native-harness-backup-source.js"; +import { withWorkspaceLockKeepalive } from "./native-workspace-lock-keepalive.js"; + +type Run = typeof heartbeatRuns.$inferSelect; +type Coordinator = typeof nativeRunFinalizations.$inferSelect; +type Lease = typeof environmentLeases.$inferSelect; +export type SettledNativeHarnessBackupOwner = { + kind: "settled_native_harness_backup"; + companyId: string; issueId: string; agentId: string; runId: string; + sessionId: string; runnerInstanceId: string; scopeDigest: string; root: string; manifestSha256: string; + selectedSessionId: string; selectedLastRunId: string; sourceLeaseId: string; providerLeaseId: string; + runDigest: string; coordinatorDigest: string; leaseDigest: string; +}; +const runDigest = (run: Run) => nativeSha256({ status: run.status, finishedAt: run.finishedAt?.toISOString(), + controllerBootId: run.controllerBootId, controllerLeaseExpiresAt: run.controllerLeaseExpiresAt?.toISOString(), + processPid: run.processPid, processGroupId: run.processGroupId, processStartedAt: run.processStartedAt?.toISOString(), + runnerInstanceId: run.runnerInstanceId, runnerProfileJson: run.runnerProfileJson }); +const coordinatorDigest = (row: Coordinator) => nativeSha256({ phase: row.phase, resultId: row.resultId, + controllerBootId: row.controllerBootId, controllerGeneration: row.controllerGeneration, recoveryState: row.recoveryState }); +const leaseDigest = (row: Lease) => nativeSha256({ heartbeatRunId: row.heartbeatRunId, status: row.status, + releasedAt: row.releasedAt?.toISOString(), cleanupStatus: row.cleanupStatus, providerLeaseId: row.providerLeaseId, + nativeHarnessBackup: row.metadata?.nativeHarnessBackup }); +const settled = (row: Coordinator) => row.leaseOwner === null && row.nextAttemptAt === null + && (row.phase === "terminal_failure" || row.phase === "committed" && row.resultId !== null); +const fail = (): never => { throw new Error("native_harness_preservation_source_unsettled"); }; + +/** Copy sealed historical files without certifying provider termination, changing + * a failed run, or taking an expired coordinator. Every publication rechecks it. */ +export async function settledNativeHarnessBackupOwner(db: Db, input: { + companyId: string; issueId: string; runId: string; sessionId: string; runnerInstanceId: string; + scopeDigest: string; root: string; manifestSha256: string; providerLeaseId: string; signal: AbortSignal; +}): Promise { + const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId), + eq(nativeRunFinalizations.issueId, input.issueId), eq(nativeRunFinalizations.runId, input.runId))); + if (!run || !coordinator || !settled(coordinator)) return null; + const selected = await db.select({ id: agentTaskSessions.id, lastRunId: agentTaskSessions.lastRunId }).from(agentTaskSessions) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, agentTaskSessions.companyId), eq(heartbeatRuns.agentId, agentTaskSessions.agentId), + eq(heartbeatRuns.id, agentTaskSessions.lastRunId))).where(and(eq(agentTaskSessions.companyId, input.companyId), + eq(agentTaskSessions.agentId, run.agentId), eq(agentTaskSessions.adapterType, "paperclip_runner"), + eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.nativeSessionId, input.sessionId))).limit(2); + if (selected.length !== 1 || !selected[0]!.lastRunId) return null; + const leases = await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), + eq(environmentLeases.issueId, input.issueId), eq(environmentLeases.providerLeaseId, input.providerLeaseId), + sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${input.manifestSha256}`)); + const lease = leases.find(row => readNativeHarnessBackupStamp(row.metadata?.nativeHarnessBackup, input.providerLeaseId)?.sessionScopeSha256 === input.scopeDigest); + if (!lease) return null; + const owner: SettledNativeHarnessBackupOwner = { kind: "settled_native_harness_backup", companyId: input.companyId, + issueId: input.issueId, agentId: run.agentId, runId: input.runId, sessionId: input.sessionId, runnerInstanceId: input.runnerInstanceId, + scopeDigest: input.scopeDigest, root: input.root, manifestSha256: input.manifestSha256, + selectedSessionId: selected[0]!.id, selectedLastRunId: selected[0]!.lastRunId, + sourceLeaseId: lease.id, providerLeaseId: input.providerLeaseId, + runDigest: runDigest(run), coordinatorDigest: coordinatorDigest(coordinator), leaseDigest: leaseDigest(lease) }; + await db.transaction(tx => assertSettledNativeHarnessBackupSource(tx as unknown as Db, owner, input.signal)); + return owner; +} + +export async function assertSettledNativeHarnessBackupSource(db: Db, owner: SettledNativeHarnessBackupOwner, + signal = AbortSignal.timeout(300_000)) { + signal.throwIfAborted(); + 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 || !settled(coordinator) || coordinatorDigest(coordinator) !== owner.coordinatorDigest) fail(); + const [selection] = await db.select().from(agentTaskSessions).where(and(eq(agentTaskSessions.companyId, owner.companyId), + eq(agentTaskSessions.agentId, owner.agentId), eq(agentTaskSessions.id, owner.selectedSessionId), + eq(agentTaskSessions.adapterType, "paperclip_runner"), + eq(agentTaskSessions.lastRunId, owner.selectedLastRunId))).for("share"); + const [selectedRun] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId), + eq(heartbeatRuns.agentId, owner.agentId), eq(heartbeatRuns.id, owner.selectedLastRunId), + eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, owner.sessionId))); + const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId), + eq(heartbeatRuns.agentId, owner.agentId), eq(heartbeatRuns.id, owner.runId), eq(heartbeatRuns.nativeIssueId, owner.issueId), + eq(heartbeatRuns.nativeSessionId, owner.sessionId), eq(heartbeatRuns.runnerInstanceId, owner.runnerInstanceId), eq(heartbeatRuns.runtimeMode, "native"))).for("share"); + const [clock] = await db.execute<{ now: Date }>(sql`select clock_timestamp() as now`); + if (!selection || !selectedRun || !run?.finishedAt || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status) + || run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date(clock.now) + || adapterExecutionControls.has(run.id) || run.runnerProfileJson?.nativeWorkspaceFinalizationOwner != null + || runDigest(run) !== owner.runDigest) 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 [lease] = await db.select().from(environmentLeases).where(and(eq(environmentLeases.id, owner.sourceLeaseId), + eq(environmentLeases.companyId, owner.companyId), eq(environmentLeases.issueId, owner.issueId), + eq(environmentLeases.providerLeaseId, owner.providerLeaseId))).for("share"); + const stamp = readNativeHarnessBackupStamp(lease?.metadata?.nativeHarnessBackup, owner.providerLeaseId); + if (!lease || !stamp || leaseDigest(lease) !== owner.leaseDigest || stamp.sessionScopeSha256 !== owner.scopeDigest + || stamp.manifestSha256 !== owner.manifestSha256 || stamp.normalizedSessionId !== owner.sessionId + || stamp.runnerInstanceId !== owner.runnerInstanceId) fail(); + if (!/^[a-f0-9]{64}$/.test(owner.scopeDigest) + || resolve(owner.root) !== resolve(nativeHarnessStateBase(), owner.scopeDigest, "failover-backups", "current")) fail(); + const [settings] = await db.execute<{ idle_timeout_ms: string }>(sql`select + extract(epoch from current_setting('idle_in_transaction_session_timeout')::interval) * 1000 as idle_timeout_ms`); + await withWorkspaceLockKeepalive(db, Number(settings.idle_timeout_ms), async () => { + const source = await readNativeHarnessBackupSource(owner.root); + if (!source || source.state.runId !== owner.runId || source.state.lifecycle !== "suspended" + || `sha256:${createHash("sha256").update(source.manifestBytes).digest("hex")}` !== owner.manifestSha256 + || !await verifyNativeHarnessBackupStampAsync(stamp, owner.providerLeaseId, signal, owner.root)) fail(); + }); +}