# Event-log projections (/docs/event-log-projections) A projection is a deterministic fold over the event log that produces a typed view. Instead of re-implementing "last message per channel" or "active tool calls" against the raw row stream in every consumer, the substrate ships the fold contract — `GraphProjection` — plus `runtime.events.iterate` and `runtime.events.fold` to drive it. The contract is stable. The built-in projections are **in soak** — their names and exact signatures may drift between releases. Depend on `GraphProjection`; treat the shipped projections as reference implementations whose API surface hasn't fully settled yet. Today's roster: | Projection | What it folds into | Status | | --------------------------------------- | ----------------------------------------------------------- | -------- | | `configProjection` | Resolved per-session config snapshot | shipping | | `messageProjection` | Ordered message log | shipping | | `toolCallProjection` | Per-turn tool-call timeline | shipping | | `jobProjection` / `createJobProjection` | Async job state per session | shipping | | `artifactProjection` | Lineage-aware artifact set + `asset.consumed` reducer | shipping | | `interruptProjection` | Active human-interrupt chain | shipping | | `subagentProjection` | Active subagent registry | shipping | | `exportProjection` | Sandbox-export lineage | shipping | | `userCardProjection` | User-card state per session | shipping | | `reconstructSessionState` | Composite function over the above (not a `GraphProjection`) | shipping | ```typescript import type { GraphProjection } from "@pleach/core"; import { configProjection, messageProjection, toolCallProjection, jobProjection, artifactProjection, interruptProjection, subagentProjection, exportProjection, userCardProjection, reconstructSessionState, } from "@pleach/core/eventLog"; ``` `runtime.events.iterate({ chatId, fromSequenceNumber?, tenantId? })` returns a paginated async-iterable over the persisted event-log rows ordered by `sequence_number`; `runtime.events.fold(projection)` composes a `GraphProjection` over that stream and returns the accumulator. The read side is **store-agnostic**: it delegates to a `HarnessEventLogReader` you inject at runtime construction (`eventReader` on the runtime config — the read-side counterpart to `eventLogWriter`). The reader is the only thing that knows your store, so you can back the same `iterate`/`fold`/resume surface with Postgres, Supabase, SQLite/pglite, or an HTTP API — and the reader is free to use whatever query mechanism your store offers (a raw `SELECT`, a PostgREST builder, a fetch). When no reader is injected, `iterate` yields nothing (a bare runtime has no event store), exactly as on the browser / `MemoryAdapter` path. See [Backing the reader](#backing-the-reader) below. See [Event log](/docs/event-log) for the underlying row stream and the durable-flush + hydration pipeline this builds on. ## The `GraphProjection` contract [#the-graphprojectiont-contract] A projection declares a stable `name`, an `initial` state factory, a `reduce` step, and an optional `finalize` step that runs once after all rows are folded. ```typescript interface GraphProjection { readonly name: string; readonly initial: () => T; readonly reduce: (acc: T, row: EventLogRow) => T; readonly finalize?: (acc: T) => T; } ``` Rules: 1. **Pure.** `reduce` is a pure function of `(acc, row)`. Same event slice in, same `T` out — that's what makes projections safe to re-derive on every render instead of caching. 2. **Order-respecting.** Rows arrive in `sequence_number` order (`(chat_id, sequence_number)` ascending). A projection that needs order-independence has to model it explicitly; the substrate doesn't shuffle for you. 3. **Total.** `reduce` must treat unknown `event_type` values as no-ops (return the accumulator unchanged), not throws — the event space accumulates new types over time. Filter to the event types you care about with an early `return acc` inside `reduce`. 4. **`finalize` is post-fold.** Use the optional `finalize(acc)` step for cross-row resolution — matching lifecycle pairs by ID, sorting, terminal-state derivation — after every row is consumed. ## Reading via `runtime.events` [#reading-via-runtimeevents] Two accessors. The first hands you the raw row stream; the second runs a projection to completion against the current event log. `runtime.events.iterate(opts)` is an async iterator over event log rows for a session, optionally scoped by `chatId`, `fromSequenceNumber`, or `tenantId`. ```typescript for await (const row of runtime.events.iterate({ chatId, fromSequenceNumber: lastSeenSequence, })) { console.log(row.event_type, row.payload); } ``` `runtime.events.fold(projection, opts)` runs a projection against the (filtered) row stream and returns the final state. ```typescript import { toolCallProjection } from "@pleach/core/eventLog"; const toolState = await runtime.events.fold(toolCallProjection, { fromSequenceNumber: turnStartSequence, }); // toolState carries the pending and completed tool calls for this turn. ``` Both accessors take the same options — `chatId`, `fromSequenceNumber`, and `tenantId` — so you can scope a projection to a session, resume from a checkpoint cursor, or filter to a single tenant without leaving the contract. ## Backing the reader [#backing-the-reader] `runtime.events.iterate` / `.fold` read whatever you wire as the runtime's `eventReader`. The contract is one method: ```typescript import type { HarnessEventLogReader } from "@pleach/core/eventLog"; const eventReader: HarnessEventLogReader = { iterate({ chatId, fromSequenceNumber, tenantId } = {}) { return { async *[Symbol.asyncIterator]() { // ANY query mechanism your store offers — here, a raw SELECT. const rows = await db.query( `SELECT * FROM harness_event_log WHERE session_id = $1 ORDER BY sequence_number ASC`, [chatId], ); for (const row of rows) yield row; // EventLogRow shape }, }; }, }; createPleachRuntime({ storage, eventReader }); ``` Each yielded row is an `EventLogRow` (the same shape `reconstructSessionState` and the shipped projections fold). Because the runtime delegates to your reader, there is no PostgREST grammar to emulate and no provider lock-in — the reader picks the cheapest query for its store. The host casts its client into this shape **at construction**, keeping `@pleach/core` itself free of any backing-store knowledge. The reader is **required** to read the durable log: call `runtime.events.iterate` / `.fold` without an `eventReader` and it throws a typed error at the call site. `@pleach/core` ships no built-in reader — there is no implicit store fallback — so a bare runtime never silently reaches for a database you didn't wire. ## Shipped projections — in soak [#shipped-projections--in-soak] Nine built-in projections ship in `@pleach/core/eventLog` (`configProjection`, `messageProjection`, `toolCallProjection`, `jobProjection`, `artifactProjection`, `interruptProjection`, `subagentProjection`, `exportProjection`, `userCardProjection`) — alongside the `reconstructSessionState` composite function. Their shapes are reference implementations; the names and exact return types may move between releases. If you build against one, plan to re-pin on upgrade. If you build against `GraphProjection` directly with your own reducer, you're insulated. ### `configProjection` [#configprojection] Folds the `session.created` lifecycle-root event into an identity snapshot per session — `sessionCreatedAt` and `sessionId`. Every other event type is a no-op; duplicate `session.created` rows collapse last-write-wins. ### `messageProjection` [#messageprojection] Folds streaming chunk events (`content.delta`) plus the terminal message events (`message.user`, `message.assistant`) into the reconstructed message stream, keyed by message ID. Pair with the `ContentLedger` accumulator from [Event log](/docs/event-log) when the consumer needs the final materialized message text rather than the per-message envelope. ### `toolCallProjection` [#toolcallprojection] Folds `tool.started`, `tool.completed`, and `tool.failed` events into the pending/completed tool-call set. Useful for "what tools are still in flight" panels and for replay parity checks. ### `jobProjection` [#jobprojection] Folds async-job lifecycle events under the `domain..job.*` namespace into a per-job state machine — `pending`, `completed`, `failed`, `timeout`. The reducer is plugin-agnostic; any plugin emitting events under the `domain..job.*` convention plugs in for free. ### `artifactProjection` [#artifactprojection] Folds artifact-emission events into the artifact set for a session. Returns a map keyed by `artifactId` with the latest envelope per artifact. ### Composite session-state reconstructor [#composite-session-state-reconstructor] `reconstructSessionState` is a **function** (not a `GraphProjection`) that takes an ordered array of event-log rows and composes the shipped projections into a single `SessionState` value — returning `null` when the stream is empty, carries no `session.created` row, or references multiple distinct sessions. Replay harnesses and `@pleach/eval` (npm name reserved; not yet shipping) use it to derive end-of-turn state from event log rows alone, with no snapshot-table dependency. ```typescript import { reconstructSessionState } from "@pleach/core/eventLog"; const rows = []; for await (const row of runtime.events.iterate({ chatId })) rows.push(row); const sessionState = reconstructSessionState(rows); // SessionState | null ``` ## `content.delta` streaming chunks [#contentdelta-streaming-chunks] Streaming model output emits one `content.delta` event per chunk — typically per token, sometimes per coarser chunk depending on the provider's stream shape. Each delta carries `messageId`, `channelId`, and the delta text. `messageProjection` folds these into the in-progress message until a terminal `message.assistant` event seals the message. The shape mirrors what the user saw at end-of-turn — same deltas in, same final message out. For the wire-level event shape, see [Stream events](/docs/stream-events). Projections read what stream events become once they land in the event log; the stream-events page documents the live wire format. ## The dual-write → dual-read → snapshot-retire ladder [#the-dual-write--dual-read--snapshot-retire-ladder] The substrate is migrating shapes that used to live in dedicated snapshot tables to the event log fold. The migration happens in three observable steps: 1. **Dual-write.** The runtime writes both the legacy snapshot and the event log rows that a projection would fold. Consumers still read the snapshot. 2. **Dual-read.** The projection runs alongside the snapshot read. Audit gates compare the two for parity over a defined window — for example, read parity over 100 consecutive turns on a representative workload. 3. **Snapshot retired.** Once parity holds, the snapshot table for that shape is retired and the projection becomes the default read path. Consumers don't drive this ladder — they observe its position in the changelog. The "in soak" label on a shipped projection means it's in step 1 or 2; "default read path" in the changelog means it's graduated. ## Soak-gated projection parity [#soak-gated-projection-parity] The dual-read step is gated by operator-side soak audits — small ledger-backed scripts that accumulate per-batch parity samples and refuse to clear the gate until the last several batches all hold clean. They take the place of a wall-clock soak window: the signal that the projection has matched the snapshot N times in a row is what the gate actually wants. Two audits ship today, both two-mode (source-text regression on the required projection sites, plus a per-batch ledger): | Audit | Watches | | --------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------- | | `audit:event-log-canonical-clean` | The canonical projection set (interrupts, subagents, exports / user-cards, hydration entry) plus the wire-layer's tenant-stamping site | | `audit:asset-consumed-projection-clean` | The `asset.consumed` dual-write fire site and projection arm | Each invocation appends a `BatchRecord` keyed on `--label` with sample size, OK count, mismatch count, error count, and the first few mismatch samples (capped so the ledger doesn't grow without bound). The `--strict` mode exits non-zero unless the last three batches all carry `mismatchCount === 0 && errorCount === 0`. Run them from the cutover PR — typically once per representative production canvas batch — before propagating a projection from "in soak" to the default read path. The source-text mode of each audit also doubles as a regression-detection floor: if a future refactor accidentally deletes the projection's emit site or fold arm, the audit fails loud before any data parity check runs. ## Custom projections [#custom-projections] Stay inside the contract and projections survive substrate upgrades. A small example — count tool failures by `toolName`: ```typescript import type { GraphProjection, EventLogRow } from "@pleach/core/eventLog"; interface ToolFailureCounts { readonly byTool: Readonly>; } const toolFailureCounts: GraphProjection = { name: "toolFailureCounts", initial: () => ({ byTool: {} }), reduce(state, row) { if (row.event_type !== "tool.failed") return state; const toolName = (row.payload as { toolName?: string }).toolName ?? "unknown"; return { byTool: { ...state.byTool, [toolName]: (state.byTool[toolName] ?? 0) + 1, }, }; }, }; const counts = await runtime.events.fold(toolFailureCounts, { fromSequenceNumber: turnStartSequence, }); // counts.byTool → { "search_corpus": 2, "fetch_url": 1 } ``` The early `return state` keeps the fold tight — `tool.failed` rows only mutate the accumulator — so the projection scales with failures, not with total event volume. ## `domain..*` event namespace [#domainplugin-event-namespace] Host plugins contribute event types under their own `domain..*` prefix. A projection written against `domain.your-plugin.foo.*` survives substrate upgrades because the core type space and the plugin type space are disjoint by construction — see `resolveDomainEventType` in [Event log](/docs/event-log). When writing a projection for a plugin you own, use the prefix as the reducer boundary: ```typescript const myPluginProjection: GraphProjection = { name: "myPlugin", initial: () => initialState, reduce(state, row) { if (!row.event_type.startsWith("domain.my-plugin.")) return state; /* ... */ return state; }, }; ``` That way a substrate release that adds new core event types can't silently feed rows into your reducer. ## What projections are NOT [#what-projections-are-not] * **Not a query language.** For SQL aggregation over the persisted `harness_event_log` table — "tool failure rate per tenant per day" and similar — see [Query](/docs/query). Projections fold the full row stream in memory; SQL is the right tool for aggregations that should run in the database. * **Not real-time subscriptions.** A projection runs to completion against the current event log. For streaming new events as they arrive (UI live updates, server-sent events), see [Stream events](/docs/stream-events). * **Not a snapshot store.** Projections compute on demand from the event log. There's no projection cache the substrate maintains for you — if you need cheap repeated reads, memoize at the consumer or pin against a checkpoint via [Checkpointing](/docs/checkpointing). ## Where to go next [#where-to-go-next]