The log and read models

Part of the Loom User Guide, section 17 (Loom 2.0 kernel hardening). Working with the durable log itself: read models (ADR 0058), live queries, the log lifecycle (K10: crypto-shredding, retention, compaction), and the wire-level ABI (T2).

Read models: durable cross-execution queries (ADR 0058)

Projections are per-execution and in-process; read models (@loom/plugin-read-model) are the durable, cross-execution query layer — “every FINDING_RECORDED for table orders, newest first”, over months of runs, answered from an index instead of a re-fold:

import { InMemoryReadModelStore, ReadModelService } from "@loom/plugin-read-model";
// durable backends: SqliteReadModelStore / PostgresReadModelStore

const readModels = new ReadModelService(new InMemoryReadModelStore());
readModels.register({
  id: "findings-by-target",
  version: "1", // ADR 0095 — bump when `extract` changes MEANING (see below)
  kinds: ["FINDING_RECORDED"],
  indexes: ["target", "severity"],
  extract: (e) => [{ fields: { target: e.payload.target, severity: e.payload.severity } }],
});

const app = defineConfig({
  // the live tail rides the same in-process feed projections use:
  observability: { onEvent: (e) => readModels.handleEvent(e) },
  /* … */
});
await readModels.catchUp([app.stores.eventStore /*, archiveStore */]); // watermark-incremental

const page = await readModels.query("findings-by-target", {
  where: { target: "orders" }, order: "desc", limit: 50,
});

A read model is a cache of the log with a watermark, never an authority: drop it and rebuild() re-derives identical state; queries are read-only and never recorded, so replay is untouched. Ingestion converges from any mix of live tail, catch-up scans, and multiple nodes (idempotent rows + monotonic per-execution watermarks — no global event order required).

Changing an extractor is silent unless you pin it. The rows the old extract wrote stay, the (modelId, executionId) watermark is already past the events that produced them, so catch-up never revisits them, and one query then answers from two extractors at once. rebuild() is the exit; version is how anyone finds out it is needed. Whatever you derive from these rows records readModels.pins(), and readModels.checkPins(recorded) compares it later — a query that clears nothing, reporting drift when an extractor changed and unpinned when one side declared no version, because two sides that declared nothing must not read as a verified match (ADR 0095). The same declaration exists on projection$ / Projection for the folds a composed surface binds to; see the composed-surface recipe.

Live queries: watching one execution as it runs (@loom/live-query)

Read models answer questions across executions. A live query is the other axis: the state of one execution, pushed again every time an event lands — a progress view, a dashboard panel, an SSE stream.

It needs no syscall and no new event type. loadState already folds an execution’s durable state on demand (bounded by snapshots), and EventStore.subscribe already says when something changed; a live query is the two composed, which keeps it a read-model concern (P10) rather than a kernel primitive.

import { liveQuery } from "@loom/live-query";

const lq = liveQuery({
  eventStore: app.stores.eventStore,
  reader: app.runtime,                        // anything with `loadState` satisfies StateReader
  onError: (err) => logger.warn({ err }),     // background re-folds — see below
});

await lq.snapshot(executionId);               // fold once, now (null if unknown)

const unsubscribe = await lq.subscribe(executionId, (state) => {
  render(state);                              // current state immediately, then on every event
});
// …later
unsubscribe();

Coalesced, latest-wins. Only one re-fold is ever in flight. Events arriving during a fold set a dirty flag and cause exactly one more fold afterwards, so a burst of N events costs O(1) extra folds rather than N — and a subscriber never sees a stale state arrive after a newer one. subscribe also pushes once immediately, so a caller is not blind until the next event.

Two different failure paths, deliberately. The first fold happens inside subscribe, which you await — so if the store is unreachable, subscribe rejects and the failure is yours to handle. Every later re-fold runs from the store’s subscribe callback with nothing awaiting it; those go to onError, and the subscription stays live — one failed re-fold costs that update, and the next event folds from the log again.

Wire onError in a long-lived process. Those background folds cannot reject to a caller, so without it a failing store read (or an onState callback that throws) is swallowed — and before it existed, it was an unhandled rejection, which terminates the process by default.

