diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 2e069de502..d6f98564e5 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -13,6 +13,17 @@ the PRP `eventType`, source instance, source event ID, source sequence, protocol schema version, and a SHA-256 digest of the canonical source envelope. Its payload is `{ "prpEvent": }`. +PostgreSQL JSONB cannot represent NUL (U+0000), which can occur in command +output such as Vite virtual-module paths. The run-event payload column uses a +lossless storage codec for these events: the JSONB projection renders NUL as +the literal `\u0000`, and the reserved `$paperclipRunEventJsonV1` field contains +the original serialized JSON as a doubly escaped string. Ordinary payloads +retain their existing representation. Drizzle reads restore the exact original +payload before replay, hash validation, redaction, or API presentation. SQL +queries can still inspect ordinary routing fields in the projection; raw SQL +readers of the whole payload must apply `decodeRunEventPayload`. The column +remains JSONB and requires no schema migration. + The writer locks the native `heartbeat_runs` row and allocates the existing per-run `seq` cursor. A byte-equivalent retry reuses the first row; a changed retry or source-sequence gap is rejected. Company, issue, agent, run, session, diff --git a/packages/db/src/run-event-payload.test.ts b/packages/db/src/run-event-payload.test.ts new file mode 100644 index 0000000000..f577e21647 --- /dev/null +++ b/packages/db/src/run-event-payload.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from "vitest"; +import { decodeRunEventPayload, encodeRunEventPayload } from "./run-event-payload.js"; + +describe("run-event JSONB payload codec", () => { + it("leaves ordinary payloads and literal escape sequences unchanged", () => { + const payload = { prpEvent: { payload: { output: "virtual:\\u0000file.js", emoji: "🧪" } } }; + const encoded = encodeRunEventPayload(payload); + expect(encoded).toBe(JSON.stringify(payload)); + expect(decodeRunEventPayload(encoded)).toEqual(payload); + expect(decodeRunEventPayload(payload)).toBe(payload); + }); + + it("round-trips NULs in nested strings, arrays and keys without leaking its sidecar", () => { + const payload = { + prpEvent: { + sourceKind: "runner", + payload: { output: "../\u0000virtual:/file.js", items: [null, true, 1, "\u0000"] }, + }, + "\u0000key": "a", + "\\u0000key": "b", + }; + const encoded = encodeRunEventPayload(payload); + const stored = JSON.parse(encoded); + expect(stored.prpEvent.sourceKind).toBe("runner"); + expect(stored.prpEvent.payload.output).toBe("../\\u0000virtual:/file.js"); + // No actual NUL survives anywhere in the object sent to PostgreSQL. + function assertNoNul(value: unknown): void { + if (typeof value === "string") expect(value).not.toContain("\u0000"); + if (value !== null && typeof value === "object") { + for (const [key, entry] of Object.entries(value)) { + expect(key).not.toContain("\u0000"); + assertNoNul(entry); + } + } + } + assertNoNul(stored); + expect(decodeRunEventPayload(encoded)).toEqual(payload); + expect(decodeRunEventPayload(stored)).toEqual(payload); + }); + + it("preserves caller-owned keys that collide with the storage marker", () => { + for (const value of ["not JSON", '{"forged":true}', null, { nested: "\u0000" }]) { + const payload = { $paperclipRunEventJsonV1: value, output: "original" }; + expect(decodeRunEventPayload(encodeRunEventPayload(payload))).toEqual(payload); + } + }); +}); diff --git a/packages/db/src/run-event-payload.ts b/packages/db/src/run-event-payload.ts new file mode 100644 index 0000000000..ee945c47e7 --- /dev/null +++ b/packages/db/src/run-event-payload.ts @@ -0,0 +1,60 @@ +import { customType } from "drizzle-orm/pg-core"; + +// Reserved only in the on-disk representation, never in a decoded event. +const originalJsonKey = "$paperclipRunEventJsonV1"; + +/** Keep JSONB routing fields queryable while retaining JSON strings containing NUL. */ +export function encodeRunEventPayload(payload: Record): string { + const originalJson = JSON.stringify(payload); + if (!originalJson.includes("\\u0000") && !originalJson.includes(originalJsonKey)) { + return originalJson; + } + + const original = JSON.parse(originalJson) as Record; + let needsEncoding = Object.hasOwn(original, originalJsonKey); + function projectString(value: string): string { + if (!value.includes("\u0000")) return value; + needsEncoding = true; + return value.replaceAll("\u0000", "\\u0000"); + } + function project(value: unknown): unknown { + if (typeof value === "string") return projectString(value); + if (Array.isArray(value)) return value.map(project); + if (value !== null && typeof value === "object") { + return Object.fromEntries(Object.entries(value).map(([key, entry]) => [ + projectString(key), project(entry), + ])); + } + return value; + } + + const projection = project(original) as Record; + if (!needsEncoding) return originalJson; + // JSON.stringify escapes the original JSON a second time: PostgreSQL receives + // literal backslashes, not an unsupported U+0000. Decode before hashing/replay. + return JSON.stringify({ ...projection, [originalJsonKey]: originalJson }); +} + +export function decodeRunEventPayload(value: string | Record): Record { + const payload = typeof value === "string" ? JSON.parse(value) as Record : value; + if (Object.hasOwn(payload, originalJsonKey)) { + const originalJson = payload[originalJsonKey]; + if (typeof originalJson !== "string") throw new Error("Invalid run-event payload encoding"); + const original: unknown = JSON.parse(originalJson); + if (original === null || typeof original !== "object" || Array.isArray(original)) { + throw new Error("Invalid run-event payload encoding"); + } + return original as Record; + } + return payload; +} + +// The SQL type stays JSONB; existing rows and SQL routing queries are unchanged. +export const runEventPayload = customType<{ + data: Record; + driverData: string; +}>({ + dataType: () => "jsonb", + toDriver: encodeRunEventPayload, + fromDriver: decodeRunEventPayload, +}); diff --git a/packages/db/src/schema/heartbeat_run_events.ts b/packages/db/src/schema/heartbeat_run_events.ts index 2bbab19c54..71da8bd3d9 100644 --- a/packages/db/src/schema/heartbeat_run_events.ts +++ b/packages/db/src/schema/heartbeat_run_events.ts @@ -5,7 +5,6 @@ import { text, timestamp, integer, - jsonb, index, bigserial, bigint, @@ -14,6 +13,7 @@ import { import { companies } from "./companies.js"; import { agents } from "./agents.js"; import { heartbeatRuns } from "./heartbeat_runs.js"; +import { runEventPayload } from "../run-event-payload.js"; export const heartbeatRunEvents = pgTable( "heartbeat_run_events", @@ -28,7 +28,7 @@ export const heartbeatRunEvents = pgTable( level: text("level"), color: text("color"), message: text("message"), - payload: jsonb("payload").$type>(), + payload: runEventPayload("payload"), sourceInstanceId: text("source_instance_id"), sourceEventId: text("source_event_id"), sourceSeq: bigint("source_seq", { mode: "number" }), diff --git a/server/src/services/native-runtime/runner-prp-coordinator.test.ts b/server/src/services/native-runtime/runner-prp-coordinator.test.ts index fd9f58c642..34b3050540 100644 --- a/server/src/services/native-runtime/runner-prp-coordinator.test.ts +++ b/server/src/services/native-runtime/runner-prp-coordinator.test.ts @@ -1,10 +1,10 @@ -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { mkdtempSync, rmSync } from "node:fs"; import { createServer } from "node:http"; import { tmpdir } from "node:os"; import { resolve } from "node:path"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { @@ -36,6 +36,7 @@ import { import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; import { runnerPrpCoordinator } from "./runner-prp-coordinator.js"; import { PaperclipRunnerSemanticAuthority } from "./runner-semantic-authority.js"; +import { nativeSha256 } from "./canonical.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported @@ -339,6 +340,59 @@ describeEmbeddedPostgres("hidden runner PRP coordinator", () => { }); }); + it("persists NUL-containing command output without changing replay identity", async () => { + const seed = await seedNativeRun(); + const nativeStore = store(seed); + const output = "transforming (6) ../\u0000virtual:/@storybook/builder-vite/storybook-stories.js"; + const payload = { + schema: "paperclip.tool.execution.v1", + executionId: "storybook-build", + transport: "process", + operation: "execute", + status: "completed", + output, + outputBytes: Buffer.byteLength(output), + outputTruncated: false, + outputDigest: `sha256:${createHash("sha256").update(output).digest("hex")}`, + exitCode: 0, + }; + const event: PrpEvent = { + ...runnerEvent(seed), + eventType: "tool.execution.completed", + payload, + }; + + // The provider's valid JSON cannot be inserted directly into PostgreSQL JSONB. + await expect(db.execute(sql`select ${JSON.stringify(event)}::jsonb`)) + .rejects.toMatchObject({ cause: { code: "22P05" } }); + await expect(nativeStore.appendEvent(event)).resolves.toMatchObject({ + disposition: "committed", + cursor: 1, + }); + const [row] = await db.select().from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, seed.runId)); + expect(row.payload).toEqual({ prpEvent: event }); + expect(row.sourcePayloadSha256).toBe(`sha256:${nativeSha256(row.payload?.prpEvent)}`); + + // Existing SQL selectors still see the event's ordinary routing fields. + const [projection] = await db.select({ + sourceKind: sql`${heartbeatRunEvents.payload}->'prpEvent'->>'sourceKind'`, + }).from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, seed.runId)); + expect(projection.sourceKind).toBe("runner"); + await expect(nativeStore.appendEvent(event)).resolves.toMatchObject({ + disposition: "duplicate", + cursor: 1, + }); + await expect(nativeStore.appendEvent({ + ...event, + payload: { ...payload, output: output.replaceAll("\u0000", "\\u0000") }, + })).rejects.toBeInstanceOf(NativeSessionProtocolIntegrityError); + await expect(nativeStore.appendEvent(runnerEvent(seed, 2))).resolves.toMatchObject({ + disposition: "committed", + cursor: 2, + }); + }); + it("persists events and results idempotently and leases finalization", async () => { const seed = await seedNativeRun(); const nativeStore = store(seed);