From d1306f338b4595aadcba27832233b2effc67901c Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:08:27 -0600 Subject: [PATCH] fix(runtime): preserve settled failed harness backups online (#19715) ## Thinking Path > - Paperclip coordinates agents, tasks, and their saved execution state. > - A replacement server needs native provider backups in shared storage. > - The online pre-copy owner accepts successful runs with a committed result. > - A sealed backup from a failed or cancelled run can still contain user history. > - The existing owner defers those files even after their source ownership is settled. > - This change adds a separate preservation owner with exact source and publication checks. > - The copy retains the failed outcome and original files while making the backup readable from a fresh home. ## Linked Issues or Issue Description Refs #19137 and #19108 for the existing pre-copy and encrypted backup owners. **What happened?** Online pre-copy reports `source_not_successful` for a sealed legacy backup from a settled failed or cancelled run. The provider files remain tied to the old disk. **Expected behavior** Copy the exact recorded backup when the source has no remaining writer ownership. Preserve the run status, warnings, result, allocation, selected session, and local files. **Steps to reproduce** 1. Create a lease-stamped suspended native backup. 2. Settle its source as a failed run with a terminal-failure coordinator and no owner. 3. Keep an unrelated execution active. 4. Run canonical online pre-copy. The previous owner defers the source. ## What Changed - Add a narrow preservation owner for terminal source files. It checks the physical finalization lock, task, coordinator, run, selected session, source lease, and source fingerprints. - Recheck that owner before the encrypted upload receipt and current-backup publication. - Reuse the finite workspace keepalive during long source checks and defer source-local invalid archives. - Add optional exact-root stamp verification. A valid previous backup cannot authorize a changed current payload. Existing callers keep their fallback behavior. - Bind the archive input to the checked owner and record an activity event after publication. - Keep pending object receipts after an uncertain or failed publication. - Add focused PostgreSQL and filesystem controls. Update the existing migration runbook. ## Verification - Real PostgreSQL baseline on unchanged production: the new positive case fails with `source_not_successful` instead of publishing the backup. - Same case passes after the change and decrypts/restores the exact provider history. - Final head 4b6cf093be: three focused suites pass, 81 tests in 62.26 seconds, including existing storage and stamp consumers. The canonical PostgreSQL wrapper exited zero and completed cleanup. - Controls cover cancelled and failed sources, real committed results, expired nonnull owners, physical owner/lock contention, late source/owner changes, previous-tree fallback, and archive root/scope binding. - Both review regressions fail with the committed pre-fix owners: a real 30-second database idle timeout expires during a 32-second exact-source check, and a stamp-valid invalid archive stops the pass. The fixed owner keeps its transaction live and copies the next valid source. Source hashes and automatic file restoration were recorded for each leg. - Final-head strict server TypeScript and module-boundary checks pass after the master rebase and required Runner/shared declaration rebuild. - No source migration, provider mutation, or archive apply was executed. ## Risks This path does not reconstruct missing backups or prove that missing provider state is recoverable. Such sources remain blockers. It does not take expired owners or report failed work as successful. Source hashing adds bounded disk work while source row locks are held; unrelated executions remain active in the PostgreSQL fixture. Existing archive size limits, authenticated encryption, object readback, and publication fencing remain in place. No schema, API route, scheduler, or rollout-policy change is included. ## Model Used OpenAI Codex (GPT-6 family; exact runtime model ID and context capacity are not exposed in this session), with reasoning, repository editing, code execution, and tool use. ## 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 - [ ] All Paperclip CI gates are green - [ ] 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 | 16 +- .../native-harness-backup-migration.ts | 25 ++- .../native-harness-backup-stamp.ts | 6 +- .../native-harness-backup-storage.ts | 30 ++- ...ss-online-preservation.integration.test.ts | 176 ++++++++++++++++++ .../native-harness-preservation-source.ts | 112 +++++++++++ 6 files changed, 355 insertions(+), 10 deletions(-) create mode 100644 server/src/services/native-runtime/native-harness-online-preservation.integration.test.ts create mode 100644 server/src/services/native-runtime/native-harness-preservation-source.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 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(); + }); +}