feat: brain.onChange — the in-process change feed for every committed mutation

Subscribe once and receive one post-commit event per affected record for
EVERY canonical write, regardless of origin: direct calls, batch methods,
transact(), imports, and Virtual Filesystem writes all funnel through the
same commit points the feed is emitted from. This is the authoritative
in-process signal for live UIs, cache invalidation, and realtime sync layers
that forward it over their own transports.

Architecture — the post-commit dual of the ifRev precommit hook: mutation
methods hand lightweight event descriptors to the commit seam
(persistSingleOp / transact's plan), which stamps the committed
{generation, timestamp}, enriches entity deletes with the record's LAST
committed state from the commit's own before-images (free — Model B reads
them anyway; removeMany's id-only deletes gain full payloads this way), and
emits only after the commit succeeds — a losing CAS or rejected batch never
announces anything. Dispatch is a microtask FIFO after the mutex releases:
commit-ordered, a slow listener never delays a write, a throwing listener is
logged and isolated, and with no subscribers the write path constructs no
events at all. Single-point emission was chosen over per-method hooks
because the aggregation-hook pattern demonstrably drifted (relation ops and
removeMany were silently missing from it).

Coverage: add/update/remove (+ cascade unrelate per deleted relationship),
relate (both edges when bidirectional)/unrelate/updateRelation, per-item
events for addMany/updateMany/relateMany/removeMany, per-item events sharing
one generation for transact(), and transitively imports + VFS. clear() and
restore() — wholesale raw-state operations outside the per-record commit
path — emit a single store-level event meaning "refetch everything".
brain.close() drops all listeners.

New module src/events/changeFeed.ts (BrainyChangeEvent + ChangeFeed,
exported from the package root); guide docs/guides/reacting-to-changes.md.
Integration suite pins the contract: per-op payload fidelity, delete
last-state payloads, batch per-item emission, one-generation transact
batches, VFS-origin events, CAS-loser silence, ordering, listener isolation,
unsubscribe, and store-level events.

docs/guides/reacting-to-changes.md

@@ -0,0 +1,114 @@
+---
+title: Reacting to Changes
+slug: guides/reacting-to-changes
+public: true
+category: guides
+template: guide
+order: 11
+description: Subscribe to every committed mutation with `brain.onChange` — the in-process change feed behind live UIs, cache invalidation, and realtime sync. Covers the event shape, delivery guarantees, and catch-up patterns.
+next:
+  - guides/optimistic-concurrency
+  - guides/snapshots-and-time-travel
+---
+
+# Reacting to Changes
+
+`brain.onChange(cb)` is Brainy's in-process change feed: subscribe once and
+receive one event per committed mutation — **every** mutation, regardless of
+how it happened. Direct calls, batch methods, `transact()`, imports, and
+Virtual Filesystem writes all funnel through the same commit point the feed is
+emitted from, so nothing slips past it.
+
+```ts
+const off = brain.onChange((e) => {
+  if (e.kind === 'entity') {
+    console.log(`${e.op} ${e.entity?.type} ${e.id} @ generation ${e.generation}`)
+  }
+})
+
+await brain.add({ data: 'Ada Lovelace', type: 'person' })
+// → "add person 0198... @ generation 42"
+
+off() // unsubscribe when done
+```
+
+## The event
+
+```ts
+interface BrainyChangeEvent {
+  kind: 'entity' | 'relation' | 'store'
+  op: 'add' | 'update' | 'remove' | 'relate' | 'unrelate' | 'updateRelation'
+    | 'clear' | 'restore'
+  id?: string
+  entity?:   { id: string; type: string; subtype?: string;
+               metadata: Record<string, unknown>; service?: string }
+  relation?: { id: string; from: string; to: string; type: string;
+               metadata?: Record<string, unknown> }
+  generation?: number
+  timestamp: number
+}
+```
+
+- **Entity events** (`add` / `update` / `remove`) carry the post-commit indexed
+  view — `type`, `subtype`, and the full custom `metadata`, so you can match
+  your own `where`-style filters against events without a read.
+- **Deletes are fully described.** A `remove` or `unrelate` event carries the
+  record's *last committed state* (sourced from the commit's own history
+  record), not just an id.
+- **Batches emit per item.** `addMany` / `updateMany` / `relateMany` /
+  `removeMany` emit one event per affected record; a `transact()` batch emits
+  one event per item, all sharing the batch's single `generation`.
+- **Cascades are visible.** Removing an entity also emits `unrelate` for each
+  relationship the delete cascaded to.
+- **Store-level events** (`kind: 'store'`) fire for the two wholesale
+  operations — `clear()` and `restore()` — and mean *"everything may have
+  changed; refetch what you care about."*
+
+## Delivery guarantees
+
+- **Post-commit only.** An aborted write — a losing
+  [`ifRev` compare-and-swap](optimistic-concurrency.md), a rejected
+  transaction — never emits. If you received the event, the write is durable.
+- **Commit-ordered.** Events arrive in the order writes committed;
+  `generation` is monotonic.
+- **Asynchronous, never blocking.** Delivery happens in a microtask after the
+  write completes. A slow listener cannot delay a write; a throwing listener
+  is logged and isolated from other listeners.
+- **Zero overhead when unused.** With no subscribers, the write path does no
+  event work at all.
+- **Fire-and-forget.** There is no replay or backpressure. For catch-up after
+  a disconnect, use the `generation` on each event together with
+  [`asOf()` / the transaction log](snapshots-and-time-travel.md): record the
+  last generation you processed, and on reconnect diff from there.
+
+## Patterns
+
+**Cache invalidation** — drop cached reads for whatever changed:
+
+```ts
+brain.onChange((e) => {
+  if (e.kind === 'store') return cache.clear()
+  if (e.id) cache.delete(e.id)
+})
+```
+
+**Live queries (notify-and-refetch)** — re-run a query when a relevant change
+lands, rather than diffing incrementally:
+
+```ts
+brain.onChange((e) => {
+  if (e.kind === 'entity' && e.entity?.type === 'order') {
+    refreshOpenOrdersView() // debounce as needed
+  }
+})
+```
+
+**Forwarding to other processes** — the feed is in-process by design. To push
+changes to browsers or other services, forward events through your own
+transport (WebSocket, SSE) from the process that owns the brain.
+
+## Lifecycle
+
+`onChange` returns an unsubscribe function — call it when tearing down a
+subscriber (for example, when evicting a pooled instance). `brain.close()`
+drops all listeners; no events are delivered for or after `close()`.