A lost wake-up costs a delay, not a hang. Cross-node delivery is not guaranteed: a dropped listener or a full Postgres NOTIFY queue loses a wake-up, and a lost one is not sent later. So each subscription also re-reads the log every pollMs (default 5000) until its state is terminal. A quiet pass emits nothing and costs at most two indexed reads. pollMs: 0 turns this off, and a lost wake-up is then only recovered by that execution’s next event. A terminal subscription stops polling, so a later re-open is only seen through a wake-up.

reader is typed as StateReader — just loadState — so a live query works over the runtime, a remote read façade, or a test double, without depending on the whole runtime.

Log lifecycle: crypto-shredding, retention & compaction (K10)

“Replay forever” collides with GDPR erasure and bounded storage. K10 gives the log a lifecycle without ever creating a silent third behavior between “replays identically” and “fails loudly”.

Pinning (K10.6, ADR 0055). Retention consults an optional PinPredicate (defineConfig({ durability: { retention: { pinned } } })) before reclaiming an execution — pinByEventKinds(store, ["FINDING_RECORDED"]) pins runs carrying your designated domain events. The companion sweepUnpinnedContent (in @loom/plugin-content) reclaims content-store blobs on a TTL while refs cited by pinned event kinds survive — across archival, re-derived from the log on every sweep (mark-and-sweep, no mutable index). Over a store with ContentStore SPI 1.3.0 (SQLite, in-memory) every erase is conditional on the blob’s touch time, so a blob re-cited mid-sweep is kept, whether the re-cite is a put or a fork/append that carries the ref (SPI 1.4.0 touch, which ContentResolvingEventStore calls). reclaimOrphans: { completeScan: true } also reclaims blobs nothing cites any more; the acknowledgement is required to erase (a bare true throws ReclaimOrphansAckError; a dryRun is exempt and is how you check) because you are asserting that you passed every archive, every snapshot store and every roots source, and in the right ORDER: each live store before its archive. A citation that enters a log without a put (a fork’s copyPrefix; an append that keeps a ref already in its payload) is protected by ContentStore SPI 1.4.0 touch: ContentResolvingEventStore touches those refs, so a fork made mid-sweep makes the conditional delete refuse. To erase, the content store must implement touch (SQLite, in-memory and CachingContentStore do) or reclaimOrphans throws ReclaimOrphansTouchError; a dryRun is exempt. The log must be written through a ContentResolvingEventStore over that store, and its snapshot store wrapped in a ContentTouchingSnapshotStore (a fork of a compacted parent keeps its pre-compaction refs only in the seed snapshot). loom does not build either wrapper for you, so an erasing pass given snapshotStores must assert it: reclaimOrphans: { completeScan: true, snapshotsTouchContent: true } (else ReclaimOrphansSnapshotAckError; the CLI flag is --snapshots-touch-content), and the store needs touchClock as well (ReclaimOrphansClockError). The guarantee is the TTL’s: the touch before a write protects while the write takes less than the margin between the store’s clock and the sweep’s cutoff, and the touch after it is only a best-effort extension (failures reach onTouchError), so an erasing reclaimOrphans pass refuses olderThanMs <= 0 or a margin under 60 s (ReclaimOrphansMarginError). Run dryRun first. roots pins addresses held in your own tables (ADR 0055 D5–D7). The joint rule the perpetual-agents series fixed: findings and their evidence are permanent; runs are archivable; raw telemetry is disposable. Give the same kinds to retention.pinnedEventTypes so compaction keeps those events in place too (ADR 0106): a pinned execution then compacts around its findings. (With no pinnedEventTypes, lifecycle.compact leaves an execution pinned pins uncompacted.) Add retention.pinByEventTypes: true and retention pins an execution holding one of those kinds too, so the kinds are declared once instead of again in a pinByEventKinds predicate. To keep a WHOLE log uncompacted — a legal hold by tag — use retention.compactionExempt, which compact always consults. The whole pattern — schedule, carry state, claim-check, pin policy — is one declaration via sentinelPreset() (@loom/plugin-sentinel), and loom new sentinel <name> scaffolds it.

Crypto-shredding (K10.2, @loom/plugin-crypto). Encrypt sensitive fields at rest under a per-subject key; satisfy erasure by destroying the key — no history rewrite:

import { EncryptingEventStore, EncryptingSnapshotStore, InMemoryKeyStore, shredSubject }
  from "@loom/plugin-crypto";

const keys = new InMemoryKeyStore(); // tests only — `loom doctor` FAILS this in an app
                                     // (keys are the erasure boundary; losing them erases)
