Bound offline log migration quiescence checks per page

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 `<run>.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.

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(),

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,