src/brainy.ts

@@ -167,6 +167,12 @@ import {
   type ImportResult
 } from './db/portableGraph.js'
 import { GenerationStore, type CommitBeforeImages } from './db/generationStore.js'
+import {
+  ChangeFeed,
+  type BrainyChangeEvent,
+  type ChangeListener,
+  type PendingChangeEvent
+} from './events/changeFeed.js'
 import { isDeterministicEmbedMode } from './embeddings/deterministicEmbedMode.js'
 import { GenerationConflictError } from './db/errors.js'
 import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError } from './errors/brainyError.js'
@@ -335,6 +341,13 @@ interface PlannedTransact {
    * commit precondition leaves them untouched.
    */
   createdNouns: Set<string>
+  /**
+   * Change-feed events, one per affected record in op order, populated by the
+   * planners ONLY when a listener is subscribed. Stamped with the batch's
+   * committed generation and emitted after `commitTransaction` returns — a
+   * rejected batch (CAS conflict, failed apply) emits nothing.
+   */
+  changeEvents: PendingChangeEvent[]
 }
 
 /**
@@ -491,6 +504,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    *  upgrade verifies + stamps; retained on failure. See {@link createMigrationBackupIfNeeded}. */
   private _migrationBackupPath: string | null = null
 
+  /**
+   * The in-process change feed behind {@link onChange}. Emitted from the
+   * commit seam ({@link persistSingleOp} / {@link transact}), so every
+   * canonical mutation — any origin — produces exactly one post-commit event
+   * per affected record. See src/events/changeFeed.ts for the delivery
+   * contract.
+   */
+  private readonly _changeFeed = new ChangeFeed()
+
   /** Set when the on-open VFS-blob adoption left one or more `_cow/` blobs it
    *  could not fully adopt (bytes or metadata missing). While true, the
    *  pre-upgrade backup is NOT auto-removed — the upgrade is not verifiably
@@ -1553,8 +1575,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
   private async persistSingleOp(
     touched: { nouns?: string[]; verbs?: string[] },
     run: TransactionFunction<void>,
-    precommit?: (before: CommitBeforeImages) => void
+    precommit?: (before: CommitBeforeImages) => void,
-  ): Promise<void> {
+    pendingEvents?: PendingChangeEvent[]
+  ): Promise<{ generation?: number; timestamp: number }> {
+    // Change-feed capture: when this write will emit, hold a reference to the
+    // commit's before-images so `remove` events can carry the record's last
+    // committed state (free — the commit reads them anyway for Model B).
+    let capturedBefore: CommitBeforeImages | undefined
+    const captureAndCheck =
+      pendingEvents && pendingEvents.length > 0
+        ? (before: CommitBeforeImages): void => {
+            capturedBefore = before
+            precommit?.(before)
+          }
+        : precommit
+
     if (!this._generationStampingActive) {
       // Init-time / infrastructure baseline write (e.g. the VFS root): apply
       // WITHOUT creating a generation. Generation 0 is the freshly-materialized
@@ -1562,7 +1597,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       // Bootstrap writes are single-threaded, so a supplied precondition is
       // honored against directly-read before-images (no commit mutex exists
       // on this path — nothing to race).
-      if (precommit) {
+      if (captureAndCheck) {
         const nouns = new Map<string, { kind: 'noun'; metadata: unknown; vector: unknown }>()
         for (const id of touched.nouns ?? []) {
           const prev = await this.storage.readNounRaw(id)
@@ -1573,18 +1608,88 @@ export class Brainy<T = any> implements BrainyInterface<T> {
           const prev = await this.storage.readVerbRaw(id)
           verbs.set(id, { kind: 'verb', metadata: prev.metadata, vector: prev.vector })
         }
-        precommit({ nouns, verbs } as CommitBeforeImages)
+        captureAndCheck({ nouns, verbs } as CommitBeforeImages)
       }
       await this.generationStore.runWithoutGeneration(() =>
         this.transactionManager.executeTransaction(run)
       )
-      return
+      const timestamp = Date.now()
+      // Bootstrap writes are not generation-stamped; emit without one.
+      this.emitCommitted(pendingEvents, capturedBefore, undefined, timestamp)
+      return { timestamp }
     }
-    await this.generationStore.commitSingleOp({
+    const receipt = await this.generationStore.commitSingleOp({
       touched,
-      precommit,
+      precommit: captureAndCheck,
       execute: () => this.transactionManager.executeTransaction(run)
     })
+    // POST-COMMIT ONLY: an aborted commit (CAS conflict, failed apply) throws
+    // above and never reaches this line — the feed cannot announce a write
+    // that did not become durable.
+    this.emitCommitted(pendingEvents, capturedBefore, receipt.generation, receipt.timestamp)
+    return receipt
+  }
+
+  /**
+   * @description Stamp pending change events with their commit receipt,
+   * enrich entity `remove` events with the record's last committed state
+   * (from the commit's before-images), and hand them to the change feed for
+   * post-mutex dispatch. No-op when the write produced no events.
+   * @param pending - Events the mutation method constructed pre-commit
+   *   (only when a listener is subscribed — see {@link ChangeFeed.hasListeners}).
+   * @param before - The commit's before-images (delete-payload source).
+   * @param generation - The committed generation (absent for bootstrap writes).
+   * @param timestamp - The commit timestamp.
+   */
+  private emitCommitted(
+    pending: PendingChangeEvent[] | undefined,
+    before: CommitBeforeImages | undefined,
+    generation: number | undefined,
+    timestamp: number
+  ): void {
+    if (!pending || pending.length === 0) return
+    const events: BrainyChangeEvent[] = pending.map((p) => {
+      let entity = p.entity
+      if (!entity && p.kind === 'entity' && p.op === 'remove' && p.id && before) {
+        const record = before.nouns.get(p.id)?.metadata as
+          | Record<string, unknown>
+          | null
+          | undefined
+        if (record) {
+          entity = this.entityViewFromRawRecord(p.id, record)
+        }
+      }
+      return {
+        ...p,
+        ...(entity && { entity }),
+        ...(generation !== undefined && { generation }),
+        timestamp
+      }
+    })
+    this._changeFeed.emit(events)
+  }
+
+  /**
+   * @description Build the change-event entity view from a stored flat
+   * metadata record (the shape before-images and pre-delete reads carry),
+   * using THE canonical reserved/custom split so the view can never drift
+   * from the live read paths.
+   * @param id - The entity's canonical UUID.
+   * @param record - The stored flat metadata record.
+   * @returns The `ChangeEventEntity` view (type, subtype, service, custom metadata).
+   */
+  private entityViewFromRawRecord(
+    id: string,
+    record: Record<string, unknown>
+  ): NonNullable<BrainyChangeEvent['entity']> {
+    const { reserved, custom } = splitNounMetadataRecord(record)
+    return {
+      id,
+      type: String(reserved.noun ?? 'unknown'),
+      ...(reserved.subtype !== undefined && { subtype: String(reserved.subtype) }),
+      metadata: custom,
+      ...(reserved.service !== undefined && { service: String(reserved.service) })
+    }
   }
 
   /**
@@ -1795,10 +1900,28 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       // or merge (upsert); a merge that then hits a concurrent delete retries
       // the insert. Each persistSingleOp call constructs fresh operations, so
       // re-running the callback is safe.
+      // Change feed: built only when someone is listening (zero-cost gate).
+      const addEvents: PendingChangeEvent[] | undefined = this._changeFeed.hasListeners
+        ? [
+            {
+              kind: 'entity',
+              op: 'add',
+              id,
+              entity: {
+                id,
+                type: String(params.type),
+                ...(params.subtype !== undefined && { subtype: String(params.subtype) }),
+                metadata: (params.metadata as Record<string, unknown>) ?? {},
+                ...(params.service !== undefined && { service: String(params.service) })
+              }
+            }
+          ]
+        : undefined
+
       const MAX_UPSERT_ATTEMPTS = 10
       for (let attempt = 0; ; attempt++) {
         try {
-          await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit)
+          await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit, addEvents)
           break
         } catch (err) {
           if (!(err instanceof InsertPreconditionExistsSignal)) {
@@ -2786,7 +2909,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
         tx.addOperation(
           new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing)
         )
-      }, casPrecommit)
+      }, casPrecommit, this._changeFeed.hasListeners
+        ? [
+            {
+              kind: 'entity',
+              op: 'update',
+              id: params.id,
+              entity: {
+                id: params.id,
+                type: String(entityForIndexing.type),
+                ...(entityForIndexing.subtype !== undefined && {
+                  subtype: String(entityForIndexing.subtype)
+                }),
+                metadata: (newMetadata as Record<string, unknown>) ?? {},
+                ...(entityForIndexing.service !== undefined && {
+                  service: String(entityForIndexing.service)
+                })
+              }
+            }
+          ]
+        : undefined)
 
       // Aggregation hook (outside transaction — derived data)
       if (this._aggregationIndex) {
@@ -2876,7 +3018,35 @@ export class Brainy<T = any> implements BrainyInterface<T> {
             new DeleteVerbMetadataOperation(this.storage, verb.id)
           )
         }
-      })
+      },
+      undefined,
+      this._changeFeed.hasListeners
+        ? [
+            // The entity delete (payload = last state, from the pre-delete read)
+            // plus one unrelate per cascade-deleted relationship.
+            {
+              kind: 'entity',
+              op: 'remove',
+              id,
+              ...(metadata && {
+                entity: this.entityViewFromRawRecord(id, metadata as Record<string, unknown>)
+              })
+            },
+            ...allVerbs.map(
+              (v): PendingChangeEvent => ({
+                kind: 'relation',
+                op: 'unrelate',
+                id: v.id,
+                relation: {
+                  id: v.id,
+                  from: v.sourceId,
+                  to: v.targetId,
+                  type: String(v.verb)
+                }
+              })
+            )
+          ]
+        : undefined)
 
       // Aggregation hook (outside transaction — derived data)
       if (this._aggregationIndex && metadata) {
@@ -3595,7 +3765,44 @@ export class Brainy<T = any> implements BrainyInterface<T> {
             )
           )
         }
-      })
+      },
+      undefined,
+      this._changeFeed.hasListeners
+        ? [
+            {
+              kind: 'relation',
+              op: 'relate',
+              id,
+              relation: {
+                id,
+                from: params.from,
+                to: params.to,
+                type: String(params.type),
+                ...(params.metadata && {
+                  metadata: params.metadata as Record<string, unknown>
+                })
+              }
+            },
+            ...(reverseId
+              ? [
+                  {
+                    kind: 'relation',
+                    op: 'relate',
+                    id: reverseId,
+                    relation: {
+                      id: reverseId,
+                      from: params.to,
+                      to: params.from,
+                      type: String(params.type),
+                      ...(params.metadata && {
+                        metadata: params.metadata as Record<string, unknown>
+                      })
+                    }
+                  } as PendingChangeEvent
+                ]
+              : [])
+          ]
+        : undefined)
 
       return id
   }
@@ -3643,7 +3850,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
         tx.addOperation(
           new DeleteVerbMetadataOperation(this.storage, id)
         )
-      })
+      },
+      undefined,
+      this._changeFeed.hasListeners && verb
+        ? [
+            {
+              kind: 'relation',
+              op: 'unrelate',
+              id,
+              relation: {
+                id,
+                from: verb.sourceId,
+                to: verb.targetId,
+                type: String(verb.verb),
+                ...(verb.metadata && {
+                  metadata: verb.metadata as Record<string, unknown>
+                })
+              }
+            }
+          ]
+        : undefined)
   }
 
   /**
@@ -3778,7 +4004,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
           )
         )
       }
-    })
+    },
+    undefined,
+    this._changeFeed.hasListeners
+      ? [
+          {
+            kind: 'relation',
+            op: 'updateRelation',
+            id: params.id,
+            relation: {
+              id: params.id,
+              from: verbForIndex.sourceId,
+              to: verbForIndex.targetId,
+              type: String(verbForIndex.verb ?? verbForIndex.type),
+              ...(verbForIndex.metadata && {
+                metadata: verbForIndex.metadata as Record<string, unknown>
+              })
+            }
+          }
+        ]
+      : undefined)
   }
 
   /**
@@ -6335,6 +6580,16 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       }
 
       try {
+        // Change feed: the BUILDER populates this (per id actually staged), so
+        // an id whose builder failed never emits — the array is read only
+        // after the chunk's commit succeeds. Verb events dedupe within the
+        // chunk (two related chunk-members see the same cascade verb twice).
+        const chunkEvents: PendingChangeEvent[] | undefined = this._changeFeed
+          .hasListeners
+          ? []
+          : undefined
+        const seenVerbEvents = new Set<string>()
+
         // Process chunk as ONE atomic Model-B generation (entities + cascade verbs).
         await this.persistSingleOp({ nouns: touchedNouns, verbs: touchedVerbs }, async (tx) => {
           for (const id of chunk) {
@@ -6373,6 +6628,35 @@ export class Brainy<T = any> implements BrainyInterface<T> {
                 )
               }
 
+              if (chunkEvents) {
+                chunkEvents.push({
+                  kind: 'entity',
+                  op: 'remove',
+                  id,
+                  ...(metadata && {
+                    entity: this.entityViewFromRawRecord(
+                      id,
+                      metadata as Record<string, unknown>
+                    )
+                  })
+                })
+                for (const verb of allVerbs) {
+                  if (seenVerbEvents.has(verb.id)) continue
+                  seenVerbEvents.add(verb.id)
+                  chunkEvents.push({
+                    kind: 'relation',
+                    op: 'unrelate',
+                    id: verb.id,
+                    relation: {
+                      id: verb.id,
+                      from: verb.sourceId,
+                      to: verb.targetId,
+                      type: String(verb.verb)
+                    }
+                  })
+                }
+              }
+
               chunkQueued.push(id)
             } catch (error) {
               chunkBuilderFailed.push({
@@ -6384,7 +6668,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
               }
             }
           }
-        })
+        }, undefined, chunkEvents)
 
         // Transaction committed — queued IDs were actually deleted
         result.successful.push(...chunkQueued)
@@ -6690,6 +6974,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       // VFS was never used, reset flag for clean state
       this._vfsInitialized = false
     }
+
+    // Change feed: clear() wipes the store wholesale (raw, not per-record) —
+    // one store-level event tells subscribers everything they held is gone.
+    this._changeFeed.emit([
+      { kind: 'store', op: 'clear', timestamp: Date.now() }
+    ])
   }
 
   // ─── Migration API ───────────────────────────────────────────────
@@ -7124,6 +7414,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       hook()
     }
 
+    // Change feed: the batch's events share its single committed generation.
+    // A rejected batch throws at commitTransaction and never reaches here.
+    this.emitCommitted(plan.changeEvents, undefined, generation, timestamp)
+
     const receipt: TransactReceipt = { generation, timestamp, ids: plan.ids }
     return this.createPinnedDb({ generation, timestamp, receipt })
   }
@@ -7667,6 +7961,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       this.index.rebuild(),
       this.graphIndex.rebuild()
     ])
+
+    // Change feed: a restore is a wholesale raw-state replacement, not a
+    // per-record commit — one store-level event tells subscribers to refetch.
+    this._changeFeed.emit([
+      { kind: 'store', op: 'restore', timestamp: Date.now() }
+    ])
   }
 
   /**
@@ -8251,7 +8551,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       touchedVerbs: [],
       postCommit: [],
       casUpdates: [],
-      createdNouns: new Set()
+      createdNouns: new Set(),
+      changeEvents: []
     }
 
     for (const op of ops) {
@@ -8445,6 +8746,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
         this._aggregationIndex.onEntityAdded(id, entityForIndexing)
       }
     })
+    if (this._changeFeed.hasListeners) {
+      plan.changeEvents.push({
+        kind: 'entity',
+        op: 'add',
+        id,
+        entity: {
+          id,
+          type: String(params.type),
+          ...(params.subtype !== undefined && { subtype: String(params.subtype) }),
+          metadata: (params.metadata as Record<string, unknown>) ?? {},
+          ...(params.service !== undefined && { service: String(params.service) })
+        }
+      })
+    }
 
     state.nouns.set(id, { metadata: storageMetadata, vector })
     state.removedNouns.delete(id)
@@ -8611,6 +8926,24 @@ export class Brainy<T = any> implements BrainyInterface<T> {
         this._aggregationIndex.onEntityUpdated(params.id, entityForIndexing, oldEntityForAgg)
       }
     })
+    if (this._changeFeed.hasListeners) {
+      plan.changeEvents.push({
+        kind: 'entity',
+        op: 'update',
+        id: params.id,
+        entity: {
+          id: params.id,
+          type: String(entityForIndexing.type),
+          ...(entityForIndexing.subtype !== undefined && {
+            subtype: String(entityForIndexing.subtype)
+          }),
+          metadata: (newMetadata as Record<string, unknown>) ?? {},
+          ...(entityForIndexing.service !== undefined && {
+            service: String(entityForIndexing.service)
+          })
+        }
+      })
+    }
 
     state.nouns.set(params.id, { metadata: updatedMetadata, vector })
     return params.id
@@ -8676,8 +9009,31 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       plan.touchedVerbs.push(verb.id)
       state.verbs.delete(verb.id)
       state.removedVerbs.add(verb.id)
+      if (this._changeFeed.hasListeners) {
+        plan.changeEvents.push({
+          kind: 'relation',
+          op: 'unrelate',
+          id: verb.id,
+          relation: {
+            id: verb.id,
+            from: verb.sourceId,
+            to: verb.targetId,
+            type: String(verb.verb)
+          }
+        })
+      }
     }
     plan.touchedNouns.push(id)
+    if (this._changeFeed.hasListeners) {
+      plan.changeEvents.push({
+        kind: 'entity',
+        op: 'remove',
+        id,
+        ...(metadata && {
+          entity: this.entityViewFromRawRecord(id, metadata as Record<string, unknown>)
+        })
+      })
+    }
 
     if (metadata) {
       // Canonical reserved/custom split — mirror of remove()'s aggregation hook.
@@ -8813,6 +9169,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     plan.touchedVerbs.push(id)
     state.verbs.set(id, verb)
     state.removedVerbs.delete(id)
+    if (this._changeFeed.hasListeners) {
+      plan.changeEvents.push({
+        kind: 'relation',
+        op: 'relate',
+        id,
+        relation: {
+          id,
+          from: params.from,
+          to: params.to,
+          type: String(params.type),
+          ...(params.metadata && { metadata: params.metadata as Record<string, unknown> })
+        }
+      })
+    }
 
     if (params.bidirectional) {
       const reverseId = uuidv4()
@@ -8841,6 +9211,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       plan.touchedVerbs.push(reverseId)
       state.verbs.set(reverseId, reverseVerb)
       state.removedVerbs.delete(reverseId)
+      if (this._changeFeed.hasListeners) {
+        plan.changeEvents.push({
+          kind: 'relation',
+          op: 'relate',
+          id: reverseId,
+          relation: {
+            id: reverseId,
+            from: params.to,
+            to: params.from,
+            type: String(params.type),
+            ...(params.metadata && { metadata: params.metadata as Record<string, unknown> })
+          }
+        })
+      }
     }
 
     return id
@@ -8869,6 +9253,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     }
     plan.operations.push(new DeleteVerbMetadataOperation(this.storage, id))
     plan.touchedVerbs.push(id)
+    if (this._changeFeed.hasListeners) {
+      plan.changeEvents.push({
+        kind: 'relation',
+        op: 'unrelate',
+        id,
+        ...(verb && {
+          relation: {
+            id,
+            from: verb.sourceId,
+            to: verb.targetId,
+            type: String(verb.verb)
+          }
+        })
+      })
+    }
 
     state.verbs.delete(id)
     state.removedVerbs.add(id)
@@ -9358,6 +9757,43 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     return this._hub
   }
 
+  /**
+   * @description Subscribe to this brain's committed mutations — the
+   * authoritative in-process change feed. Fires exactly once per affected
+   * record for EVERY canonical write, regardless of origin: direct API calls,
+   * batch methods, `transact()`, imports, the Virtual Filesystem, and
+   * native-accelerated deployments all funnel through the same commit point
+   * this feed is emitted from.
+   *
+   * Delivery contract (see {@link BrainyChangeEvent}):
+   * - **Post-commit only** — an aborted write (e.g. a losing `ifRev`
+   *   compare-and-swap) never emits.
+   * - **Commit-ordered, asynchronous** — events arrive in the order writes
+   *   became durable, dispatched in a microtask so a slow listener never
+   *   delays a write. A throwing listener is isolated (logged, others
+   *   unaffected).
+   * - **Fully described** — `remove`/`unrelate` events carry the record's
+   *   LAST committed state, and batch operations emit one event per item.
+   * - **Fire-and-forget** — no replay or backpressure. Each event carries its
+   *   committed `generation`, so catch-up after a gap can be built on
+   *   {@link transactionLog} / {@link asOf}.
+   * - **Zero overhead when unused** — with no subscribers the write path does
+   *   no event work at all.
+   *
+   * @param listener - Called once per committed mutation.
+   * @returns An unsubscribe function — call it when done (e.g. on teardown of
+   *   a pooled instance) to stop delivery.
+   * @example
+   * const off = brain.onChange((e) => {
+   *   if (e.kind === 'entity') console.log(e.op, e.entity?.type, e.id)
+   * })
+   * await brain.add({ data: 'hello', type: 'document' }) // → "add document <id>"
+   * off()
+   */
+  onChange(listener: ChangeListener): () => void {
+    return this._changeFeed.subscribe(listener)
+  }
+
   /**
    * Get Triple Intelligence System
    * Advanced pattern recognition and relationship analysis
@@ -14587,6 +15023,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    * This ensures deferred persistence mode data is saved
    */
   async close(): Promise<void> {
+    // Change-feed teardown: no events are delivered for or after close().
+    this._changeFeed.close()
+
     // Phase 0a: Persist buffered single-op generation history (async
     // group-commit) before anything else, so a clean close never drops history
     // the caller already observed. No-op when nothing is pending or read-only.

src/events/changeFeed.ts

@@ -0,0 +1,171 @@
+/**
+ * @module events/changeFeed
+ * @description The in-process change feed behind {@link Brainy.onChange} — the
+ * authoritative "something committed" signal for every canonical mutation,
+ * regardless of origin (direct API calls, batches, transactions, imports, the
+ * VFS, or an accelerated native deployment: all of them funnel through the
+ * same generation-store commit points this feed is emitted from).
+ *
+ * Design properties:
+ * - **Post-commit only.** Events are enqueued after the commit succeeds, so an
+ *   aborted commit (a losing `ifRev` CAS, a failed transaction) never emits.
+ * - **Commit-ordered.** Events are enqueued in commit order and dispatched
+ *   FIFO, so a subscriber observes mutations in the order they became durable.
+ * - **Never blocks the write path.** Dispatch happens in a microtask after the
+ *   committing call returns; a slow or throwing listener cannot delay or fail
+ *   a write. Listener errors are isolated per listener and per event.
+ * - **Zero cost when unused.** Callers consult {@link ChangeFeed.hasListeners}
+ *   before constructing event payloads; with no subscribers the write path
+ *   does no event work at all.
+ * - **Fire-and-forget.** No acknowledgement, backpressure, or replay. Events
+ *   carry the committed `generation`, so a consumer that needs catch-up
+ *   semantics can pair the live feed with `transactionLog()` / `asOf()`.
+ */
+
+/** The post-commit view of an entity carried by entity change events. */
+export interface ChangeEventEntity {
+  /** The entity's canonical UUID. */
+  id: string
+  /** The entity's NounType string (e.g. `'person'`, `'document'`). */
+  type: string
+  /** The entity's subtype, when set. */
+  subtype?: string
+  /** The entity's custom (indexed) metadata fields. */
+  metadata: Record<string, unknown>
+  /** The writing service, when set. */
+  service?: string
+}
+
+/** The post-commit view of a relationship carried by relation change events. */
+export interface ChangeEventRelation {
+  /** The relationship's canonical UUID. */
+  id: string
+  /** Source entity UUID. */
+  from: string
+  /** Target entity UUID. */
+  to: string
+  /** The relationship's VerbType string (e.g. `'contains'`). */
+  type: string
+  /** The relationship's custom metadata fields, when present. */
+  metadata?: Record<string, unknown>
+}
+
+/**
+ * One committed mutation, as delivered to {@link Brainy.onChange} listeners.
+ *
+ * Exactly one of `entity` / `relation` is populated for `kind: 'entity'` /
+ * `kind: 'relation'` events. `kind: 'store'` events (`clear` / `restore`)
+ * carry neither — they mean "the whole store changed; refetch what you care
+ * about".
+ *
+ * For `op: 'remove'` / `op: 'unrelate'` the payload is the record's LAST
+ * committed state (sourced from the commit's own before-image), so deletes are
+ * fully described rather than id-only.
+ */
+export interface BrainyChangeEvent {
+  /** What changed: one record, one relationship, or the whole store. */
+  kind: 'entity' | 'relation' | 'store'
+  /** The mutation, in Brainy's own API vocabulary. */
+  op:
+    | 'add'
+    | 'update'
+    | 'remove'
+    | 'relate'
+    | 'unrelate'
+    | 'updateRelation'
+    | 'clear'
+    | 'restore'
+  /** The mutated record's id (absent for store-level events). */
+  id?: string
+  /** Post-commit entity view (entity events only). */
+  entity?: ChangeEventEntity
+  /** Post-commit relation view (relation events only). */
+  relation?: ChangeEventRelation
+  /**
+   * The committed generation this mutation belongs to (Model B: every write
+   * is a generation; a transaction's items share one). Absent for store-level
+   * events and init-time bootstrap writes, which are not generation-stamped.
+   */
+  generation?: number
+  /** Commit timestamp (ms since epoch). */
+  timestamp: number
+}
+
+/** Listener signature for {@link Brainy.onChange}. */
+export type ChangeListener = (event: BrainyChangeEvent) => void
+
+/**
+ * A change event as constructed by a mutation method BEFORE its commit: the
+ * commit seam stamps `generation`/`timestamp` after the write becomes
+ * durable. An entity `remove` descriptor may omit `entity` — the seam fills
+ * it from the commit's own before-image (the record's last committed state).
+ */
+export type PendingChangeEvent = Omit<BrainyChangeEvent, 'generation' | 'timestamp'>
+
+/**
+ * @description Listener registry + commit-ordered async dispatcher for
+ * {@link BrainyChangeEvent}s. One instance per {@link Brainy}.
+ */
+export class ChangeFeed {
+  private listeners = new Set<ChangeListener>()
+  private queue: BrainyChangeEvent[] = []
+  private draining = false
+
+  /** Whether any listener is subscribed — the write path's zero-cost gate. */
+  get hasListeners(): boolean {
+    return this.listeners.size > 0
+  }
+
+  /**
+   * @description Subscribe to committed mutations.
+   * @param listener - Called once per committed mutation, in commit order.
+   * @returns An unsubscribe function.
+   */
+  subscribe(listener: ChangeListener): () => void {
+    this.listeners.add(listener)
+    return () => {
+      this.listeners.delete(listener)
+    }
+  }
+
+  /**
+   * @description Enqueue committed events and schedule dispatch. Call ONLY
+   * after the commit has succeeded — this feed must never announce a write
+   * that did not become durable. Safe to call with an empty array.
+   * @param events - The committed mutations, in commit order.
+   */
+  emit(events: BrainyChangeEvent[]): void {
+    if (events.length === 0 || this.listeners.size === 0) return
+    this.queue.push(...events)
+    if (!this.draining) {
+      this.draining = true
+      queueMicrotask(() => this.drain())
+    }
+  }
+
+  /** Deliver everything queued, FIFO, isolating listener errors. */
+  private drain(): void {
+    try {
+      while (this.queue.length > 0) {
+        const event = this.queue.shift()!
+        for (const listener of this.listeners) {
+          try {
+            listener(event)
+          } catch (err) {
+            // A subscriber's bug must never affect the write path or its
+            // sibling subscribers.
+            console.error('[Brainy] onChange listener threw:', err)
+          }
+        }
+      }
+    } finally {
+      this.draining = false
+    }
+  }
+
+  /** Drop all listeners (brain close/teardown). Queued events are discarded. */
+  close(): void {
+    this.listeners.clear()
+    this.queue.length = 0
+  }
+}

src/index.ts

@@ -15,6 +15,14 @@ import { Brainy } from './brainy.js'
 
 export { Brainy }
 
+// The in-process change feed (brain.onChange) — event + listener types.
+export type {
+  BrainyChangeEvent,
+  ChangeEventEntity,
+  ChangeEventRelation,
+  ChangeListener
+} from './events/changeFeed.js'
+
 // Export diagnostics result type
 export type { DiagnosticsResult } from './brainy.js'
 

tests/integration/onchange-feed.test.ts

@@ -0,0 +1,228 @@
+/**
+ * @module tests/integration/onchange-feed
+ * @description The `brain.onChange` in-process change feed — the authoritative
+ * post-commit signal for every canonical mutation, regardless of origin.
+ * Pins the delivery contract:
+ *  - every operation fires exactly once per affected record with the
+ *    post-commit payload (deletes carry the record's LAST state);
+ *  - batch operations emit one event per item; transact items share one
+ *    generation;
+ *  - VFS writes (a Tier-1 blind spot for router-synthesized events) emit;
+ *  - events arrive in commit order with monotonic generations;
+ *  - a losing `ifRev` CAS emits NOTHING (aborted commits are invisible);
+ *  - a throwing listener is isolated; unsubscribe stops delivery;
+ *  - clear()/restore() emit one store-level "refetch everything" event.
+ */
+import { describe, it, expect, beforeEach, afterEach } from 'vitest'
+import * as fs from 'node:fs'
+import * as os from 'node:os'
+import * as path from 'node:path'
+import { Brainy } from '../../src/brainy.js'
+import type { BrainyChangeEvent } from '../../src/events/changeFeed.js'
+import { NounType, VerbType } from '../../src/types/graphTypes.js'
+
+let seq = 0
+const freshId = (): string =>
+  `00000000-0000-4000-8000-${(++seq).toString(16).padStart(12, '0')}`
+
+/** Let the microtask-dispatched feed drain. */
+const drained = (): Promise<void> => new Promise((r) => setTimeout(r, 0))
+
+describe('brain.onChange — the in-process change feed', () => {
+  let dir: string
+  let brain: any
+  let events: BrainyChangeEvent[]
+  let off: () => void
+
+  beforeEach(async () => {
+    process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
+    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-onchange-'))
+    brain = new Brainy({
+      requireSubtype: false,
+      storage: { type: 'filesystem', path: dir },
+      dimensions: 384,
+      silent: true
+    })
+    await brain.init()
+    events = []
+    off = brain.onChange((e: BrainyChangeEvent) => events.push(e))
+  })
+
+  afterEach(async () => {
+    off()
+    await brain.close()
+    fs.rmSync(dir, { recursive: true, force: true })
+  })
+
+  it('add → update → remove: one fully-described event each; remove carries the last state', async () => {
+    const id = await brain.add({
+      id: freshId(),
+      data: 'a document',
+      type: NounType.Document,
+      metadata: { status: 'draft' }
+    })
+    await brain.update({ id, metadata: { status: 'published' } })
+    await brain.remove(id)
+    await drained()
+
+    expect(events.map((e) => e.op)).toEqual(['add', 'update', 'remove'])
+    expect(events.every((e) => e.kind === 'entity' && e.id === id)).toBe(true)
+
+    expect(events[0].entity).toMatchObject({ id, type: 'document', metadata: { status: 'draft' } })
+    expect(events[1].entity).toMatchObject({ id, metadata: { status: 'published' } })
+    // The delete payload is the record's LAST committed state.
+    expect(events[2].entity).toMatchObject({ id, type: 'document', metadata: { status: 'published' } })
+
+    // Post-commit ordering: generations are monotonic.
+    const gens = events.map((e) => e.generation!)
+    expect([...gens].sort((a, b) => a - b)).toEqual(gens)
+    expect(new Set(gens).size).toBe(3)
+  })
+
+  it('relate → updateRelation → unrelate: relation events with endpoints + type', async () => {
+    const a = await brain.add({ id: freshId(), data: 'a', type: NounType.Thing })
+    const b = await brain.add({ id: freshId(), data: 'b', type: NounType.Thing })
+    events.length = 0
+
+    const rel = await brain.relate({ from: a, to: b, type: VerbType.RelatedTo, metadata: { w: 1 } })
+    await brain.updateRelation({ id: rel, metadata: { w: 2 } })
+    await brain.unrelate(rel)
+    await drained()
+
+    expect(events.map((e) => e.op)).toEqual(['relate', 'updateRelation', 'unrelate'])
+    expect(events.every((e) => e.kind === 'relation' && e.id === rel)).toBe(true)
+    for (const e of events) {
+      expect(e.relation).toMatchObject({ id: rel, from: a, to: b })
+    }
+  })
+
+  it('remove() cascades: deleting an entity emits unrelate for its relationships too', async () => {
+    const a = await brain.add({ id: freshId(), data: 'src', type: NounType.Thing })
+    const b = await brain.add({ id: freshId(), data: 'tgt', type: NounType.Thing })
+    const rel = await brain.relate({ from: a, to: b, type: VerbType.Contains })
+    events.length = 0
+
+    await brain.remove(a)
+    await drained()
+
+    const removeEvent = events.find((e) => e.op === 'remove')
+    const unrelateEvent = events.find((e) => e.op === 'unrelate')
+    expect(removeEvent).toMatchObject({ kind: 'entity', id: a })
+    expect(unrelateEvent).toMatchObject({
+      kind: 'relation',
+      id: rel,
+      relation: { from: a, to: b }
+    })
+    // Entity + cascaded verb share the one remove() generation.
+    expect(removeEvent!.generation).toBe(unrelateEvent!.generation)
+  })
+
+  it('batches emit one event per item; removeMany deletes carry payloads', async () => {
+    const ids = [freshId(), freshId(), freshId()]
+    await brain.addMany({
+      items: ids.map((id, i) => ({
+        id,
+        data: `item ${i}`,
+        type: NounType.Thing,
+        metadata: { n: i }
+      }))
+    })
+    await drained()
+    expect(events.filter((e) => e.op === 'add').length).toBe(3)
+
+    events.length = 0
+    await brain.removeMany({ ids })
+    await drained()
+    const removes = events.filter((e) => e.op === 'remove')
+    expect(removes.length).toBe(3)
+    // Every delete is fully described (enriched, not id-only).
+    for (const e of removes) {
+      expect(e.entity).toBeDefined()
+      expect(e.entity!.type).toBe('thing')
+      expect(typeof e.entity!.metadata.n).toBe('number')
+    }
+  })
+
+  it('transact(): per-item events sharing ONE generation; a rejected batch emits nothing', async () => {
+    const x = freshId()
+    const y = freshId()
+    await brain.transact([
+      { op: 'add', id: x, data: 'x', type: NounType.Thing, metadata: { k: 1 } },
+      { op: 'add', id: y, data: 'y', type: NounType.Thing },
+      { op: 'relate', from: x, to: y, type: VerbType.RelatedTo }
+    ])
+    await drained()
+
+    expect(events.map((e) => e.op)).toEqual(['add', 'add', 'relate'])
+    expect(new Set(events.map((e) => e.generation)).size).toBe(1)
+
+    // Rejected batch (stale per-op CAS) → zero events.
+    events.length = 0
+    await expect(
+      brain.transact([{ op: 'update', id: x, metadata: { k: 2 }, ifRev: 999 }])
+    ).rejects.toMatchObject({ name: 'RevisionConflictError' })
+    await drained()
+    expect(events.length).toBe(0)
+  })
+
+  it('a losing ifRev CAS update emits NOTHING; the winner emits once', async () => {
+    const id = await brain.add({ id: freshId(), data: 'contended', type: NounType.Thing })
+    const rev = (await brain.get(id))._rev
+    events.length = 0
+
+    const results = await Promise.allSettled(
+      Array.from({ length: 4 }, (_, i) =>
+        brain.update({ id, metadata: { w: i }, merge: false, ifRev: rev })
+      )
+    )
+    await drained()
+
+    expect(results.filter((r) => r.status === 'fulfilled').length).toBe(1)
+    expect(events.filter((e) => e.op === 'update').length).toBe(1) // exactly the winner
+  })
+
+  it('VFS writes emit (the router-synthesis blind spot)', async () => {
+    await brain.vfs.writeFile('/notes/hello.txt', 'hello feed')
+    await drained()
+    // A VFS write is entities + containment relations under the hood — the
+    // feed sees it because VFS delegates to canonical brain methods.
+    expect(events.length).toBeGreaterThan(0)
+    expect(events.some((e) => e.kind === 'entity' && e.op === 'add')).toBe(true)
+  })
+
+  it('listener errors are isolated; unsubscribe stops delivery', async () => {
+    const good: BrainyChangeEvent[] = []
+    const offBad = brain.onChange(() => {
+      throw new Error('subscriber bug')
+    })
+    const offGood = brain.onChange((e: BrainyChangeEvent) => good.push(e))
+
+    const id = await brain.add({ id: freshId(), data: 'p', type: NounType.Thing })
+    await drained()
+    expect(good.length).toBeGreaterThan(0) // sibling unaffected by the throwing listener
+
+    offBad()
+    offGood()
+    good.length = 0
+    events.length = 0
+    off() // unsubscribe the outer listener too
+    await brain.update({ id, metadata: { after: true } })
+    await drained()
+    expect(events.length).toBe(0)
+    expect(good.length).toBe(0)
+
+    // Re-subscribe for afterEach symmetry.
+    off = brain.onChange((e: BrainyChangeEvent) => events.push(e))
+  })
+
+  it('clear() emits one store-level event meaning "refetch everything"', async () => {
+    await brain.add({ id: freshId(), data: 'doomed', type: NounType.Thing })
+    events.length = 0
+    await brain.clear()
+    await drained()
+    const store = events.filter((e) => e.kind === 'store')
+    expect(store).toHaveLength(1)
+    expect(store[0].op).toBe('clear')
+    expect(store[0].generation).toBeUndefined()
+  })
+})