const eventStore = new EncryptingEventStore(inner, keys, {
  sensitive: { STEP_COMPLETED: ["output.ssn"] },      // dot-paths; only these encrypt (A4)
});
const snapshotStore = new EncryptingSnapshotStore(innerSnaps, keys, {
  sensitive: ["variables._input.ssn"],                // snapshots fold plaintext — cover them too (A1)
});

// `snapshotStore` is the RAW one: the tombstone follows the folded tip, which a compacted log's
// last event can sit below (ADR 0106). Required when `inner` can truncatePrefix.
await shredSubject(keys, inner, executionId, { reason: "gdpr-erasure", snapshotStore: innerSnaps });
// key destroyed; PAYLOAD_SHREDDED tombstoned at shred time (A3); any decrypting read now
// throws PayloadShreddedError — loud, never a silent divergence.

Snapshots are covered (or the secret survives the fold — erasure theater); a fork shares the parent’s subject so its copied ciphertext stays readable and shreds with the parent (A2).

Retention & compaction (K10.1 / K10.4, LogLifecycle). A kernel reclaims resources:

import { LogLifecycle } from "@loom/workflow-runtime";

const life = new LogLifecycle(eventStore, snapshotStore, {
  archive: { eventStore: archiveEvents, snapshotStore: archiveSnaps }, // A6 — archive first
  compactionArchive, // ADR 0138 — where `compact` MOVES the events it removes (an ArchiveStore)
});

// Reclaim a terminal execution older than its window (archive-before-delete, so the
// learning plane + deploy-sim corpus aren't starved):
await life.applyRetention(executionId, { terminalTtlMs: 30 * 86_400_000 }, Date.now());

// Compact a very long execution: archive the prefix, checkpoint, then truncate it (sequence 0
// preserved); `fold(snapshot, suffix) === fold(events)` still holds, so it replays byte-identically:
await life.compact(executionId);

// Audit: the moved events, byte-exact, digest-verified; with the surviving log this is the
// original log. Replay, resume and `readRange` never read the archive.
const moved = await life.readArchived(executionId);
const original = await life.readWithArchive(executionId);

An app-owned retention policy (ADR 0145, ADR 0146). Nothing is reclaimed by default, and nothing runs on a timer: a policy states what you intend, and your app or an operator runs the passes. One validated policy drives all three, so the windows cannot drift apart, and a content.unpinnedAfterMs below terminalTtlMs throws RetentionPolicyError at construction.

import { retentionPolicy } from "@loom/workflow-runtime";
import { sweepUnpinnedContent } from "@loom/plugin-content";

const DAY = 86_400_000;
const policy = retentionPolicy({
  terminalTtlMs: 90 * DAY,
  snapshots: { keepLatest: 3 },              // never by age: a compacted run's checkpoint IS its state
  content: { unpinnedAfterMs: 120 * DAY },
});

// Preview first. `evaluate` applies every gate `applyRetention` does and changes nothing.
if ((await life.evaluate(executionId, policy, Date.now())) === "reclaimable") {
  await life.applyRetention(executionId, policy, Date.now());
}
await life.pruneSnapshots(executionId, { policy });

// Content is swept across the whole store, not per execution, and only reclaims refs the app marked
// `reclaimable`; a retained blob is never reclaimed by age.
const report = await sweepUnpinnedContent({ eventStores: [eventStore], contentStore, policy });

A terminal execution’s citations age from its end, not from when they were written, so a reclaimable blob cited on day 0 by a run that ended on day 60 is held until day 180 under this policy (heldByEndClock in the report). A running or re-opened execution keeps the citation clock. From the command line, loom retention run --policy <module> and loom sweep content --policy <module> run the same passes, dry by default and --apply to reclaim. The retention-policy example in loom-examples runs all of it on a simulated clock.

