diff --git a/doc/architecture/paperclip-runner-compatibility.md b/doc/architecture/paperclip-runner-compatibility.md index effabc1eba4..59a047303e5 100644 --- a/doc/architecture/paperclip-runner-compatibility.md +++ b/doc/architecture/paperclip-runner-compatibility.md @@ -216,6 +216,24 @@ Catalog generation and production authorization are separate steps. effect. - Receipts redact credentials, private provider payloads, and hidden identity. +## Durable journal proof reads + +The durable writer and controller accept the same 192 MiB raw journal limit. +Identity, cleanup, warm-transition, and quiescent recovery reads scan the JSON +with a fixed 48 KiB input buffer. They retain only the fields used to prove +ownership, completion, and settled side effects. Generic tool output and task +text do not become a second in-memory journal. One synchronous scan runs at a +time; every retained node and string shares an 8 MiB conservative allocation +budget, including fields materialized after command or event classification. + +Proof reads validate discarded JSON as well as retained fields. They preserve +original types, duplicate-key last-wins behavior, complete semantic evidence, +and complete `run.attach` commands and results. The full raw file digest covers +all omitted bytes. Warm-transition fingerprints keep their existing byte-exact +algorithm. Symlinks, changing files, invalid JSON, excessive nesting, or excess +essential evidence fail closed. A projection is read-only evidence and must +never be written back as a replacement journal. + ## Required compatibility matrix Each runner-related pull request updates only rows that it can execute. The diff --git a/server/src/services/native-runtime/native-journal-projection.test.ts b/server/src/services/native-runtime/native-journal-projection.test.ts new file mode 100644 index 00000000000..eef76ad4732 --- /dev/null +++ b/server/src/services/native-runtime/native-journal-projection.test.ts @@ -0,0 +1,228 @@ +import { createHash } from "node:crypto"; +import { + mkdtempSync, + mkdirSync, + readFileSync, + rmSync, + symlinkSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + readNativeJournalProjection, + scanNativeStateFile, + NATIVE_JOURNAL_PROJECTION_BUDGET, +} from "./native-journal-projection.js"; +import { runnerdStateProvesIncompleteBootstrap } from "./native-session-executor.js"; +const roots: string[] = []; +afterEach(() => { + vi.restoreAllMocks(); + for (const root of roots.splice(0)) + rmSync(root, { recursive: true, force: true }); +}); +function fixture(raw: string) { + const root = mkdtempSync(join(tmpdir(), "journal-projection-")); + roots.push(root); + mkdirSync(join(root, "control-plane")); + const path = join(root, "control-plane/control-plane-state.json"); + writeFileSync(path, raw); + return { root, path }; +} +const event = (type: string, payload: unknown) => ({ + eventType: type, + sourceSeq: 1, + sourceEventId: "event-1", + priority: 1, + deliveryCount: 1, + logicalEffectCount: 1, + envelope: { + runId: "run-1", + normalizedSessionId: "session-1", + payload: { eventType: type, runId: "run-1", payload }, + }, +}); +describe("bounded native journal proof projection", () => { + it("does not materialize large tool history during a real controller admission read", () => { + const { root } = fixture( + JSON.stringify({ + schema: "paperclip.runner.durable.control-plane-state.v1", + identity: { runId: "run-1" }, + connectionCount: 0, + commands: [], + committedEvents: [ + event("item.completed", { text: "x".repeat(24 * 1024 * 1024) }), + ], + }), + ); + const parse = JSON.parse; + const sizes: number[] = []; + vi.spyOn(JSON, "parse").mockImplementation((text, reviver) => { + sizes.push(text.length); + return parse(text, reviver); + }); + expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(false); + expect(Math.max(...sizes)).toBeLessThan(NATIVE_JOURNAL_PROJECTION_BUDGET); + }); + it("hashes omitted bytes while retaining semantic and process proof", () => { + const semantic = { + semantic_tool: { + operationId: "paperclip_finish", + input: { summary: "exact" }, + }, + }; + const state = { + schema: "schema", + identity: { runId: "exact" }, + commands: [ + { + type: "run.attach", + payload: { nested: { prompt: "must retain" } }, + result: { status: "completed" }, + }, + { + type: "turn.start", + payload: { prompt: "discard" }, + result: { result: { providerTurnId: "turn" } }, + }, + ], + committedEvents: [ + event("item.completed", { + text: "x".repeat(3 * 1024 * 1024), + processId: 42, + }), + event("semantic_tool.input", semantic), + ], + }; + const raw = JSON.stringify(state), + { path } = fixture(raw), + proof = readNativeJournalProjection(path); + expect(proof.sha256).toBe(createHash("sha256").update(raw).digest("hex")); + expect(proof.retainedBudgetBytes).toBeLessThan(32 * 1024); + expect(proof.value).toMatchObject({ + commands: [ + { + payload: state.commands[0]!.payload, + result: state.commands[0]!.result, + }, + { payload: {}, result: { result: { providerTurnId: "turn" } } }, + ], + committedEvents: [ + { envelope: { payload: { payload: { processId: 42 } } } }, + { envelope: { payload: { payload: semantic } } }, + ], + }); + expect(JSON.stringify(proof.value)).not.toContain('"text"'); + }); + it("preserves duplicate-key last-wins semantics and original malformed types", () => { + const { path } = fixture( + '{"schema":"old","schema":"new","identity":{"runId":"wrong","runId":"right","__proto__":{"polluted":true}},"commands":[{"type":"run.attach","payload":null,"result":[]},{"type":"turn.start","payload":"bad","result":false}],"committedEvents":[{"envelope":null}]}', + ); + const proof = readNativeJournalProjection(path).value; + expect(proof).toMatchObject({ + schema: "new", + identity: { runId: "right" }, + commands: [ + { payload: null, result: [] }, + { payload: "bad", result: false }, + ], + committedEvents: [{ envelope: null }], + }); + expect(({} as Record).polluted).toBeUndefined(); + }); + it.each([ + '{"identity":{},"ignored":"bad\\q"}', + '{"identity":{},"ignored":[1,]}', + '{"identity":{},"ignored":01}', + '{"identity":{},"ignored":"unterminated}', + '{"identity":{}} trailing', + '{"identity":{},"ignored":{"x":1,}}', + ])("rejects malformed JSON even in discarded bytes: %s", (raw) => { + const { path } = fixture(raw); + expect(() => readNativeJournalProjection(path, "identity")).toThrow(); + }); + it("rejects excessive essential evidence before parsing a large token", () => { + const { path } = fixture( + JSON.stringify({ + schema: "s", + identity: { runId: "x".repeat(4 * 1024 * 1024) }, + }), + ); + expect(() => readNativeJournalProjection(path)).toThrow( + "native_journal_projection_budget_exceeded", + ); + }); + it("bounds complete semantic evidence without silently dropping it", () => { + const { path } = fixture( + JSON.stringify({ + schema: "s", + commands: [ + { + type: "semantic_tool.result", + payload: { input: "x".repeat(4 * 1024 * 1024) }, + result: null, + }, + ], + }), + ); + expect(() => readNativeJournalProjection(path)).toThrow( + "native_journal_projection_budget_exceeded", + ); + }); + it("keeps the turn identity without materializing unrelated turn result text", () => { + const { path } = fixture( + JSON.stringify({ + commands: [ + { + type: "turn.start", + payload: {}, + result: { + result: { + providerTurnId: "turn-1", + output: "x".repeat(12 * 1024 * 1024), + }, + }, + }, + ], + }), + ); + const proof = readNativeJournalProjection(path); + expect(proof.value).toMatchObject({ + commands: [{ result: { result: { providerTurnId: "turn-1" } } }], + }); + expect(proof.retainedBudgetBytes).toBeLessThan(4096); + }); + it("rejects excessive nesting even in discarded history", () => { + const { path } = fixture( + '{"ignored":' + "[".repeat(130) + "0" + "]".repeat(130) + "}", + ); + expect(() => readNativeJournalProjection(path, "identity")).toThrow( + "native_journal_projection_budget_exceeded", + ); + }); + it("rejects symlinks and detects file changes during incremental scans", () => { + const { path, root } = fixture('{"identity":{}}'); + const link = join(root, "link"); + symlinkSync(path, link); + expect(() => readNativeJournalProjection(link)).toThrow(); + expect(() => + scanNativeStateFile(path, 1024, () => + writeFileSync(path, "changed and larger"), + ), + ).toThrow("native_state_file_changed"); + }); + it("streams byte-identical base64 chunks, including arbitrary tails", () => { + for (const size of [0, 1, 2, 3, 49151, 49152, 49153, 100000]) { + const { path } = fixture("x".repeat(size)); + let encoded = ""; + const proof = scanNativeStateFile( + path, + 200000, + (chunk) => (encoded += chunk.toString("base64")), + ); + expect(encoded).toBe(readFileSync(path).toString("base64")); + expect(proof.byteSize).toBe(size); + } + }); +}); diff --git a/server/src/services/native-runtime/native-journal-projection.ts b/server/src/services/native-runtime/native-journal-projection.ts new file mode 100644 index 00000000000..b3f39b00365 --- /dev/null +++ b/server/src/services/native-runtime/native-journal-projection.ts @@ -0,0 +1,389 @@ +import { createHash } from "node:crypto"; +import { closeSync, constants, fstatSync, openSync, readSync } from "node:fs"; +import { DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES } from "../../vendor/paperclip-runner/index.js"; + +// These are read-only proof projections, never replacement journal contents. +// The raw writer limit still applies, including to bytes omitted from the proof. +export const NATIVE_JOURNAL_PROJECTION_BUDGET = 8 * 1024 * 1024; +const CHUNK = 48 * 1024; // divisible by three for the existing base64 fingerprint +const OMIT = Symbol("omitted"); +const REF = Symbol("raw JSON span"); +type Span = { [REF]: true; start: number; end: number }; +type Mode = "keep" | "skip" | "span"; +type Select = (path: string[]) => Mode; +const object = (v: unknown): v is Record => + v !== null && typeof v === "object" && !Array.isArray(v); +const isSpan = (v: unknown): v is Span => object(v) && REF in v; +const full: Select = () => "keep"; +const stateFields = new Set([ + "schema", + "identity", + "tickets", + "leases", + "commands", + "committedEvents", + "ackedSourceSeq", + "connectionCount", + "commandDeliveryCounts", + "runAttachTemplate", + "warmTransition", + "completedWarmTransition", + "replayDeliveries", + "duplicateCommandResults", + "freshBootstraps", + "malformedFrames", +]); +const essentialEvents = new Set([ + "session.started", + "session.resumed", + "session.reconciled", + "turn.accepted", + "semantic_tool.input", + "semantic_tool.result", + "mcp_app.tool_input", + "harness.diagnostic", +]); + +/** One synchronous scanner can execute at a time. All retained nodes/strings, + * including later materialized spans, share the same conservative byte budget. */ +class Budget { + used = 0; + charge(bytes: number) { + this.used += bytes; + if (this.used > NATIVE_JOURNAL_PROJECTION_BUDGET) + throw new Error("native_journal_projection_budget_exceeded"); + } +} + +class JsonScanner { + private buffer = Buffer.allocUnsafe(CHUNK); + private begin = -1; + private length = 0; + offset: number; + constructor( + private fd: number, + private end: number, + private budget: Budget, + start = 0, + private digest?: ReturnType, + ) { + this.offset = start; + } + private peek(): number { + if (this.offset >= this.end) return -1; + if (this.offset < this.begin || this.offset >= this.begin + this.length) { + this.begin = this.offset; + this.length = readSync( + this.fd, + this.buffer, + 0, + Math.min(CHUNK, this.end - this.offset), + this.offset, + ); + if (!this.length) throw new Error("native_state_file_changed"); + this.digest?.update(this.buffer.subarray(0, this.length)); + } + return this.buffer[this.offset - this.begin]!; + } + private space() { + while ([32, 9, 10, 13].includes(this.peek())) this.offset++; + } + private expect(byte: number) { + if (this.peek() !== byte) throw new Error("native_journal_invalid_json"); + this.offset++; + } + private token(start: number, end: number, key = false): unknown { + const size = end - start; + if (key && size > 4096) + throw new Error("native_journal_projection_budget_exceeded"); + this.budget.charge(size * 3 + 64); + const bytes = Buffer.allocUnsafe(size); + let count = 0; + while (count < size) { + const read = readSync(this.fd, bytes, count, size - count, start + count); + if (!read) throw new Error("native_state_file_changed"); + count += read; + } + return JSON.parse(bytes.toString("utf8")); + } + private string(capture: boolean, key = false): unknown { + const start = this.offset; + this.expect(34); + for (;;) { + const c = this.peek(); + this.offset++; + if (c === 34) break; + if (c < 32) throw new Error("native_journal_invalid_json"); + if (c === 92) { + const escape = this.peek(); + this.offset++; + if (escape === 117) { + for (let n = 0; n < 4; n++) { + const hex = this.peek(); + if ( + !( + (hex >= 48 && hex <= 57) || + (hex >= 65 && hex <= 70) || + (hex >= 97 && hex <= 102) + ) + ) + throw new Error("native_journal_invalid_json"); + this.offset++; + } + } else if (![34, 92, 47, 98, 102, 110, 114, 116].includes(escape)) + throw new Error("native_journal_invalid_json"); + } + } + return capture ? this.token(start, this.offset, key) : OMIT; + } + value(path: string[], select: Select, depth = 0, forced?: Mode): unknown { + if (depth > 128) + throw new Error("native_journal_projection_budget_exceeded"); + this.space(); + const start = this.offset, + mode = forced ?? select(path), + c = this.peek(); + if (mode === "span") { + this.value(path, select, depth, "skip"); + this.budget.charge(96); + return { [REF]: true, start, end: this.offset } satisfies Span; + } + const keep = mode === "keep"; + if (keep) this.budget.charge(64); + if (c === 34) return this.string(keep); + if (c === 123 || c === 91) { + const array = c === 91; + this.offset++; + this.space(); + const result: unknown[] | Record = array ? [] : {}; + const close = array ? 93 : 125; + if (this.peek() === close) { + this.offset++; + return keep ? result : OMIT; + } + let index = 0; + for (;;) { + this.space(); + let key: string; + if (array) key = String(index++); + else { + const parsed = this.string(keep, true); + key = keep ? (parsed as string) : ""; + this.space(); + this.expect(58); + } + const child = this.value( + keep ? [...path, key] : path, + select, + depth + 1, + keep ? undefined : "skip", + ); + if (keep && child !== OMIT) + Object.defineProperty(result, key, { + value: child, + configurable: true, + enumerable: true, + writable: true, + }); + this.space(); + if (this.peek() === close) { + this.offset++; + return keep ? result : OMIT; + } + this.expect(44); + } + } + for (const [first, literal, value] of [ + [116, "true", true], + [102, "false", false], + [110, "null", null], + ] as const) { + if (c === first) { + for (const ch of literal) this.expect(ch.charCodeAt(0)); + return keep ? value : OMIT; + } + } + let number = ""; + while ( + this.peek() >= 0 && + ![32, 9, 10, 13, 44, 93, 125].includes(this.peek()) + ) { + if (number.length >= 128) throw new Error("native_journal_invalid_json"); + number += String.fromCharCode(this.peek()); + this.offset++; + } + if (!/^-?(?:0|[1-9]\d*)(?:\.\d+)?(?:[eE][+-]?\d+)?$/.test(number)) + throw new Error("native_journal_invalid_json"); + return keep ? JSON.parse(number) : OMIT; + } + parse(select: Select): unknown { + const result = this.value([], select); + this.space(); + if (this.offset !== this.end) + throw new Error("native_journal_invalid_json"); + return result; + } +} + +function unchanged( + before: ReturnType, + after: ReturnType, +) { + if ( + before.size !== after.size || + before.mtimeMs !== after.mtimeMs || + before.ino !== after.ino + ) + throw new Error("native_state_file_changed"); +} + +export function readNativeJournalProjection( + path: string, + purpose: "identity" | "evidence" = "evidence", +) { + const fd = openSync(path, constants.O_RDONLY | constants.O_NOFOLLOW); + try { + const before = fstatSync(fd); + if ( + !before.isFile() || + before.size > DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES + ) + throw new Error("native_journal_too_large"); + const budget = new Budget(), + digest = createHash("sha256"); + const select: Select = (parts) => { + if ( + parts.length === 1 && + (purpose === "identity" + ? !["schema", "identity"].includes(parts[0]!) + : !stateFields.has(parts[0]!)) + ) + return "skip"; + if ( + parts[0] === "commands" && + parts.length === 3 && + ["payload", "result"].includes(parts[2]!) + ) + return "span"; + if ( + parts[0] === "committedEvents" && + parts.length === 5 && + parts.slice(2).join("/") === "envelope/payload/payload" + ) + return "span"; + return "keep"; + }; + const value = new JsonScanner(fd, before.size, budget, 0, digest).parse( + select, + ); + const materialize = (v: unknown, selection: Select): unknown => + isSpan(v) + ? new JsonScanner(fd, v.end, budget, v.start).parse(selection) + : v; + // A projection preserves original primitive/array/object types. It must not + // turn malformed authority into a valid empty object. + const fields = + (names: string[]): Select => + (parts) => + parts.length === 1 && !names.includes(parts[0]!) ? "skip" : "keep"; + if (object(value)) { + if (Array.isArray(value.commands)) + for (const command of value.commands) { + if (!object(command)) continue; + const complete = + command.type === "run.attach" || + command.type === "semantic_tool.result"; + if ("payload" in command) + command.payload = materialize( + command.payload, + complete + ? full + : fields( + command.type === "run.prepare" + ? ["completionContract"] + : [], + ), + ); + if ("result" in command) + command.result = materialize( + command.result, + complete + ? full + : command.type === "turn.start" + ? (parts) => + (parts.length === 1 && parts[0] !== "result") || + (parts.length === 2 && parts[1] !== "providerTurnId") + ? "skip" + : "keep" + : fields([]), + ); + } + if (Array.isArray(value.committedEvents)) + for (const entry of value.committedEvents) { + if ( + !object(entry) || + !object(entry.envelope) || + !object(entry.envelope.payload) + ) + continue; + const event = entry.envelope.payload; + if ("payload" in event) + event.payload = materialize( + event.payload, + essentialEvents.has(String(event.eventType)) + ? full + : fields(["processId"]), + ); + } + } + unchanged(before, fstatSync(fd)); + return { + value, + sha256: digest.digest("hex"), + byteSize: before.size, + retainedBudgetBytes: budget.used, + }; + } finally { + closeSync(fd); + } +} + +/** Hash the complete source, including discarded history, without retaining it. */ +export function scanNativeStateFile( + path: string, + maximum: number, + consume?: (chunk: Buffer) => void, +) { + const fd = openSync(path, constants.O_RDONLY | constants.O_NOFOLLOW); + try { + const before = fstatSync(fd); + if (!before.isFile() || before.size > maximum) + throw new Error("native_state_file_too_large"); + const buffer = Buffer.allocUnsafe(CHUNK), + digest = createHash("sha256"); + let offset = 0; + while (offset < before.size) { + const length = Math.min(CHUNK, before.size - offset); + let count = 0; + while (count < length) { + const read = readSync( + fd, + buffer, + count, + length - count, + offset + count, + ); + if (!read) throw new Error("native_state_file_changed"); + count += read; + } + const chunk = buffer.subarray(0, count); + digest.update(chunk); + consume?.(chunk); + offset += count; + } + unchanged(before, fstatSync(fd)); + return { sha256: digest.digest("hex"), byteSize: before.size }; + } finally { + closeSync(fd); + } +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 9f79d00df23..b5ae041c94c 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { readNativeJournalProjection, scanNativeStateFile } from "./native-journal-projection.js"; import { isSupportedRemoteCodexVersion, parseCodexCliVersion, @@ -1790,19 +1791,13 @@ function cleanupStateSnapshot(root: string, providerFile = "codex-provider-state throw new Error("native_cleanup_maintenance_unproven"); } const files = [...CLEANUP_CANONICAL_FILES.slice(0, 2), `runner/${providerFile}`]; - const bytes = files.map((file) => - readBoundedNativeFile( - resolve(root, file), - nativeStateFileMaxBytes(file), - "native_cleanup_maintenance_unproven", - ), + const controlProof = readNativeJournalProjection(resolve(root, files[0]!)); + const bytes = files.slice(1).map((file) => + readBoundedNativeFile(resolve(root, file), NATIVE_RUNNER_STATE_MAX_BYTES, "native_cleanup_maintenance_unproven"), ); - const [control, runner, provider] = bytes.map((value) => - record(JSON.parse(value.toString("utf8"))), - ); - const fileSha256 = bytes.map((value) => - createHash("sha256").update(value).digest("hex"), - ) as [string, string, string]; + const control = record(controlProof.value); + const [runner, provider] = bytes.map(value => record(JSON.parse(value.toString("utf8")))); + const fileSha256 = [controlProof.sha256, ...bytes.map(value => createHash("sha256").update(value).digest("hex"))] as [string, string, string]; return { control: control!, runner: runner!, @@ -4063,44 +4058,30 @@ async function verifyPriorRunnerdStateForSessionScope(input: { function hasRetainedWarmTransitionEvidence(root: string): boolean { for (const [directory, filename, maximum] of [ - [ - "control-plane", - "control-plane-state.json", - DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, - ], + ["control-plane", "control-plane-state.json", DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES], ["runner", "runner-state.json", NATIVE_RUNNER_STATE_MAX_BYTES], ] as const) { const path = resolve(root, directory, filename); if (!lstatSync(path, { throwIfNoEntry: false })) continue; - let bytes: string; try { - bytes = readBoundedNativeFile( - path, - maximum, - "runner_state_too_large", - ).toString("utf8"); + const state = record(directory === "control-plane" + ? 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 { - // The ordinary verifier still owns unreadable legacy state. Inspect the - // other file before deciding whether this is a forward-protocol fence. - continue; - } - try { - const state = record(JSON.parse(bytes)); - 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 { - // This is detection only, never admission. A damaged forward receipt - // must remain available to its owner rather than become legacy state. - if ( - bytes.includes("warm-transition.v1") || - bytes.includes('"warmTransition"') - ) - return true; + // 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. */ } } } return false; @@ -4121,25 +4102,19 @@ function readWarmTransitionSnapshot(root: string) { ) { throw new Error("native_runner_warm_transition_recovery_unproven"); } - const core = readBoundedNativeFile( - resolve(root, "control-plane", "control-plane-state.json"), - DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, - "native_runner_warm_transition_recovery_unproven", - ); - const runner = readBoundedNativeFile( - resolve(root, "runner", "runner-state.json"), - NATIVE_RUNNER_STATE_MAX_BYTES, - "native_runner_warm_transition_recovery_unproven", - ); try { - return { - controlPlaneState: JSON.parse(core.toString("utf8")) as unknown, - runnerState: JSON.parse(runner.toString("utf8")) as unknown, - stateFingerprint: nativeSha256([ - core.toString("base64"), - runner.toString("base64"), - ]), - }; + const corePath = resolve(root, "control-plane", "control-plane-state.json"); + const runnerPath = resolve(root, "runner", "runner-state.json"); + const core = 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 { throw new Error("native_runner_warm_transition_recovery_unproven"); } @@ -4706,7 +4681,7 @@ async function recoverQuiescentRunnerdState(input: { const verified: Array<{ root: string; runnerBytes: string; - controlBytes: string; + controlSha256: string; providerBytes: string; runner: Record; runId: string; @@ -4729,13 +4704,11 @@ async function recoverQuiescentRunnerdState(input: { ).toString("utf8"); }; const runnerBytes = readState("runner", "runner-state.json"); - const controlBytes = readState( - "control-plane", - "control-plane-state.json", - ); + const controlProof = 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)); - const control = record(JSON.parse(controlBytes)); + const control = record(controlProof.value); const provider = record(JSON.parse(providerBytes)); const commands = control.commands; const events = control.committedEvents; @@ -4857,15 +4830,14 @@ async function recoverQuiescentRunnerdState(input: { // No mutation if any evidence changed while the database was read. if ( runnerBytes !== readState("runner", "runner-state.json") || - controlBytes !== - readState("control-plane", "control-plane-state.json") || + controlSha256 !== 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; verified.push({ root, runnerBytes, - controlBytes, + controlSha256, providerBytes, runner, runId: identity.runId, @@ -4896,14 +4868,12 @@ async function recoverQuiescentRunnerdState(input: { for (const [relativePath, expected] of [ ["runner/runner-state.json", candidate.runnerBytes], ["runner/codex-provider-state.json", candidate.providerBytes], - ["control-plane/control-plane-state.json", candidate.controlBytes], + ["control-plane/control-plane-state.json", candidate.controlSha256], ]) { if ( - readBoundedNativeFile( - resolve(candidate.root, relativePath!), - nativeStateFileMaxBytes(relativePath!), - "recovery_state_too_large", - ).toString("utf8") !== expected + (relativePath === "control-plane/control-plane-state.json" + ? 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"); } @@ -4960,15 +4930,7 @@ export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { "control-plane", "control-plane-state.json", ); - const state = record( - JSON.parse( - readBoundedNativeFile( - statePath, - DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, - "runner_durable_identity_too_large", - ).toString("utf8"), - ), - ); + const state = record(readNativeJournalProjection(statePath).value); const commands = Array.isArray(state.commands) ? state.commands.map(record) : []; @@ -5013,15 +4975,7 @@ function readRunnerdDurableIdentity( const statePath = resolve(controlPlaneRoot, "control-plane-state.json"); if (!existsSync(statePath)) return null; try { - const state = record( - JSON.parse( - readBoundedNativeFile( - statePath, - DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, - "runner_durable_identity_too_large", - ).toString("utf8"), - ), - ); + const state = record(readNativeJournalProjection(statePath, "identity").value); if (state.schema !== RUNNERD_CONTROL_PLANE_STATE_SCHEMA) return null; return record(state.identity); } catch {