diff --git a/server/src/services/native-runtime/native-controller-migration.integration.test.ts b/server/src/services/native-runtime/native-controller-migration.integration.test.ts index 69005e82bb3..693022dd224 100644 --- a/server/src/services/native-runtime/native-controller-migration.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-migration.integration.test.ts @@ -11,6 +11,7 @@ import * as sessionResume from "./native-session-resume.js"; import { nativeControllerMigrationDescriptor } from "./native-session-executor.js"; import { hasLocalNativeControllerObligations, migrateNativeControllers, prestageNativeControllers } from "./native-controller-migration.js"; import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint, nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; +import * as checkpointReader from "./native-runner-checkpoint.js"; import { currentNativeControllerIdentity } from "./native-restart-recovery.js"; import { createNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js"; @@ -189,8 +190,11 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv await f.db.insert(agentTaskSessions).values(Array.from({ length: 15 }, () => ({ companyId: f.companyId, agentId: f.agentId, adapterType: "paperclip_runner", taskKey: randomUUID(), lastRunId: f.runId }))); const original = sessionResume.findNativeSessionResumeRun; + const originalCheckpoint = checkpointReader.hasNativeRunnerCheckpoint; let started = 0, finished = 0, inFlight = 0, peak = 0, fail = false; + let checkpointStarted = 0, checkpointFinished = 0, checkpointInFlight = 0, checkpointPeak = 0, checkpointFail = false; const reason = new Error("retained source reader failed"); + const checkpointReason = new Error("durable checkpoint reader failed"); const spy = vi.spyOn(sessionResume, "findNativeSessionResumeRun").mockImplementation(async (...args) => { const first = ++started === 1; peak = Math.max(peak, ++inFlight); try { @@ -199,16 +203,33 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv return await original(...args); } finally { inFlight--; finished++; } }); + const checkpointSpy = vi.spyOn(checkpointReader, "hasNativeRunnerCheckpoint").mockImplementation(async (...args) => { + const first = ++checkpointStarted === 1; + checkpointPeak = Math.max(checkpointPeak, ++checkpointInFlight); + try { + await new Promise(resolve => setTimeout(resolve, first && checkpointFail ? 1 : 20)); + if (first && checkpointFail) throw checkpointReason; + return await originalCheckpoint(...args); + } finally { checkpointInFlight--; checkpointFinished++; } + }); try { const before = performance.now(); expect(await hasLocalNativeControllerObligations(f.db)).toBe(false); - console.info(JSON.stringify({ state: "retained_source_readers", elapsedMs: performance.now() - before, peak, started, finished })); + console.info(JSON.stringify({ state: "retained_source_readers", elapsedMs: performance.now() - before, peak, started, finished, + checkpointStarted, checkpointFinished, checkpointPeak })); expect(started).toBe(16); expect(finished).toBe(16); expect(peak).toBeGreaterThan(1); expect(peak).toBeLessThanOrEqual(8); + expect(checkpointStarted).toBe(16); expect(checkpointFinished).toBe(16); + expect(checkpointPeak).toBeGreaterThan(1); expect(checkpointPeak).toBeLessThanOrEqual(8); started = finished = peak = 0; fail = true; await expect(hasLocalNativeControllerObligations(f.db)).rejects.toBe(reason); expect(started).toBe(8); expect(finished).toBe(8); expect(inFlight).toBe(0); + started = finished = peak = 0; fail = false; + checkpointStarted = checkpointFinished = checkpointPeak = 0; checkpointFail = true; + await expect(hasLocalNativeControllerObligations(f.db)).rejects.toBe(checkpointReason); + expect(checkpointStarted).toBe(8); expect(checkpointFinished).toBe(8); expect(checkpointInFlight).toBe(0); + expect(started).toBe(8); expect(finished).toBe(8); expect(inFlight).toBe(0); expect(await readFile(f.file, "utf8")).toBe(f.source); - } finally { spy.mockRestore(); } + } finally { checkpointSpy.mockRestore(); spy.mockRestore(); } }); if (scenario === "active_durable") it("accepts a complete active database authority without terminal metadata and refuses stale or incomplete handoff proof", async () => { const f = await fixture(1, false, true), controller = await currentNativeControllerIdentity(); diff --git a/server/src/services/native-runtime/native-controller-migration.ts b/server/src/services/native-runtime/native-controller-migration.ts index 4c74b2e3608..2c127058e67 100644 --- a/server/src/services/native-runtime/native-controller-migration.ts +++ b/server/src/services/native-runtime/native-controller-migration.ts @@ -77,7 +77,7 @@ async function hasConfirmedOlderController(db: Db, selected: Selected, } catch { return false; } } -async function* retainedSources(db: Db, signal?: AbortSignal, concurrency = 1) { +async function* retainedSources(db: Db, signal?: AbortSignal, concurrency = 1, inspectDatabaseRecovery = false) { let after: string | undefined; for (;;) { signal?.throwIfAborted(); @@ -89,15 +89,20 @@ async function* retainedSources(db: Db, signal?: AbortSignal, concurrency = 1) { ${after ? sql`and s.id > ${after}::uuid` : sql``} order by s.id limit 100`); for (let offset = 0; offset < rows.length; offset += concurrency) { signal?.throwIfAborted(); - const batch = await Promise.allSettled(rows.slice(offset, offset + concurrency).map(async selected => ({ - selected, source: await findNativeSessionResumeRun(db, { companyId: selected.companyId, agentId: selected.agentId, + const batch = await Promise.allSettled(rows.slice(offset, offset + concurrency).map(async selected => { + const source = await findNativeSessionResumeRun(db, { companyId: selected.companyId, agentId: selected.agentId, issueId: selected.issueId, normalizedSessionId: selected.sessionId, - beforeCreatedAt: new Date(selected.createdAt), includeProgressBarrier: true }), - }))); + beforeCreatedAt: new Date(selected.createdAt), includeProgressBarrier: true }); + const databaseRecoveryPreserved = Boolean(inspectDatabaseRecovery && source && source.status !== "running" + && readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint) + && await hasNativeRunnerCheckpoint(db, { companyId: selected.companyId, issueId: selected.issueId, runId: source.id })); + return { selected, source, databaseRecoveryPreserved }; + })); for (const result of batch) { signal?.throwIfAborted(); if (result.status === "rejected") throw result.reason; - if (result.value.source) yield { selected: result.value.selected, source: result.value.source }; + if (result.value.source) yield { selected: result.value.selected, source: result.value.source, + databaseRecoveryPreserved: result.value.databaseRecoveryPreserved }; } } if (rows.length < 100) return; @@ -256,11 +261,9 @@ export async function hasLocalNativeControllerObligations(db: Db, storage?: Stor if (error.code === "ENOENT") return [] as string[]; throw error; }); if (!roots.length) return false; - for await (const { selected, source } of retainedSources(db, undefined, 8)) { + for await (const { selected, source, databaseRecoveryPreserved } of retainedSources(db, undefined, 8, true)) { const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id }; - if (source.status !== "running" - && readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint) - && await hasNativeRunnerCheckpoint(db, binding)) continue; + if (databaseRecoveryPreserved) continue; if (source.status === "failed" && source.runnerProfileJson.sessionCheckpoint == null && preservedFailedCheckpoint(source, selected.sessionId, await loadNativeRunnerCheckpoint(db, binding))) continue; const paths = await localPaths(source); 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 15787198a77..0e62a9bd809 100644 --- a/server/src/services/native-runtime/native-harness-backup-migration.ts +++ b/server/src/services/native-runtime/native-harness-backup-migration.ts @@ -12,6 +12,8 @@ import { NativeHarnessBackupSourceError, readNativeHarnessBackupSource } from ". import { settledNativeHarnessBackupOwner } from "./native-harness-preservation-source.js"; import { hasValidNativeHarnessFilesystemIndependence } from "./native-harness-filesystem-independence.js"; +const MAX_PARALLEL_INDEPENDENCE_CHECKS = 4; + export async function hasLocalNativeHarnessObligations(db: Db): Promise { let afterSession = "", afterScope = ""; for (;;) { @@ -31,9 +33,19 @@ export async function hasLocalNativeHarnessObligations(db: Db): Promise and b.scope_digest = e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256' and b.status = 'current') and (s.id::text, coalesce(e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256', '')) > (${afterSession}, ${afterScope}) order by selected_session_id, scope_digest limit 100`); - for (const row of rows) if (!row.has_receipt || !await hasValidNativeHarnessFilesystemIndependence(db, { - companyId: row.company_id, selectedSessionId: row.selected_session_id, scopeDigest: row.scope_digest, - })) return true; + let next = 0, invalid = false; + const checked = await Promise.allSettled(Array.from({ length: Math.min(MAX_PARALLEL_INDEPENDENCE_CHECKS, rows.length) }, async () => { + while (!invalid && next < rows.length) { + const row = rows[next++]!; + try { + if (!row.has_receipt || !await hasValidNativeHarnessFilesystemIndependence(db, { + companyId: row.company_id, selectedSessionId: row.selected_session_id, scopeDigest: row.scope_digest, + })) invalid = true; + } catch (error) { invalid = true; throw error; } + } + })); + for (const result of checked) if (result.status === "rejected") throw result.reason; + if (invalid) return true; if (rows.length < 100) return false; afterSession = rows.at(-1)!.selected_session_id; afterScope = rows.at(-1)!.scope_digest; } diff --git a/server/src/services/native-runtime/native-harness-filesystem-independence.integration.test.ts b/server/src/services/native-runtime/native-harness-filesystem-independence.integration.test.ts index 3d25873d71c..799f9ea88ad 100644 --- a/server/src/services/native-runtime/native-harness-filesystem-independence.integration.test.ts +++ b/server/src/services/native-runtime/native-harness-filesystem-independence.integration.test.ts @@ -18,6 +18,7 @@ import { preserveRuntimeSkillAssets } from "./runtime-skill-preservation.js"; import { nativeControllerMigrationDescriptor, scopedRunnerdStateRoot } from "./native-session-executor.js"; import { hasLocalNativeHarnessObligations } from "./native-harness-backup-migration.js"; import { preserveNativeHarnessFilesystemIndependence, hasValidNativeHarnessFilesystemIndependence } from "./native-harness-filesystem-independence.js"; +import * as filesystemIndependence from "./native-harness-filesystem-independence.js"; import { originalHarnessVolumeAnchorPath, verifyOriginalHarnessVolumeSource, type OriginalHarnessVolumeAnchor, type ReviewedOriginalHarnessVolumeWitness } from "./native-harness-original-volume.js"; import { removeRuntimeAssetTree } from "./runtime-asset-retention.js"; @@ -197,6 +198,64 @@ describeEmbeddedPostgres("owner-verified native harness filesystem independence" return { ...f, execution, controllerPath, raw, supervisor, supervisorPath: descriptor.supervisorPath, identity }; } + it("validates retained WEB receipts with bounded parallel database proofs", async () => { + const fixtures = []; + for (let index = 0; index < 5; index++) { + const f = await controllerFixture(true, "none"); + await preserveNativeHarnessFilesystemIndependence(db, { ...f.input, mode: "web_evidence" }); + fixtures.push(f); + } + const validate = filesystemIndependence.hasValidNativeHarnessFilesystemIndependence; + let active = 0, maximum = 0, completed = 0, release!: () => void; + const blocked = new Promise(resolve => { release = resolve; }); + const reader = vi.spyOn(filesystemIndependence, "hasValidNativeHarnessFilesystemIndependence").mockImplementation(async (...args) => { + active++; maximum = Math.max(maximum, active); + try { await blocked; const valid = await validate(...args); completed++; return valid; } + finally { active--; } + }); + const scan = hasLocalNativeHarnessObligations(db); + try { + await vi.waitFor(() => expect(active).toBe(4)); + expect(reader).toHaveBeenCalledTimes(4); + release(); + expect(await scan).toBe(false); + expect(maximum).toBe(4); expect(completed).toBe(fixtures.length); expect(active).toBe(0); + expect(reader).toHaveBeenCalledTimes(fixtures.length); + } finally { release(); await scan; } + }); + + it.each(["invalid", "error"] as const)("joins started WEB proofs after an %s result without admitting more", async outcome => { + for (let index = 0; index < 5; index++) { + const f = await controllerFixture(true, "none"); + await preserveNativeHarnessFilesystemIndependence(db, { ...f.input, mode: "web_evidence" }); + } + const validate = filesystemIndependence.hasValidNativeHarnessFilesystemIndependence; + let started = 0, active = 0, settled = false, releaseFirst!: () => void, releasePeers!: () => void, failed!: () => void; + const first = new Promise(resolve => { releaseFirst = resolve; }); + const peers = new Promise(resolve => { releasePeers = resolve; }); + const failure = new Promise(resolve => { failed = resolve; }); + const expected = new Error("fixture_independence_reader_failed"); + vi.spyOn(filesystemIndependence, "hasValidNativeHarnessFilesystemIndependence").mockImplementation(async (...args) => { + const index = started++; active++; + try { + const valid = await validate(...args); + if (index === 0) { await first; failed(); if (outcome === "error") throw expected; return false; } + await peers; return valid; + } finally { active--; } + }); + const scan = hasLocalNativeHarnessObligations(db); + void scan.then(() => { settled = true; }, () => { settled = true; }); + try { + await vi.waitFor(() => expect(started).toBe(4)); + releaseFirst(); await failure; await new Promise(resolve => setImmediate(resolve)); + expect(settled).toBe(false); expect(started).toBe(4); expect(active).toBe(3); + releasePeers(); + if (outcome === "error") await expect(scan).rejects.toBe(expected); + else expect(await scan).toBe(true); + expect(active).toBe(0); expect(started).toBe(4); + } finally { releaseFirst(); releasePeers(); await scan.catch(() => undefined); } + }); + it("preserves WEB-only evidence with a succeeded source and unresolved external lease", async () => { const f = await controllerFixture(true, "deleted"); if (!("runtimeContext" in f.execution)) throw new Error("fixture runtime context missing");