Compaction moves events, it never deletes them (ADR 0138). compact appends the exact envelopes it removes to an ArchiveStore (compactionArchive; in-memory, SQLite and Postgres backends ship, and sqliteStorage() / postgresStorage() / kind: "memory" provide one) BEFORE it truncates, and it refuses (CompactionArchiveRequiredError, nothing changed) when none is configured. A crash between the two leaves a duplicate in the archive, never a gap, and a retry is idempotent. Over an encrypting event store the archive must be an EncryptingArchiveStore on the SAME KeyStore (PlaintextArchiveError otherwise), so a shred erases the archived copy too. Pass the archive to sweepUnpinnedContent({ archiveStores }) so a blob only an archived event cites is not reclaimed, and verify an audit hash chain across the compaction with verifyChain(id, { archived }) (hashChainedEventStore(..., { allowCompaction: true })). An erasing sweepUnpinnedContent checks every compacted log’s removed range against the given archiveStores’ headers and refuses (ArchiveStoresRequiredError) when it is not covered - [] no longer suffices. For logs compacted before the archive existed (their history is already gone) pass acknowledgeUnarchivedLogs: true (“erasing their orphan blobs is accepted”; the executions are reported in unarchivedAcknowledged); a fork child of a compacted parent needs it too. loom sweep content needs --compaction-archive <archive.db> (default: archive.db beside --db; --config wires the app’s) and, for such logs, --acknowledge-unarchived. The tenant guard is opt-in: a multi-tenant deployment must set durability.retention.tenant (or use tenant-scoped stores) to get the ArchiveTenantMismatchError refusals; an undeclared, unscoped stack reads as single-tenant. A log compacted before the archive existed keeps its old checkpoint as a legacy-checkpoint segment (readLegacyCheckpoints); readArchived / readWithArchive report a range only that checkpoint remembers (ArchiveIncompleteError, archiveCoverage) instead of leaving a silent hole. A segment is bound to its log’s incarnation, so an execution id reused after a delete never inherits the old archive; over a tenant-scoped event store the archive must be a TenantScopedArchiveStore for the same tenant.

The durable inbox’s streams are archived the same way (ADR 0139). trimInbox, the per-topic trim, the quarantine trim and a retired chain’s delete MOVE the envelopes (ids, timestamps, sender metadata - the payload survives in MESSAGE_CONSUMED) into the SAME archive, keyed by the reserved stream id, and refuse without one: through defineConfig the compaction archive is handed to the runtime (inboxArchive), and defineConfig THROWS when the inbox is on and no archive is resolvable (kernel.inbox.archive: "none" is the explicit opt-out: nothing is trimmed or deleted, said at boot, and sends continue until inboxMaxRetained refuses them). A hand-built runtime with the inbox on and no inboxArchive is refused at construction (InboxArchiveRequiredError) unless its event store is in-memory or it says inboxArchive: "none"; a trim that still reaches the runtime without an archive refuses with ReservedStreamArchiveRequiredError, reported per chain through onStorageError. Read it back with life.readReservedWithArchive(streamId) / life.reservedArchiveCoverage(streamId); exportSubject returns it as reservedStreams; a re-created stream starts above the archive, and a straggler lands in a late run (__loom_late__:<stream>@<n>). The terminal delete is bounded by what was archived (EventStore.deleteThrough; a store without it keeps the stream), the inbox needs an archive at construction (inboxArchive, defaulted for an in-memory store, "none" to decline), and erasing a subject deletes its archived copies from a plaintext archive. Over a hash-chained store the terminal inbox delete needs hashChainedEventStore(inner, { allowCompaction: true }) (without it the delete refuses and the streams are kept). A sealing deployment seals each message’s sender (actor, tenant, onBehalfOf) under an opaque handle derived from KeyStore material (senderSubjectOf: false opts out and defineConfig warns), so eraseSubject also erases the subject AS A SENDER and an erased sender reads { erased: true }; it stays blocked (SenderErasedError) until app.reinstateSender(actor, tenant?, { by }) journals a reinstatement under a new handle. KeyStore.getOrCreate must be an atomic create-if-absent across nodes (an upsert lets two nodes mint two sender seeds, and erasure would destroy only one): runKeyStoreConformance has a concurrent case that a non-atomic store fails. app.recoverArchiveRedactions() restores a redaction that a crash left staged.

Both run over optional store SPI (deleteExecution / truncatePrefix, on in-memory, SQLite, and Postgres); a store that omits them can’t be swept. A compacted execution resumes, and compacts again, exactly as if it had never been compacted (ADR 0106), and the event kinds in retention.pinnedEventTypes survive compaction in place (truncatePrefixKeeping, SPI 1.5.0). Over an encrypting store, the destination must share its KeyStore: compact throws UnsafeCompactionError unless the snapshot store is an EncryptingSnapshotStore on the same keys, and applyRetention returns "unsafe-archive" for an archive that isn’t. Otherwise the decrypted fold or events would sit outside the reach of a key shred (ADR 0022 Amendment).

