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<T> — 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<T>; treat the shipped projections as reference implementations whose API surface hasn't fully settled yet. Today's roster:

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<T> 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 below.

See Event log for the underlying row stream and the durable-flush + hydration pipeline this builds on.

The GraphProjection<T> 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.

interface GraphProjection<T> {
  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

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.

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.

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

runtime.events.iterate / .fold read whatever you wire as the runtime's eventReader. The contract is one method:

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

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<T> directly with your own reducer, you're insulated.

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

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 when the consumer needs the final materialized message text rather than the per-message envelope.

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

Folds async-job lifecycle events under the domain.<plugin>.job.* namespace into a per-job state machine — pending, completed, failed, timeout. The reducer is plugin-agnostic; any plugin emitting events under the domain.<plugin>.job.* convention plugs in for free.

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

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.

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

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. 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 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

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):

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

Stay inside the contract and projections survive substrate upgrades. A small example — count tool failures by toolName:

import type { GraphProjection, EventLogRow } from "@pleach/core/eventLog";

interface ToolFailureCounts {
  readonly byTool: Readonly<Record<string, number>>;
}

const toolFailureCounts: GraphProjection<ToolFailureCounts> = {
  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.<plugin>.* event namespace

Host plugins contribute event types under their own domain.<plugin>.* 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.

When writing a projection for a plugin you own, use the prefix as the reducer boundary:

const myPluginProjection: GraphProjection<MyState> = {
  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

• 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. 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.
• 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.

Where to go next