mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 18:19:05 +02:00
fix(runtime): bound durable journal proof memory
Stream proof reads through a fixed buffer, retain only authority evidence, and preserve full-file digests and warm-transition fingerprints. Fail closed when essential evidence exceeds the bounded projection budget. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
fee9aa657e
commit
594ab9e629
4 files changed
+684
-95
No files matched your search
@@ -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
|
||||
|
||||
@@ -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<string, unknown>).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);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown> =>
|
||||
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<typeof createHash>,
|
||||
) {
|
||||
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<string, unknown> = 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<typeof fstatSync>,
|
||||
after: ReturnType<typeof fstatSync>,
|
||||
) {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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<string, unknown>;
|
||||
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 {
|
||||
|
||||
Reference in new issue
Block a user