Redaction markers (K10.3). Where crypto-shredding stores an encrypted secret in the log, redaction keeps a secret out of the log entirely — a step carries a deterministic reference, resolved to the real value only at effect execution:

import { redactionMiddleware, secretRef, InMemorySecretResolver } from "@loom/middleware";
const resolver = new InMemorySecretResolver(/* vault/KMS in production */);
new EffectRuntime([redactionMiddleware({ resolver })]);

// flow code passes a REFERENCE — STEP_STARTED logs the ref, never the secret:
await ctx.run(chargeStep, { amount, apiKey: secretRef("stripe-key") });
// the middleware resolves it into the value the TOOL sees; the durable log only
// ever contains { __loomSecret: "stripe-key" }. redactOutput does the same for
// secrets a tool RETURNS (replaced by a deterministic ref before STEP_COMPLETED).

Encryption at rest (K10.5). The two seams above are application-layer (Loom encrypts fields / keeps secrets out of the log). Encrypting the store files themselves — against a stolen disk or leaked backup — is a storage-layer concern: volume encryption (FileVault/LUKS/encrypted-EBS for SQLite; managed-DB storage encryption or LUKS for Postgres), or SQLCipher/TDE for per-file/per-cluster keys. All three layers stack; see Encryption at rest (loom/docs/loom-2.0/11-encryption-at-rest.md) for the SQLite and Postgres configuration and when to use which.

Classify once, derive the wiring (K10-R1/R2/R6). The hand-authored sensitive: {…} maps above are the low level. In practice, mark the field once on its schema and every mechanism derives its wiring — encryption, redaction, and egress scrubbing:

import { sensitive, secret, internal, externalized } from "@loom/classify";

const output = z.object({
  ssn: sensitive(z.string()),        // encrypt-at-rest + crypto-shreddable + scrubbed from egress
  apiToken: secret(z.string()),      // never persisted (substituted via secretRef)
  note: internal(z.string()),        // logged, but never egressed externally
  doc: externalized(z.object({…})),  // stored in a content store, log keeps a ref (ADR 0046)
});

// One key wires secret redaction; telemetry egress is auto-scrubbed for marked fields:
const app = defineConfig({
  storage,
  flows,
  security: { secrets: { resolver } },
  observability: { telemetry: [datadog] },
});

See Data classification for the full taxonomy + derive helpers, and the content-store seam for externalized().

Data-subject rights (T3, @loom/compliance). Event-sourcing makes GDPR Art. 15/20 (access, portability) and Art. 17 (erasure) a fold. Wire a subjectOf resolver (and tenantOf for multi-tenant isolation) and get a live subject index + export + erasure:

const app = defineConfig({ storage, flows,
  security: { compliance: { subjectOf: (e) => e.payload.input?.userId, erase: { shred } } } });

await app.exportSubject("user-42");             // Art. 15 access + Art. 20 portability
await app.eraseSubject("user-42");              // Art. 17 — liveness-gated (never shreds a live fold)

Operator sweeps over cold storage: loom compliance report (subject inventory) and loom retention run. See Compliance.

Wire-level ABI (T2)

@loom/wire-abi lets a non-TypeScript client (any language, even a shell script) drive executions over JSON-RPC. The wire protocol is a versioned projection of the K1 syscalls — generated from KERNEL_SYSCALLS, so the hand-written ctx surface stays authoritative:

import { handleRpc, WIRE_ABI, FOLD_VECTORS, runFoldVectors } from "@loom/wire-abi";

// Mount the carrier on any transport (node:http, the A2A server, worker-server):
const response = await handleRpc(runtime, jsonRpcRequest);
// methods: execution.start / execution.resume / execution.getState / execution.cancel
// A wire client starts/resumes/reads; the KERNEL folds and runs every ring (K2/K3/K6/K7/K12)
// on its side of the wire — a wire client gains no bypass. The returned fold is
// byte-identical to an in-process run.

Cross-language conformance is executable: FOLD_VECTORS are plain-JSON event-log → expected- fold fixtures any implementation can run (runFoldVectors is the TS reference), extending the K1/K5 pins from “the fold understands these types” to “the fold produces exactly this state”.

---

Back to User Guide section 17.