From 91e9f6686d233545140c9a7d8d74a79951f1af1e Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 20 Aug 2026 12:19:34 +0200 Subject: [PATCH 1/5] feat: persist per-turn privacy receipts with operator API (#757) --- docs/CHANGELOG.md | 16 ++ docs/ai-act-transparency.md | 9 + docs/middleware-agent-handoff.md | 17 ++ middleware/.env.example | 6 + middleware/migrations/0039_turn_receipts.sql | 34 +++ .../src/buildOrchestrator.ts | 8 + .../harness-orchestrator/src/orchestrator.ts | 49 ++++ .../harness-orchestrator/src/plugin.ts | 9 + middleware/packages/plugin-api/src/index.ts | 3 + .../plugin-api/src/turnReceiptStore.ts | 55 ++++ middleware/src/config.ts | 5 + middleware/src/index.ts | 23 ++ middleware/src/receipts/routes.ts | 164 ++++++++++++ middleware/src/receipts/store.ts | 112 ++++++++ .../turnReceiptPersistence.test.ts | 154 +++++++++++ middleware/test/turnReceipts.test.ts | 250 ++++++++++++++++++ web-ui/app/_components/Nav.tsx | 2 + web-ui/app/_lib/receipts.ts | 66 +++++ .../receipts/_components/ReceiptsList.tsx | 120 +++++++++ web-ui/app/operator/receipts/page.tsx | 55 ++++ web-ui/messages/de.json | 15 ++ web-ui/messages/en.json | 15 ++ 22 files changed, 1187 insertions(+) create mode 100644 middleware/migrations/0039_turn_receipts.sql create mode 100644 middleware/packages/plugin-api/src/turnReceiptStore.ts create mode 100644 middleware/src/receipts/routes.ts create mode 100644 middleware/src/receipts/store.ts create mode 100644 middleware/test/orchestrator/turnReceiptPersistence.test.ts create mode 100644 middleware/test/turnReceipts.test.ts create mode 100644 web-ui/app/_lib/receipts.ts create mode 100644 web-ui/app/operator/receipts/_components/ReceiptsList.tsx create mode 100644 web-ui/app/operator/receipts/page.tsx diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 702b84f23..01c6c0341 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -18,6 +18,22 @@ entry. See `CONTRIBUTING.md` § Releases & changelog. ## [Unreleased] +### Added — persistent per-turn privacy receipts (#757) + +- **`turn_receipts` (migration `0039`).** The per-turn `PrivacyReceipt` is no + longer ephemeral UI state: every completed turn writes its PII-free receipt + (counts + routing metadata, never a value) synchronously to Postgres — no + optional graph sink, no user-cluster precondition. Failures are counted and + logged, never silent. Schema note: `turn_id` unique (idempotent on replayed + `done` events); retention bounded by the new `RECEIPT_RETENTION_DAYS` + (default 90) via an unref'd reaper anchored on the DB clock. +- **Operator surface.** Auth-gated `GET /api/v1/operator/receipts` (+ + `/:turnId`) with composite-keyset pagination, and a web-ui page under + `/operator/receipts` rendering the exact receipt card the user saw. +- Not yet tamper-evident — hash chaining, signatures, and verification are + #758/#761; `docs/ai-act-transparency.md` §6 keeps "cryptographically + verifiable" a non-claim until they ship. + ### Added — a command policy that reads what a command actually does - **Shell-normalizing command policy (#580).** Command gating is not diff --git a/docs/ai-act-transparency.md b/docs/ai-act-transparency.md index 85188d2ac..d9daaa72d 100644 --- a/docs/ai-act-transparency.md +++ b/docs/ai-act-transparency.md @@ -158,6 +158,15 @@ User-Cluster-Knoten existiert — der Normalfall für jeden Kanal außer dem Bro fehlender Trace heißt "nicht aufgezeichnet", nie "diesen Turn gab es nicht". Entschieden und begründet in #684; jeder Ausfall wird seitdem gezählt und protokolliert. +**Seit #757 gibt es daneben einen persistierten Per-Turn-Receipt** (`turn_receipts`, +Migration `0039`): der PII-freie Privacy-Receipt jedes abgeschlossenen Turns wird auf dem +Postgres-Backend synchron gespeichert — ohne Graph-Sink, ohne User-Cluster-Vorbedingung — +und ist unter `/api/v1/operator/receipts` (auth-gated) sowie im Operator-UI abrufbar. +Fehlschläge werden gezählt und protokolliert, nie still verworfen. Der Receipt ist ein +*Record*, aber (noch) nicht manipulationssicher: Hash-Verkettung, Signaturen und +Verifikation sind #758/#761 — bis dahin bleibt „kryptographisch nachweisbar" eine +Nicht-Zusage. + **C2PA ist offen.** Im Code existiert keine C2PA-Implementierung. Für Bilder wäre das der naheliegende nächste Schritt; heute ist es keine Zusage, sondern ein offener Punkt. diff --git a/docs/middleware-agent-handoff.md b/docs/middleware-agent-handoff.md index 952acefdb..55d6e41c6 100644 --- a/docs/middleware-agent-handoff.md +++ b/docs/middleware-agent-handoff.md @@ -1103,6 +1103,21 @@ Emission-Asserts in `test/cliBridge/loopbackMcpServer.test.ts` und --- +### Turn-Receipts (#757) — persistierte Per-Turn-Privacy-Receipts + +Jeder abgeschlossene Turn persistiert seinen PII-freien `PrivacyReceipt` +synchron nach `turn_receipts` (Migration `0039`, Postgres-Backend only). Der +Orchestrator löst den Store late-bound über den Service +`turnReceiptStore` auf (Kernel provided in `index.ts`, gleiches Muster wie +`privacyRedact`); ohne Service bleiben Receipts ephemer. Fehlschläge werden +gezählt (`persistFailures` in `src/receipts/store.ts`) und greppbar geloggt +(`turn-receipt persist failed`), scheitern aber nie den Turn. Read-API: +auth-gated **`GET /api/v1/operator/receipts`** (Liste, Composite-Keyset-Cursor +`(created_at, id)`) und **`GET /api/v1/operator/receipts/:turnId`**; UI unter +`/operator/receipts`. Retention: `RECEIPT_RETENTION_DAYS` (Default 90), +Reaper mit Eager-Boot-Tick, Cutoff auf der DB-Uhr. Tests: +`test/turnReceipts.test.ts`, `test/orchestrator/turnReceiptPersistence.test.ts`. + ## 4. Migration Managed Agents → Lokal ### Warum migriert @@ -1452,6 +1467,8 @@ SKILLS_DIR=../skills # relativ zum middleware root MEMORY_DIR=./.memory MEMORY_SEED_DIR=./seed/memory MEMORY_SEED_MODE=missing # missing | overwrite | skip +# Turn receipts (#757) +RECEIPT_RETENTION_DAYS=90 # bounded retention for turn_receipts # Odoo ODOO_URL, ODOO_DB, ODOO_LOGIN, ODOO_API_KEY ODOO_PROXY_MAX_BYTES=500000 diff --git a/middleware/.env.example b/middleware/.env.example index 0d5a5f026..f240b5b2b 100644 --- a/middleware/.env.example +++ b/middleware/.env.example @@ -236,6 +236,12 @@ DEV_PLATFORM_EVENT_RETENTION_DAYS=30 # so the credential is INSIDE the runner. Off by default; boot REFUSES the flag # unless DEV_PLATFORM_SUBSCRIPTION_ACK is also set. DEV_PLATFORM_SUBSCRIPTION_MODE=false + +# --- Turn receipts (#757) --------------------------------------------------- +# Every completed turn persists its PII-free privacy receipt (turn_receipts, +# Postgres backend only). turn_id + scope are personal-data linkage, so rows +# are reaped after this many days. +RECEIPT_RETENTION_DAYS=90 # DEV_PLATFORM_SUBSCRIPTION_ACK= # required acknowledgment string when SUBSCRIPTION_MODE=true # --- Conductor generic webhooks (issue #437) -------------------------------- diff --git a/middleware/migrations/0039_turn_receipts.sql b/middleware/migrations/0039_turn_receipts.sql new file mode 100644 index 000000000..9f5ec70b4 --- /dev/null +++ b/middleware/migrations/0039_turn_receipts.sql @@ -0,0 +1,34 @@ +-- #757 — persistent per-turn audit receipts. +-- +-- The per-turn PrivacyReceipt was, until now, attached to the `done` event and +-- discarded — nothing let an operator answer "what did the system disclose or +-- mask for turn X last Tuesday?". This table persists it, one row per turn, +-- written synchronously by the orchestrator at turn end (no optional graph +-- sink, no user-cluster precondition — deliberately NOT the RunTrace, which is +-- best-effort telemetry). +-- +-- The payload is PII-free by construction (counts + verb names, see +-- plugin-api/src/privacyReceipt.ts); turn_id + scope are personal-data +-- LINKAGE, so retention is bounded by RECEIPT_RETENTION_DAYS via a reaper. +-- +-- 0038 is reserved for the Satellites epic (#746); this series continues at 0039. + +CREATE TABLE IF NOT EXISTS turn_receipts ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + turn_id TEXT NOT NULL, + session_scope TEXT, + channel TEXT, + model TEXT, + receipt JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- Idempotence key: a replayed `done` event must not duplicate the row. +CREATE UNIQUE INDEX IF NOT EXISTS turn_receipts_turn_id + ON turn_receipts (turn_id); + +-- Operator list view: newest first, optionally filtered by scope. +CREATE INDEX IF NOT EXISTS turn_receipts_scope_created + ON turn_receipts (session_scope, created_at DESC); +CREATE INDEX IF NOT EXISTS turn_receipts_created + ON turn_receipts (created_at DESC); diff --git a/middleware/packages/harness-orchestrator/src/buildOrchestrator.ts b/middleware/packages/harness-orchestrator/src/buildOrchestrator.ts index a0e7223cf..d20f3537f 100644 --- a/middleware/packages/harness-orchestrator/src/buildOrchestrator.ts +++ b/middleware/packages/harness-orchestrator/src/buildOrchestrator.ts @@ -33,6 +33,7 @@ import type { ProcessMemoryService, ResponseGuardService, SessionBriefingService, + TurnReceiptStore, } from '@omadia/plugin-api'; import type { VerifierBundle } from '@omadia/verifier'; import type { Pool } from 'pg'; @@ -140,6 +141,10 @@ export interface OrchestratorDeps { readonly responseGuard: () => ResponseGuardService | undefined; /** Late-bound `privacy.redact@1` lookup (see `OrchestratorOptions`). */ readonly privacyGuard: () => PrivacyGuardService | undefined; + /** #757 — late-bound persistent per-turn receipt store lookup (see + * `OrchestratorOptions.turnReceiptStore`). Optional: absent ⇒ receipts + * stay ephemeral. */ + readonly turnReceiptStore?: () => TurnReceiptStore | undefined; /** * Slice 2.5 — cross-plugin runtime-config lookup for the privacy bypass * resolver (see `OrchestratorOptions.pluginConfigGet`). Wired from the @@ -393,6 +398,9 @@ export function buildOrchestratorForAgent( ...(deps.embeddingClient ? { embeddingClient: deps.embeddingClient } : {}), responseGuard: deps.responseGuard, privacyGuard: deps.privacyGuard, + ...(deps.turnReceiptStore + ? { turnReceiptStore: deps.turnReceiptStore } + : {}), ...(deps.pluginConfigGet ? { pluginConfigGet: deps.pluginConfigGet } : {}), diff --git a/middleware/packages/harness-orchestrator/src/orchestrator.ts b/middleware/packages/harness-orchestrator/src/orchestrator.ts index 3ab64ffb0..de4b3ddd3 100644 --- a/middleware/packages/harness-orchestrator/src/orchestrator.ts +++ b/middleware/packages/harness-orchestrator/src/orchestrator.ts @@ -139,9 +139,11 @@ import type { PalaiaExcerpt, PalaiaExcerptExtractor, PrivacyGuardService, + PrivacyReceipt, ProcessMemoryService, ResponseGuardService, SessionBriefingService, + TurnReceiptStore, } from '@omadia/plugin-api'; import { agentScopePrefix, @@ -557,6 +559,16 @@ export interface OrchestratorOptions { * attached to the returned `ChatTurnResult.privacyReceipt`. */ privacyGuard?: () => PrivacyGuardService | undefined; + /** + * #757 — persistent per-turn receipt store lookup. Same late-bound thunk + * shape as `privacyGuard` (the kernel provides the service once its pg + * pool resolves; a per-turn lookup needs no restart). When present, every + * receipt `finalizeTurn` emits is ALSO persisted before the `done` event + * is considered flushed; persistence failure is logged + counted by the + * store, never fails the turn. Absent ⇒ receipts stay ephemeral + * (pre-#757 behaviour: UI-only). + */ + turnReceiptStore?: () => TurnReceiptStore | undefined; /** * Slice 2.5 — cross-plugin runtime-config lookup for the privacy * dispatch hook. Given `(agentId, configKey)` returns the operator-set @@ -1747,6 +1759,7 @@ export class Orchestrator { private readonly bookMeetingTool: BookMeetingTool | undefined; private readonly responseGuard: (() => ResponseGuardService | undefined) | undefined; private readonly privacyGuard: (() => PrivacyGuardService | undefined) | undefined; + private readonly turnReceiptStore: (() => TurnReceiptStore | undefined) | undefined; /** Slice 2.5 — cross-plugin runtime-config lookup (see OrchestratorOptions). */ private readonly pluginConfigGet: | ((agentId: string, configKey: string) => unknown | undefined) @@ -1871,6 +1884,7 @@ export class Orchestrator { this.bookMeetingTool = options.bookMeetingTool; this.responseGuard = options.responseGuard; this.privacyGuard = options.privacyGuard; + this.turnReceiptStore = options.turnReceiptStore; this.pluginConfigGet = options.pluginConfigGet; this.isPluginToolsReady = options.isPluginToolsReady; this.nudgeRegistry = options.nudgeRegistry; @@ -3327,6 +3341,7 @@ export class Orchestrator { try { const receipt = await privacyHandle.finalize(input.userMessage); if (receipt) { + await this.persistTurnReceipt(turnId, input, receipt); return { ...result, privacyReceipt: receipt }; } } catch (err) { @@ -3341,6 +3356,38 @@ export class Orchestrator { ); } + /** + * #757 — persist the turn's privacy receipt into the kernel-provided + * store, when one is wired. Called at every site that obtains a receipt + * from `finalizeTurn` (non-streaming, direct-line, streaming done) so a + * receipt that reaches the user also reaches the record. Never fails the + * turn: the user's answer outranks the audit row; the store counts the + * failure (`persistFailures`) and this logs it greppably — the exact + * inversion of the RunTrace defect (#684), where the drop was invisible. + */ + private async persistTurnReceipt( + turnId: string, + input: ChatTurnInput, + receipt: PrivacyReceipt, + ): Promise { + const store = this.turnReceiptStore?.(); + if (!store) return; + try { + await store.record({ + turnId, + sessionScope: input.sessionScope, + channel: input.channelIdentity?.channelKind, + model: this.model, + receipt, + }); + } catch (err) { + console.error( + `[orchestrator] turn-receipt persist failed for turn ${turnId}:`, + err, + ); + } + } + /** * #332 Layer 2 — Direct Line. When the USER directs input at a named * specialist (`@omadia #strategist `), the HARNESS — not the LLM — @@ -4894,6 +4941,7 @@ export class Orchestrator { try { const receipt = await privacyHandle.finalize(input.userMessage); if (receipt) { + await this.persistTurnReceipt(turnId, input, receipt); doneEvent = { ...doneEvent, privacyReceipt: receipt }; } } catch (err) { @@ -4995,6 +5043,7 @@ export class Orchestrator { try { const receipt = await privacyHandle.finalize(input.userMessage); if (receipt) { + await this.persistTurnReceipt(turnId, input, receipt); doneEvent = { ...doneEvent, privacyReceipt: receipt }; } } catch (err) { diff --git a/middleware/packages/harness-orchestrator/src/plugin.ts b/middleware/packages/harness-orchestrator/src/plugin.ts index acc84f656..8aea7c483 100644 --- a/middleware/packages/harness-orchestrator/src/plugin.ts +++ b/middleware/packages/harness-orchestrator/src/plugin.ts @@ -52,7 +52,9 @@ import type { MemoryStore, PluginContext } from '@omadia/plugin-api'; import { PRIVACY_REDACT_SERVICE_NAME, RESPONSE_GUARD_SERVICE_NAME, + TURN_RECEIPT_STORE_SERVICE_NAME, type PrivacyGuardService, + type TurnReceiptStore, } from '@omadia/plugin-api'; import type { VerifierBundle } from '@omadia/verifier'; @@ -587,6 +589,12 @@ export async function activate( const privacyGuardGetter = (): PrivacyGuardService | undefined => ctx.services.get(PRIVACY_REDACT_SERVICE_NAME); + // #757 — persistent per-turn receipt store, published by the kernel once + // its pg pool resolves (in-memory backend: never provided, receipts stay + // ephemeral). Same late-bound shape as the two getters above. + const turnReceiptStoreGetter = (): TurnReceiptStore | undefined => + ctx.services.get(TURN_RECEIPT_STORE_SERVICE_NAME); + // Slice 2.5 — cross-plugin runtime-config reader for the privacy // bypass resolver. Published by the kernel at boot // (`middleware/src/index.ts:installedPluginConfigReader`). When absent @@ -829,6 +837,7 @@ export async function activate( nudgeRegistry, responseGuard: responseGuardGetter, privacyGuard: privacyGuardGetter, + turnReceiptStore: turnReceiptStoreGetter, ...(pluginConfigGet ? { pluginConfigGet } : {}), ...(isPluginToolsReady ? { isPluginToolsReady } : {}), ...(contextRetriever ? { contextRetriever } : {}), diff --git a/middleware/packages/plugin-api/src/index.ts b/middleware/packages/plugin-api/src/index.ts index 2a2d454a3..fe8a2ae85 100644 --- a/middleware/packages/plugin-api/src/index.ts +++ b/middleware/packages/plugin-api/src/index.ts @@ -58,6 +58,9 @@ export * from './agentPriorities.js'; // renderers build against the fixtures here. export * from './privacyReceipt.js'; export * from './privacyReceiptFixtures.js'; +// #757 — persistent per-turn audit receipts (kernel-provided store the +// orchestrator resolves late-bound; receipts stay ephemeral when absent). +export * from './turnReceiptStore.js'; // Slice 2.5 — operator-owned per-plugin Privacy Mode contract. Shared by // the orchestrator dispatch hook (resolves mode at dispatch time) and the diff --git a/middleware/packages/plugin-api/src/turnReceiptStore.ts b/middleware/packages/plugin-api/src/turnReceiptStore.ts new file mode 100644 index 000000000..d69e4acc9 --- /dev/null +++ b/middleware/packages/plugin-api/src/turnReceiptStore.ts @@ -0,0 +1,55 @@ +/** + * #757 — persistent per-turn audit receipts. + * + * The `PrivacyReceipt` (see `privacyReceipt.ts`) is emitted once per turn and + * was, until #757, attached to the `done` event and then gone — an operator + * could never answer "what did the system disclose or mask for turn X last + * Tuesday?". This service persists that receipt, guaranteed per turn, into a + * kernel-owned store (`turn_receipts`, migration `0039`). + * + * Deliberately NOT the RunTrace: the trace is best-effort telemetry behind an + * optional graph sink (`runTraceObservability.ts` documents why it must not be + * promised as a record). This store has no user-cluster precondition and no + * optional sink — a turn either lands here or the failure is counted and + * logged loudly. + * + * The record stays PII-free by construction: it carries the receipt's counts + * plus routing metadata (turn id, session scope, channel kind, model). The + * `turnId`+`scope` pair is personal-data *linkage*, so retention is bounded + * (`RECEIPT_RETENTION_DAYS`) and enforced by a reaper. + */ + +import type { PrivacyReceipt } from './privacyReceipt.js'; + +/** + * Service-registry name the kernel publishes its store under. The + * harness-orchestrator resolves it late-bound (per turn, like + * `privacyRedact`) so boot order does not matter; absent service — e.g. the + * in-memory backend, unit tests — means receipts stay ephemeral, exactly the + * pre-#757 behaviour. + */ +export const TURN_RECEIPT_STORE_SERVICE_NAME = 'turnReceiptStore'; + +/** One persisted per-turn receipt row. PII-free: counts + routing metadata. */ +export interface TurnReceiptRecordInput { + /** The orchestrator's per-turn id — unique key of the row. */ + readonly turnId: string; + /** Session-transcript bucket the turn ran in (`ChatTurnInput.sessionScope`). */ + readonly sessionScope?: string; + /** Channel kind when the dispatcher mapped one (`channelIdentity.channelKind`). */ + readonly channel?: string; + /** Model id the orchestrator was configured with for this turn. */ + readonly model?: string; + /** The turn's aggregated privacy receipt, verbatim. */ + readonly receipt: PrivacyReceipt; +} + +export interface TurnReceiptStore { + /** + * Persist one turn's receipt. Idempotent on `turnId` (a replayed `done` + * event must not duplicate the row). Implementations throw on storage + * failure — the caller decides whether the turn survives (it does; the + * failure is counted and logged, never swallowed silently). + */ + record(entry: TurnReceiptRecordInput): Promise; +} diff --git a/middleware/src/config.ts b/middleware/src/config.ts index c259d28ae..28fe8b406 100644 --- a/middleware/src/config.ts +++ b/middleware/src/config.ts @@ -283,6 +283,11 @@ const ConfigSchema = z.object({ // origin in dev, while the webhook route only ever lives on the middleware). CONDUCTOR_WEBHOOK_PUBLIC_BASE_URL: z.string().url().optional(), + // #757 — bounded retention for persisted per-turn privacy receipts + // (`turn_receipts`). The payload is PII-free (counts only) but turn_id + + // scope are personal-data linkage, so rows are reaped after this many days. + RECEIPT_RETENTION_DAYS: z.coerce.number().int().positive().default(90), + // Epic #470 W4 — default per-job LLM cost budget (USD) applied when neither the // job nor its repo sets one (spec §5). Token budgets have NO default: they are // enforced only when explicitly set on the job or repo. diff --git a/middleware/src/index.ts b/middleware/src/index.ts index 178198e41..c3e67b2b3 100644 --- a/middleware/src/index.ts +++ b/middleware/src/index.ts @@ -31,6 +31,9 @@ import { createMemoryBackendRouter } from './routes/memoryBackend.js'; import { createChatRouter } from './routes/chat.js'; import { createOperatorAgentsRouter } from './routes/operatorAgents.js'; import { wireConductor, AwaitNotPendingError, AwaitResponderNotHolderError, ConductorRoleStore } from './conductor/index.js'; +import { TURN_RECEIPT_STORE_SERVICE_NAME } from '@omadia/plugin-api'; +import { PgTurnReceiptStore, startTurnReceiptReaper } from './receipts/store.js'; +import { createReceiptRoutes } from './receipts/routes.js'; import { bindingKeyForTurn } from './conductor/principalId.js'; import { createOperatorChannelsRouter } from './routes/operatorChannels.js'; import { createAgentBuilderRouter } from './routes/agentBuilder.js'; @@ -3482,6 +3485,26 @@ async function main(): Promise { log: (m) => console.log(m), }); console.log('[middleware] conductor wired at /api/v1/operator/conductors/* (auth-gated)'); + + // #757 — persistent per-turn privacy receipts. The store is published as + // a service the orchestrator resolves late-bound at turn end (same shape + // as `privacyRedact`); the read API mounts auth-gated next to the + // Conductor's operator surface; retention is enforced by an unref'd + // reaper (`RECEIPT_RETENTION_DAYS`). Postgres-only by construction — + // on the in-memory backend none of this wiring runs and receipts stay + // ephemeral, exactly the pre-#757 behaviour. + serviceRegistry.provide( + TURN_RECEIPT_STORE_SERVICE_NAME, + new PgTurnReceiptStore(graphPool), + ); + app.use('/api/v1/operator/receipts', requireAuth, createReceiptRoutes(graphPool)); + startTurnReceiptReaper(graphPool, { + retentionDays: config.RECEIPT_RETENTION_DAYS, + }); + console.log( + `[middleware] turn receipts wired at /api/v1/operator/receipts (auth-gated, retention ${config.RECEIPT_RETENTION_DAYS}d)`, + ); + const userStore = new UserStore(graphPool); const bootstrapResult = await runAuthBootstrap({ diff --git a/middleware/src/receipts/routes.ts b/middleware/src/receipts/routes.ts new file mode 100644 index 000000000..e90021398 --- /dev/null +++ b/middleware/src/receipts/routes.ts @@ -0,0 +1,164 @@ +/** + * #757 — operator read API for persisted per-turn receipts. + * + * Mounted under `/api/v1/operator/receipts` behind `requireAuth` (same + * posture as the Conductor's operator API — mounting happens at the caller, + * `middleware/src/index.ts`, which owns the auth middleware). + * + * Read-only by design: rows are written solely by the orchestrator's turn + * path. Query params are validated at the boundary; the receipt payload is + * PII-free by construction, so no masking is owed on this surface. + * + * Pagination is a composite keyset cursor `(created_at, id)`. A bare + * `created_at` cursor would lose rows on this surface twice over: exact-tie + * rows cut off by LIMIT would be skipped for good, and node-postgres + * truncates pg's microseconds to JS milliseconds, so a `< cursor` filter on + * a round-tripped ISO string swallows the sub-millisecond remainder. The + * cursor therefore carries pg's own text form of the timestamp (full + * microseconds, cast back with `::timestamptz`) plus the row id as the + * tiebreaker — "no receipt silently disappears" is this feature's promise, + * and paging is not allowed to break it. + */ + +import { Router, type Request, type Response } from 'express'; +import type { Pool } from 'pg'; +import { z } from 'zod'; + +const LIST_MAX_LIMIT = 100; + +const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; + +/** Opaque list cursor: `|`. */ +function parseCursor(raw: string): { ts: string; id: string } | undefined { + const sep = raw.lastIndexOf('|'); + if (sep <= 0) return undefined; + const ts = raw.slice(0, sep); + const id = raw.slice(sep + 1); + if (!UUID_RE.test(id)) return undefined; + if (ts.length === 0 || ts.length > 64 || Number.isNaN(Date.parse(ts))) { + return undefined; + } + return { ts, id }; +} + +const listQuerySchema = z.object({ + scope: z.string().min(1).max(512).optional(), + from: z.coerce.date().optional(), + to: z.coerce.date().optional(), + limit: z.coerce.number().int().min(1).max(LIST_MAX_LIMIT).default(25), + cursor: z + .string() + .max(128) + .transform((raw, ctx) => { + const parsed = parseCursor(raw); + if (!parsed) { + ctx.addIssue({ code: 'custom', message: 'malformed cursor' }); + return z.NEVER; + } + return parsed; + }) + .optional(), +}); + +interface TurnReceiptRow { + id: string; + turn_id: string; + session_scope: string | null; + channel: string | null; + model: string | null; + receipt: unknown; + created_at: Date; + /** pg's own text rendering of created_at — microsecond-exact, used only + * to build the outgoing cursor. */ + created_at_cursor: string; +} + +function toApiShape(row: TurnReceiptRow): Record { + return { + turnId: row.turn_id, + sessionScope: row.session_scope ?? undefined, + channel: row.channel ?? undefined, + model: row.model ?? undefined, + receipt: row.receipt, + createdAt: row.created_at.toISOString(), + }; +} + +const SELECT_COLUMNS = `id, turn_id, session_scope, channel, model, receipt, + created_at, created_at::text AS created_at_cursor`; + +export function createReceiptRoutes(pool: Pool): Router { + const router = Router(); + + router.get('/', async (req: Request, res: Response) => { + const parsed = listQuerySchema.safeParse(req.query); + if (!parsed.success) { + res.status(400).json({ error: 'invalid_query', detail: parsed.error.issues }); + return; + } + const { scope, from, to, limit, cursor } = parsed.data; + const where: string[] = []; + const params: unknown[] = []; + const add = (clause: string, value: unknown): void => { + params.push(value); + where.push(clause.replace('?', `$${params.length}`)); + }; + if (scope) add('session_scope = ?', scope); + if (from) add('created_at >= ?', from); + if (to) add('created_at <= ?', to); + if (cursor) { + // Composite keyset: strictly after the boundary row in (created_at + // DESC, id DESC) order — ascending row comparison inverts to `<`. + params.push(cursor.ts, cursor.id); + where.push( + `(created_at, id) < ($${params.length - 1}::timestamptz, $${params.length}::uuid)`, + ); + } + params.push(limit); + const sql = `SELECT ${SELECT_COLUMNS} + FROM turn_receipts + ${where.length > 0 ? `WHERE ${where.join(' AND ')}` : ''} + ORDER BY created_at DESC, id DESC + LIMIT $${params.length}`; + try { + const result = await pool.query(sql, params); + const items = result.rows.map(toApiShape); + // Next-page cursor only when the page was full — a short page is the end. + const lastRow = result.rows[result.rows.length - 1]; + const nextCursor = + result.rows.length === limit && lastRow + ? `${lastRow.created_at_cursor}|${lastRow.id}` + : undefined; + res.json({ items, ...(nextCursor ? { nextCursor } : {}) }); + } catch (err) { + console.error('[receipts] list query failed:', err); + res.status(500).json({ error: 'receipts_query_failed' }); + } + }); + + router.get('/:turnId', async (req: Request, res: Response) => { + const turnId = String(req.params.turnId ?? ''); + if (turnId.length === 0 || turnId.length > 512) { + res.status(400).json({ error: 'invalid_turn_id' }); + return; + } + try { + const result = await pool.query( + `SELECT ${SELECT_COLUMNS} + FROM turn_receipts WHERE turn_id = $1`, + [turnId], + ); + const row = result.rows[0]; + if (!row) { + res.status(404).json({ error: 'receipt_not_found' }); + return; + } + res.json(toApiShape(row)); + } catch (err) { + console.error('[receipts] get query failed:', err); + res.status(500).json({ error: 'receipts_query_failed' }); + } + }); + + return router; +} diff --git a/middleware/src/receipts/store.ts b/middleware/src/receipts/store.ts new file mode 100644 index 000000000..697e5a0b8 --- /dev/null +++ b/middleware/src/receipts/store.ts @@ -0,0 +1,112 @@ +/** + * #757 — Postgres-backed persistent per-turn receipt store. + * + * Backs the `turnReceiptStore` service the orchestrator resolves late-bound + * at turn end (see `plugin-api/src/turnReceiptStore.ts` for the contract and + * why this is deliberately NOT the RunTrace). Postgres-only, like the + * Conductor: on the in-memory backend the service is simply never provided + * and receipts stay ephemeral. + * + * Failure posture: `record()` throws to its caller; the orchestrator counts + * and logs the failure but never fails the turn over it — the user's answer + * outranks the audit row, and the loss is observable (`persistFailures`), + * never silent. That is the exact inversion of the RunTrace defect (#684): + * there the drop was invisible; here it is counted. + */ + +import type { Pool } from 'pg'; +import type { + TurnReceiptRecordInput, + TurnReceiptStore, +} from '@omadia/plugin-api'; + +/** Process-wide failure counters, exported for /health-style introspection + * and asserted in tests. Mirrors `runTraceObservability.ts`'s "count what + * would otherwise be silently incomplete" obligation. */ +export interface TurnReceiptObservability { + persisted: number; + persistFailures: number; +} + +const counters: TurnReceiptObservability = { persisted: 0, persistFailures: 0 }; + +export function turnReceiptCounters(): Readonly { + return counters; +} + +/** Test seam only. */ +export function resetTurnReceiptCounters(): void { + counters.persisted = 0; + counters.persistFailures = 0; +} + +export class PgTurnReceiptStore implements TurnReceiptStore { + constructor(private readonly pool: Pool) {} + + async record(entry: TurnReceiptRecordInput): Promise { + try { + // Idempotent on turn_id: a replayed `done` event (retry, double flush) + // must not duplicate the row; first write wins. + const result = await this.pool.query( + `INSERT INTO turn_receipts (turn_id, session_scope, channel, model, receipt) + VALUES ($1, $2, $3, $4, $5::jsonb) + ON CONFLICT (turn_id) DO NOTHING`, + [ + entry.turnId, + entry.sessionScope ?? null, + entry.channel ?? null, + entry.model ?? null, + JSON.stringify(entry.receipt), + ], + ); + // A replayed turn hits DO NOTHING (rowCount 0) — that is not a + // persist, and the counter must not overstate the record. + if ((result.rowCount ?? 0) > 0) { + counters.persisted += 1; + } + } catch (err) { + counters.persistFailures += 1; + throw err; + } + } +} + +/** + * Retention reaper: deletes receipt rows older than `retentionDays`. Runs on + * an unref'd interval so it never keeps the process alive; each tick is + * best-effort and logged on failure. The `created_at` anchor is the DB's own + * clock (`NOW()`), never the process clock — the reaper-clock-race lesson + * from #709: the anchor must not be a value the test (or a lagging process) + * also controls. + */ +export function startTurnReceiptReaper( + pool: Pool, + opts: { retentionDays: number; intervalMs?: number }, +): { stop: () => void } { + const intervalMs = opts.intervalMs ?? 6 * 60 * 60 * 1000; // 6h + const tick = async (): Promise => { + try { + const result = await pool.query( + `DELETE FROM turn_receipts + WHERE created_at < NOW() - make_interval(days => $1)`, + [opts.retentionDays], + ); + if ((result.rowCount ?? 0) > 0) { + console.log( + `[receipts] reaper removed ${result.rowCount} receipt(s) past ${opts.retentionDays}d retention`, + ); + } + } catch (err) { + console.error('[receipts] retention reaper tick failed:', err); + } + }; + const timer = setInterval(() => void tick(), intervalMs); + timer.unref(); + // Eager first pass: plugin activation (which applies the numbered + // migrations, incl. `0039_turn_receipts`) runs well before this wiring in + // `index.ts`, so the table exists here — and without the boot tick a + // process restarting more often than the interval would never enforce + // retention at all. + void tick(); + return { stop: () => clearInterval(timer) }; +} diff --git a/middleware/test/orchestrator/turnReceiptPersistence.test.ts b/middleware/test/orchestrator/turnReceiptPersistence.test.ts new file mode 100644 index 000000000..ba2314722 --- /dev/null +++ b/middleware/test/orchestrator/turnReceiptPersistence.test.ts @@ -0,0 +1,154 @@ +/** + * #757 — orchestrator-side wiring: a receipt that reaches the user also + * reaches the persistent store, and a failing store never fails the turn. + * + * Setup mirrors `promptMaskPipeline.test.ts`: the REAL privacy-guard service + * with `mask_user_prompt` forced on produces a real receipt (masked email + * span) on a real `runTurn`; the store is a fake capturing `record()` calls. + */ + +import { strict as assert } from 'node:assert'; +import { describe, it } from 'node:test'; + +import type { + LlmProvider, + LlmRequest, + LlmResponse, + LlmStreamEvent, +} from '@omadia/llm-provider'; +import { NativeToolRegistry, Orchestrator } from '@omadia/orchestrator'; +import type { TurnReceiptRecordInput } from '@omadia/plugin-api'; +import { createPrivacyGuardService } from '@omadia/plugin-privacy-guard/dist/index.js'; + +const providerCapabilities = { + tools: true, + vision: true, + streaming: true, + promptCaching: true, + forcedToolChoice: true, + parallelToolCalls: true, +} as const; + +const RAW_EMAIL = 'anna.schmidt@firma.de'; + +function maskingService(): ReturnType { + return createPrivacyGuardService({ + readConfig: (key: string) => (key === 'mask_user_prompt' ? 'on' : undefined), + }); +} + +function textResponse(text: string): LlmResponse { + return { + content: [{ type: 'text', text }], + finishReason: 'stop', + providerFinishReason: 'end_turn', + model: 'test', + usage: { inputTokens: 10, outputTokens: 1, cacheReadTokens: 0, cacheWriteTokens: 0 }, + }; +} + +function staticProvider(): LlmProvider { + const provider = { + id: 'anthropic', + capabilities: providerCapabilities, + complete: async (_req: LlmRequest): Promise => + textResponse('Alles klar.'), + stream: (): AsyncIterable => { + throw new Error('staticProvider: stream() not scripted'); + }, + classifyError: () => ({ retryable: false, kind: 'other' as const }), + }; + return provider as unknown as LlmProvider; +} + +type OrchestratorOptions = ConstructorParameters[0]; + +const sessionLogger = { + log: async (): Promise<{ turnExternalId: string }> => ({ + turnExternalId: 'turn:sess-1:t1', + }), +} as unknown as OrchestratorOptions['sessionLogger']; + +function buildOrch(store: { + record: (entry: TurnReceiptRecordInput) => Promise; +}): Orchestrator { + return new Orchestrator({ + provider: staticProvider(), + model: 'test-model', + maxTokens: 1024, + maxToolIterations: 3, + domainTools: [], + nativeToolRegistry: new NativeToolRegistry(), + sessionLogger, + privacyGuard: () => maskingService(), + turnReceiptStore: () => store, + }); +} + +describe('#757 turn-receipt persistence — orchestrator wiring', () => { + it('persists the receipt the turn attaches, with routing metadata', async () => { + const recorded: TurnReceiptRecordInput[] = []; + const orch = buildOrch({ + record: async (entry) => { + recorded.push(entry); + }, + }); + const result = await orch.runTurn({ + userMessage: `Bitte schreibe an ${RAW_EMAIL}.`, + sessionScope: 'sess-1', + userId: 'u1', + channelIdentity: { channelKind: 'teams', channelUserId: 'aad-1' }, + }); + assert.ok(result.privacyReceipt, 'turn must attach a receipt'); + assert.equal(recorded.length, 1, 'exactly one persisted receipt per turn'); + const entry = recorded[0]!; + assert.ok(entry.turnId.length > 0, 'a non-empty turn id must be recorded'); + assert.equal(entry.sessionScope, 'sess-1'); + assert.equal(entry.channel, 'teams'); + assert.equal(entry.model, 'test-model'); + // The persisted receipt is the SAME object the user saw — no divergence + // between UI truth and record truth. + assert.deepEqual(entry.receipt, result.privacyReceipt); + const spans = entry.receipt.maskedPromptSpans ?? []; + assert.ok( + spans.some((s) => s.type === 'email'), + 'persisted receipt must carry the masked email span', + ); + }); + + it('a throwing store is loud but never fails the turn', async () => { + const orch = buildOrch({ + record: async () => { + throw new Error('pg down'); + }, + }); + const result = await orch.runTurn({ + userMessage: `Bitte schreibe an ${RAW_EMAIL}.`, + sessionScope: 'sess-1', + userId: 'u1', + }); + // The user's answer outranks the audit row. + assert.ok(result.answer.length > 0); + assert.ok(result.privacyReceipt, 'receipt still reaches the user'); + }); + + it('no store wired ⇒ byte-identical pre-#757 behaviour', async () => { + const orch = new Orchestrator({ + provider: staticProvider(), + model: 'test-model', + maxTokens: 1024, + maxToolIterations: 3, + domainTools: [], + nativeToolRegistry: new NativeToolRegistry(), + sessionLogger, + privacyGuard: () => maskingService(), + // deliberately: no turnReceiptStore option + }); + const result = await orch.runTurn({ + userMessage: `Bitte schreibe an ${RAW_EMAIL}.`, + sessionScope: 'sess-1', + userId: 'u1', + }); + assert.ok(result.privacyReceipt); + }); +}); diff --git a/middleware/test/turnReceipts.test.ts b/middleware/test/turnReceipts.test.ts new file mode 100644 index 000000000..92ddfdd3d --- /dev/null +++ b/middleware/test/turnReceipts.test.ts @@ -0,0 +1,250 @@ +/** + * #757 — persistent per-turn receipts: kernel store, retention reaper, and + * operator read API. The orchestrator-side wiring (receipt reaches the store + * on a real turn) lives in `test/orchestrator/turnReceiptPersistence.test.ts`; + * here the units are exercised against a fake pg pool. + */ + +import { strict as assert } from 'node:assert'; +import { after, describe, it } from 'node:test'; +import type { Server } from 'node:http'; +import type { AddressInfo } from 'node:net'; + +import express from 'express'; +import type { Pool } from 'pg'; + +import { + PgTurnReceiptStore, + resetTurnReceiptCounters, + startTurnReceiptReaper, + turnReceiptCounters, +} from '../src/receipts/store.js'; +import { createReceiptRoutes } from '../src/receipts/routes.js'; + +interface RecordedQuery { + sql: string; + params: unknown[]; +} + +/** Minimal fake pg pool: records queries, answers from a script. */ +function fakePool( + handler: (sql: string, params: unknown[]) => { rows?: unknown[]; rowCount?: number } | Error, +): { pool: Pool; queries: RecordedQuery[] } { + const queries: RecordedQuery[] = []; + const pool = { + query: async (sql: string, params: unknown[] = []) => { + queries.push({ sql, params }); + const result = handler(sql, params); + if (result instanceof Error) throw result; + return { rows: result.rows ?? [], rowCount: result.rowCount ?? (result.rows?.length ?? 0) }; + }, + } as unknown as Pool; + return { pool, queries }; +} + +const RECEIPT = { + datasetsInterned: 2, + fieldsMasked: 7, + fieldsCleartext: 3, + verbsExecuted: ['v4_sort'], + pseudonymProjectionUsed: false, +}; + +describe('#757 PgTurnReceiptStore', () => { + it('inserts one idempotent row per turn and counts the persist', async () => { + resetTurnReceiptCounters(); + const { pool, queries } = fakePool(() => ({ rowCount: 1 })); + const store = new PgTurnReceiptStore(pool); + await store.record({ + turnId: 't-1', + sessionScope: 'sess-1', + channel: 'teams', + model: 'claude-test', + receipt: RECEIPT, + }); + assert.equal(queries.length, 1); + const q = queries[0]!; + assert.match(q.sql, /INSERT INTO turn_receipts/); + // Idempotence is the SQL's job: a replayed done event must hit the + // turn_id conflict target, not add a second row. + assert.match(q.sql, /ON CONFLICT \(turn_id\) DO NOTHING/); + assert.deepEqual(q.params.slice(0, 4), ['t-1', 'sess-1', 'teams', 'claude-test']); + assert.deepEqual(JSON.parse(q.params[4] as string), RECEIPT); + assert.equal(turnReceiptCounters().persisted, 1); + assert.equal(turnReceiptCounters().persistFailures, 0); + }); + + it('counts a storage failure and rethrows (caller decides turn fate)', async () => { + resetTurnReceiptCounters(); + const { pool } = fakePool(() => new Error('pg down')); + const store = new PgTurnReceiptStore(pool); + await assert.rejects( + store.record({ turnId: 't-2', receipt: RECEIPT }), + /pg down/, + ); + assert.equal(turnReceiptCounters().persistFailures, 1); + assert.equal(turnReceiptCounters().persisted, 0); + }); + + it('stores absent metadata as NULL, not as the string "undefined"', async () => { + resetTurnReceiptCounters(); + const { pool, queries } = fakePool(() => ({ rowCount: 1 })); + await new PgTurnReceiptStore(pool).record({ turnId: 't-3', receipt: RECEIPT }); + assert.deepEqual(queries[0]!.params.slice(0, 4), ['t-3', null, null, null]); + }); + + it('a replayed turn (ON CONFLICT no-op) does not inflate the persisted counter', async () => { + resetTurnReceiptCounters(); + const { pool } = fakePool(() => ({ rowCount: 0 })); + await new PgTurnReceiptStore(pool).record({ turnId: 't-4', receipt: RECEIPT }); + assert.equal(turnReceiptCounters().persisted, 0); + assert.equal(turnReceiptCounters().persistFailures, 0); + }); +}); + +describe('#757 retention reaper', () => { + it('deletes past-retention rows eagerly at start, anchored on the DB clock', async () => { + const { pool, queries } = fakePool(() => ({ rowCount: 2 })); + const reaper = startTurnReceiptReaper(pool, { retentionDays: 30, intervalMs: 60_000 }); + try { + // Eager boot tick: a process restarting more often than the interval + // must still enforce retention — so the first pass fires immediately, + // without waiting for the (long) interval. + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.ok(queries.length >= 1, 'reaper must have ticked at least once'); + const q = queries[0]!; + assert.match(q.sql, /DELETE FROM turn_receipts/); + // The cutoff must be computed from NOW() in the database — never a + // process-clock timestamp parameter (#709: anchor ≠ what a lagging + // process controls). + assert.match(q.sql, /NOW\(\) - make_interval/); + assert.deepEqual(q.params, [30]); + } finally { + reaper.stop(); + } + }); + + it('a failing tick logs but never throws out of the timer', async () => { + const { pool, queries } = fakePool(() => new Error('relation missing')); + const reaper = startTurnReceiptReaper(pool, { retentionDays: 30, intervalMs: 5 }); + try { + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.ok(queries.length >= 1); + // Reaching this line without an unhandled rejection IS the assertion. + } finally { + reaper.stop(); + } + }); +}); + +describe('#757 operator receipts API', () => { + const servers: Server[] = []; + after(async () => { + await Promise.all( + servers.map((s) => new Promise((resolve) => s.close(() => resolve()))), + ); + }); + + const ROW_ID = '0f5c1c3a-1111-4222-8333-444455556666'; + const ROW = { + id: ROW_ID, + turn_id: 't-1', + session_scope: 'sess-1', + channel: 'teams', + model: 'claude-test', + receipt: RECEIPT, + created_at: new Date('2026-08-20T10:00:00.123Z'), + // pg's own text rendering — microsecond-exact, beyond JS Date precision. + created_at_cursor: '2026-08-20 10:00:00.123456+00', + }; + + async function serve( + handler: Parameters[0], + ): Promise<{ baseUrl: string; queries: RecordedQuery[] }> { + const { pool, queries } = fakePool(handler); + const app = express(); + app.use('/api/v1/operator/receipts', createReceiptRoutes(pool)); + const server: Server = await new Promise((resolve) => { + const s = app.listen(0, '127.0.0.1', () => resolve(s)); + }); + servers.push(server); + const port = (server.address() as AddressInfo).port; + return { baseUrl: `http://127.0.0.1:${String(port)}/api/v1/operator/receipts`, queries }; + } + + it('lists receipts newest-first with camelCase mapping and no cursor on a short page', async () => { + const { baseUrl } = await serve(() => ({ rows: [ROW] })); + const res = await fetch(`${baseUrl}?limit=25`); + assert.equal(res.status, 200); + const body = (await res.json()) as { items: Array>; nextCursor?: string }; + assert.equal(body.items.length, 1); + assert.equal(body.items[0]!.turnId, 't-1'); + assert.equal(body.items[0]!.sessionScope, 'sess-1'); + assert.equal(body.items[0]!.createdAt, '2026-08-20T10:00:00.123Z'); + // The keyset internals never leak into the item shape. + assert.equal('id' in body.items[0]!, false); + assert.equal('created_at_cursor' in body.items[0]!, false); + assert.deepEqual(body.items[0]!.receipt, RECEIPT); + assert.equal(body.nextCursor, undefined); + }); + + it('emits a composite keyset nextCursor exactly when the page is full', async () => { + const { baseUrl } = await serve(() => ({ rows: [ROW] })); + const res = await fetch(`${baseUrl}?limit=1`); + const body = (await res.json()) as { nextCursor?: string }; + // pg's microsecond-exact text stamp + row id — a bare ISO(ms) cursor + // would lose exact-tie and truncation-gap rows at page boundaries. + assert.equal(body.nextCursor, `2026-08-20 10:00:00.123456+00|${ROW_ID}`); + }); + + it('threads scope + composite cursor filters into the query parameters', async () => { + const { baseUrl, queries } = await serve(() => ({ rows: [] })); + const cursor = encodeURIComponent(`2026-08-20 10:00:00.123456+00|${ROW_ID}`); + const res = await fetch(`${baseUrl}?scope=sess-1&cursor=${cursor}&limit=10`); + assert.equal(res.status, 200); + const q = queries[0]!; + assert.ok(q.params.includes('sess-1')); + assert.ok(q.params.includes('2026-08-20 10:00:00.123456+00')); + assert.ok(q.params.includes(ROW_ID)); + assert.equal(q.params[q.params.length - 1], 10); + assert.match(q.sql, /session_scope = \$1/); + assert.match(q.sql, /\(created_at, id\) < \(\$2::timestamptz, \$3::uuid\)/); + assert.match(q.sql, /ORDER BY created_at DESC, id DESC/); + }); + + it('rejects a malformed cursor with 400 before the pool is touched', async () => { + const { baseUrl, queries } = await serve(() => ({ rows: [] })); + for (const bad of ['not-a-cursor', '2026-08-20 10:00:00+00|not-a-uuid', `|${ROW_ID}`]) { + const res = await fetch(`${baseUrl}?cursor=${encodeURIComponent(bad)}`); + assert.equal(res.status, 400, `cursor ${JSON.stringify(bad)} must 400`); + } + assert.equal(queries.length, 0); + }); + + it('rejects an invalid limit with 400 instead of clamping silently', async () => { + const { baseUrl, queries } = await serve(() => ({ rows: [] })); + const res = await fetch(`${baseUrl}?limit=5000`); + assert.equal(res.status, 400); + assert.equal(queries.length, 0, 'invalid input must never reach the pool'); + }); + + it('serves a single receipt by turn id and 404s an unknown one', async () => { + const { baseUrl } = await serve((_sql, params) => + params[0] === 't-1' ? { rows: [ROW] } : { rows: [] }, + ); + const hit = await fetch(`${baseUrl}/t-1`); + assert.equal(hit.status, 200); + assert.equal(((await hit.json()) as { turnId: string }).turnId, 't-1'); + const miss = await fetch(`${baseUrl}/t-unknown`); + assert.equal(miss.status, 404); + }); + + it('maps a pool failure to 500 without leaking the error', async () => { + const { baseUrl } = await serve(() => new Error('secret dsn in message')); + const res = await fetch(`${baseUrl}`); + assert.equal(res.status, 500); + const body = (await res.json()) as { error: string }; + assert.equal(body.error, 'receipts_query_failed'); + assert.ok(!JSON.stringify(body).includes('secret dsn')); + }); +}); diff --git a/web-ui/app/_components/Nav.tsx b/web-ui/app/_components/Nav.tsx index af107b75a..9c6dbfa03 100644 --- a/web-ui/app/_components/Nav.tsx +++ b/web-ui/app/_components/Nav.tsx @@ -73,6 +73,8 @@ const NAV: readonly NavItem[] = [ // operator-facing configuration surfaces, same audience as Admin/System. { kind: 'link', href: '/operator/agents', key: 'agentsCluster' }, { kind: 'link', href: '/conductor', key: 'conductor' }, + // #757 — persisted per-turn privacy receipts, same operator audience. + { kind: 'link', href: '/operator/receipts', key: 'receipts' }, // Dev Platform used to be hardcoded here. It is now contributed at // runtime (middleware registers it while DEV_PLATFORM_ENABLED), so the // entry disappears when the feature is off — see mergeNav below. diff --git a/web-ui/app/_lib/receipts.ts b/web-ui/app/_lib/receipts.ts new file mode 100644 index 000000000..19778506b --- /dev/null +++ b/web-ui/app/_lib/receipts.ts @@ -0,0 +1,66 @@ +import { ApiError } from './api'; +import type { PrivacyReceipt } from './chatSessions'; + +/** + * #757 — typed client for the operator receipts REST surface + * (`/api/v1/operator/receipts`). Mirrors the `channels.ts` pattern: works + * from both server components (MIDDLEWARE_URL + forwarded cookies) and the + * browser (`/bot-api` catch-all route). + */ + +function botApi(path: string): string { + if (typeof window !== 'undefined') { + return `/bot-api${path}`; + } + const base = process.env['MIDDLEWARE_URL'] ?? 'http://localhost:3979'; + return `${base}/api${path}`; +} + +async function forwardCookieHeader(): Promise> { + if (typeof window !== 'undefined') return {}; + try { + const mod = await import('next/headers'); + const jar = await mod.cookies(); + const cookieHeader = jar + .getAll() + .map((c) => `${c.name}=${c.value}`) + .join('; '); + return cookieHeader ? { cookie: cookieHeader } : {}; + } catch { + return {}; + } +} + +export interface TurnReceiptDto { + turnId: string; + sessionScope?: string; + channel?: string; + model?: string; + receipt: PrivacyReceipt; + createdAt: string; +} + +export interface ReceiptsPageDto { + items: TurnReceiptDto[]; + nextCursor?: string; +} + +export async function listReceipts(opts?: { + scope?: string; + cursor?: string; + limit?: number; +}): Promise { + const params = new URLSearchParams(); + if (opts?.scope) params.set('scope', opts.scope); + if (opts?.cursor) params.set('cursor', opts.cursor); + if (opts?.limit) params.set('limit', String(opts.limit)); + const qs = params.size > 0 ? `?${params.toString()}` : ''; + const res = await fetch(botApi(`/v1/operator/receipts${qs}`), { + headers: { ...(await forwardCookieHeader()) }, + cache: 'no-store', + }); + if (!res.ok) { + throw new ApiError(res.status, `receipts list failed: ${String(res.status)}`); + } + return (await res.json()) as ReceiptsPageDto; +} diff --git a/web-ui/app/operator/receipts/_components/ReceiptsList.tsx b/web-ui/app/operator/receipts/_components/ReceiptsList.tsx new file mode 100644 index 000000000..1a396ccf0 --- /dev/null +++ b/web-ui/app/operator/receipts/_components/ReceiptsList.tsx @@ -0,0 +1,120 @@ +'use client'; + +import { useState } from 'react'; +import { useFormatter, useTranslations } from 'next-intl'; + +import { PrivacyReceiptCard } from '../../../_components/chat/PrivacyReceiptCard'; +import { + listReceipts, + type ReceiptsPageDto, + type TurnReceiptDto, +} from '../../../_lib/receipts'; + +interface ReceiptsListProps { + initial: ReceiptsPageDto; +} + +/** + * #757 — expandable per-turn receipt rows with cursor pagination. Each row + * shows routing metadata (when, scope, channel, model) plus the shield's + * headline counts; expanding renders the exact `PrivacyReceiptCard` the + * user saw in the chat — the record and the UI share one truth. + */ +export function ReceiptsList({ initial }: ReceiptsListProps): React.ReactElement { + const t = useTranslations('operatorReceipts'); + const format = useFormatter(); + const [items, setItems] = useState(initial.items); + const [nextCursor, setNextCursor] = useState(initial.nextCursor); + const [loading, setLoading] = useState(false); + const [loadError, setLoadError] = useState(null); + + async function loadMore(): Promise { + if (!nextCursor || loading) return; + setLoading(true); + setLoadError(null); + try { + const page = await listReceipts({ cursor: nextCursor, limit: 25 }); + setItems((prev) => [...prev, ...page.items]); + setNextCursor(page.nextCursor); + } catch { + setLoadError(t('loadError')); + } finally { + setLoading(false); + } + } + + if (items.length === 0) { + return ( +

+ {t('empty')} +

+ ); + } + + return ( +
+
    + {items.map((item) => ( +
  • +
    + + + {item.sessionScope ? ( + + {t('scopeLabel')}: {item.sessionScope} + + ) : null} + {item.channel ? ( + + {t('channelLabel')}: {item.channel} + + ) : null} + {item.model ? ( + + {t('modelLabel')}: {item.model} + + ) : null} + + {t('summaryCounts', { + masked: item.receipt.fieldsMasked, + datasets: item.receipt.datasetsInterned, + })} + + +
    + +

    + {t('turnIdLabel')}: {item.turnId} +

    +
    +
    +
    • + ))} +
+ {loadError ? ( +

{loadError}

+ ) : null} + {nextCursor ? ( + + ) : null} +
+ ); +} diff --git a/web-ui/app/operator/receipts/page.tsx b/web-ui/app/operator/receipts/page.tsx new file mode 100644 index 000000000..5d7e5bfa6 --- /dev/null +++ b/web-ui/app/operator/receipts/page.tsx @@ -0,0 +1,55 @@ +import type { Metadata } from 'next'; +import { getTranslations } from 'next-intl/server'; + +import { redirectIfUnauthorized } from '../../_lib/authRedirect'; +import { listReceipts, type ReceiptsPageDto } from '../../_lib/receipts'; +import { ReceiptsList } from './_components/ReceiptsList'; + +/** + * #757 — operator-facing list of persisted per-turn privacy receipts. + * + * Every completed turn writes its PII-free receipt to the middleware's + * `turn_receipts` store; this page is the read surface: what did the shield + * intern, mask, and (where the operator opted into bypass) pass through — + * per turn, after the fact, not only while the answer was on screen. + */ + +export async function generateMetadata(): Promise { + const t = await getTranslations('operatorReceipts'); + return { title: t('metaTitle') }; +} + +export const dynamic = 'force-dynamic'; + +export default async function OperatorReceiptsPage(): Promise { + const t = await getTranslations('operatorReceipts'); + let initial: ReceiptsPageDto | null = null; + let loadError: string | null = null; + try { + initial = await listReceipts({ limit: 25 }); + } catch (err) { + await redirectIfUnauthorized(err); + // Catalog key first (checklist rule 3) — the raw error is a technical + // detail that belongs in the server log, not as the page's primary text. + console.error('[operator/receipts] initial load failed:', err); + loadError = t('loadError'); + } + + return ( +
+
+

{t('title')}

+

+ {t('subtitle')} +

+
+ {loadError ? ( +
+ {loadError} +
+ ) : ( + + )} +
+ ); +} diff --git a/web-ui/messages/de.json b/web-ui/messages/de.json index 013a0241f..7b81d6a16 100644 --- a/web-ui/messages/de.json +++ b/web-ui/messages/de.json @@ -428,6 +428,7 @@ "graph": "Graph", "routines": "Routinen", "conductor": "Conductor", + "receipts": "Receipts", "admin": "Admin", "system": "System", "agentsCluster": "Orchestratoren", @@ -707,6 +708,20 @@ "templatePendingEmpty": "Keine Vorlagen warten auf Review", "templateNoMatches": "Keine Vorlagen entsprechen deinen Filtern." }, + "operatorReceipts": { + "metaTitle": "Receipts · omadia", + "title": "Privacy-Receipts", + "subtitle": "Jeder abgeschlossene Turn schreibt seinen PII-freien Privacy-Receipt hierher — was der Shield interniert, maskiert und durchgereicht hat. Der Beleg überlebt die Chat-Ansicht; die Aufbewahrung begrenzt RECEIPT_RETENTION_DAYS der Middleware.", + "loadError": "Receipts konnten nicht geladen werden", + "empty": "Noch keine Receipts vorhanden. Sobald ein Turn mit Privacy-Shield-Aktivität abschließt, erscheint er hier.", + "scopeLabel": "Scope", + "channelLabel": "Channel", + "modelLabel": "Modell", + "turnIdLabel": "Turn", + "summaryCounts": "{masked} maskiert · {datasets} Datasets", + "loadMore": "Mehr laden", + "loadingMore": "Lädt…" + }, "operatorChannels": { "metaTitle": "Channels · omadia", "title": "Channels", diff --git a/web-ui/messages/en.json b/web-ui/messages/en.json index c6c9b3279..a56ac0aac 100644 --- a/web-ui/messages/en.json +++ b/web-ui/messages/en.json @@ -428,6 +428,7 @@ "graph": "Graph", "routines": "Routines", "conductor": "Conductor", + "receipts": "Receipts", "admin": "Admin", "system": "System", "agentsCluster": "Orchestrators", @@ -707,6 +708,20 @@ "templatePendingEmpty": "No templates waiting for review", "templateNoMatches": "No templates match your filters." }, + "operatorReceipts": { + "metaTitle": "Receipts · omadia", + "title": "Privacy receipts", + "subtitle": "Every completed turn writes its PII-free privacy receipt here — what the shield interned, masked, and passed through. The record survives the chat view; retention is bounded by the middleware's RECEIPT_RETENTION_DAYS.", + "loadError": "Failed to load receipts", + "empty": "No receipts recorded yet. Receipts appear here as soon as a turn with privacy-shield activity completes.", + "scopeLabel": "Scope", + "channelLabel": "Channel", + "modelLabel": "Model", + "turnIdLabel": "Turn", + "summaryCounts": "{masked} masked · {datasets} datasets", + "loadMore": "Load more", + "loadingMore": "Loading…" + }, "operatorChannels": { "metaTitle": "Channels · omadia", "title": "Channels", From 872e6c0c3ec87710ebb621192bb4f2165745b6ac Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 20 Aug 2026 12:22:16 +0200 Subject: [PATCH 2/5] fix: use canonical Button for receipts load-more (lint) --- .../operator/receipts/_components/ReceiptsList.tsx | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/web-ui/app/operator/receipts/_components/ReceiptsList.tsx b/web-ui/app/operator/receipts/_components/ReceiptsList.tsx index 1a396ccf0..14bad08ef 100644 --- a/web-ui/app/operator/receipts/_components/ReceiptsList.tsx +++ b/web-ui/app/operator/receipts/_components/ReceiptsList.tsx @@ -4,6 +4,7 @@ import { useState } from 'react'; import { useFormatter, useTranslations } from 'next-intl'; import { PrivacyReceiptCard } from '../../../_components/chat/PrivacyReceiptCard'; +import { Button } from '../../../_components/ui/Button'; import { listReceipts, type ReceiptsPageDto, @@ -106,14 +107,16 @@ export function ReceiptsList({ initial }: ReceiptsListProps): React.ReactElement

{loadError}

) : null} {nextCursor ? ( - + {t('loadMore')} + ) : null} ); From b5726ca07e55d5410b2f59a9f19573d7729f6b0f Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 20 Aug 2026 13:46:35 +0200 Subject: [PATCH 3/5] feat: tamper-evident receipt chain with signed checkpoints (#758) --- docs/CHANGELOG.md | 28 ++ docs/ai-act-transparency.md | 12 +- docs/middleware-agent-handoff.md | 16 ++ docs/security-architecture.md | 17 ++ middleware/.env.example | 11 + .../migrations/0041_receipt_hash_chain.sql | 69 +++++ .../scripts/generate-audit-signing-key.mjs | 25 ++ middleware/src/config.ts | 12 + middleware/src/index.ts | 28 ++ middleware/src/receipts/chain.ts | 123 +++++++++ middleware/src/receipts/checkpoints.ts | 175 ++++++++++++ middleware/src/receipts/store.ts | 81 +++++- middleware/test/receiptHashChain.test.ts | 256 ++++++++++++++++++ middleware/test/turnReceipts.test.ts | 46 +++- 14 files changed, 872 insertions(+), 27 deletions(-) create mode 100644 middleware/migrations/0041_receipt_hash_chain.sql create mode 100644 middleware/scripts/generate-audit-signing-key.mjs create mode 100644 middleware/src/receipts/chain.ts create mode 100644 middleware/src/receipts/checkpoints.ts create mode 100644 middleware/test/receiptHashChain.test.ts diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 01c6c0341..4c9476c82 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -18,6 +18,34 @@ entry. See `CONTRIBUTING.md` § Releases & changelog. ## [Unreleased] +### Added — tamper-evident receipt chain: hash chaining + signed checkpoints (#758) + +- **Hash chain (migration `0041`).** Every persisted receipt row now joins a + per-stream chain: `entry_hash = sha256(stream ‖ seq ‖ prev_hash ‖ + canonical(payload))`, appends serialized through a `FOR UPDATE`-locked + stream head so concurrent turns form one linear chain. Editing row *n* + breaks the copy of its hash stored in row *n+1* — visible to every later + entry. Replayed turns roll the whole transaction back (no phantom head + movement). UPDATE on `turn_receipts` is trigger-forbidden (defence in + depth; the chain is the proof); DELETE stays legal for retention, and + deletions show as seq gaps. +- **Ed25519 checkpoints.** On an interval (`AUDIT_CHECKPOINT_INTERVAL_MINUTES`, + default 60) the stream head is signed with a key held ONLY in + env/secret-manager (`AUDIT_SIGNING_KEY` — never in Postgres, or the admin + the chain defends against could re-sign a rewritten chain). Optional + external anchor file (`AUDIT_ANCHOR_PATH`, JSONL) for WORM storage. + Keygen: `node scripts/generate-audit-signing-key.mjs`. Public key + + fingerprint served at `GET /api/v1/operator/provenance/public-key`. +- **Verification foundation** (`verifyChainSegment`) ships with tamper tests + (edit → `hash_mismatch` at the exact seq; delete → `seq_gap`; forged + suffix → `link_mismatch`); the operator-facing verify surface (endpoint, + signed export, offline verifier, UI) is #761 — until it ships, + "cryptographically verifiable" remains a non-claim + (`docs/ai-act-transparency.md`). +- Known limitations, stated: detection not prevention; per-row time is + anchored by checkpoint cadence, not per-row (`created_at` is outside the + hash); pre-chain rows carry NULL chain columns ("pre-chain era"). + ### Added — persistent per-turn privacy receipts (#757) - **`turn_receipts` (migration `0039`).** The per-turn `PrivacyReceipt` is no diff --git a/docs/ai-act-transparency.md b/docs/ai-act-transparency.md index d9daaa72d..6f91f8f7e 100644 --- a/docs/ai-act-transparency.md +++ b/docs/ai-act-transparency.md @@ -162,10 +162,14 @@ und begründet in #684; jeder Ausfall wird seitdem gezählt und protokolliert. Migration `0039`): der PII-freie Privacy-Receipt jedes abgeschlossenen Turns wird auf dem Postgres-Backend synchron gespeichert — ohne Graph-Sink, ohne User-Cluster-Vorbedingung — und ist unter `/api/v1/operator/receipts` (auth-gated) sowie im Operator-UI abrufbar. -Fehlschläge werden gezählt und protokolliert, nie still verworfen. Der Receipt ist ein -*Record*, aber (noch) nicht manipulationssicher: Hash-Verkettung, Signaturen und -Verifikation sind #758/#761 — bis dahin bleibt „kryptographisch nachweisbar" eine -Nicht-Zusage. +Fehlschläge werden gezählt und protokolliert, nie still verworfen. **Seit #758 ist der +Record hash-verkettet und checkpoint-signiert** (Migration `0041`: `entry_hash` über +`prev_hash` verkettet, Ed25519-Checkpoints mit Schlüssel außerhalb der DB, optionaler +externer Anker) — eine nachträgliche Änderung bricht die Kette sichtbar. Was noch fehlt, +ist die **Verifikations-Fläche** (#761: Verify-Endpoint, signierter Export, +Offline-Verifier, UI). Bis #761 shipped, bleibt „kryptographisch nachweisbar" öffentlich +eine Nicht-Zusage — intern ist der Mechanismus da, aber ein Nachweis, den nur wir führen +können, ist noch kein Nachweis. **C2PA ist offen.** Im Code existiert keine C2PA-Implementierung. Für Bilder wäre das der naheliegende nächste Schritt; heute ist es keine Zusage, sondern ein offener Punkt. diff --git a/docs/middleware-agent-handoff.md b/docs/middleware-agent-handoff.md index 55d6e41c6..a6a320d89 100644 --- a/docs/middleware-agent-handoff.md +++ b/docs/middleware-agent-handoff.md @@ -1118,6 +1118,22 @@ auth-gated **`GET /api/v1/operator/receipts`** (Liste, Composite-Keyset-Cursor Reaper mit Eager-Boot-Tick, Cutoff auf der DB-Uhr. Tests: `test/turnReceipts.test.ts`, `test/orchestrator/turnReceiptPersistence.test.ts`. +### Receipt-Hash-Kette + signierte Checkpoints (#758) + +`turn_receipts` ist seit Migration `0041` hash-verkettet: `entry_hash = +sha256(stream ‖ seq ‖ prev_hash ‖ canonical(payload))`, Appends serialisiert +über `audit_stream_heads` (FOR UPDATE — eine lineare Kette, keine Forks); +Replay ⇒ kompletter Rollback. UPDATE per Trigger verboten, DELETE bleibt für +Retention erlaubt (Lücken sind detektierbar). Ed25519-Checkpoints +(`src/receipts/checkpoints.ts`): Key NUR in Env (`AUDIT_SIGNING_KEY`, +Keygen `scripts/generate-audit-signing-key.mjs`), Intervall +`AUDIT_CHECKPOINT_INTERVAL_MINUTES` (60), externer Anker `AUDIT_ANCHOR_PATH` +(JSONL). Public Key: **`GET /api/v1/operator/provenance/public-key`**. +Verify-Grundstein `verifyChainSegment` in `src/receipts/chain.ts` (Tamper- +Tests in `test/receiptHashChain.test.ts`); die Operator-Verify-Fläche ist +#761. Zeitanker: Checkpoint-Kadenz, nicht pro Zeile (`created_at` ist +außerhalb des Hashes — bewusst, Doku im Migration-Header). + ## 4. Migration Managed Agents → Lokal ### Warum migriert diff --git a/docs/security-architecture.md b/docs/security-architecture.md index 6876a42db..c2c96b344 100644 --- a/docs/security-architecture.md +++ b/docs/security-architecture.md @@ -197,6 +197,23 @@ the connection. A subscription URL is also checked at creation time (`assertOutboundUrlAllowed`), so an operator gets an immediate 400 rather than only discovering the block on the first delivery attempt. +## 7b. Tamper-evident receipt chain (#758) + +The per-turn receipt record (`turn_receipts`, #757) is hash-chained: each +row's `entry_hash` covers its canonical payload plus the previous row's +hash, appends serialized through a locked stream head. Editing a row breaks +the copy of its hash stored in the next row — the chain visibly breaks for +every later entry. Periodic Ed25519 checkpoints sign the head with a key +held **only** in env/secret-manager (`AUDIT_SIGNING_KEY`) — never in +Postgres, or the DB admin the chain defends against could re-sign a +rewritten chain — optionally anchored to an external append-only file +(`AUDIT_ANCHOR_PATH`) for WORM storage. Threat model: **detection, not +prevention** — wholesale destruction shows as sequence gaps and orphaned +checkpoints; per-row timestamps are anchored by checkpoint cadence, not +per-row. UPDATE on the table is trigger-forbidden as defence in depth; +DELETE stays legal for bounded retention. The operator verify surface +(endpoint, signed export, offline verifier) is #761. + ## 8. What lives in the vault At a minimum, your deployment vault holds: diff --git a/middleware/.env.example b/middleware/.env.example index f240b5b2b..8ebd620cd 100644 --- a/middleware/.env.example +++ b/middleware/.env.example @@ -242,6 +242,17 @@ DEV_PLATFORM_SUBSCRIPTION_MODE=false # Postgres backend only). turn_id + scope are personal-data linkage, so rows # are reaped after this many days. RECEIPT_RETENTION_DAYS=90 + +# --- Receipt hash chain checkpoints (#758) ---------------------------------- +# Ed25519 private key (base64 PKCS#8 DER) for signing chain checkpoints — +# generate with `node scripts/generate-audit-signing-key.mjs`. Keep it in a +# secret manager / env, NEVER in the database. Absent = chain builds, no +# signed checkpoints (logged loudly at boot). +# AUDIT_SIGNING_KEY= +AUDIT_CHECKPOINT_INTERVAL_MINUTES=60 +# Optional external anchor file (JSONL, append-only) outside the DB — point +# at storage the DB admin cannot rewrite (WORM / synced bucket). +# AUDIT_ANCHOR_PATH=/data/audit-anchors.jsonl # DEV_PLATFORM_SUBSCRIPTION_ACK= # required acknowledgment string when SUBSCRIPTION_MODE=true # --- Conductor generic webhooks (issue #437) -------------------------------- diff --git a/middleware/migrations/0041_receipt_hash_chain.sql b/middleware/migrations/0041_receipt_hash_chain.sql new file mode 100644 index 000000000..863f59173 --- /dev/null +++ b/middleware/migrations/0041_receipt_hash_chain.sql @@ -0,0 +1,69 @@ +-- #758 — tamper-evident receipt chain: hash chaining + signed checkpoints. +-- +-- Builds directly on `turn_receipts` (0039, #757). Every receipt row joins a +-- per-stream hash chain: `entry_hash = sha256(canonical(payload) || prev_hash +-- || seq)`. Editing row n breaks the match stored in row n+1 — the chain +-- visibly breaks for every later entry. Periodic Ed25519 checkpoints sign +-- (stream, seq, head_hash) with a key held OUTSIDE the database, so the +-- whole chain cannot be silently rewritten either. +-- +-- Threat model: DETECTION of after-the-fact modification, not prevention. +-- Wholesale destruction shows as seq gaps + orphaned checkpoints. +-- +-- Retention interplay: the reaper (#757) legitimately DELETEs expired rows, +-- so DELETE stays allowed and deletions are detectable (seq gaps below the +-- oldest surviving row are expected exactly up to the retention horizon). +-- UPDATE is never legitimate on this table — blocked by trigger below +-- (defence in depth: an admin can drop the trigger; the chain is the proof). +-- +-- Rows written before this migration (or while chaining was not yet active) +-- have NULL chain columns — the "pre-chain era", which a verifier reports as +-- unverifiable rather than broken. +-- +-- Numbering: 0039 = #757 (turn_receipts), 0040 = #760 (privacy_miss_reports). + +ALTER TABLE turn_receipts + ADD COLUMN IF NOT EXISTS stream_id TEXT, + ADD COLUMN IF NOT EXISTS seq BIGINT, + ADD COLUMN IF NOT EXISTS prev_hash BYTEA, + ADD COLUMN IF NOT EXISTS entry_hash BYTEA, + ADD COLUMN IF NOT EXISTS hash_version SMALLINT; + +CREATE UNIQUE INDEX IF NOT EXISTS turn_receipts_stream_seq + ON turn_receipts (stream_id, seq); + +-- Serialization point for chain appends: one row per stream, locked +-- FOR UPDATE inside the insert transaction so concurrent appends line up +-- into a single linear chain (no forks). +CREATE TABLE IF NOT EXISTS audit_stream_heads ( + stream_id TEXT PRIMARY KEY, + head_seq BIGINT NOT NULL, + head_hash BYTEA NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- Signed checkpoints: Ed25519 over (stream_id, seq, head_hash, signed_at). +-- The private key lives in env/secret manager, NEVER in this database — +-- otherwise the admin we defend against could re-sign a rewritten chain. +CREATE TABLE IF NOT EXISTS audit_checkpoints ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + stream_id TEXT NOT NULL, + seq BIGINT NOT NULL, + head_hash BYTEA NOT NULL, + signed_at TIMESTAMPTZ NOT NULL, + signature BYTEA NOT NULL, + public_key_fingerprint TEXT NOT NULL, + UNIQUE (stream_id, seq) +); + +-- Defence in depth, not the proof: UPDATE is never legitimate on receipts. +CREATE OR REPLACE FUNCTION turn_receipts_forbid_update() RETURNS trigger AS $$ +BEGIN + RAISE EXCEPTION 'turn_receipts is append-only: UPDATE is forbidden (#758)'; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS turn_receipts_no_update ON turn_receipts; +CREATE TRIGGER turn_receipts_no_update + BEFORE UPDATE ON turn_receipts + FOR EACH ROW EXECUTE FUNCTION turn_receipts_forbid_update(); diff --git a/middleware/scripts/generate-audit-signing-key.mjs b/middleware/scripts/generate-audit-signing-key.mjs new file mode 100644 index 000000000..38797929e --- /dev/null +++ b/middleware/scripts/generate-audit-signing-key.mjs @@ -0,0 +1,25 @@ +#!/usr/bin/env node +/** + * #758 — generate the Ed25519 checkpoint-signing keypair. + * + * Prints the PRIVATE key as base64 PKCS#8 DER (the AUDIT_SIGNING_KEY value) + * and the PUBLIC key as PEM + fingerprint (hand the public half to auditors + * out-of-band; pin the fingerprint). Store the private key in your secret + * manager / env — NEVER in the database the chain defends. + */ + +import { createHash, generateKeyPairSync } from 'node:crypto'; + +const { privateKey, publicKey } = generateKeyPairSync('ed25519'); + +const privateDer = privateKey.export({ format: 'der', type: 'pkcs8' }); +const publicPem = publicKey.export({ format: 'pem', type: 'spki' }).toString(); +const fingerprint = createHash('sha256') + .update(publicKey.export({ format: 'der', type: 'spki' })) + .digest('hex'); + +console.log('AUDIT_SIGNING_KEY (private — keep in secret manager / env):\n'); +console.log(privateDer.toString('base64')); +console.log('\nPublic key (share with auditors out-of-band):\n'); +console.log(publicPem); +console.log(`Public key fingerprint (sha256 of SPKI DER):\n\n${fingerprint}`); diff --git a/middleware/src/config.ts b/middleware/src/config.ts index 28fe8b406..2a8f8edd9 100644 --- a/middleware/src/config.ts +++ b/middleware/src/config.ts @@ -288,6 +288,18 @@ const ConfigSchema = z.object({ // scope are personal-data linkage, so rows are reaped after this many days. RECEIPT_RETENTION_DAYS: z.coerce.number().int().positive().default(90), + // #758 — Ed25519 checkpoint signing for the receipt hash chain. The + // private key (base64 PKCS#8 DER; generate with + // scripts/generate-audit-signing-key.mjs) lives HERE — env / secret + // manager — never in Postgres: the admin the chain defends against must + // not be able to re-sign a rewritten chain. Absent ⇒ the chain still + // builds, but no signed checkpoints are produced (logged loudly at boot). + AUDIT_SIGNING_KEY: z.string().optional(), + AUDIT_CHECKPOINT_INTERVAL_MINUTES: z.coerce.number().int().positive().default(60), + // Optional external anchor: checkpoint JSONL appended OUTSIDE the DB — + // point it at storage the DB admin cannot rewrite (WORM/S3 sync). + AUDIT_ANCHOR_PATH: z.string().optional(), + // Epic #470 W4 — default per-job LLM cost budget (USD) applied when neither the // job nor its repo sets one (spec §5). Token budgets have NO default: they are // enforced only when explicitly set on the job or repo. diff --git a/middleware/src/index.ts b/middleware/src/index.ts index c3e67b2b3..3fabad445 100644 --- a/middleware/src/index.ts +++ b/middleware/src/index.ts @@ -34,6 +34,7 @@ import { wireConductor, AwaitNotPendingError, AwaitResponderNotHolderError, Cond import { TURN_RECEIPT_STORE_SERVICE_NAME } from '@omadia/plugin-api'; import { PgTurnReceiptStore, startTurnReceiptReaper } from './receipts/store.js'; import { createReceiptRoutes } from './receipts/routes.js'; +import { loadCheckpointSigner, startCheckpointWorker } from './receipts/checkpoints.js'; import { bindingKeyForTurn } from './conductor/principalId.js'; import { createOperatorChannelsRouter } from './routes/operatorChannels.js'; import { createAgentBuilderRouter } from './routes/agentBuilder.js'; @@ -3505,6 +3506,33 @@ async function main(): Promise { `[middleware] turn receipts wired at /api/v1/operator/receipts (auth-gated, retention ${config.RECEIPT_RETENTION_DAYS}d)`, ); + // #758 — signed checkpoints over the receipt hash chain. The chain + // itself always builds (the store appends chained rows unconditionally); + // signing is the layer that needs the operator-held key. Absent key ⇒ + // loud boot log, not a silent no-op. + if (config.AUDIT_SIGNING_KEY) { + const signer = loadCheckpointSigner(config.AUDIT_SIGNING_KEY); + startCheckpointWorker(graphPool, signer, { + intervalMs: config.AUDIT_CHECKPOINT_INTERVAL_MINUTES * 60_000, + ...(config.AUDIT_ANCHOR_PATH ? { anchorPath: config.AUDIT_ANCHOR_PATH } : {}), + }); + app.get('/api/v1/operator/provenance/public-key', requireAuth, (_req, res) => { + res.json({ + publicKeyPem: signer.publicKeyPem, + fingerprint: signer.publicKeyFingerprint, + checkpointIntervalMinutes: config.AUDIT_CHECKPOINT_INTERVAL_MINUTES, + anchorConfigured: Boolean(config.AUDIT_ANCHOR_PATH), + }); + }); + console.log( + `[middleware] audit checkpoints wired (every ${config.AUDIT_CHECKPOINT_INTERVAL_MINUTES}min, fingerprint ${signer.publicKeyFingerprint.slice(0, 16)}…${config.AUDIT_ANCHOR_PATH ? ', external anchor on' : ''})`, + ); + } else { + console.warn( + '[middleware] AUDIT_SIGNING_KEY not set — receipt chain builds WITHOUT signed checkpoints; generate a key with scripts/generate-audit-signing-key.mjs (#758)', + ); + } + const userStore = new UserStore(graphPool); const bootstrapResult = await runAuthBootstrap({ diff --git a/middleware/src/receipts/chain.ts b/middleware/src/receipts/chain.ts new file mode 100644 index 000000000..6bc69aa3e --- /dev/null +++ b/middleware/src/receipts/chain.ts @@ -0,0 +1,123 @@ +/** + * #758 — receipt hash chain: canonicalization, entry hashing, and segment + * verification. Pure functions — the transactional append lives in + * `store.ts`, the signing in `checkpoints.ts`. + * + * Mechanism (the sentence for the docs): every entry carries the fingerprint + * of its predecessor; editing entry n changes its hash, which no longer + * matches the copy stored in entry n+1 — the chain visibly breaks for every + * later entry. Signed checkpoints anchored outside the DB mean the whole + * chain cannot be silently rewritten either. Detection, not prevention. + */ + +import { createHash } from 'node:crypto'; + +/** Bump when the canonicalization or hash input layout changes; stored per + * row so old chains stay verifiable under their own rules. */ +export const HASH_VERSION = 1; + +/** The one stream #758 ships. More streams (admin_audit, …) join later. */ +export const RECEIPT_STREAM_ID = 'receipts'; + +/** + * Canonical JSON — the RFC 8785 (JCS) subset this codebase needs: objects + * with lexicographically sorted keys (code-unit order), arrays in order, + * `undefined` object members dropped (JSON.stringify semantics), numbers as + * JSON.stringify renders them. Payloads here are JSONB round-trips — + * strings, bounded numbers, booleans, null, plain objects/arrays — so the + * full-JCS number edge cases (±0, exponents beyond double round-trip) cannot + * arise; `hash_version` exists for the day that changes. + */ +export function canonicalJson(value: unknown): string { + if (value === null || typeof value !== 'object') { + return JSON.stringify(value) ?? 'null'; + } + if (Array.isArray(value)) { + return `[${value.map((v) => canonicalJson(v === undefined ? null : v)).join(',')}]`; + } + const record = value as Record; + const keys = Object.keys(record) + .filter((k) => record[k] !== undefined) + .sort(); + const body = keys.map((k) => `${JSON.stringify(k)}:${canonicalJson(record[k])}`).join(','); + return `{${body}}`; +} + +/** Genesis hash of a stream: sha256 over the stream id itself — a fixed, + * reproducible starting anchor an offline verifier can recompute. */ +export function genesisHash(streamId: string): Buffer { + return createHash('sha256').update(`genesis:${streamId}`, 'utf-8').digest(); +} + +export interface ChainEntryInput { + readonly streamId: string; + readonly seq: number; + readonly prevHash: Buffer; + /** The canonical payload — for receipts: {turnId, sessionScope, channel, + * model, receipt}. Never includes DB-generated values (created_at): + * time is anchored by checkpoint cadence, not per-row (documented). */ + readonly payload: unknown; +} + +/** entry_hash = sha256(streamId \n seq \n hex(prevHash) \n canonical(payload)). + * Newline framing keeps fields from bleeding into each other. */ +export function computeEntryHash(input: ChainEntryInput): Buffer { + return createHash('sha256') + .update( + `${input.streamId}\n${String(input.seq)}\n${input.prevHash.toString('hex')}\n${canonicalJson(input.payload)}`, + 'utf-8', + ) + .digest(); +} + +export interface ChainRow { + readonly seq: number; + readonly prevHash: Buffer; + readonly entryHash: Buffer; + readonly payload: unknown; +} + +export type ChainBreakKind = 'hash_mismatch' | 'link_mismatch' | 'seq_gap'; + +export interface ChainVerdict { + readonly ok: boolean; + readonly checkedEntries: number; + readonly firstBrokenSeq?: number; + readonly breakKind?: ChainBreakKind; +} + +/** + * Verify a contiguous ascending-`seq` segment. `trustedPrevHash` is the hash + * the first row must link to: the genesis hash when the segment starts at + * seq 1, or a hash vouched for out-of-band (a signed checkpoint) when the + * prefix was reaped by retention. Foundation for the #761 verify surface; + * shipped here so the chain is testable (tamper tests) from day one. + */ +export function verifyChainSegment( + streamId: string, + rows: readonly ChainRow[], + trustedPrevHash: Buffer, +): ChainVerdict { + let prev = trustedPrevHash; + let expectedSeq: number | undefined; + for (const row of rows) { + if (expectedSeq !== undefined && row.seq !== expectedSeq) { + return { ok: false, checkedEntries: rows.indexOf(row), firstBrokenSeq: row.seq, breakKind: 'seq_gap' }; + } + if (!row.prevHash.equals(prev)) { + return { ok: false, checkedEntries: rows.indexOf(row), firstBrokenSeq: row.seq, breakKind: 'link_mismatch' }; + } + const recomputed = computeEntryHash({ + streamId, + seq: row.seq, + prevHash: row.prevHash, + payload: row.payload, + }); + if (!recomputed.equals(row.entryHash)) { + return { ok: false, checkedEntries: rows.indexOf(row), firstBrokenSeq: row.seq, breakKind: 'hash_mismatch' }; + } + prev = row.entryHash; + expectedSeq = row.seq + 1; + } + return { ok: true, checkedEntries: rows.length }; +} diff --git a/middleware/src/receipts/checkpoints.ts b/middleware/src/receipts/checkpoints.ts new file mode 100644 index 000000000..9c09c1de1 --- /dev/null +++ b/middleware/src/receipts/checkpoints.ts @@ -0,0 +1,175 @@ +/** + * #758 — signed audit checkpoints. On an interval, sign the receipt stream's + * current head `(stream_id, seq, head_hash, signed_at)` with Ed25519 and + * persist the checkpoint — plus, when configured, append it to an external + * anchor file (JSONL) OUTSIDE the database, suitable for shipping to WORM + * storage. The private key comes from the environment / secret manager and + * is NEVER stored in Postgres: the admin the chain defends against must not + * be able to re-sign a rewritten chain. + * + * Key format: base64-encoded PKCS#8 DER Ed25519 private key — generate with + * `node scripts/generate-audit-signing-key.mjs`. + */ + +import { appendFile } from 'node:fs/promises'; +import { + createHash, + createPrivateKey, + createPublicKey, + sign as edSign, + type KeyObject, +} from 'node:crypto'; +import type { Pool } from 'pg'; + +import { RECEIPT_STREAM_ID } from './chain.js'; + +export interface CheckpointSigner { + readonly privateKey: KeyObject; + readonly publicKeyPem: string; + /** sha256 over the SPKI DER of the public key, hex — the stable id an + * offline verifier pins. */ + readonly publicKeyFingerprint: string; +} + +/** Parse the configured signing key. Throws with a actionable message on a + * malformed key — a silently-disabled signer would be the #640 no-op. */ +export function loadCheckpointSigner(privateKeyBase64: string): CheckpointSigner { + let privateKey: KeyObject; + try { + privateKey = createPrivateKey({ + key: Buffer.from(privateKeyBase64, 'base64'), + format: 'der', + type: 'pkcs8', + }); + } catch (err) { + throw new Error( + `AUDIT_SIGNING_KEY is not a base64 PKCS#8 Ed25519 private key (generate one with scripts/generate-audit-signing-key.mjs): ${err instanceof Error ? err.message : String(err)}`, + ); + } + if (privateKey.asymmetricKeyType !== 'ed25519') { + throw new Error( + `AUDIT_SIGNING_KEY must be an Ed25519 key, got '${privateKey.asymmetricKeyType ?? 'unknown'}'`, + ); + } + const publicKey = createPublicKey(privateKey); + const spkiDer = publicKey.export({ format: 'der', type: 'spki' }); + return { + privateKey, + publicKeyPem: publicKey.export({ format: 'pem', type: 'spki' }).toString(), + publicKeyFingerprint: createHash('sha256').update(spkiDer).digest('hex'), + }; +} + +/** The exact bytes a checkpoint signature covers. Newline-framed, hex for + * the hash — reproducible by an offline verifier with no library beyond + * node:crypto. */ +export function checkpointSigningInput(input: { + streamId: string; + seq: number; + headHash: Buffer; + signedAtIso: string; +}): Buffer { + return Buffer.from( + `omadia-audit-checkpoint-v1\n${input.streamId}\n${String(input.seq)}\n${input.headHash.toString('hex')}\n${input.signedAtIso}`, + 'utf-8', + ); +} + +export interface CheckpointRecord { + readonly streamId: string; + readonly seq: number; + readonly headHashHex: string; + readonly signedAtIso: string; + readonly signatureBase64: string; + readonly publicKeyFingerprint: string; +} + +/** + * One checkpoint pass: read the stream head; if it advanced past the last + * checkpoint, sign and persist (+ optionally anchor). Exported for tests and + * for an eager boot pass. Returns the record written, or undefined when the + * head has not moved (no pointless duplicate checkpoints). + */ +export async function runCheckpointPass( + pool: Pool, + signer: CheckpointSigner, + opts: { anchorPath?: string; now?: () => Date }, +): Promise { + const head = await pool.query<{ head_seq: string; head_hash: Buffer }>( + `SELECT head_seq, head_hash FROM audit_stream_heads WHERE stream_id = $1`, + [RECEIPT_STREAM_ID], + ); + const row = head.rows[0]; + if (!row) return undefined; // nothing recorded yet + const seq = Number(row.head_seq); + const last = await pool.query<{ seq: string }>( + `SELECT MAX(seq)::text AS seq FROM audit_checkpoints WHERE stream_id = $1`, + [RECEIPT_STREAM_ID], + ); + const lastSeq = last.rows[0]?.seq ? Number(last.rows[0].seq) : 0; + if (seq <= lastSeq) return undefined; + + const signedAt = (opts.now?.() ?? new Date()).toISOString(); + const signature = edSign( + null, // Ed25519: algorithm must be null/undefined + checkpointSigningInput({ + streamId: RECEIPT_STREAM_ID, + seq, + headHash: row.head_hash, + signedAtIso: signedAt, + }), + signer.privateKey, + ); + await pool.query( + `INSERT INTO audit_checkpoints + (stream_id, seq, head_hash, signed_at, signature, public_key_fingerprint) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (stream_id, seq) DO NOTHING`, + [RECEIPT_STREAM_ID, seq, row.head_hash, signedAt, signature, signer.publicKeyFingerprint], + ); + const record: CheckpointRecord = { + streamId: RECEIPT_STREAM_ID, + seq, + headHashHex: row.head_hash.toString('hex'), + signedAtIso: signedAt, + signatureBase64: signature.toString('base64'), + publicKeyFingerprint: signer.publicKeyFingerprint, + }; + if (opts.anchorPath) { + // External anchor: append-only JSONL outside the DB. Failure is loud but + // non-fatal — the in-DB checkpoint stands, and a missing anchor line is + // itself detectable when comparing the two. + try { + await appendFile(opts.anchorPath, `${JSON.stringify(record)}\n`, 'utf-8'); + } catch (err) { + console.error('[receipts] checkpoint anchor append failed:', err); + } + } + return record; +} + +/** Interval worker — unref'd, eager first pass (the table exists: plugin + * activation applies migrations well before this wiring, same reasoning as + * the retention reaper). */ +export function startCheckpointWorker( + pool: Pool, + signer: CheckpointSigner, + opts: { intervalMs: number; anchorPath?: string }, +): { stop: () => void } { + const tick = async (): Promise => { + try { + const record = await runCheckpointPass(pool, signer, { anchorPath: opts.anchorPath }); + if (record) { + console.log( + `[receipts] checkpoint signed: stream=${record.streamId} seq=${String(record.seq)} fingerprint=${record.publicKeyFingerprint.slice(0, 16)}…`, + ); + } + } catch (err) { + console.error('[receipts] checkpoint pass failed:', err); + } + }; + const timer = setInterval(() => void tick(), opts.intervalMs); + timer.unref(); + void tick(); + return { stop: () => clearInterval(timer) }; +} diff --git a/middleware/src/receipts/store.ts b/middleware/src/receipts/store.ts index 697e5a0b8..15a61a541 100644 --- a/middleware/src/receipts/store.ts +++ b/middleware/src/receipts/store.ts @@ -20,6 +20,13 @@ import type { TurnReceiptStore, } from '@omadia/plugin-api'; +import { + HASH_VERSION, + RECEIPT_STREAM_ID, + computeEntryHash, + genesisHash, +} from './chain.js'; + /** Process-wide failure counters, exported for /health-style introspection * and asserted in tests. Mirrors `runTraceObservability.ts`'s "count what * would otherwise be silently incomplete" obligation. */ @@ -40,16 +47,51 @@ export function resetTurnReceiptCounters(): void { counters.persistFailures = 0; } +/** #758 — the canonical hash payload of a receipt row. NEVER includes + * DB-generated values (created_at): time is anchored by checkpoint cadence, + * not per-row. Exported so the #761 verifier recomputes the identical shape. */ +export function receiptChainPayload(entry: TurnReceiptRecordInput): unknown { + return { + turnId: entry.turnId, + sessionScope: entry.sessionScope ?? null, + channel: entry.channel ?? null, + model: entry.model ?? null, + receipt: entry.receipt, + }; +} + export class PgTurnReceiptStore implements TurnReceiptStore { constructor(private readonly pool: Pool) {} async record(entry: TurnReceiptRecordInput): Promise { + const client = await this.pool.connect(); try { - // Idempotent on turn_id: a replayed `done` event (retry, double flush) - // must not duplicate the row; first write wins. - const result = await this.pool.query( - `INSERT INTO turn_receipts (turn_id, session_scope, channel, model, receipt) - VALUES ($1, $2, $3, $4, $5::jsonb) + // #758 — chained append. One transaction: lock the stream head + // (FOR UPDATE serializes concurrent appends into a single linear + // chain — no forks), compute seq/prev, insert, advance the head. + // Idempotence on turn_id is preserved: a replayed `done` event hits + // DO NOTHING, and then the head must NOT advance — the transaction + // rolls back to keep head and rows consistent. + await client.query('BEGIN'); + const headRes = await client.query<{ head_seq: string; head_hash: Buffer }>( + `SELECT head_seq, head_hash FROM audit_stream_heads + WHERE stream_id = $1 FOR UPDATE`, + [RECEIPT_STREAM_ID], + ); + const head = headRes.rows[0]; + const prevHash = head ? head.head_hash : genesisHash(RECEIPT_STREAM_ID); + const seq = head ? Number(head.head_seq) + 1 : 1; + const entryHash = computeEntryHash({ + streamId: RECEIPT_STREAM_ID, + seq, + prevHash, + payload: receiptChainPayload(entry), + }); + const inserted = await client.query( + `INSERT INTO turn_receipts + (turn_id, session_scope, channel, model, receipt, + stream_id, seq, prev_hash, entry_hash, hash_version) + VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9, $10) ON CONFLICT (turn_id) DO NOTHING`, [ entry.turnId, @@ -57,16 +99,37 @@ export class PgTurnReceiptStore implements TurnReceiptStore { entry.channel ?? null, entry.model ?? null, JSON.stringify(entry.receipt), + RECEIPT_STREAM_ID, + seq, + prevHash, + entryHash, + HASH_VERSION, ], ); - // A replayed turn hits DO NOTHING (rowCount 0) — that is not a - // persist, and the counter must not overstate the record. - if ((result.rowCount ?? 0) > 0) { - counters.persisted += 1; + if ((inserted.rowCount ?? 0) === 0) { + // Replayed turn: no row, no head movement, no counter. + await client.query('ROLLBACK'); + return; } + await client.query( + `INSERT INTO audit_stream_heads (stream_id, head_seq, head_hash, updated_at) + VALUES ($1, $2, $3, NOW()) + ON CONFLICT (stream_id) + DO UPDATE SET head_seq = EXCLUDED.head_seq, head_hash = EXCLUDED.head_hash, updated_at = NOW()`, + [RECEIPT_STREAM_ID, seq, entryHash], + ); + await client.query('COMMIT'); + counters.persisted += 1; } catch (err) { + try { + await client.query('ROLLBACK'); + } catch { + /* connection-level failure — nothing further to roll back */ + } counters.persistFailures += 1; throw err; + } finally { + client.release(); } } } diff --git a/middleware/test/receiptHashChain.test.ts b/middleware/test/receiptHashChain.test.ts new file mode 100644 index 000000000..a24843ecb --- /dev/null +++ b/middleware/test/receiptHashChain.test.ts @@ -0,0 +1,256 @@ +/** + * #758 — receipt hash chain: canonicalization, entry hashing, segment + * verification (tamper tests), the transactional chained append, and + * Ed25519 checkpoint signing. All against in-memory fakes; the pg wire + * behaviour (FOR UPDATE serialization) is modelled by a stateful fake pool. + */ + +import { strict as assert } from 'node:assert'; +import { generateKeyPairSync, verify as edVerify, createPublicKey } from 'node:crypto'; +import { mkdtempSync, readFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, it } from 'node:test'; +import type { Pool } from 'pg'; + +import { + RECEIPT_STREAM_ID, + canonicalJson, + computeEntryHash, + genesisHash, + verifyChainSegment, + type ChainRow, +} from '../src/receipts/chain.js'; +import { + PgTurnReceiptStore, + receiptChainPayload, + resetTurnReceiptCounters, + turnReceiptCounters, +} from '../src/receipts/store.js'; +import { + checkpointSigningInput, + loadCheckpointSigner, + runCheckpointPass, +} from '../src/receipts/checkpoints.js'; + +const RECEIPT = { + datasetsInterned: 1, + fieldsMasked: 4, + fieldsCleartext: 2, + verbsExecuted: ['v4_sort'], + pseudonymProjectionUsed: false, +}; + +describe('#758 canonicalJson', () => { + it('is key-order independent, drops undefined members, keeps array order', () => { + const a = canonicalJson({ b: 1, a: { y: [3, 1], x: 'ü' }, skip: undefined }); + const b = canonicalJson({ a: { x: 'ü', y: [3, 1] }, b: 1 }); + assert.equal(a, b); + assert.equal(a, '{"a":{"x":"ü","y":[3,1]},"b":1}'); + }); +}); + +describe('#758 computeEntryHash + verifyChainSegment', () => { + function buildChain(payloads: unknown[]): ChainRow[] { + const rows: ChainRow[] = []; + let prev = genesisHash(RECEIPT_STREAM_ID); + payloads.forEach((payload, i) => { + const seq = i + 1; + const entryHash = computeEntryHash({ streamId: RECEIPT_STREAM_ID, seq, prevHash: prev, payload }); + rows.push({ seq, prevHash: prev, entryHash, payload }); + prev = entryHash; + }); + return rows; + } + + it('a well-formed chain verifies end to end', () => { + const rows = buildChain([{ n: 1 }, { n: 2 }, { n: 3 }]); + const verdict = verifyChainSegment(RECEIPT_STREAM_ID, rows, genesisHash(RECEIPT_STREAM_ID)); + assert.deepEqual(verdict, { ok: true, checkedEntries: 3 }); + }); + + it('TAMPER: editing a mid-chain payload reports hash_mismatch at exactly that seq', () => { + const rows = buildChain([{ n: 1 }, { n: 2 }, { n: 3 }]); + const tampered = rows.map((r) => (r.seq === 2 ? { ...r, payload: { n: 99 } } : r)); + const verdict = verifyChainSegment(RECEIPT_STREAM_ID, tampered, genesisHash(RECEIPT_STREAM_ID)); + assert.equal(verdict.ok, false); + assert.equal(verdict.firstBrokenSeq, 2); + assert.equal(verdict.breakKind, 'hash_mismatch'); + }); + + it('TAMPER: deleting a mid-chain row reports seq_gap', () => { + const rows = buildChain([{ n: 1 }, { n: 2 }, { n: 3 }]); + const withGap = [rows[0]!, rows[2]!]; + const verdict = verifyChainSegment(RECEIPT_STREAM_ID, withGap, genesisHash(RECEIPT_STREAM_ID)); + assert.equal(verdict.ok, false); + assert.equal(verdict.firstBrokenSeq, 3); + assert.equal(verdict.breakKind, 'seq_gap'); + }); + + it('TAMPER: a re-written chain suffix still fails against the trusted genesis (link_mismatch)', () => { + const rows = buildChain([{ n: 1 }]); + const forged = buildChain([{ n: 999 }]).map((r) => ({ ...r, prevHash: rows[0]!.entryHash })); + const verdict = verifyChainSegment(RECEIPT_STREAM_ID, forged, genesisHash(RECEIPT_STREAM_ID)); + assert.equal(verdict.ok, false); + assert.equal(verdict.breakKind, 'link_mismatch'); + }); +}); + +// ── stateful fake pool modelling the chained-append transaction ──────────── + +interface FakeRow { + turn_id: string; + seq: number; + prev_hash: Buffer; + entry_hash: Buffer; + payload: unknown; +} + +function chainFakePool(): { + pool: Pool; + rows: FakeRow[]; + head: () => { head_seq: number; head_hash: Buffer } | undefined; +} { + const rows: FakeRow[] = []; + const turnIds = new Set(); + let head: { head_seq: number; head_hash: Buffer } | undefined; + const query = async (sql: string, params: unknown[] = []) => { + if (sql.startsWith('BEGIN') || sql.startsWith('COMMIT') || sql.startsWith('ROLLBACK')) { + // The fake applies writes immediately; ROLLBACK-safety is asserted via + // the head/rows invariants in the replay test below. + return { rows: [], rowCount: 0 }; + } + if (sql.includes('FROM audit_stream_heads') && sql.includes('FOR UPDATE')) { + return { rows: head ? [{ head_seq: String(head.head_seq), head_hash: head.head_hash }] : [], rowCount: head ? 1 : 0 }; + } + if (sql.includes('INSERT INTO turn_receipts')) { + const turnId = params[0] as string; + if (turnIds.has(turnId)) return { rows: [], rowCount: 0 }; // ON CONFLICT DO NOTHING + turnIds.add(turnId); + rows.push({ + turn_id: turnId, + seq: params[6] as number, + prev_hash: params[7] as Buffer, + entry_hash: params[8] as Buffer, + payload: JSON.parse(params[4] as string), + }); + return { rows: [], rowCount: 1 }; + } + if (sql.includes('INSERT INTO audit_stream_heads')) { + head = { head_seq: params[1] as number, head_hash: params[2] as Buffer }; + return { rows: [], rowCount: 1 }; + } + throw new Error(`chainFakePool: unscripted SQL: ${sql.slice(0, 80)}`); + }; + const client = { query, release: () => undefined }; + const pool = { connect: async () => client, query } as unknown as Pool; + return { pool, rows, head: () => head }; +} + +describe('#758 PgTurnReceiptStore chained append', () => { + it('three records form a verifiable linear chain from the genesis hash', async () => { + resetTurnReceiptCounters(); + const { pool, rows } = chainFakePool(); + const store = new PgTurnReceiptStore(pool); + for (const id of ['t-1', 't-2', 't-3']) { + await store.record({ turnId: id, sessionScope: 's', receipt: RECEIPT }); + } + assert.equal(rows.length, 3); + const chainRows: ChainRow[] = rows.map((r) => ({ + seq: r.seq, + prevHash: r.prev_hash, + entryHash: r.entry_hash, + // The verifier recomputes from the SAME payload shape the store hashed. + payload: receiptChainPayload({ turnId: r.turn_id, sessionScope: 's', receipt: r.payload as never }), + })); + const verdict = verifyChainSegment(RECEIPT_STREAM_ID, chainRows, genesisHash(RECEIPT_STREAM_ID)); + assert.deepEqual(verdict, { ok: true, checkedEntries: 3 }); + assert.equal(turnReceiptCounters().persisted, 3); + }); + + it('a replayed turn advances neither rows, nor head, nor the counter', async () => { + resetTurnReceiptCounters(); + const { pool, rows, head } = chainFakePool(); + const store = new PgTurnReceiptStore(pool); + await store.record({ turnId: 't-1', receipt: RECEIPT }); + const headAfterFirst = head()!.head_hash; + await store.record({ turnId: 't-1', receipt: RECEIPT }); // replay + assert.equal(rows.length, 1); + assert.ok(head()!.head_hash.equals(headAfterFirst), 'head must not move on a replay'); + assert.equal(turnReceiptCounters().persisted, 1); + }); +}); + +describe('#758 checkpoint signing', () => { + function makeKey(): string { + const { privateKey } = generateKeyPairSync('ed25519'); + return privateKey.export({ format: 'der', type: 'pkcs8' }).toString('base64'); + } + + it('loadCheckpointSigner rejects a non-Ed25519 key loudly', () => { + const { privateKey } = generateKeyPairSync('rsa', { modulusLength: 2048 }); + const rsaB64 = privateKey.export({ format: 'der', type: 'pkcs8' }).toString('base64'); + assert.throws(() => loadCheckpointSigner(rsaB64), /must be an Ed25519 key/); + assert.throws(() => loadCheckpointSigner('not-a-key'), /AUDIT_SIGNING_KEY/); + }); + + function checkpointFakePool(headSeq: number | undefined): { pool: Pool; inserted: unknown[][] } { + const inserted: unknown[][] = []; + let lastCheckpointSeq: number | undefined; + const pool = { + query: async (sql: string, params: unknown[] = []) => { + if (sql.includes('FROM audit_stream_heads')) { + return headSeq === undefined + ? { rows: [], rowCount: 0 } + : { rows: [{ head_seq: String(headSeq), head_hash: genesisHash('x') }], rowCount: 1 }; + } + if (sql.includes('MAX(seq)')) { + return { rows: [{ seq: lastCheckpointSeq === undefined ? null : String(lastCheckpointSeq) }], rowCount: 1 }; + } + if (sql.includes('INSERT INTO audit_checkpoints')) { + inserted.push(params); + lastCheckpointSeq = params[1] as number; + return { rows: [], rowCount: 1 }; + } + throw new Error(`checkpointFakePool: unscripted SQL: ${sql.slice(0, 60)}`); + }, + } as unknown as Pool; + return { pool, inserted }; + } + + it('signs the head, is verifiable with the public key, anchors externally, and never duplicates', async () => { + const signer = loadCheckpointSigner(makeKey()); + const { pool, inserted } = checkpointFakePool(7); + const anchorPath = join(mkdtempSync(join(tmpdir(), 'anchor-')), 'anchors.jsonl'); + const record = await runCheckpointPass(pool, signer, { anchorPath }); + assert.ok(record); + assert.equal(record.seq, 7); + // Signature verifies against the exported public key over the documented input. + const ok = edVerify( + null, + checkpointSigningInput({ + streamId: record.streamId, + seq: record.seq, + headHash: Buffer.from(record.headHashHex, 'hex'), + signedAtIso: record.signedAtIso, + }), + createPublicKey(signer.publicKeyPem), + Buffer.from(record.signatureBase64, 'base64'), + ); + assert.equal(ok, true, 'checkpoint signature must verify with the public key'); + // External anchor line landed and round-trips. + const anchored = JSON.parse(readFileSync(anchorPath, 'utf-8').trim()) as { seq: number }; + assert.equal(anchored.seq, 7); + // Head unchanged ⇒ second pass writes nothing (no duplicate checkpoints). + const again = await runCheckpointPass(pool, signer, { anchorPath }); + assert.equal(again, undefined); + assert.equal(inserted.length, 1); + }); + + it('a stream with no head yet produces no checkpoint (nothing to certify)', async () => { + const signer = loadCheckpointSigner(makeKey()); + const { pool, inserted } = checkpointFakePool(undefined); + assert.equal(await runCheckpointPass(pool, signer, {}), undefined); + assert.equal(inserted.length, 0); + }); +}); diff --git a/middleware/test/turnReceipts.test.ts b/middleware/test/turnReceipts.test.ts index 92ddfdd3d..77e997bc7 100644 --- a/middleware/test/turnReceipts.test.ts +++ b/middleware/test/turnReceipts.test.ts @@ -31,14 +31,16 @@ function fakePool( handler: (sql: string, params: unknown[]) => { rows?: unknown[]; rowCount?: number } | Error, ): { pool: Pool; queries: RecordedQuery[] } { const queries: RecordedQuery[] = []; - const pool = { - query: async (sql: string, params: unknown[] = []) => { - queries.push({ sql, params }); - const result = handler(sql, params); - if (result instanceof Error) throw result; - return { rows: result.rows ?? [], rowCount: result.rowCount ?? (result.rows?.length ?? 0) }; - }, - } as unknown as Pool; + const query = async (sql: string, params: unknown[] = []) => { + queries.push({ sql, params }); + const result = handler(sql, params); + if (result instanceof Error) throw result; + return { rows: result.rows ?? [], rowCount: result.rowCount ?? (result.rows?.length ?? 0) }; + }; + // #758 — the store's chained append runs on a dedicated client (BEGIN … + // COMMIT); route/reaper paths keep using pool.query directly. + const client = { query, release: () => undefined }; + const pool = { query, connect: async () => client } as unknown as Pool; return { pool, queries }; } @@ -62,14 +64,24 @@ describe('#757 PgTurnReceiptStore', () => { model: 'claude-test', receipt: RECEIPT, }); - assert.equal(queries.length, 1); - const q = queries[0]!; - assert.match(q.sql, /INSERT INTO turn_receipts/); + const q = queries.find((x) => x.sql.includes('INSERT INTO turn_receipts'))!; + assert.ok(q, 'a receipt INSERT must run'); // Idempotence is the SQL's job: a replayed done event must hit the // turn_id conflict target, not add a second row. assert.match(q.sql, /ON CONFLICT \(turn_id\) DO NOTHING/); assert.deepEqual(q.params.slice(0, 4), ['t-1', 'sess-1', 'teams', 'claude-test']); assert.deepEqual(JSON.parse(q.params[4] as string), RECEIPT); + // #758 — the row joins the hash chain: stream, seq 1 (genesis append), + // 32-byte prev/entry hashes, hash version. + assert.equal(q.params[5], 'receipts'); + assert.equal(q.params[6], 1); + assert.equal((q.params[7] as Buffer).length, 32); + assert.equal((q.params[8] as Buffer).length, 32); + assert.equal(q.params[9], 1); + assert.ok( + queries.some((x) => x.sql.includes('INSERT INTO audit_stream_heads')), + 'the stream head must advance with the row', + ); assert.equal(turnReceiptCounters().persisted, 1); assert.equal(turnReceiptCounters().persistFailures, 0); }); @@ -90,15 +102,21 @@ describe('#757 PgTurnReceiptStore', () => { resetTurnReceiptCounters(); const { pool, queries } = fakePool(() => ({ rowCount: 1 })); await new PgTurnReceiptStore(pool).record({ turnId: 't-3', receipt: RECEIPT }); - assert.deepEqual(queries[0]!.params.slice(0, 4), ['t-3', null, null, null]); + const q = queries.find((x) => x.sql.includes('INSERT INTO turn_receipts'))!; + assert.deepEqual(q.params.slice(0, 4), ['t-3', null, null, null]); }); - it('a replayed turn (ON CONFLICT no-op) does not inflate the persisted counter', async () => { + it('a replayed turn (ON CONFLICT no-op) rolls back: no counter, no head movement', async () => { resetTurnReceiptCounters(); - const { pool } = fakePool(() => ({ rowCount: 0 })); + const { pool, queries } = fakePool((sql) => + sql.includes('INSERT INTO turn_receipts') ? { rowCount: 0 } : { rowCount: 1, rows: [] }, + ); await new PgTurnReceiptStore(pool).record({ turnId: 't-4', receipt: RECEIPT }); assert.equal(turnReceiptCounters().persisted, 0); assert.equal(turnReceiptCounters().persistFailures, 0); + // #758 — the head must not advance for a row that was never inserted. + assert.ok(!queries.some((x) => x.sql.includes('INSERT INTO audit_stream_heads'))); + assert.ok(queries.some((x) => x.sql.startsWith('ROLLBACK'))); }); }); From 0f1e7fd6c8bdce9b7b7414e77d3d01c186752500 Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 20 Aug 2026 13:51:09 +0200 Subject: [PATCH 4/5] fix: seed genesis head, gate checkpoint anchor, normalize hash payload --- docs/middleware-agent-handoff.md | 5 ++- docs/security-architecture.md | 10 +++++ .../migrations/0041_receipt_hash_chain.sql | 13 +++++++ middleware/src/index.ts | 34 +++++++++++------ middleware/src/receipts/checkpoints.ts | 7 +++- middleware/src/receipts/store.ts | 24 ++++++++---- middleware/test/receiptHashChain.test.ts | 37 +++++++++++++++++++ 7 files changed, 108 insertions(+), 22 deletions(-) diff --git a/docs/middleware-agent-handoff.md b/docs/middleware-agent-handoff.md index a6a320d89..989fc200e 100644 --- a/docs/middleware-agent-handoff.md +++ b/docs/middleware-agent-handoff.md @@ -1132,7 +1132,10 @@ Keygen `scripts/generate-audit-signing-key.mjs`), Intervall Verify-Grundstein `verifyChainSegment` in `src/receipts/chain.ts` (Tamper- Tests in `test/receiptHashChain.test.ts`); die Operator-Verify-Fläche ist #761. Zeitanker: Checkpoint-Kadenz, nicht pro Zeile (`created_at` ist -außerhalb des Hashes — bewusst, Doku im Migration-Header). +außerhalb des Hashes — bewusst, begründet in `src/receipts/chain.ts` + +`receiptChainPayload` in `store.ts`). ⚠️ #761-Pflicht: Retention-Lücken +gegen die Checkpoint-Zeitachse prüfen (Backdating-Laundering-Kanal, s. +security-architecture §7b). ## 4. Migration Managed Agents → Lokal diff --git a/docs/security-architecture.md b/docs/security-architecture.md index c2c96b344..f353e5446 100644 --- a/docs/security-architecture.md +++ b/docs/security-architecture.md @@ -214,6 +214,16 @@ per-row. UPDATE on the table is trigger-forbidden as defence in depth; DELETE stays legal for bounded retention. The operator verify surface (endpoint, signed export, offline verifier) is #761. +One consequence to state explicitly: because `created_at` sits outside the +hash and DELETE is legal, an admin who drops the trigger could backdate +`created_at` and let the reaper delete a row early — presenting the gap as +legal retention. The mitigation is the checkpoint timeline: a row with +`seq ≤` a checkpoint's seq provably existed by that checkpoint's signed +time, so **the #761 verifier MUST check every retention gap's age against +the checkpoint timeline** (a gap younger than the retention window measured +in checkpoint time is a finding, not retention). Recorded as a hard +requirement on #761. + ## 8. What lives in the vault At a minimum, your deployment vault holds: diff --git a/middleware/migrations/0041_receipt_hash_chain.sql b/middleware/migrations/0041_receipt_hash_chain.sql index 863f59173..aec3c9fef 100644 --- a/middleware/migrations/0041_receipt_hash_chain.sql +++ b/middleware/migrations/0041_receipt_hash_chain.sql @@ -42,6 +42,19 @@ CREATE TABLE IF NOT EXISTS audit_stream_heads ( updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); +-- Seed the receipts head at genesis (review H1): `SELECT … FOR UPDATE` on a +-- row that does not exist locks NOTHING, so on a fresh deployment two +-- concurrent FIRST appends would both compute seq=1 and the loser's receipt +-- would be permanently lost on the unique index. With the row pre-seeded the +-- lock always has something to grab. head_seq 0 + the genesis hash keep the +-- store's `seq = head_seq + 1` arithmetic identical. +-- The literal is sha256('genesis:receipts') — reproduce with: +-- node -e "console.log(require('node:crypto').createHash('sha256').update('genesis:receipts','utf-8').digest('hex'))" +-- (hard-coded rather than pgcrypto's digest() so the migration needs no extension). +INSERT INTO audit_stream_heads (stream_id, head_seq, head_hash) +VALUES ('receipts', 0, '\xb69452622fd89eb75373337022abd13f81da4da98bdad81955868609bbe42ac2') +ON CONFLICT (stream_id) DO NOTHING; + -- Signed checkpoints: Ed25519 over (stream_id, seq, head_hash, signed_at). -- The private key lives in env/secret manager, NEVER in this database — -- otherwise the admin we defend against could re-sign a rewritten chain. diff --git a/middleware/src/index.ts b/middleware/src/index.ts index 3fabad445..2cd2cd5e9 100644 --- a/middleware/src/index.ts +++ b/middleware/src/index.ts @@ -3510,28 +3510,38 @@ async function main(): Promise { // itself always builds (the store appends chained rows unconditionally); // signing is the layer that needs the operator-held key. Absent key ⇒ // loud boot log, not a silent no-op. - if (config.AUDIT_SIGNING_KEY) { - const signer = loadCheckpointSigner(config.AUDIT_SIGNING_KEY); - startCheckpointWorker(graphPool, signer, { + const checkpointSigner = config.AUDIT_SIGNING_KEY + ? loadCheckpointSigner(config.AUDIT_SIGNING_KEY) + : undefined; + if (checkpointSigner) { + startCheckpointWorker(graphPool, checkpointSigner, { intervalMs: config.AUDIT_CHECKPOINT_INTERVAL_MINUTES * 60_000, ...(config.AUDIT_ANCHOR_PATH ? { anchorPath: config.AUDIT_ANCHOR_PATH } : {}), }); - app.get('/api/v1/operator/provenance/public-key', requireAuth, (_req, res) => { - res.json({ - publicKeyPem: signer.publicKeyPem, - fingerprint: signer.publicKeyFingerprint, - checkpointIntervalMinutes: config.AUDIT_CHECKPOINT_INTERVAL_MINUTES, - anchorConfigured: Boolean(config.AUDIT_ANCHOR_PATH), - }); - }); console.log( - `[middleware] audit checkpoints wired (every ${config.AUDIT_CHECKPOINT_INTERVAL_MINUTES}min, fingerprint ${signer.publicKeyFingerprint.slice(0, 16)}…${config.AUDIT_ANCHOR_PATH ? ', external anchor on' : ''})`, + `[middleware] audit checkpoints wired (every ${config.AUDIT_CHECKPOINT_INTERVAL_MINUTES}min, fingerprint ${checkpointSigner.publicKeyFingerprint.slice(0, 16)}…${config.AUDIT_ANCHOR_PATH ? ', external anchor on' : ''})`, ); } else { console.warn( '[middleware] AUDIT_SIGNING_KEY not set — receipt chain builds WITHOUT signed checkpoints; generate a key with scripts/generate-audit-signing-key.mjs (#758)', ); } + // Always-on (review LOW): a keyless deployment answers `configured:false` + // instead of an undifferentiated 404 — #761 tooling can discover the + // posture either way. + app.get('/api/v1/operator/provenance/public-key', requireAuth, (_req, res) => { + res.json({ + configured: Boolean(checkpointSigner), + ...(checkpointSigner + ? { + publicKeyPem: checkpointSigner.publicKeyPem, + fingerprint: checkpointSigner.publicKeyFingerprint, + } + : {}), + checkpointIntervalMinutes: config.AUDIT_CHECKPOINT_INTERVAL_MINUTES, + anchorConfigured: Boolean(config.AUDIT_ANCHOR_PATH), + }); + }); const userStore = new UserStore(graphPool); diff --git a/middleware/src/receipts/checkpoints.ts b/middleware/src/receipts/checkpoints.ts index 9c09c1de1..a878f9c6d 100644 --- a/middleware/src/receipts/checkpoints.ts +++ b/middleware/src/receipts/checkpoints.ts @@ -120,13 +120,18 @@ export async function runCheckpointPass( }), signer.privateKey, ); - await pool.query( + const insertRes = await pool.query( `INSERT INTO audit_checkpoints (stream_id, seq, head_hash, signed_at, signature, public_key_fingerprint) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (stream_id, seq) DO NOTHING`, [RECEIPT_STREAM_ID, seq, row.head_hash, signedAt, signature, signer.publicKeyFingerprint], ); + // Replica race (review M1): the loser of the (stream, seq) conflict must + // NOT anchor or report its own differently-timestamped signature — the + // anchor file has to correspond to what the DB actually stored, or the + // promised anchor↔DB comparison flags phantom "tampering". + if ((insertRes.rowCount ?? 0) === 0) return undefined; const record: CheckpointRecord = { streamId: RECEIPT_STREAM_ID, seq, diff --git a/middleware/src/receipts/store.ts b/middleware/src/receipts/store.ts index 15a61a541..338469379 100644 --- a/middleware/src/receipts/store.ts +++ b/middleware/src/receipts/store.ts @@ -49,15 +49,23 @@ export function resetTurnReceiptCounters(): void { /** #758 — the canonical hash payload of a receipt row. NEVER includes * DB-generated values (created_at): time is anchored by checkpoint cadence, - * not per-row. Exported so the #761 verifier recomputes the identical shape. */ + * not per-row. Exported so the #761 verifier recomputes the identical shape. + * + * The JSON round-trip is load-bearing (review M3): the verifier recomputes + * from the stored JSONB, which honored `toJSON` at write time — hashing the + * live object would canonicalize e.g. a Date to `{}` while the row stores + * its ISO string, a guaranteed spurious mismatch. Round-tripping here makes + * hash input and stored row see the identical plain-JSON value. */ export function receiptChainPayload(entry: TurnReceiptRecordInput): unknown { - return { - turnId: entry.turnId, - sessionScope: entry.sessionScope ?? null, - channel: entry.channel ?? null, - model: entry.model ?? null, - receipt: entry.receipt, - }; + return JSON.parse( + JSON.stringify({ + turnId: entry.turnId, + sessionScope: entry.sessionScope ?? null, + channel: entry.channel ?? null, + model: entry.model ?? null, + receipt: entry.receipt, + }), + ); } export class PgTurnReceiptStore implements TurnReceiptStore { diff --git a/middleware/test/receiptHashChain.test.ts b/middleware/test/receiptHashChain.test.ts index a24843ecb..9da162c0a 100644 --- a/middleware/test/receiptHashChain.test.ts +++ b/middleware/test/receiptHashChain.test.ts @@ -168,6 +168,23 @@ describe('#758 PgTurnReceiptStore chained append', () => { assert.equal(turnReceiptCounters().persisted, 3); }); + it('a migration-seeded head (seq 0, genesis hash) yields the identical first append', async () => { + // Review H1 — 0041 seeds the head row so FOR UPDATE always has a row to + // lock. The seeded state (0, genesis) must produce byte-identical chain + // rows to the pre-seed fallback path. + resetTurnReceiptCounters(); + const { pool, rows } = chainFakePool(); + // Simulate the seed by priming the fake's head before any append. + await pool.query( + 'INSERT INTO audit_stream_heads (stream_id, head_seq, head_hash, updated_at) VALUES ($1,$2,$3,NOW())', + [RECEIPT_STREAM_ID, 0, genesisHash(RECEIPT_STREAM_ID)], + ); + await new PgTurnReceiptStore(pool).record({ turnId: 't-1', sessionScope: 's', receipt: RECEIPT }); + assert.equal(rows.length, 1); + assert.equal(rows[0]!.seq, 1); + assert.ok(rows[0]!.prev_hash.equals(genesisHash(RECEIPT_STREAM_ID))); + }); + it('a replayed turn advances neither rows, nor head, nor the counter', async () => { resetTurnReceiptCounters(); const { pool, rows, head } = chainFakePool(); @@ -253,4 +270,24 @@ describe('#758 checkpoint signing', () => { assert.equal(await runCheckpointPass(pool, signer, {}), undefined); assert.equal(inserted.length, 0); }); + + it('the LOSING replica of a checkpoint race neither anchors nor reports (review M1)', async () => { + const signer = loadCheckpointSigner(makeKey()); + // Fake where the checkpoint INSERT loses the (stream, seq) conflict. + const pool = { + query: async (sql: string) => { + if (sql.includes('FROM audit_stream_heads')) { + return { rows: [{ head_seq: '7', head_hash: genesisHash('x') }], rowCount: 1 }; + } + if (sql.includes('MAX(seq)')) return { rows: [{ seq: null }], rowCount: 1 }; + if (sql.includes('INSERT INTO audit_checkpoints')) return { rows: [], rowCount: 0 }; + throw new Error('unscripted'); + }, + } as unknown as Pool; + const anchorPath = join(mkdtempSync(join(tmpdir(), 'anchor-loser-')), 'anchors.jsonl'); + const record = await runCheckpointPass(pool, signer, { anchorPath }); + assert.equal(record, undefined, 'the loser must not report a checkpoint'); + // The anchor file must not exist — nothing was appended. + assert.throws(() => readFileSync(anchorPath, 'utf-8')); + }); }); From 952c4f25a39493241b7134638fa68782bc54611a Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 20 Aug 2026 14:20:53 +0200 Subject: [PATCH 5/5] fix: attach cause to signer parse error (preserve-caught-error lint) --- middleware/src/receipts/checkpoints.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/middleware/src/receipts/checkpoints.ts b/middleware/src/receipts/checkpoints.ts index a878f9c6d..ce76aa580 100644 --- a/middleware/src/receipts/checkpoints.ts +++ b/middleware/src/receipts/checkpoints.ts @@ -44,6 +44,7 @@ export function loadCheckpointSigner(privateKeyBase64: string): CheckpointSigner } catch (err) { throw new Error( `AUDIT_SIGNING_KEY is not a base64 PKCS#8 Ed25519 private key (generate one with scripts/generate-audit-signing-key.mjs): ${err instanceof Error ? err.message : String(err)}`, + { cause: err }, ); } if (privateKey.asymmetricKeyType !== 'ed25519') {