From 93e2f348cf23a459b6874be4b132e21f4186d85d Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 4 Nov 2026 17:21:19 -0600 Subject: [PATCH] feat(runtime): pre-copy settled harness history while serving Co-Authored-By: Paperclip --- ...-existing-workspace-ephemeral-migration.md | 36 ++++ server/scripts/bundle-runtime.mjs | 8 +- server/scripts/bundle-runtime.test.mjs | 9 + .../maintenance/native-harness-prestage.ts | 36 ++++ .../native-harness-backup-migration.ts | 177 +++++++++++++----- ...harness-backup-storage.integration.test.ts | 71 ++++++- .../native-harness-backup-storage.ts | 45 ++++- 7 files changed, 323 insertions(+), 59 deletions(-) create mode 100644 server/src/maintenance/native-harness-prestage.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 8ba9c30ed72..90b15c61114 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -214,3 +214,39 @@ source, export-failure cause or unknown recovery kind retains the existing blocker unless the importer already records that exact legacy recovery source. Live migration keeps its recovery exclusion. Recheck recovery evidence before publishing the saved workspace. + +### Online preparation for large provider histories + +Parallel offline transfer alone did not fit the copied staging dataset in the +maintenance window. A resumed attempt reached 1,389 of 1,953 sources. Keep the +offline deadline and move successful released history before draining the server. +The inventory contains 1,802 such backups, approximately 48.5 GB of the 54.8 GB +declared source bytes. This is preparatory publication under recorded runtime +authority, not permission to detach a volume. + +Run the finite operator entry from a normal released image: + +```sh +PAPERCLIP_NATIVE_HARNESS_PRESTAGE=true node --max-old-space-size=512 server/dist-runtime/native-harness-prestage.mjs +``` + +The entry loads the stored Cloud configuration and requires shared object storage. +It processes one source at a time, has a 90-minute deadline, and stops on shutdown. +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 +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 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 +database, records the separate preparation phase, and then proves the normal +bundled offline entry and empty-home restoration. Preparation alone cannot pass +enrollment or active-agent release acceptance. diff --git a/server/scripts/bundle-runtime.mjs b/server/scripts/bundle-runtime.mjs index e36a8c8749a..09a25fe7a72 100644 --- a/server/scripts/bundle-runtime.mjs +++ b/server/scripts/bundle-runtime.mjs @@ -71,11 +71,12 @@ function externalizeThirdParty(isBundled, runtimeDir) { }; } -export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) { +export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, nativeHarnessPrestageEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) { await rm(outdir, { recursive: true, force: true }); const result = await build({ entryPoints: { index: "paperclip-runtime-entry", server: entry, ...(maintenanceEntry ? { "agent-files": maintenanceEntry } : {}), - ...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}) }, + ...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}), + ...(nativeHarnessPrestageEntry ? { "native-harness-prestage": nativeHarnessPrestageEntry } : {}) }, outdir, outExtension: { ".js": ".mjs" }, chunkNames: "chunk-[hash]", @@ -105,7 +106,8 @@ async function main() { const started = performance.now(); const summary = await bundleRuntime({ entry: join(serverDir, "dist", "index.js"), outdir: join(serverDir, "dist-runtime"), maintenanceEntry: join(serverDir, "dist", "maintenance", "agent-files.js"), - ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js") }); + ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js"), + nativeHarnessPrestageEntry: join(serverDir, "dist", "maintenance", "native-harness-prestage.js") }); console.log(`[bundle-runtime] ${summary.inputs} inputs -> ${summary.chunks} chunks in ${Math.round(performance.now() - started)} ms: ${summary.outfile}`); } diff --git a/server/scripts/bundle-runtime.test.mjs b/server/scripts/bundle-runtime.test.mjs index adacbca2871..f0d28cad9d7 100644 --- a/server/scripts/bundle-runtime.test.mjs +++ b/server/scripts/bundle-runtime.test.mjs @@ -89,6 +89,15 @@ describe("bundleRuntime", () => { expect(JSON.parse(stdout)).toEqual({ migrated: 42 }); }); + it("packages online harness preparation as a standalone plain-Node entry", async () => { + const nativeHarnessPrestageEntry = path.join(root, "server", "dist", "prestage.js"); + write(nativeHarnessPrestageEntry, 'import { values } from "@paperclipai/fixture"; console.log(JSON.stringify({ copied: values.doubled, enrollmentReady: false }));'); + const outdir = path.join(root, "server", "dist-runtime"); + await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir, nativeHarnessPrestageEntry }); + const { stdout } = await execFileAsync(process.execPath, [path.join(outdir, "native-harness-prestage.mjs")], { timeout: 30_000 }); + expect(JSON.parse(stdout)).toEqual({ copied: 42, enrollmentReady: false }); + }); + it("bundles workspace TypeScript and runs with plain node", async () => { const outdir = path.join(root, "server", "dist-runtime"); const summary = await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir }); diff --git a/server/src/maintenance/native-harness-prestage.ts b/server/src/maintenance/native-harness-prestage.ts new file mode 100644 index 00000000000..11ce3894d98 --- /dev/null +++ b/server/src/maintenance/native-harness-prestage.ts @@ -0,0 +1,36 @@ +import { closeRegisteredClients, createDb } from "@paperclipai/db"; +import { cloudRuntimeConfigService } from "../services/cloud-runtime-config.js"; +import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js"; +import { isCloudManagedInstance } from "../services/cloud-instance.js"; +import { getStorageService } from "../storage/index.js"; +import { prestageNativeHarnessBackups } from "../services/native-runtime/native-harness-backup-migration.js"; + +// Operator-only finite preparation. No task drain, scheduler, app routes, +// enrollment export, or assertion that other executions have stopped. +if (process.env.PAPERCLIP_NATIVE_HARNESS_PRESTAGE !== "true") throw new Error("Explicit harness pre-copy mode is required"); +const url = process.env.DATABASE_URL; +if (!url) throw new Error("Harness pre-copy requires an explicit database"); +const cancellation = new AbortController(); +const stop = () => cancellation.abort(new Error("native_harness_prestage_stopped")); +const deadline = setTimeout(() => cancellation.abort(new Error("native_harness_prestage_deadline")), 90 * 60_000); +process.once("SIGINT", stop); process.once("SIGTERM", stop); +const db = createDb(url, { maxConnections: 2 }); +try { + if (isCloudManagedInstance()) { + const config = await cloudRuntimeConfigService(db).inspectStored(); + if (!config) throw new Error("cloud_runtime_configuration_missing"); + setCloudRuntimeConfigSnapshot(config); + } + const storage = getStorageService(); + if (storage.provider !== "s3") throw new Error("shared_object_storage_required"); + const counts = await prestageNativeHarnessBackups(db, { storage, signal: cancellation.signal }); + console.log(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", counts, enrollmentReady: false })); +} catch (error) { + const message = error instanceof Error ? error.message : "native_harness_prestage_failed"; + const failure = /^[a-z_]{1,80}$/.test(message) ? message : "native_harness_prestage_failed"; + console.error(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", state: "blocked", failure, enrollmentReady: false })); + process.exitCode = 1; +} finally { + clearTimeout(deadline); process.off("SIGINT", stop); process.off("SIGTERM", stop); + await closeRegisteredClients(url, { scope: "database" }); +} 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 3cb5050b962..5d7ee216fd4 100644 --- a/server/src/services/native-runtime/native-harness-backup-migration.ts +++ b/server/src/services/native-runtime/native-harness-backup-migration.ts @@ -1,11 +1,11 @@ import { createHash } from "node:crypto"; import { lstat, readFile, readdir } from "node:fs/promises"; import { join } from "node:path"; -import { and, eq, sql } from "drizzle-orm"; -import { environmentLeases, heartbeatRuns, nativeHarnessBackups, type Db } from "@paperclipai/db"; +import { and, eq, inArray, or, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; import type { StorageService } from "../../storage/types.js"; import { nativeHarnessStateBase, readNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js"; -import { persistNativeHarnessBackup, verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; +import { nativeHarnessBackupOwner, persistNativeHarnessBackup, verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; import { MAX_NATIVE_HARNESS_BYTES, MAX_NATIVE_HARNESS_OBJECT_BYTES, nativeHarnessArchiveSizeLimit } from "./native-harness-backup-archive.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; @@ -61,16 +61,34 @@ async function prepareSource(base: string, scopeDigest: string) { type Source = NonNullable>>; -async function migrateSource(db: Db, input: { storage: StorageService; signal: AbortSignal }, source: Source) { - input.signal.throwIfAborted(); - const { root, scopeDigest, manifestSha256 } = source; +async function sourceRun(db: Db, source: Source) { const [run] = await db.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId, - sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns) + sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId, + status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt }).from(heartbeatRuns) .where(eq(heartbeatRuns.id, source.runId)); - if (!run) return 0; - if (!run.issueId || run.sessionId !== source.sessionId || run.runnerId !== source.runnerId) { + if (!run) return null; + if (!run.issueId || !run.sessionId || !run.runnerId || run.sessionId !== source.sessionId || run.runnerId !== source.runnerId) { throw new Error("native_harness_migration_binding_mismatch"); } + return { ...run, issueId: run.issueId, sessionId: run.sessionId, runnerId: run.runnerId }; +} + +async function verifiedSourceLease(db: Db, source: Source, run: NonNullable>>, signal: AbortSignal) { + const leases = await db.select({ stamp: sql`${environmentLeases.metadata}->'nativeHarnessBackup'`, + providerLeaseId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.issueId, run.issueId), + sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${source.manifestSha256}`)); + for (const lease of leases) { + const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? ""); + if (stamp?.sessionScopeSha256 === source.scopeDigest && await verifyNativeHarnessBackupStampAsync(stamp, lease.providerLeaseId!, signal)) return lease.providerLeaseId!; + } + return null; +} + +async function migrateSource(db: Db, input: { storage: StorageService; signal: AbortSignal }, source: Source) { + input.signal.throwIfAborted(); + const { root, scopeDigest, manifestSha256 } = source, run = await sourceRun(db, source); + if (!run) return 0; const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId), eq(nativeHarnessBackups.scopeDigest, scopeDigest), eq(nativeHarnessBackups.status, "current"))); if (current) { @@ -79,57 +97,120 @@ async function migrateSource(db: Db, input: { storage: StorageService; signal: A scopeDigest, manifestSha256: current.manifestSha256, signal: input.signal }); return 1; } - const leases = await db.select({ stamp: sql`${environmentLeases.metadata}->'nativeHarnessBackup'`, - providerLeaseId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and( - eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.issueId, run.issueId), - sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${manifestSha256}`)); - let valid = false; - for (const lease of leases) { - const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? ""); - if (stamp?.sessionScopeSha256 === scopeDigest && await verifyNativeHarnessBackupStampAsync(stamp, lease.providerLeaseId!, input.signal)) { valid = true; break; } - } - if (!valid) throw new Error("native_harness_migration_backup_unverified"); + if (!await verifiedSourceLease(db, source, run, input.signal)) throw new Error("native_harness_migration_backup_unverified"); await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest, signal: input.signal, maxArchiveBytes: source.archiveLimit, owner: { kind: "offline_migration", writersStopped: true, companyId: run.companyId, issueId: run.issueId, runId: source.runId } }); return 1; } +async function mapSources(input: { signal: AbortSignal; concurrency?: number; + unavailableSource?: (error: unknown) => T | undefined }, visit: (source: Source, signal: AbortSignal) => Promise) { + const base = nativeHarnessStateBase(); + const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return [] as string[]; throw error; + }); + let reserved = 0, wave: Source[] = []; + const results: T[] = []; + const flush = async () => { + const cancellation = new AbortController(); + const signal = AbortSignal.any([input.signal, cancellation.signal]); + const settled = await Promise.allSettled(wave.map(async source => { + try { return await visit(source, signal); } + catch (error) { cancellation.abort(error); throw error; } + })); + // Never return failure while a peer can still upload or publish. Partial + // completed transfers remain valid for the next offline attempt. + if (cancellation.signal.aborted) throw cancellation.signal.reason; + for (const result of settled) { + if (result.status === "rejected") throw result.reason; + results.push(result.value); + } + wave = []; reserved = 0; + }; + for (const scopeDigest of directories.filter(name => /^[a-f0-9]{64}$/.test(name)).sort()) { + input.signal.throwIfAborted(); + let source; + try { source = await prepareSource(base, scopeDigest); } + catch (error) { + input.signal.throwIfAborted(); + const deferred = input.unavailableSource?.(error); + if (deferred === undefined) throw error; + results.push(deferred); continue; + } + if (!source) continue; + const reservation = 2 * source.archiveLimit; + if (wave.length && (wave.length === (input.concurrency ?? MAX_PARALLEL_BACKUPS) || reserved + reservation > MAX_SCRATCH_BYTES)) await flush(); + wave.push(source); reserved += reservation; + } + if (wave.length) await flush(); + return results; +} + export async function migrateNativeHarnessBackups(db: Db, input: { storage: StorageService; writersStopped: true; signal: AbortSignal; }): Promise { if (!input.writersStopped) throw new Error("native_harness_migration_requires_offline_writers"); const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1); if (active) throw new Error("native_harness_migration_requires_offline_writers"); - const base = nativeHarnessStateBase(); - const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => { - if (error.code === "ENOENT") return [] as string[]; throw error; + const counts = await mapSources(input, (source, signal) => migrateSource(db, { storage: input.storage, signal }, source)); + return counts.reduce((total, count) => total + count, 0); +} + +/** Copy settled successful history using its released runtime owner. This is + * preparatory work, never an offline-writer assertion or enrollment receipt. */ +export async function prestageNativeHarnessBackups(db: Db, input: { storage: StorageService; signal: AbortSignal }) { + // Keep preparatory work below offline throughput so the serving process can + // continue handling real work. This finite operation has no idle schedule. + const outcomes = await mapSources({ ...input, concurrency: 1, unavailableSource: error => + (error as NodeJS.ErrnoException)?.code === "ENOENT" || error instanceof Error && error.message === "native_harness_migration_source_unsettled" + ? "source_unavailable" : undefined }, async (source, signal) => { + signal.throwIfAborted(); + const run = await sourceRun(db, source).catch(error => { + if (error instanceof Error && error.message === "native_harness_migration_binding_mismatch") return null; + throw error; + }); + if (!run) return "missing_run"; + const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId), + eq(nativeHarnessBackups.scopeDigest, source.scopeDigest), eq(nativeHarnessBackups.status, "current"))); + if (current) { + if (current.manifestSha256 !== source.manifestSha256) return "durable_source_changed"; + await verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: run.companyId, + scopeDigest: source.scopeDigest, manifestSha256: source.manifestSha256, signal }); + return "current"; + } + if (run.status !== "succeeded" || !run.finishedAt) return "source_not_successful"; + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.issueId), + eq(nativeRunFinalizations.runId, source.runId))); + if (coordinator?.phase !== "committed" || coordinator.leaseOwner || !coordinator.resultId || !coordinator.controllerBootId) return "owner_not_released"; + const [task] = await db.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues) + .where(and(eq(issues.companyId, run.companyId), eq(issues.id, run.issueId))); + const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), inArray(heartbeatRuns.status, ["queued", "running"]), + or(eq(heartbeatRuns.nativeIssueId, run.issueId), eq(heartbeatRuns.issueId, run.issueId), eq(heartbeatRuns.nativeSessionId, run.sessionId)))).limit(1); + if (!task || task.executionRunId || task.checkoutRunId || active) return "source_active"; + const providerLeaseId = await verifiedSourceLease(db, source, run, signal); + if (!providerLeaseId) return "source_unverified"; + try { + const owner = await nativeHarnessBackupOwner(db, { companyId: run.companyId, issueId: run.issueId, + runId: source.runId, bootId: coordinator.controllerBootId }); + await persistNativeHarnessBackup({ db, storage: input.storage, root: source.root, scopeDigest: source.scopeDigest, + owner, signal, maxArchiveBytes: source.archiveLimit, + inactiveSource: { normalizedSessionId: run.sessionId, runnerInstanceId: run.runnerId, providerLeaseId } }); + return "published"; + } catch (error) { + signal.throwIfAborted(); + if ((error as NodeJS.ErrnoException)?.code === "ENOENT" || error instanceof Error && ["native_runner_checkpoint_owner_lost", "native_harness_storage_owner_missing", + "native_harness_storage_prestage_source_active", "native_harness_storage_prestage_source_changed", + "native_harness_storage_owner_superseded", "native_harness_archive_invalid"].includes(error.message)) return "source_changed"; + throw error; + } }); - let count = 0, reserved = 0, wave: Source[] = []; - const flush = async () => { - const cancellation = new AbortController(); - const signal = AbortSignal.any([input.signal, cancellation.signal]); - const settled = await Promise.allSettled(wave.map(async source => { - try { return await migrateSource(db, { ...input, signal }, source); } - catch (error) { cancellation.abort(error); throw error; } - })); - // Never return failure while a peer can still upload or publish. Partial - // completed transfers remain valid for the next offline attempt. - if (cancellation.signal.aborted) throw cancellation.signal.reason; - for (const result of settled) { - if (result.status === "rejected") throw result.reason; - count += result.value; - } - wave = []; reserved = 0; - }; - for (const scopeDigest of directories.filter(name => /^[a-f0-9]{64}$/.test(name)).sort()) { - input.signal.throwIfAborted(); - const source = await prepareSource(base, scopeDigest); - if (!source) continue; - const reservation = 2 * source.archiveLimit; - if (wave.length && (wave.length === MAX_PARALLEL_BACKUPS || reserved + reservation > MAX_SCRATCH_BYTES)) await flush(); - wave.push(source); reserved += reservation; + const counts = { inspected: outcomes.length, published: 0, current: 0, deferred: {} as Record }; + for (const outcome of outcomes) { + if (outcome === "published" || outcome === "current") counts[outcome]++; + else counts.deferred[outcome] = (counts.deferred[outcome] ?? 0) + 1; } - if (wave.length) await flush(); - return count; + return counts; } diff --git a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts index c66b503e2dd..6474ed3381b 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts @@ -12,7 +12,7 @@ import { nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNat import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; import { buildNativeHarnessBackupManifest } from "./native-session-executor.js"; import { createNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js"; -import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups } from "./native-harness-backup-migration.js"; +import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups, prestageNativeHarnessBackups } from "./native-harness-backup-migration.js"; import { nativeHarnessBackupExpirationService } from "./native-harness-backup-expiration.js"; import { legacyControllerBootId } from "../legacy-controller-lease.js"; @@ -67,7 +67,7 @@ describeEmbeddedPostgres("durable native provider backups", () => { await f.db.update(heartbeatRuns).set({ nativeSessionId: f.identity.normalizedSessionId, runnerInstanceId }).where(eq(heartbeatRuns.id, f.runId)); await f.db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId, providerLeaseId, status: "expired", metadata: { nativeHarnessBackup: stamp } }); - return { ...f, home, root }; + return { ...f, home, root, runnerInstanceId, providerLeaseId }; } async function offlineFixtures(count: number) { const base = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(base); await mkdir(base); @@ -179,6 +179,73 @@ describeEmbeddedPostgres("durable native provider backups", () => { leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); return result; } + async function inactiveFixture() { + const [f] = await offlineFixtures(1); + await commitWarmRun(f); + await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + const owner = await nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, bootId: environment.current.successor.bootId }); + return { ...f, owner, inactiveSource: { normalizedSessionId: f.identity.normalizedSessionId, + runnerInstanceId: f.runnerInstanceId, providerLeaseId: f.providerLeaseId } }; + } + it("pre-copies released history and restores its bytes while an unrelated execution stays active", async () => { + const f = await inactiveFixture(), active = await environment.seedRun(randomUUID()); + const input = { storage: f.storage, signal: AbortSignal.timeout(15_000) }; + expect(await prestageNativeHarnessBackups(f.db, input)).toEqual({ inspected: 1, published: 1, current: 0, deferred: {} }); + expect(await prestageNativeHarnessBackups(f.db, input)).toEqual({ inspected: 1, published: 0, current: 1, deferred: {} }); + await expect(migrateNativeHarnessBackups(f.db, { ...input, writersStopped: true })).rejects.toThrow("requires_offline_writers"); + expect((await active.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, active.runId)))[0].status).toBe("running"); + await removeNativeHarnessBackup(f.home); + const replacement = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(replacement); + expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement })).toBe(true); + expect(JSON.parse(await readFile(join(replacement, "current", "codex-home", "sessions", "history.json"), "utf8"))).toEqual(f.identity); + }); + it.each(["queued", "running"] as const)("defers pre-copy when the same task has a %s execution", async status => { + const f = await inactiveFixture(), put = vi.spyOn(f.storage, "putFile"); + await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId, agentId: environment.current.agentId, + issueId: f.issueId, nativeIssueId: f.issueId, nativeSessionId: f.identity.normalizedSessionId, status, runtimeMode: "native" }); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_active: 1 } }); + await expect(persistNativeHarnessBackup(f)).rejects.toThrow("native_harness_storage_prestage_source_active"); + expect(put).not.toHaveBeenCalled(); + expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]); + }); + it("defers failed history without treating pre-copy as offline authority", async () => { + const [f] = await offlineFixtures(1); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_not_successful: 1 } }); + expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]); + }); + it("defers a source being replaced online while offline migration still refuses it", async () => { + const f = await inactiveFixture(); + await writeFile(join(f.root, "runner", "runner-state.json"), JSON.stringify({ runId: f.runId, lifecycle: "running" })); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_unavailable: 1 } }); + await expect(migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) })) + .rejects.toThrow("native_harness_migration_source_unsettled"); + }); + it.each(["source_advanced", "queued_turn", "lease_revoked", "owner_changed"])("keeps the published source when %s happens during pre-copy upload", async change => { + const f = await inactiveFixture(); await persistNativeHarnessBackup(f); + const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); + const put = f.storage.putFile.bind(f.storage); + vi.spyOn(f.storage, "putFile").mockImplementation(async input => { + const result = await put(input); + if (change === "source_advanced") await writeFile(join(f.root, "manifest.json"), (await readFile(join(f.root, "manifest.json"), "utf8")) + "\n"); + if (change === "queued_turn") await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId, + agentId: environment.current.agentId, issueId: f.issueId, status: "queued" }); + if (change === "lease_revoked") await f.db.update(environmentLeases).set({ metadata: {} }) + .where(eq(environmentLeases.providerLeaseId, f.providerLeaseId)); + if (change === "owner_changed") await f.db.update(nativeRunFinalizations).set({ controllerGeneration: 2 }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + return result; + }); + await expect(persistNativeHarnessBackup(f)).rejects.toThrow(change === "owner_changed" ? "native_runner_checkpoint_owner_lost" + : change === "queued_turn" ? "native_harness_storage_prestage_source_active" : "native_harness_storage_prestage_source_changed"); + const rows = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); + expect(rows.filter(row => row.status === "current").map(row => row.id)).toEqual([before.id]); + expect(rows.filter(row => row.status === "pending")).toHaveLength(1); + expect(await readFile(join(f.root, "manifest.json"), "utf8")).toContain("paperclip.native-harness-backup.v1"); + }); it.each(["idle", "preparing", "native"])("preserves a released committed warm session with a %s successor", async state => { const f = await fixture(); const result = await commitWarmRun(f); 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 8a9880f8785..c235e620d62 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -6,7 +6,7 @@ import { join } from "node:path"; import os from "node:os"; import { pipeline } from "node:stream/promises"; import { Transform } from "node:stream"; -import { and, eq, inArray, sql } from "drizzle-orm"; +import { and, eq, inArray, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import type { StorageService } from "../../storage/types.js"; @@ -28,6 +28,8 @@ type Reference = Binding & { type Row = typeof nativeHarnessBackups.$inferSelect; type OfflineOwner = { kind: "offline_migration"; companyId: string; issueId: string; runId: string; writersStopped: true; terminalFailureOnly?: true }; +type InactiveSource = { root: string; manifestSha256: string; scopeDigest: string; + normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string }; const fail = (reason: string): never => { throw new Error(`native_harness_storage_${reason}`); }; const aad = (binding: Binding) => Buffer.from(JSON.stringify([SCHEMA, binding.companyId, binding.scopeDigest, binding.id])); const scope = (binding: Pick) => and( @@ -89,9 +91,11 @@ 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, + inactiveSource?: InactiveSource) { + if (inactiveSource && "kind" in owner) fail("prestage_requires_runtime_owner"); const checkpointOwner = "kind" in owner ? null : await lockNativeRunnerCheckpointOwner(tx, owner); - const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) + 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"); const authorityRunId = "kind" in owner ? owner.runId : owner.authorityRunId; if (!task) fail("owner_superseded"); @@ -138,15 +142,44 @@ async function fence(tx: Parameters[0]>[0], owner: if (owner.leaseOwner === null && !committed) fail("owner_missing"); if (task.executionRunId && task.executionRunId !== authorityRunId && !committed) fail("owner_superseded"); await validateNativeRunnerCheckpointOwner(tx, owner, owned, false); + if (inactiveSource) { + if (!committed || task.executionRunId || task.checkoutRunId) fail("prestage_source_active"); + const [active] = await tx.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, inactiveSource.normalizedSessionId)))).limit(1); + if (active) fail("prestage_source_active"); + const [binding] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, owner.authorityRunId), eq(heartbeatRuns.companyId, owner.companyId), + eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, inactiveSource.normalizedSessionId), + eq(heartbeatRuns.runnerInstanceId, inactiveSource.runnerInstanceId))); + const [lease] = await tx.select({ stamp: sql`${environmentLeases.metadata}->'nativeHarnessBackup'` }) + .from(environmentLeases).where(and(eq(environmentLeases.companyId, owner.companyId), + eq(environmentLeases.issueId, owner.issueId), eq(environmentLeases.providerLeaseId, inactiveSource.providerLeaseId), + sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${inactiveSource.manifestSha256}`)).limit(1); + const stamp = readNativeHarnessBackupStamp(lease?.stamp, inactiveSource.providerLeaseId); + if (!binding || stamp?.sessionScopeSha256 !== inactiveSource.scopeDigest + || stamp.normalizedSessionId !== inactiveSource.normalizedSessionId || stamp.runnerInstanceId !== inactiveSource.runnerInstanceId) { + fail("prestage_source_changed"); + } + // The task share lock serializes publication with execution checkout. A + // turn that ran during upload also changes the immutable backup manifest. + const manifest = await readFile(join(inactiveSource.root, "manifest.json")); + const state = JSON.parse(await readFile(join(inactiveSource.root, "runner", "runner-state.json"), "utf8")); + if (`sha256:${createHash("sha256").update(manifest).digest("hex")}` !== inactiveSource.manifestSha256 + || state.runId !== owner.authorityRunId || state.lifecycle !== "suspended") fail("prestage_source_changed"); + } } export async function persistNativeHarnessBackup(input: { db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner; scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number; + inactiveSource?: { normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string }; }): Promise { if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid"); const manifestBytes = await readFile(join(input.root, "manifest.json")); const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`; + 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); @@ -163,7 +196,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); + await fence(tx, input.owner, inactiveSource); await tx.insert(nativeHarnessBackups).values({ ...binding, issueId: input.owner.issueId, runId: input.owner.runId, manifestSha256, reference: ref }); }); @@ -172,7 +205,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); + await fence(tx, input.owner, inactiveSource); 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"); @@ -182,7 +215,7 @@ 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); + await fence(tx, input.owner, inactiveSource); 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() })