From b96e80dbb3c56c0d7f9ab686def4d0aab63c3673 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 16:17:43 -0600 Subject: [PATCH] Bound offline log migration quiescence checks per page --- doc/run-evidence-storage.md | 3 +++ ...run-evidence-migration.integration.test.ts | 27 +++++++++++++++++++ server/src/services/run-evidence-migration.ts | 16 +++++++---- 3 files changed, 41 insertions(+), 5 deletions(-) diff --git a/doc/run-evidence-storage.md b/doc/run-evidence-storage.md index fc6cc353966..ff2662df10c 100644 --- a/doc/run-evidence-storage.md +++ b/doc/run-evidence-storage.md @@ -244,6 +244,9 @@ The offline ephemeral-enrollment process can also preserve unfinalized native local files as read-only `s3_archive` logs. This path requires `PAPERCLIP_EPHEMERAL_MIGRATION=true`, all old deployments stopped, no queued or running execution anywhere in the instance, and expired writer/finalizer leases. +The global execution check runs at page boundaries; per-run ownership and +metadata checks still fence every publication. New execution fails the page and +enrollment, while verified copies remain available for a safe retry. It copies exact surviving bytes to a separate `.archive` journal, checks that the local source did not change, and verifies the remote SHA-256 before switching the run's reference. It supports terminal `interrupted` runs as well. diff --git a/server/src/services/run-evidence-migration.integration.test.ts b/server/src/services/run-evidence-migration.integration.test.ts index 03723d2742e..7a57c72d91b 100644 --- a/server/src/services/run-evidence-migration.integration.test.ts +++ b/server/src/services/run-evidence-migration.integration.test.ts @@ -162,6 +162,33 @@ describe.skipIf(!endpoint || !databaseUrl)("run evidence migration (Postgres and if (source.source !== "local") throw new Error("expected real local source"); await expect(source.assertUnchanged()).rejects.toThrow("legacy_log_source_changed"); }); + it("fails the offline page if new execution appears during publication, and resumes verified history", async () => { + vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), activeId = randomUUID(); + const logRef = `${companyId}/${agentId}/${runId}.ndjson`, bytes = "old unfinished history"; + await fs.mkdir(path.dirname(path.join(directory, "logs", logRef)), { recursive: true }); + await fs.writeFile(path.join(directory, "logs", logRef), bytes); + await db.execute(sql`insert into heartbeat_runs(id,company_id,agent_id,status,runtime_mode,log_store,log_ref) + values(${runId},${companyId},${agentId},'failed','native','local_file',${logRef})`); + // Fault injection at the real PostgreSQL publication boundary. The old + // run is still terminal; a new run violates the global offline contract. + await db.$client.unsafe(`create function admit_during_publication() returns trigger language plpgsql as $$ + begin if new.id='${runId}' then insert into heartbeat_runs(id,company_id,agent_id,status) + values('${activeId}','${companyId}','${agentId}','running'); end if; return new; end $$; + create trigger admit_during_publication before update on heartbeat_runs + for each row execute function admit_during_publication()`); + const options = { apply: true, companyId, kinds: ["log"] as const, preserveHistoricalFiles: true }; + try { + await expect(migrateRunEvidence(db, options)).rejects.toThrow("active_execution_requires_drain"); + const [row] = await db.$client.unsafe("select status,log_store,log_bytes from heartbeat_runs where id=$1", [runId]); + expect(row).toMatchObject({ status: "failed", log_store: "s3_archive", log_bytes: null }); + } finally { + await db.$client.unsafe("drop trigger admit_during_publication on heartbeat_runs; drop function admit_during_publication()"); + await db.execute(sql`delete from heartbeat_runs where id=${activeId}`); + } + expect((await migrateRunEvidence(db, options)).items).toEqual([]); + vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "false"); + }); it("preserves exact historical bytes and expiry, while refusing active/missing/corrupt sources", async () => { const company = randomUUID(), agent = randomUUID(), diff --git a/server/src/services/run-evidence-migration.ts b/server/src/services/run-evidence-migration.ts index 60661857818..66d0f786323 100644 --- a/server/src/services/run-evidence-migration.ts +++ b/server/src/services/run-evidence-migration.ts @@ -60,14 +60,17 @@ export async function migrateRunEvidence( const config = resolveRunEvidenceStorage(); if (!config) throw new Error("Run evidence migration requires instance S3 storage"); const preserveHistoricalFiles = options.preserveHistoricalFiles === true; - const noActiveRuns = sql`not exists (select 1 from ${heartbeatRuns} where ${inArray(heartbeatRuns.status, ["queued", "running"])})`; + async function assertNoActiveRuns() { + options.signal?.throwIfAborted(); + const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(inArray(heartbeatRuns.status, ["queued", "running"])).limit(1); + if (active) throw new Error("active_execution_requires_drain"); + } if (preserveHistoricalFiles) { if (process.env.PAPERCLIP_EPHEMERAL_MIGRATION !== "true") { throw new Error("historical_log_archive_requires_offline_migration"); } - const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) - .where(inArray(heartbeatRuns.status, ["queued", "running"])).limit(1); - if (active) throw new Error("active_execution_requires_drain"); + await assertNoActiveRuns(); } function drainedWriter(now = new Date()) { return and( @@ -81,7 +84,6 @@ export async function migrateRunEvidence( eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId), gt(nativeRunFinalizations.leaseExpiresAt, now), )})`, - preserveHistoricalFiles ? noActiveRuns : undefined, ); } async function assertDrained(runId: string, companyId: string) { @@ -452,6 +454,10 @@ export async function migrateRunEvidence( item.reason = error instanceof Error ? error.message : "trace_migration_failed"; } } + // Cloud has stopped every old process. Check that global prerequisite at + // page boundaries, not with an instance-wide scan at every per-run CAS. + // If it changes, fail enrollment; already verified copies remain resumable. + if (preserveHistoricalFiles) await assertNoActiveRuns(); return { items, nextRunId: logs.length === limit ? logs.at(-1)!.id : undefined,