From e44ab4082491a754ca8a353718f82f05ad529e3e Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 27 Sep 2026 20:32:03 -0500 Subject: [PATCH] fix(runtime): await bounded journal proofs with ownership rechecks Co-Authored-By: Paperclip --- server/src/services/chat-channels.ts | 8 +- .../explicit-native-continuation.test.ts | 2 +- .../services/explicit-native-continuation.ts | 4 +- .../src/services/native-local-process-stop.ts | 4 +- .../native-runtime/native-safe-replacement.ts | 4 +- .../native-session-executor.test.ts | 123 ++++++- .../native-runtime/native-session-executor.ts | 308 +++++++++++------- 7 files changed, 310 insertions(+), 143 deletions(-) diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index dd3694dc6ab..52db9448b1e 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -12269,7 +12269,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { await failedNativeRetryCoordinator(tx, run, source.issueId); if (retiredOwner) { if ( - !nativePreProviderRetryAfterCleanupStateIsSafe({ + !(await nativePreProviderRetryAfterCleanupStateIsSafe({ failedExecution: run.runnerProfileJson?.nativeExecutionInput, retiredExecution: retiredOwner.predecessor.runnerProfileJson?.nativeExecutionInput, @@ -12285,7 +12285,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { processPid: run.processPid!, processGroupId: run.processGroupId!, receipt: retiredOwner.receipt, - }) + })) ) throw failedChatRetryDenied(); return; @@ -12299,7 +12299,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { (checkpoint?.providerSessionId != null && (typeof checkpoint.providerSessionId !== "string" || checkpoint.providerSessionId.trim().length === 0)) || - !nativeFailedRunRetryStateIsSafe({ + !(await nativeFailedRunRetryStateIsSafe({ execution: run.runnerProfileJson?.nativeExecutionInput, companyId, issueId: source.issueId, @@ -12321,7 +12321,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { | "bootstrap_retry" | "exact_checkpoint_resume", allowVerifiedBackup: leases.length > 0, - }) + })) ) throw failedChatRetryDenied(); } diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 3c8c8691370..4832f654537 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -379,7 +379,7 @@ const support = await getEmbeddedPostgresTestSupport(); "requires exact local cleanup for a new turn after worker loss (%s)", async mode => { const f = await seed(); await db.update(heartbeatRuns).set({ errorCode: "native_session_cleanup_quarantined" }).where(eq(heartbeatRuns.id, f.sourceRunId)); - const retire = vi.fn(() => mode !== "changed"); + const retire = vi.fn(async () => mode !== "changed"); const verify = vi.spyOn(nativeExecutor, "verifyStoppedNativeSessionForContinuation").mockResolvedValue( mode === "unproven" ? null : { evidence: { runId: f.sourceRunId, schema: "paperclip.stopped_native_conversation.v1" }, retire }); try { diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 5c6c0349d17..10a82aa0d50 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -147,7 +147,7 @@ export async function admitExplicitNativeContinuation(input: { if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start."); const sources: Run[] = []; - const stoppedSessions: Array<{ evidence: Record; retire: () => boolean }> = []; + const stoppedSessions: Array<{ evidence: Record; retire: () => Promise }> = []; const cancelledStartupIds = new Set(); for (const action of actions) { const runId = action.evidence.runId ?? action.evidence.sourceRunId; @@ -267,7 +267,7 @@ export async function admitExplicitNativeContinuation(input: { summary: null, exposeLowTrustRaw: false }); if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) }; for (const stopped of stoppedSessions) { - if (!stopped.retire()) return blocked("local_cleanup", "The previous provider cleanup changed. Your message is saved."); + if (!await stopped.retire()) return blocked("local_cleanup", "The previous provider cleanup changed. Your message is saved."); await appendHeartbeatRunEvent(db, { companyId, runId: String(stopped.evidence.runId), agentId, eventType: "native.stopped_conversation_verified", stream: "system", level: "info", diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 29cbafdf0bf..b0572776052 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -79,7 +79,7 @@ export async function hasHistoricalSuspendedNativeSession(db: Db, run: typeof he if (checkpoint?.providerSessionId != null && (typeof checkpoint.providerSessionId !== "string" || !checkpoint.providerSessionId.trim())) return false; const { nativeFailedRunRetryStateIsSafe } = await import("./native-runtime/native-session-executor.js"); - return nativeFailedRunRetryStateIsSafe({ + return (await nativeFailedRunRetryStateIsSafe({ execution: run.runnerProfileJson?.nativeExecutionInput, companyId: run.companyId, issueId: run.nativeIssueId, agentId: run.agentId, runId: run.id, nativeSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId, @@ -87,7 +87,7 @@ export async function hasHistoricalSuspendedNativeSession(db: Db, run: typeof he providerSessionId: typeof checkpoint?.sessionId === "string" ? checkpoint.sessionId : null, providerBackendSessionId: typeof checkpoint?.providerSessionId === "string" ? checkpoint.providerSessionId : null, recoveryMode: "exact_checkpoint_resume", allowVerifiedBackup: false, - }); + })); } /** Recover the exact stopped identity after the mutable run fields were cleared. */ diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index 0f8a98f8fc5..4c695ac7545 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -41,7 +41,7 @@ export async function reconcileSafeNativeReplacements( options: { verifyStoppedSession?: (run: typeof heartbeatRuns.$inferSelect) => Promise<{ evidence: Record; - retire: () => boolean; + retire: () => boolean | Promise; } | null>; /** Test fault injection at durability boundaries; never exposed by an API. */ failpoint?: (phase: "successor_inserted" | "lineage_committed") => void; @@ -358,7 +358,7 @@ export async function reconcileSafeNativeReplacements( } if (stoppedSession) { // If the last ownership proof changes, roll back the status restoration. - if (!stoppedSession.retire()) throw new Error("native_replacement_stopped_session_changed"); + if (!await stoppedSession.retire()) throw new Error("native_replacement_stopped_session_changed"); await appendHeartbeatRunEvent(tx as unknown as Db, { companyId: run.companyId, runId: run.id, agentId: run.agentId, eventType: "native.stopped_text_turn_verified", stream: "system", level: "info", diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 9be9567e0c8..fa21d302bd2 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -1,3 +1,4 @@ +import * as nativeJournalProof from "./native-journal-projection-async.js"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { access, @@ -935,7 +936,7 @@ describe("native incomplete-bootstrap evidence", () => { }; try { await writeFile(statePath, JSON.stringify(base)); - expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(true); + expect((await runnerdStateProvesIncompleteBootstrap(root))).toBe(true); for (const ambiguous of [ { ...base, connectionCount: 1 }, @@ -950,7 +951,7 @@ describe("native incomplete-bootstrap evidence", () => { }, ]) { await writeFile(statePath, JSON.stringify(ambiguous)); - expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(false); + expect((await runnerdStateProvesIncompleteBootstrap(root))).toBe(false); } } finally { await rm(root, { recursive: true, force: true }); @@ -3048,6 +3049,42 @@ describe("retained native cleanup activation", () => { ); for (const [file, data] of source) await writeFile(join(quarantine, file), JSON.stringify(data)); + if (mode === "empty_root") { + const projectedHistory: Record & { commands: Array> } = structuredClone(source[0][1]); + const output = "x".repeat(384 * 1024); + for (let index = 0; index < 24; index++) { + const ordinary = { + type: "semantic_tool.result", status: "completed", + payload: { callId: `read-${index}`, operationId: "paperclip_read", input: { output }, correlation, + sourceEventId: `read-input-${index}`, sourceEventType: "semantic_tool.input", isError: false }, + result: { result: { callId: `read-${index}` } }, + }; + expect(Buffer.byteLength(JSON.stringify(ordinary))).toBeLessThan(1024 * 1024 - 4096); + projectedHistory.commands.push(ordinary); + } + const controlPath = join(quarantine, "control-plane/control-plane-state.json"); + const bytes = JSON.stringify(projectedHistory); + expect(Buffer.byteLength(bytes)).toBeGreaterThan(8 * 1024 * 1024); + await writeFile(controlPath, bytes); + try { + const proof = await nativeJournalProof.readNativeJournalProjection(controlPath); + const matches = (control: Record) => retainedNativeCleanupJournalMatches({ + run, execution, accepted, control, providerSessionId: "exact-thread", providerAccountSessionId, + persistedEvents: [identityRow], + }); + expect(matches(proof.value as Record)).toBe(true); + for (const corrupt of ["input", "correlation"] as const) { + const changed = structuredClone(proof.value) as typeof projectedHistory; + const finish = changed.commands.find(command => (command.payload as Record | undefined)?.operationId === "paperclip_finish")!; + const payload = finish.payload as { input: Record; correlation: { runId: string } }; + if (corrupt === "input") payload.input.summary = "changed"; + else payload.correlation.runId = "foreign-run"; + expect(matches(changed)).toBe(false); + } + } finally { + await writeFile(controlPath, JSON.stringify(source[0][1])); + } + } await mkdir(join(quarantine, "codex-home/sessions"), { recursive: true, }); @@ -3996,7 +4033,7 @@ describe("stopped native conversation physical cleanup", () => { expect(proof).not.toBeNull(); if (mode === "changed_state") await writeFile(runnerPath, "{}"); const kill = mode === "changed_pid" ? vi.spyOn(process, "kill").mockReturnValue(true) : null; - try { expect(proof!.retire()).toBe(["stopped", "replacement", "normalized_session_receipt"].includes(mode)); } finally { kill?.mockRestore(); } + try { expect(await proof!.retire()).toBe(["stopped", "replacement", "normalized_session_receipt"].includes(mode)); } finally { kill?.mockRestore(); } expect(await readFile(join(root, `runner/${provider}-provider-state.json`), "utf8")).toBe(JSON.stringify(providerState)); } else expect(proof).toBeNull(); } finally { @@ -4021,6 +4058,8 @@ describe("explicit failed native retry physical evidence", () => { "active_provider", "ambiguous_provider", "live_pid", + "owner_started_during_journal_read", + "directory_replaced_during_journal_read", "symlink", "bootstrap", "quarantined_bootstrap", @@ -4079,6 +4118,8 @@ describe("explicit failed native retry physical evidence", () => { "distinct_account", "null_account", ].includes(kind); + let restoreRead: (() => void) | undefined; + let restoreKill: (() => void) | undefined; try { const identity = { runId: execution.binding.runId, @@ -4142,13 +4183,32 @@ describe("explicit failed native retry physical evidence", () => { ); } const before = await readdir(stateBase); + if (kind === "owner_started_during_journal_read" || kind === "directory_replaced_during_journal_read") { + const read = nativeJournalProof.readNativeJournalProjection; + let replacements = 0; + const spy = vi.spyOn(nativeJournalProof, "readNativeJournalProjection").mockImplementation(async (...args) => { + const proof = await read(...args); + if (kind === "owner_started_during_journal_read") { + const kill = vi.spyOn(process, "kill").mockReturnValue(true); + restoreKill = () => kill.mockRestore(); + } else { + const control = join(root, "control-plane"); + const archived = join(root, `changed-control-${replacements++}`); + await rename(control, archived); + await mkdir(control); + await writeFile(join(control, "control-plane-state.json"), await readFile(join(archived, "control-plane-state.json"))); + } + return proof; + }); + restoreRead = () => spy.mockRestore(); + } const retryInput = { execution, ...execution.binding, nativeSessionId: execution.session.normalizedSessionId!, runnerInstanceId: kind === "wrong_runner" ? "another-runner" : "runner-retry", - processPid: kind === "live_pid" ? process.pid : null, + processPid: kind === "live_pid" ? process.pid : kind === "owner_started_during_journal_read" ? 99_999_999 : null, providerSessionId: expectedThread, providerBackendSessionId: expectedAccount, processGroupId: null, @@ -4158,7 +4218,7 @@ describe("explicit failed native retry physical evidence", () => { allowVerifiedBackup: false, }; expect - .soft(nativeFailedRunRetryStateIsSafe(retryInput)) + .soft((await nativeFailedRunRetryStateIsSafe(retryInput))) .toBe(retryable || kind === "bootstrap"); if (!bootstrap && kind !== "symlink") { const files = [ @@ -4207,26 +4267,26 @@ describe("explicit failed native retry physical evidence", () => { processGroupId: 99_999_999, receipt, }; - expect(nativePreProviderRetryAfterCleanupStateIsSafe(input)).toBe( + expect((await nativePreProviderRetryAfterCleanupStateIsSafe(input))).toBe( retryable, ); expect( - nativePreProviderRetryAfterCleanupStateIsSafe({ + (await nativePreProviderRetryAfterCleanupStateIsSafe({ ...input, receipt: { ...receipt, settledFingerprint: "changed" }, - }), + })), ).toBe(false); expect( - nativePreProviderRetryAfterCleanupStateIsSafe({ + (await nativePreProviderRetryAfterCleanupStateIsSafe({ ...input, receipt: { ...receipt, sourceFingerprint: undefined }, - }), + })), ).toBe(false); expect( - nativePreProviderRetryAfterCleanupStateIsSafe({ + (await nativePreProviderRetryAfterCleanupStateIsSafe({ ...input, receipt: { ...receipt, requestId: "" }, - }), + })), ).toBe(false); } expect(await readdir(stateBase)).toEqual(before); @@ -4235,6 +4295,8 @@ describe("explicit failed native retry physical evidence", () => { await access(join(root, "control-plane", "control-plane-state.json")), ).toBeUndefined(); } finally { + restoreRead?.(); + restoreKill?.(); if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; await rm(stateBase, { recursive: true, force: true }); @@ -8711,6 +8773,8 @@ describe("runnerd provider runtime wiring", () => { { commandCount: 24, fault: null }, { commandCount: 4, fault: "oversized" }, { commandCount: 4, fault: "wrong identity" }, + { commandCount: 4, fault: "native_journal_worker_busy" }, + { commandCount: 4, fault: "native_journal_worker_timeout" }, ])( "admits a valid large control-plane journal while preserving bounds and identity ($commandCount commands, $fault)", async ({ commandCount, fault }) => { @@ -8744,6 +8808,7 @@ describe("runnerd provider runtime wiring", () => { status: "succeeded", runnerProfileJson: { nativeExecutionInput: prior }, }]) }) }) }), } as unknown as Db; + let restoreRead: (() => void) | undefined; try { state.createBackend.mockClear(); state.createTransport.mockClear(); @@ -8781,10 +8846,26 @@ describe("runnerd provider runtime wiring", () => { } state.createBackend.mockClear(); state.createTransport.mockClear(); + if (fault?.startsWith("native_journal_worker_")) { + const read = nativeJournalProof.readNativeJournalProjection; + const spy = vi.spyOn(nativeJournalProof, "readNativeJournalProjection").mockImplementation(async (path, purpose) => { + if (purpose === "identity") throw new Error(fault); + return read(path, purpose); + }); + restoreRead = () => spy.mockRestore(); + } const continuation = createRunnerdBackend({ db: priorRunDb, execution: current, runnerInstanceId: "new-heartbeat-runner", runnerExecutionTarget: remoteTarget, }); + if (fault?.startsWith("native_journal_worker_")) { + await expect(continuation).rejects.toThrow(fault); + expect(state.createBackend).not.toHaveBeenCalled(); + expect(state.createTransport).not.toHaveBeenCalled(); + expect((await readFile(controlPath)).equals(controlBytes)).toBe(true); + await expect(access(join(stateBase, "quarantine"))).rejects.toThrow(); + return; + } if (fault !== null) { await expect(continuation).rejects.toThrow("runner_state_identity_mismatch"); expect(state.createBackend).not.toHaveBeenCalled(); @@ -8801,6 +8882,7 @@ describe("runnerd provider runtime wiring", () => { expect((await readFile(controlPath)).equals(controlBytes)).toBe(true); await expect(access(join(stateBase, "quarantine"))).rejects.toThrow(); } finally { + restoreRead?.(); if (previousStateDirectory === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previousStateDirectory; await rm(stateBase, { recursive: true, force: true }); @@ -9025,6 +9107,7 @@ describe("runnerd provider runtime wiring", () => { "ambiguous turn start", "state symlink", "newer provider checkpoint", + "directory replaced during final hash", ])( "automatically recovers only a proven settled local session: %s", async (scenario) => { @@ -9073,6 +9156,7 @@ describe("runnerd provider runtime wiring", () => { }); } const onLog = vi.fn(async () => {}); + let finalHashSpy: ReturnType | undefined; const profile = { nativeExecutionInput: scenario === "wrong scope" @@ -9219,6 +9303,20 @@ describe("runnerd provider runtime wiring", () => { ); }; await writeCandidate(candidate); + if (scenario === "directory replaced during final hash") { + const scan = nativeJournalProof.scanNativeStateFile; + let reads = 0; + finalHashSpy = vi.spyOn(nativeJournalProof, "scanNativeStateFile").mockImplementation(async (...args) => { + const proof = await scan(...args); + if (args[0] === join(candidate, "control-plane", "control-plane-state.json") && ++reads === 2) { + const bytes = await readFile(args[0]); + await rename(join(candidate, "control-plane"), join(candidate, "original-control-plane")); + await mkdir(join(candidate, "control-plane")); + await writeFile(args[0], bytes); + } + return proof; + }); + } if (scenario === "state symlink") { await rename( join(candidate, "runner", "runner-state.json"), @@ -9306,6 +9404,7 @@ describe("runnerd provider runtime wiring", () => { expect(state.createBackend).not.toHaveBeenCalled(); } } finally { + finalHashSpy?.mockRestore(); processKill.mockRestore(); if (previousStateDirectory === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index b5ae041c94c..c075a218a02 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,4 +1,4 @@ -import { readNativeJournalProjection, scanNativeStateFile } from "./native-journal-projection.js"; +import { readNativeJournalProjection, scanNativeStateFile, nativeStateFileIncludes, nativeStateBase64Fingerprint } from "./native-journal-projection-async.js"; import { isSupportedRemoteCodexVersion, parseCodexCliVersion, @@ -1620,17 +1620,19 @@ function legacyRunnerdStateRoot(execution: NativeExecutionInput): string { ); } -function migrateLegacyRunnerdStateRoot(input: { +async function migrateLegacyRunnerdStateRoot(input: { legacy: string; scoped: string; execution: NativeExecutionInput; verifiedPriorRunId?: string; -}): string | null { +}): Promise { if (!existsSync(input.legacy)) return null; if (!isSafeNativeStateDirectory(input.legacy)) { throw new Error("runner_state_directory_unsafe"); } - const legacyIdentity = readRunnerdDurableIdentity(input.legacy); + const legacyDirectoryIdentity = nativeStateDirectoryIdentity(input.legacy); + const legacyIdentity = (await readRunnerdDurableIdentity(input.legacy)); + assertNativeStateDirectoryIdentity(input.legacy, legacyDirectoryIdentity); const exactRun = durableIdentityMatchesExecution( legacyIdentity, input.execution, @@ -1666,7 +1668,7 @@ function migrateLegacyRunnerdStateRoot(input: { // directory is authoritative once it exists; otherwise preserve the // original migration failure rather than silently starting empty state. if (!isSafeNativeStateDirectory(input.scoped)) throw error; - const scopedIdentity = readRunnerdDurableIdentity(input.scoped); + const scopedIdentity = (await readRunnerdDurableIdentity(input.scoped)); const exactScopedRun = durableIdentityMatchesExecution( scopedIdentity, input.execution, @@ -1782,7 +1784,7 @@ function nativeStateFileMaxBytes(relativePath: string): number { : NATIVE_RUNNER_STATE_MAX_BYTES; } -function cleanupStateSnapshot(root: string, providerFile = "codex-provider-state.json") { +async function cleanupStateSnapshot(root: string, providerFile = "codex-provider-state.json") { if ( ![root, resolve(root, "runner"), resolve(root, "control-plane")].every( isSafeNativeStateDirectory, @@ -1790,8 +1792,11 @@ function cleanupStateSnapshot(root: string, providerFile = "codex-provider-state ) { throw new Error("native_cleanup_maintenance_unproven"); } + const directoryIdentities = [root, resolve(root, "runner"), resolve(root, "control-plane")] + .map(path => ({ path, identity: nativeStateDirectoryIdentity(path) })); const files = [...CLEANUP_CANONICAL_FILES.slice(0, 2), `runner/${providerFile}`]; - const controlProof = readNativeJournalProjection(resolve(root, files[0]!)); + const controlProof = (await readNativeJournalProjection(resolve(root, files[0]!))); + for (const directory of directoryIdentities) assertNativeStateDirectoryIdentity(directory.path, directory.identity); const bytes = files.slice(1).map((file) => readBoundedNativeFile(resolve(root, file), NATIVE_RUNNER_STATE_MAX_BYTES, "native_cleanup_maintenance_unproven"), ); @@ -2392,7 +2397,7 @@ export function rebaseRetainedNativeCleanupProviderHome( export async function verifyStoppedNativeSessionForContinuation( db: Db, run: typeof heartbeatRuns.$inferSelect, -): Promise<{ evidence: Record; retire: () => boolean } | null> { +): Promise<{ evidence: Record; retire: () => Promise } | null> { try { if (run.runtimeMode !== "native" || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status) || !run.finishedAt || !run.nativeIssueId || !run.nativeSessionId || !run.runnerInstanceId) return null; @@ -2413,7 +2418,7 @@ export async function verifyStoppedNativeSessionForContinuation( if (!stopped) return null; const root = scopedRunnerdStateRoot(execution); const providerFile = runnerProviderStateFilename(execution); - const snapshot = cleanupStateSnapshot(root, providerFile); + const snapshot = (await cleanupStateSnapshot(root, providerFile)); const identity = record(snapshot.control.identity); if (!durableIdentityMatchesExecution(identity, execution) || identity.runnerInstanceId !== run.runnerInstanceId || !["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].every(key => @@ -2479,14 +2484,14 @@ export async function verifyStoppedNativeSessionForContinuation( if (!cleanupProcessAbsent(pid)) return null; providerPids.add(pid); } - const unchanged = () => idle() && cleanupProcessAbsent(stopped.processPid) && - [...providerPids].every(cleanupProcessAbsent) && cleanupStateSnapshot(root, providerFile).fingerprint === snapshot.fingerprint; + const unchanged = async () => (await cleanupStateSnapshot(root, providerFile)).fingerprint === snapshot.fingerprint && + idle() && cleanupProcessAbsent(stopped.processPid) && [...providerPids].every(cleanupProcessAbsent); return { evidence: { schema: "paperclip.stopped_native_conversation.v1", runId: run.id, nativeSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId, processPid: stopped.processPid, providerPids: [...providerPids], stateFingerprint: snapshot.fingerprint }, - retire: () => { - try { return unchanged() && completeTerminatedLocalNativeSessionCleanup({ + retire: async () => { + try { return await unchanged() && completeTerminatedLocalNativeSessionCleanup({ companyId: run.companyId, runId: run.id, runnerInstanceId: run.runnerInstanceId!, }); } catch { return false; } }, @@ -2500,7 +2505,7 @@ export async function verifyStoppedNativeSessionForContinuation( export async function verifyStoppedNativeSessionForReplacement( db: Db, run: typeof heartbeatRuns.$inferSelect, -): Promise<{ evidence: Record; retire: () => boolean } | null> { +): Promise<{ evidence: Record; retire: () => Promise } | null> { try { if (run.runtimeMode !== "native" || run.status !== "failed" || !run.finishedAt || !run.nativeIssueId || !run.nativeSessionId || !run.runnerInstanceId) return null; @@ -2520,7 +2525,7 @@ export async function verifyStoppedNativeSessionForReplacement( const stopped = await readNativeLocalProcessStop(db, run.companyId, run.id); if (!stopped) return null; const root = scopedRunnerdStateRoot(execution); - const snapshot = cleanupStateSnapshot(root); + const snapshot = (await cleanupStateSnapshot(root)); const identity = record(snapshot.control.identity); if (!durableIdentityMatchesExecution(identity, execution) || identity.runnerInstanceId !== run.runnerInstanceId || !["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].every(key => @@ -2609,8 +2614,8 @@ export async function verifyStoppedNativeSessionForReplacement( completedTaskControlCallIds: completedTaskControlCalls.map(call => call.callId) }; return { evidence, - retire: () => idle() && cleanupProcessAbsent(stopped.processPid) && cleanupProcessAbsent(provider.processId) && - cleanupStateSnapshot(root).fingerprint === snapshot.fingerprint && + retire: async () => (await cleanupStateSnapshot(root)).fingerprint === snapshot.fingerprint && + idle() && cleanupProcessAbsent(stopped.processPid) && cleanupProcessAbsent(provider.processId) && nativeSha256(readBoundedNativeFile(rolloutPath, 32 * 1024 * 1024, "native_crash_inventory_unproven").toString("utf8")) === rolloutSha256 && completeTerminatedLocalNativeSessionCleanup({ companyId: run.companyId, runId: run.id, runnerInstanceId: run.runnerInstanceId! }), }; @@ -2647,7 +2652,7 @@ export async function reconcileRetainedNativeSessionCleanup( quarantine: string; root: string; emptyRoot: ReturnType; - source: ReturnType; + source: Awaited>; providerHome: ReturnType; sourceArchive: { intent: Record; @@ -2656,7 +2661,7 @@ export async function reconcileRetainedNativeSessionCleanup( } | null; copySource: { directory: string; - snapshot: ReturnType; + snapshot: Awaited>; requestId: string; } | null; providerPid: number; @@ -2887,7 +2892,7 @@ export async function reconcileRetainedNativeSessionCleanup( ) return null; } - const source = cleanupStateSnapshot(sourceDirectory); + const source = (await cleanupStateSnapshot(sourceDirectory)); const identity = record(source.control.identity); if ( !durableIdentityMatchesExecution(identity, execution) || @@ -2975,7 +2980,7 @@ export async function reconcileRetainedNativeSessionCleanup( return null; let copySource: { directory: string; - snapshot: ReturnType; + snapshot: Awaited>; requestId: string; } | null = null; const providerHome = cleanupProviderHomeSnapshot( @@ -3032,7 +3037,7 @@ export async function reconcileRetainedNativeSessionCleanup( }) ) return null; - const attempted = cleanupStateSnapshot(directory); + const attempted = (await cleanupStateSnapshot(directory)); const receipts = await tx .select() .from(heartbeatRunEvents) @@ -3173,7 +3178,7 @@ export async function reconcileRetainedNativeSessionCleanup( const authorize = async () => { await assertLease(); if ( - cleanupStateSnapshot(owned.quarantine).fingerprint !== + (await cleanupStateSnapshot(owned.quarantine)).fingerprint !== owned.source.fingerprint || cleanupProviderHomeSnapshot( resolve(owned.quarantine, "codex-home"), @@ -3181,7 +3186,7 @@ export async function reconcileRetainedNativeSessionCleanup( ).metadataFingerprint !== owned.providerHome.metadataFingerprint || (owned.copySource && (!retainedRunnerdMaintenanceIsIdle(owned.copySource.directory) || - cleanupStateSnapshot(owned.copySource.directory).fingerprint !== + (await cleanupStateSnapshot(owned.copySource.directory)).fingerprint !== owned.copySource.snapshot.fingerprint || lstatSync( resolve(owned.copySource.directory, CLEANUP_ACTIVATION_FILE), @@ -3191,6 +3196,7 @@ export async function reconcileRetainedNativeSessionCleanup( canonicalJson(owned.emptyRoot) ) throw denied(); + await assertLease(); }; const appendMaintenanceHistory = async (entry: Record) => { const history = await db.transaction(async (tx) => { @@ -3237,7 +3243,7 @@ export async function reconcileRetainedNativeSessionCleanup( if ( canonicalJson(cleanupArchiveRootIdentity(sourceDirectory)) !== canonicalJson(archive.intent.rootIdentity) || - cleanupStateSnapshot(sourceDirectory).fingerprint !== + (await cleanupStateSnapshot(sourceDirectory)).fingerprint !== owned.source.fingerprint || cleanupProviderHomeSnapshot( resolve(sourceDirectory, "codex-home"), @@ -3248,6 +3254,7 @@ export async function reconcileRetainedNativeSessionCleanup( : existsSync(owned.root)) ) throw denied(); + await assertLease(); if (archive.fromCanonical) { const parent = dirname(owned.quarantine); mkdirSync(parent, { recursive: true, mode: 0o700 }); @@ -3262,7 +3269,7 @@ export async function reconcileRetainedNativeSessionCleanup( true, ); if ( - cleanupStateSnapshot(owned.quarantine).fingerprint !== + (await cleanupStateSnapshot(owned.quarantine)).fingerprint !== owned.source.fingerprint || archivedHome.fingerprint !== owned.providerHome.fingerprint || canonicalJson(cleanupArchiveRootIdentity(owned.quarantine)) !== @@ -3685,7 +3692,7 @@ async function assertCleanupActivationCommitted( entry.kind === "native_cleanup_maintenance" && entry.requestId === marker.requestId, ); - const snapshot = cleanupStateSnapshot(root); + const snapshot = (await cleanupStateSnapshot(root)); if ( marker.settledProviderHomeFingerprint !== undefined || receipt?.settledProviderHomeFingerprint !== undefined @@ -3725,10 +3732,34 @@ async function assertCleanupActivationCommitted( rmSync(path); } +function failedNativeRetryOwnersStopped(execution: NativeExecutionInput, processPid: number | null, processGroupId: number | null): boolean { + const scope = nativeSessionScopeKey(execution); + if ( + executingRunnerdSessionScopes.has(scope) || + initializingSessionToolAuthorities.has(scope) || + warmNativeSessions.has(scope) + ) + return false; + for (const [id, group] of [ + [processPid, false], + [processGroupId, true], + ] as const) { + if (id === null) continue; + if (!Number.isSafeInteger(id) || id <= 0) return false; + try { + process.kill(group ? -id : id, 0); + return false; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ESRCH") return false; + } + } + return true; +} + /** Read-only admission evidence for an explicit retry of a terminal failed * run. Never migrate/archive state, release an owner, or contact a provider. * Normal executor admission independently verifies the state again. */ -export function nativeFailedRunRetryStateIsSafe(input: { +export async function nativeFailedRunRetryStateIsSafe(input: { execution: unknown; companyId: string; issueId: string; @@ -3742,7 +3773,7 @@ export function nativeFailedRunRetryStateIsSafe(input: { processGroupId: number | null; recoveryMode: "bootstrap_retry" | "exact_checkpoint_resume"; allowVerifiedBackup: boolean; -}): boolean { +}): Promise { try { const execution = parseNativeExecutionInput(input.execution); if ( @@ -3754,26 +3785,7 @@ export function nativeFailedRunRetryStateIsSafe(input: { !input.runnerInstanceId ) return false; - const scope = nativeSessionScopeKey(execution); - if ( - executingRunnerdSessionScopes.has(scope) || - initializingSessionToolAuthorities.has(scope) || - warmNativeSessions.has(scope) - ) - return false; - for (const [id, group] of [ - [input.processPid, false], - [input.processGroupId, true], - ] as const) { - if (id === null) continue; - if (!Number.isSafeInteger(id) || id <= 0) return false; - try { - process.kill(group ? -id : id, 0); - return false; - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "ESRCH") return false; - } - } + if (!failedNativeRetryOwnersStopped(execution, input.processPid, input.processGroupId)) return false; const root = scopedRunnerdStateRoot(execution); if (!lstatSync(root, { throwIfNoEntry: false })) { if ( @@ -3812,7 +3824,9 @@ export function nativeFailedRunRetryStateIsSafe(input: { } return true; } - const identity = readRunnerdDurableIdentity(root); + const identity = (await readRunnerdDurableIdentity(root)); + // Worker reads yield: a new in-process owner or recycled PID invalidates the earlier stop proof. + if (!failedNativeRetryOwnersStopped(execution, input.processPid, input.processGroupId)) return false; if ( !durableIdentityMatchesSession(identity, execution) || !durableIdentityMatchesExecution(identity, execution) || @@ -3895,7 +3909,7 @@ export function nativeFailedRunRetryStateIsSafe(input: { /** Physical half of retrying a request rejected before provider admission. * The caller separately proves the failed receipt/coordinator has no native * events or result and selects exactly one committed cleanup owner. */ -export function nativePreProviderRetryAfterCleanupStateIsSafe(input: { +export async function nativePreProviderRetryAfterCleanupStateIsSafe(input: { failedExecution: unknown; retiredExecution: unknown; companyId: string; @@ -3910,7 +3924,7 @@ export function nativePreProviderRetryAfterCleanupStateIsSafe(input: { processPid: number; processGroupId: number; receipt: Record; -}): boolean { +}): Promise { try { const failed = parseNativeExecutionInput(input.failedExecution); const retired = parseNativeExecutionInput(input.retiredExecution); @@ -3934,7 +3948,7 @@ export function nativePreProviderRetryAfterCleanupStateIsSafe(input: { !/^[a-f0-9]{64}$/.test(input.receipt.sourceFingerprint) || typeof input.receipt.settledFingerprint !== "string" || !/^[a-f0-9]{64}$/.test(input.receipt.settledFingerprint) || - !nativeFailedRunRetryStateIsSafe({ + !(await nativeFailedRunRetryStateIsSafe({ execution: retired, companyId: input.companyId, issueId: input.issueId, @@ -3948,7 +3962,7 @@ export function nativePreProviderRetryAfterCleanupStateIsSafe(input: { providerBackendSessionId: input.providerBackendSessionId, recoveryMode: "exact_checkpoint_resume", allowVerifiedBackup: false, - }) + })) ) return false; const root = scopedRunnerdStateRoot(retired); @@ -3978,8 +3992,9 @@ export function nativePreProviderRetryAfterCleanupStateIsSafe(input: { return false; } return ( - cleanupStateSnapshot(root).fingerprint === - input.receipt.settledFingerprint + (await cleanupStateSnapshot(root)).fingerprint === + input.receipt.settledFingerprint && + failedNativeRetryOwnersStopped(retired, input.processPid, input.processGroupId) ); } catch { return false; @@ -4056,7 +4071,7 @@ async function verifyPriorRunnerdStateForSessionScope(input: { } } -function hasRetainedWarmTransitionEvidence(root: string): boolean { +async function hasRetainedWarmTransitionEvidence(root: string): Promise { for (const [directory, filename, maximum] of [ ["control-plane", "control-plane-state.json", DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES], ["runner", "runner-state.json", NATIVE_RUNNER_STATE_MAX_BYTES], @@ -4065,23 +4080,21 @@ function hasRetainedWarmTransitionEvidence(root: string): boolean { if (!lstatSync(path, { throwIfNoEntry: false })) continue; try { const state = record(directory === "control-plane" - ? readNativeJournalProjection(path).value + ? (await readNativeJournalProjection(path)).value : JSON.parse(readBoundedNativeFile(path, maximum, "runner_state_too_large").toString("utf8"))); if (Object.prototype.hasOwnProperty.call(state, "warmTransition") || state.schema === "paperclip.runner.durable.control-plane-state.warm-transition.v1" || state.schema === "paperclip.runner.durable.state.warm-transition.v1") return true; - } catch { + } catch (error) { + if (nativeJournalProofUnavailable(error)) throw error; // Detection is never admission. Preserve the malformed-forward-receipt // fence without materializing its potentially large discarded history. - let tail = "", found = false; try { - scanNativeStateFile(path, maximum, chunk => { - const text = tail + chunk.toString("utf8"); - if (text.includes("warm-transition.v1") || text.includes('"warmTransition"')) found = true; - tail = text.slice(-32); - }); - if (found) return true; - } catch { /* The ordinary verifier owns unreadable legacy state. */ } + if (await nativeStateFileIncludes(path, maximum, ["warm-transition.v1", '\"warmTransition\"'])) return true; + } catch (error) { + if (nativeJournalProofUnavailable(error)) throw error; + // The ordinary verifier owns unreadable legacy state. + } } } return false; @@ -4094,7 +4107,7 @@ type VerifiedWarmTransitionBinding = { stateFingerprint: string; }; -function readWarmTransitionSnapshot(root: string) { +async function readWarmTransitionSnapshot(root: string) { if ( ![root, resolve(root, "control-plane"), resolve(root, "runner")].every( isSafeNativeStateDirectory, @@ -4102,20 +4115,23 @@ function readWarmTransitionSnapshot(root: string) { ) { throw new Error("native_runner_warm_transition_recovery_unproven"); } + const directoryIdentities = [root, resolve(root, "runner"), resolve(root, "control-plane")] + .map(path => ({ path, identity: nativeStateDirectoryIdentity(path) })); try { const corePath = resolve(root, "control-plane", "control-plane-state.json"); const runnerPath = resolve(root, "runner", "runner-state.json"); - const core = readNativeJournalProjection(corePath); + const core = (await readNativeJournalProjection(corePath)); const runner = readBoundedNativeFile(runnerPath, NATIVE_RUNNER_STATE_MAX_BYTES, "native_runner_warm_transition_recovery_unproven"); - // Preserve the original canonical JSON [base64(core), base64(runner)] hash. - const digest = createHash("sha256").update('["'); - const rawCore = scanNativeStateFile(corePath, DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, chunk => { digest.update(chunk.toString("base64")); }); - digest.update('","'); - const rawRunner = scanNativeStateFile(runnerPath, NATIVE_RUNNER_STATE_MAX_BYTES, chunk => { digest.update(chunk.toString("base64")); }); - digest.update('"]'); - if (rawCore.sha256 !== core.sha256 || rawRunner.sha256 !== createHash("sha256").update(runner).digest("hex")) throw new Error("native_state_file_changed"); - return { controlPlaneState: core.value, runnerState: JSON.parse(runner.toString("utf8")) as unknown, stateFingerprint: digest.digest("hex") }; - } catch { + // The worker preserves the original canonical JSON [base64(core), base64(runner)] hash. + const fingerprint = await nativeStateBase64Fingerprint([ + { path: corePath, maximum: DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES }, + { path: runnerPath, maximum: NATIVE_RUNNER_STATE_MAX_BYTES }, + ]); + for (const directory of directoryIdentities) assertNativeStateDirectoryIdentity(directory.path, directory.identity); + if (fingerprint.fileSha256[0] !== core.sha256 || fingerprint.fileSha256[1] !== createHash("sha256").update(runner).digest("hex")) throw new Error("native_state_file_changed"); + return { controlPlaneState: core.value, runnerState: JSON.parse(runner.toString("utf8")) as unknown, stateFingerprint: fingerprint.sha256 }; + } catch (error) { + if (nativeJournalProofUnavailable(error)) throw error; throw new Error("native_runner_warm_transition_recovery_unproven"); } } @@ -4143,11 +4159,11 @@ async function verifyWarmTransitionRestart(input: { ) return deny(); const root = scopedRunnerdStateRoot(input.execution); - const snapshot = readWarmTransitionSnapshot(root); + const snapshot = (await readWarmTransitionSnapshot(root)); const artifact = readRunnerdArtifactBinding(resolvePaperclipRunnerBinary()); const controller = await currentNativeControllerIdentity(); const binding = input.execution.binding; - let finalAuthorityCheck: (() => boolean) | undefined; + let finalAuthorityCheck: (() => Promise) | undefined; const verified = await input.db.transaction(async (tx) => { const current = await tx .select() @@ -4439,9 +4455,9 @@ async function verifyWarmTransitionRestart(input: { root, input.execution, ); - finalAuthorityCheck = () => { + finalAuthorityCheck = async () => { if ( - readWarmTransitionSnapshot(root).stateFingerprint !== + (await readWarmTransitionSnapshot(root)).stateFingerprint !== snapshot.stateFingerprint ) return false; @@ -4463,7 +4479,7 @@ async function verifyWarmTransitionRestart(input: { finalProof.receipt.leaseExpiresAtUnixMs > Date.now() ); }; - if (!finalAuthorityCheck()) return deny(); + if (!await finalAuthorityCheck()) return deny(); return { runnerInstanceId: current.runnerInstanceId, environmentLeaseId: expectedEnvironmentLeaseId, @@ -4473,7 +4489,7 @@ async function verifyWarmTransitionRestart(input: { }); // A lock wait or the transaction completion itself can outlive a lease. // Never carry a timestamp sampled before those awaits into admission. - if (!finalAuthorityCheck?.()) return deny(); + if (!await finalAuthorityCheck?.()) return deny(); return verified; } @@ -4510,7 +4526,10 @@ async function migrateRunnerdStateRootForExecution(input: { if (!isSafeNativeStateDirectory(scoped)) { throw new Error("runner_state_directory_unsafe"); } - if (hasRetainedWarmTransitionEvidence(scoped)) { + const scopedDirectoryIdentity = nativeStateDirectoryIdentity(scoped); + const retainedWarmTransition = await hasRetainedWarmTransitionEvidence(scoped); + assertNativeStateDirectoryIdentity(scoped, scopedDirectoryIdentity); + if (retainedWarmTransition) { // A copied receipt is not restart/process authority. Keep both valid // unsupported and invalid forward evidence out of destructive legacy // migration until the exact transition admission path proves its owner. @@ -4518,7 +4537,8 @@ async function migrateRunnerdStateRootForExecution(input: { return verified; } await assertCleanupActivationCommitted(input.db, scoped, input.execution); - const identity = readRunnerdDurableIdentity(scoped); + const identity = (await readRunnerdDurableIdentity(scoped)); + assertNativeStateDirectoryIdentity(scoped, scopedDirectoryIdentity); if (!identity) { if (input.restartRecovery?.kind !== "reattach_existing_runner" && input.restartRecovery?.kind !== "reattach_remote_runner") { @@ -4534,7 +4554,9 @@ async function migrateRunnerdStateRootForExecution(input: { throw new Error("runner_state_identity_mismatch"); } if (input.restartRecovery?.kind === "bootstrap_incomplete") { - if (runnerdStateProvesIncompleteBootstrap(scoped)) { + const incompleteBootstrap = await runnerdStateProvesIncompleteBootstrap(scoped); + assertNativeStateDirectoryIdentity(scoped, scopedDirectoryIdentity); + if (incompleteBootstrap) { quarantineRunnerdStateRoot(scoped, "identity_indeterminate"); return; } @@ -4573,6 +4595,7 @@ async function migrateRunnerdStateRootForExecution(input: { allowVerifiedBackup: input.allowVerifiedBackup, allowRetainedWarmRunner: input.allowRetainedWarmRunner, }); + assertNativeStateDirectoryIdentity(scoped, scopedDirectoryIdentity); if ( verification !== "verified" && verification !== "retained_warm_runner" @@ -4595,7 +4618,9 @@ async function migrateRunnerdStateRootForExecution(input: { legacyRunnerdStateRoot(input.execution), ]) { if (!existsSync(legacy)) continue; - const identity = readRunnerdDurableIdentity(legacy); + const legacyDirectoryIdentity = nativeStateDirectoryIdentity(legacy); + const identity = (await readRunnerdDurableIdentity(legacy)); + assertNativeStateDirectoryIdentity(legacy, legacyDirectoryIdentity); if ( !identity || !durableIdentityMatchesSession(identity, input.execution) @@ -4615,6 +4640,7 @@ async function migrateRunnerdStateRootForExecution(input: { allowVerifiedBackup: input.allowVerifiedBackup, allowRetainedWarmRunner: input.allowRetainedWarmRunner, }); + assertNativeStateDirectoryIdentity(legacy, legacyDirectoryIdentity); if ( verification !== "verified" && verification !== "retained_warm_runner" @@ -4630,12 +4656,12 @@ async function migrateRunnerdStateRootForExecution(input: { } verifiedPriorRunId = identity.runId; } - migrateLegacyRunnerdStateRoot({ + (await migrateLegacyRunnerdStateRoot({ legacy, scoped, execution: input.execution, ...(verifiedPriorRunId ? { verifiedPriorRunId } : {}), - }); + })); return; } } @@ -4687,13 +4713,17 @@ async function recoverQuiescentRunnerdState(input: { runId: string; processPid: number; processGroupId: number; + providerProcessIds: unknown[]; + directoryIdentities: Array<{ path: string; identity: NativeStateDirectoryIdentity }>; }> = []; for (const root of candidates) { if (!isSafeNativeStateDirectory(root)) continue; - const identity = readRunnerdDurableIdentity(root); + const identity = (await readRunnerdDurableIdentity(root)); if (!durableIdentityMatchesSession(identity, input.execution)) continue; if (identity.runId === input.execution.binding.runId) continue; try { + const directoryIdentities = [root, resolve(root, "runner"), resolve(root, "control-plane")] + .map(path => ({ path, identity: nativeStateDirectoryIdentity(path) })); const readState = (directory: string, name: string) => { if (!isSafeNativeStateDirectory(resolve(root, directory))) throw new Error("unsafe_recovery_state"); @@ -4704,7 +4734,7 @@ async function recoverQuiescentRunnerdState(input: { ).toString("utf8"); }; const runnerBytes = readState("runner", "runner-state.json"); - const controlProof = readNativeJournalProjection(resolve(root, "control-plane", "control-plane-state.json")); + const controlProof = (await readNativeJournalProjection(resolve(root, "control-plane", "control-plane-state.json"))); const controlSha256 = controlProof.sha256; const providerBytes = readState("runner", "codex-provider-state.json"); const runner = record(JSON.parse(runnerBytes)); @@ -4830,10 +4860,12 @@ async function recoverQuiescentRunnerdState(input: { // No mutation if any evidence changed while the database was read. if ( runnerBytes !== readState("runner", "runner-state.json") || - controlSha256 !== scanNativeStateFile(resolve(root, "control-plane", "control-plane-state.json"), DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES).sha256 || + controlSha256 !== (await scanNativeStateFile(resolve(root, "control-plane", "control-plane-state.json"), DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES)).sha256 || providerBytes !== readState("runner", "codex-provider-state.json") ) continue; + for (const directory of directoryIdentities) assertNativeStateDirectoryIdentity(directory.path, directory.identity); + if (processIds.some(pid => !localProcessDefinitelyGone(pid))) continue; verified.push({ root, runnerBytes, @@ -4843,8 +4875,11 @@ async function recoverQuiescentRunnerdState(input: { runId: identity.runId, processPid: prior.processPid!, processGroupId: prior.processGroupId!, + providerProcessIds: processIds, + directoryIdentities, }); - } catch { + } catch (error) { + if (nativeJournalProofUnavailable(error)) throw error; // Unreadable, unsafe, unscoped, or unprovable state stays quarantined. } } @@ -4872,15 +4907,17 @@ async function recoverQuiescentRunnerdState(input: { ]) { if ( (relativePath === "control-plane/control-plane-state.json" - ? scanNativeStateFile(resolve(candidate.root, relativePath), DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES).sha256 + ? (await scanNativeStateFile(resolve(candidate.root, relativePath), DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES)).sha256 : readBoundedNativeFile(resolve(candidate.root, relativePath!), NATIVE_RUNNER_STATE_MAX_BYTES, "recovery_state_too_large").toString("utf8")) !== expected ) { throw new Error("runner_state_identity_mismatch"); } } + for (const directory of candidate.directoryIdentities) assertNativeStateDirectoryIdentity(directory.path, directory.identity); if ( !localProcessDefinitelyGone(candidate.processPid) || - !localProcessDefinitelyGone(candidate.processGroupId, true) + !localProcessDefinitelyGone(candidate.processGroupId, true) || + candidate.providerProcessIds.some(pid => !localProcessDefinitelyGone(pid)) ) { throw new Error("runner_state_identity_mismatch"); } @@ -4923,14 +4960,14 @@ function localProcessDefinitelyGone(value: unknown, group = false): boolean { } } -export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { +export async function runnerdStateProvesIncompleteBootstrap(root: string): Promise { try { const statePath = resolve( root, "control-plane", "control-plane-state.json", ); - const state = record(readNativeJournalProjection(statePath).value); + const state = record((await readNativeJournalProjection(statePath)).value); const commands = Array.isArray(state.commands) ? state.commands.map(record) : []; @@ -4948,11 +4985,29 @@ export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { committedEvents.length === 0 && onlyUnconsumedBootstrapCommands ); - } catch { + } catch (error) { + if (nativeJournalProofUnavailable(error)) throw error; return false; } } +function nativeJournalProofUnavailable(error: unknown): boolean { + return error instanceof Error && (error.message.startsWith("native_journal_worker_") + || error.message === "native_state_file_changed" + || error.message === "native_journal_projection_budget_exceeded"); +} + +type NativeStateDirectoryIdentity = { dev: number; ino: number }; +function nativeStateDirectoryIdentity(path: string): NativeStateDirectoryIdentity { + const stat = lstatSync(path); + if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("runner_state_directory_unsafe"); + return { dev: stat.dev, ino: stat.ino }; +} +function assertNativeStateDirectoryIdentity(path: string, expected: NativeStateDirectoryIdentity): void { + const current = nativeStateDirectoryIdentity(path); + if (current.dev !== expected.dev || current.ino !== expected.ino) throw new Error("runner_state_directory_unsafe"); +} + function isSafeNativeStateDirectory(path: string): boolean { if (!existsSync(path)) return false; const stats = lstatSync(path); @@ -4966,20 +5021,28 @@ type RunnerdDurableIdentity = Record & { environmentLeaseId: string; }; -function readRunnerdDurableIdentity( +async function readRunnerdDurableIdentity( root: string, -): Record | null { +): Promise | null> { if (!isSafeNativeStateDirectory(root)) return null; const controlPlaneRoot = resolve(root, "control-plane"); if (!isSafeNativeStateDirectory(controlPlaneRoot)) return null; const statePath = resolve(controlPlaneRoot, "control-plane-state.json"); if (!existsSync(statePath)) return null; + const rootIdentity = nativeStateDirectoryIdentity(root); + const controlIdentity = nativeStateDirectoryIdentity(controlPlaneRoot); try { - const state = record(readNativeJournalProjection(statePath, "identity").value); + const state = record((await readNativeJournalProjection(statePath, "identity")).value); if (state.schema !== RUNNERD_CONTROL_PLANE_STATE_SCHEMA) return null; return record(state.identity); - } catch { + } catch (error) { + // Resource pressure or a changing snapshot is not corrupt durable identity. + if (nativeJournalProofUnavailable(error)) throw error; return null; + } finally { + // A replaced directory is unsafe, not merely missing identity that can be quarantined. + assertNativeStateDirectoryIdentity(root, rootIdentity); + assertNativeStateDirectoryIdentity(controlPlaneRoot, controlIdentity); } } @@ -5016,7 +5079,7 @@ function durableIdentityMatchesSession( * session scope. Ambiguous company/session-only state can never be claimed by * another agent, workspace, or provider profile. */ -function runnerdStateRoot(execution: NativeExecutionInput): string { +async function runnerdStateRoot(execution: NativeExecutionInput): Promise { const scoped = scopedRunnerdStateRoot(execution); if (existsSync(scoped)) { if (!isSafeNativeStateDirectory(scoped)) { @@ -5028,33 +5091,36 @@ function runnerdStateRoot(execution: NativeExecutionInput): string { legacyCompanyRunnerdStateRoot(execution), legacyRunnerdStateRoot(execution), ]) { - const migrated = migrateLegacyRunnerdStateRoot({ + const migrated = (await migrateLegacyRunnerdStateRoot({ legacy, scoped, execution, - }); - if (migrated) return migrated; + })); + if (migrated) { + nativeStateDirectoryIdentity(migrated); + return migrated; + } } return scoped; } -function loadRunnerdDurableBinding( +async function loadRunnerdDurableBinding( execution: NativeExecutionInput, transition?: VerifiedWarmTransitionBinding, -): { +): Promise<{ runnerInstanceId: string; environmentLeaseId: string; -} | null { +} | null> { if (transition) { if ( - readWarmTransitionSnapshot(runnerdStateRoot(execution)) + (await readWarmTransitionSnapshot((await runnerdStateRoot(execution)))) .stateFingerprint !== transition.stateFingerprint ) { throw new Error("native_runner_warm_transition_recovery_unproven"); } return transition; } - const identity = readRunnerdDurableIdentity(runnerdStateRoot(execution)); + const identity = (await readRunnerdDurableIdentity((await runnerdStateRoot(execution)))); // The run id is intentionally different during a continuation. Reuse only // the verified runner/lease binding from the same company-scoped durable // session root; rotateLocalAuthorityEpoch then requires that exact binding @@ -7318,7 +7384,7 @@ async function executePaperclipNativeSessionWithinScope( }); } const durableRunnerBinding = input.useRunnerd - ? loadRunnerdDurableBinding(input.execution, retainedTransition) + ? (await loadRunnerdDurableBinding(input.execution, retainedTransition)) : null; const effectiveRunnerInstanceId = durableRunnerBinding?.runnerInstanceId ?? input.runnerInstanceId; @@ -10385,7 +10451,7 @@ export async function createRunnerdBackend(input: { if ( executingRunnerdSessionScopes.get(sessionScopeId) !== input.execution.binding.runId || - hasRetainedWarmTransitionEvidence(scopedRunnerdStateRoot(input.execution)) + (await hasRetainedWarmTransitionEvidence(scopedRunnerdStateRoot(input.execution))) ) { retainedTransition = await migrateRunnerdStateRootForExecution({ db: input.db, @@ -10510,10 +10576,10 @@ async function createRunnerdBackendWithinSessionClaim( authorityEpoch.revoke(); throw error; } - const root = runnerdStateRoot(input.execution); - const durableIdentity = readRunnerdDurableIdentity(root); + const root = (await runnerdStateRoot(input.execution)); + const durableIdentity = (await readRunnerdDurableIdentity(root)); const durableBinding = retainedTransition - ? loadRunnerdDurableBinding(input.execution, retainedTransition) + ? (await loadRunnerdDurableBinding(input.execution, retainedTransition)) : durableIdentityMatchesSession(durableIdentity, input.execution) ? durableIdentity : null; @@ -11307,11 +11373,11 @@ async function createRunnerdBackendWithinSessionClaim( const claimUntouchedSessionInResumedLease = async (): Promise => { if (!remoteCommandRunner || !remoteRuntimeRoot || !remoteSessionRoot) return false; - const identity = readRunnerdDurableIdentity(root); + const identity = (await readRunnerdDurableIdentity(root)); if (!durableIdentityMatchesExecution(identity, input.execution) || identity?.runnerInstanceId !== effectiveRunnerInstanceId || identity?.environmentLeaseId !== effectiveEnvironmentLeaseId || - !runnerdStateProvesIncompleteBootstrap(root)) return false; + !(await runnerdStateProvesIncompleteBootstrap(root))) return false; // A reusable workspace may have failed before any harness was created. // Claim this exact new session atomically under readable real directories. // Missing files inside an existing session never authorize a fresh start. @@ -12064,9 +12130,11 @@ async function createRunnerdBackendWithinSessionClaim( PAPERCLIP_WORKSPACE_CWD: input.execution.workspace.cwd, }; const archiveContinuityState = async () => { - if (hasRetainedWarmTransitionEvidence(root)) { + const rootIdentity = nativeStateDirectoryIdentity(root); + if ((await hasRetainedWarmTransitionEvidence(root))) { throw new Error("native_runner_warm_transition_recovery_unproven"); } + assertNativeStateDirectoryIdentity(root, rootIdentity); const archiveToken = `${Date.now()}-${randomUUID()}`; const archiveRoot = resolve(root, "continuity-breaks", archiveToken); mkdirSync(archiveRoot, { recursive: true, mode: 0o700 });