feat(engine): the wiring wave — stamps ride every flush, provider generations, waitForIndexed, adopt-backfill, match-all serves
Some checks failed
CI / Node 22 (push) Has been cancelled
CI / Node 24 (push) Has been cancelled
CI / Bun (latest) (push) Has been cancelled

- Watermark stamping fans out at flush: all three projections stamped
  with the committed generation before their flushes persist.
- waitForIndexed(path?, {generation, timeoutMs}) — the one honest read
  barrier for write-then-recall consumers; typed timeout error carries
  the pending count and names the gauge; getIndexStatus() gains
  per-projection gauges. awaitPendingEmbeds() unchanged underneath.
- adoptLogAuthority() self-backfills curable divergences (pre-log
  records, witness drift) by identity re-commit before flipping — a
  fresh brain flips clean; log-ahead divergences still refuse loudly.
- The verification oracle gains VERB legs (all four divergence classes;
  unwired = honest verbsChecked: 0, never a scope claim).
- find({where: {}}) match-all serves (was silent-empty, warm AND cold;
  same fix in count/streaming/subgraph seeding); removeMany({where:{}})
  refuses typed — a match-all bulk delete must be explicit.
- Aggregation native envelope stamped via noteSourceGeneration before
  serializeState; the native-blob restore gates through the same
  adoption verdict as caller-side state (the unconditional adopt dies).
- LC8 pinned: a wholesale directory move opens and serves identically
  across all three intelligences, with history traveling.

Gates: unit 2031/2031 (156 files) · integration 812 (91 files) ·
conformance 27/27.
This commit is contained in:
David Snelling 2026-08-10 10:55:11 -07:00
parent b35d87a7ab
commit b53e6e8987
10 changed files with 1234 additions and 30 deletions

View file

@ -25,7 +25,7 @@ import {
} from './storage/brainFormat.js'
import type { BrainFormat } from './storage/brainFormat.js'
import { StorageAdapter, Vector, DistanceFunction, EmbeddingFunction, GraphVerb, STANDARD_ENTITY_FIELDS } from './coreTypes.js'
import type { HNSWNounWithMetadata, HNSWVerbWithMetadata, EntityVisibility } from './coreTypes.js'
import type { HNSWNoun, HNSWNounWithMetadata, HNSWVerbWithMetadata, EntityVisibility } from './coreTypes.js'
import {
defaultEmbeddingFunction,
cosineDistance,
@ -161,6 +161,8 @@ import { AggregationIndex } from './aggregation/AggregationIndex.js'
import { AggregateMaterializer } from './aggregation/materializer.js'
import type { AggregateDefinition, AggregateQueryParams, AggregateResult } from './types/brainy.types.js'
import type { MigrationProgress } from './types/brainy.types.js'
import type { IndexedProjectionPath, WaitForIndexedOptions } from './types/brainy.types.js'
import { WaitForIndexedTimeoutError } from './types/brainy.types.js'
import { resolveJsHnswConfig, DEFAULT_RECALL } from './utils/recallPreset.js'
import * as fs from 'node:fs'
import * as os from 'node:os'
@ -1273,6 +1275,34 @@ export class Brainy<T = any> implements BrainyInterface<T> {
this.graphIndex = graphIndex
}
// Fact-log v2 mint seam: after-image records carry minted dense ints,
// and the ONE authority for those assignments is the metadata index's
// id mapper (append-only getOrAssign — a rebuilt mapper reproduces
// them exactly). The generation store cannot know the mapper, so the
// mint thunk is injected here, immediately after the index is ready;
// installing it is what flips the fact log's LIVE writes to the v2
// segment format. A configuration whose mapper is unavailable throws
// at mint time — an int of 0 is never written.
this.generationStore.setIntMinter((kind, id) => {
const mapper = this.metadataIndex?.getIdMapper?.()
if (!mapper || typeof mapper.getOrAssign !== 'function') {
throw new Error(
`fact log v2: cannot mint the ${kind} int for ${id} — the metadata index's ` +
`id mapper is unavailable on this configuration; refusing to write an ` +
`after-image without a reproducible int`
)
}
const minted = mapper.getOrAssign(id, undefined)
const asBigint = typeof minted === 'bigint' ? minted : BigInt(minted)
if (asBigint <= 0n) {
throw new Error(
`fact log v2: the id mapper minted ${asBigint} for ${kind} ${id}` +
`minted ints are positive; refusing to write`
)
}
return asBigint
})
// Eager cold-load (readiness contract). A provider that persists its
// derived state exposes init?(): trigger the load NOW — AFTER
// metadataIndex.init() above (the id-mapper is hydrated first, so a
@ -2040,6 +2070,116 @@ export class Brainy<T = any> implements BrainyInterface<T> {
return this._pendingEmbedIds.size
}
/**
* THE READ BARRIER: wait until a projection or every projection has
* caught up to the CURRENT committed head, so a write-then-recall caller
* has ONE honest await instead of a sleep-and-hope.
*
* Legs:
* - `'semantic'` waits for the deferred-embedding backlog to drain
* (delegates to {@link awaitPendingEmbeds}, which keeps working
* unchanged as this leg's engine). After it resolves, every previously
* acknowledged write is vector-searchable.
* - `'metadata'` / `'graph'` / `'aggregation'` resolve IMMEDIATELY by
* design today: these projections are updated inside the write path, so
* by the time a write's promise resolves they already reflect it. Their
* asynchrony arrives with the log-authority read path; the door's shape
* freezes now so callers written against it keep working unchanged when
* those legs become real waits.
* - no argument every projection at the head; today that reduces to the
* semantic drain (the only asynchronous projection in the current
* architecture).
*
* `opts.generation`: resolve as soon as the projection's watermark has
* reached that committed generation. The pending-embed set carries no
* generation stamps today, so the refinement is conservative an empty
* backlog resolves immediately (the watermark is at the head, hence any
* committed generation); a non-empty backlog waits for the full drain, a
* SUPERSET of the requested wait, never a partial one.
*
* `opts.timeoutMs`: on expiry the promise REJECTS with
* {@link WaitForIndexedTimeoutError} typed, carrying the leg and the
* still-pending embed count, and naming the gauge to check
* (`getIndexStatus().projections.semantic.pendingEmbeds`). Never a silent
* partial wait: a timeout means the projection has NOT caught up.
*
* @example Write, then semantically recall no polling, no sleeps
* ```typescript
* const id = await brain.add({
* data: 'quarterly revenue narrative',
* type: NounType.Document,
* deferEmbedding: true,
* metadata: { kind: 'report' }
* })
* await brain.waitForIndexed('semantic') // the barrier: vector landed + indexed
* const hits = await brain.find({ query: 'revenue report', searchMode: 'semantic' })
* // `id` is eligible to appear in `hits` — the recall is honest, not lucky.
* ```
*
* @param path - The projection to wait on; omit to wait on all of them.
* @param opts - Optional `generation` watermark target and `timeoutMs` bound.
* @throws {WaitForIndexedTimeoutError} When `timeoutMs` expires before the
* projection catches up.
*/
public async waitForIndexed(
path?: IndexedProjectionPath,
opts?: WaitForIndexedOptions
): Promise<void> {
await this.ensureInitialized()
// Synchronous projections: updated inside the write path today, so an
// acknowledged write is already reflected — resolve immediately BY
// DESIGN (honest, not a stub). When the log-authority read path makes
// these legs asynchronous, only this body changes; the door's shape is
// frozen now.
if (path === 'metadata' || path === 'graph' || path === 'aggregation') {
return
}
// 'semantic' — or no-arg, which today reduces to it: the deferred-embed
// backlog is the only asynchronous projection in the current
// architecture.
// Generation refinement (conservative — see JSDoc): an empty backlog
// means the semantic watermark is at the head, hence ≥ any committed G.
if (opts?.generation !== undefined && this._pendingEmbedIds.size === 0) {
return
}
const timeoutMs = opts?.timeoutMs
const drained = this.awaitPendingEmbeds()
if (timeoutMs === undefined) {
return drained
}
// Typed timeout: reject LOUDLY with the leg + the live backlog gauge.
// (`drained` never rejects — the worker catches its own failures — so
// abandoning it on timeout cannot leak an unhandled rejection; the
// backlog keeps draining in the background.)
let timer: ReturnType<typeof setTimeout> | undefined
try {
await Promise.race([
drained,
new Promise<never>((_, reject) => {
timer = setTimeout(
() =>
reject(
new WaitForIndexedTimeoutError(
path ?? 'all',
timeoutMs,
this._pendingEmbedIds.size
)
),
timeoutMs
)
;(timer as { unref?: () => void }).unref?.()
})
])
} finally {
if (timer !== undefined) clearTimeout(timer)
}
}
/**
* @description The write-side persistence trigger (policy `'auto'`): count
* the committed write, kick a single-flight BACKGROUND flush when the
@ -6129,6 +6269,24 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
// MATCH-ALL NORMALIZATION (served-or-refused law): an empty `where: {}`
// carries zero predicates, so it MUST route exactly like an absent `where`.
// Left in place it reads as "filter criteria present" below, builds an
// empty index filter, and `getIdsForFilter({})` answers `[]` by contract —
// a silent empty on a query that semantically matches everything (worst on
// a freshly reopened brain, where it masquerades as data loss; on the
// vector path it short-circuits `find({ query, where: {} })` to `[]`).
// Dropped here, ONCE, before branch selection: the query takes the
// unfiltered match-all branch below, which serves from truth-complete
// sources — a storage page bounded to the offset+limit window (never a
// full walk), or the column store's top-K sort when orderBy is present.
// Every delegating surface (Db pins via host.find, pagination.find,
// streaming.search, subgraph query seeding) inherits this routing.
if (params.where !== undefined && !whereConstrains(params.where)) {
const { where: _emptyWhere, ...rest } = params
params = rest as FindParams<T>
}
// Zero-config validation (static import for performance)
validateFindParams(params)
@ -7049,6 +7207,18 @@ export class Brainy<T = any> implements BrainyInterface<T> {
`An empty selector would silently delete nothing — refusing.`
)
}
// An empty `where: {}` carries zero predicates. find() serves it as
// MATCH-ALL (the served-or-refused law), which on this destructive path
// would silently become "delete up to `limit` arbitrary rows". A bulk
// delete of everything must be asked for explicitly (type selector, real
// predicates, or ids) — refuse the ambiguous shape loudly.
if (params.where && !params.ids && !params.type && !whereConstrains(params.where)) {
throw new Error(
`removeMany() received where: {} — an empty filter matches EVERYTHING, ` +
`and a match-all bulk delete must be explicit. Pass real predicates, ` +
`a { type }, or { ids }; to clear the store use clear().`
)
}
if (params.ids && params.ids.length === 0) {
throw new Error(
`removeMany() received ids: [] — an empty id list deletes nothing. ` +
@ -7814,7 +7984,89 @@ export class Brainy<T = any> implements BrainyInterface<T> {
async adoptLogAuthority(): Promise<OracleReport> {
await this.ensureInitialized()
this.assertWritable('adoptLogAuthority')
const report = await this.verifyLogAuthority()
let report = await this.verifyLogAuthority()
// BASELINE BACKFILL: curable divergences are rows whose CANONICAL truth
// simply never reached the log — pre-log records (e.g. the generation-0
// VFS root, or a brain older than its log) and witness drift from
// maintenance that rewrote canonical outside a generation. The cure is
// an identity re-commit: any generational touch of the row makes the
// commit fact capture the CURRENT canonical bytes (the fact reads
// canonical back after execute), so the log converges on witness truth.
// Log-AHEAD divergences (log-live-canonical-absent /
// log-tombstone-canonical-present) are NOT curable by backfill — the
// log claims things the witness denies — and refuse loudly below.
let passes = 0
while (report.verdict === 'red' && passes < 5) {
passes++
const curable = report.mismatches.filter(
(m) => m.reason === 'pre-log-record' || m.reason === 'state-differs'
)
const incurable = report.mismatches.filter(
(m) => m.reason !== 'pre-log-record' && m.reason !== 'state-differs'
)
if (incurable.length > 0) {
throw new Error(
`adoptLogAuthority(): the log claims state the canonical witness denies ` +
`(${incurable.length} divergence(s); first: ${incurable[0].reason} on ` +
`${incurable[0].id}) — backfill cannot cure a log-ahead divergence. ` +
`Investigate before flipping; the witness remains authoritative.`
)
}
if (curable.length === 0) break
prodLog.info(
`[Brainy] adoptLogAuthority: baseline backfill pass ${passes} — re-committing ` +
`${curable.length} row(s) whose canonical truth never reached the log`
)
for (const m of curable) {
const raw = await this.storage.readNounRaw(m.id)
if (raw.metadata === null && raw.vector === null) continue // vanished since the scan
// IDENTITY re-commit: preserve the stored vector-file wrapper AS-IS —
// the denormalized enumeration fields and the embedding floats ride
// through, because a backfill must never DEGRADE the row it cures
// (a skeleton rewrite would drop the row's floats and its enumerable
// fields, and a later log replay could only reproduce the metadata
// leg's hydration). The wrapper's floats sit nested under `vector`
// (canonical noun vector files hold the denormalized noun, not a
// bare array); adjacency legs stay in SaveNounOperation's
// placeholder shape (the vector index owns them).
const wrapper =
raw.vector !== null && typeof raw.vector === 'object' && !Array.isArray(raw.vector)
? (raw.vector as Record<string, unknown>)
: null
const vector = Array.isArray(raw.vector)
? (raw.vector as number[])
: Array.isArray(wrapper?.vector)
? (wrapper!.vector as number[])
: []
await this.persistSingleOp({ nouns: [m.id] }, async (tx) => {
tx.addOperation(
new SaveNounOperation(this.storage, {
...(wrapper ?? {}),
id: m.id,
vector,
connections: new Map(),
level: typeof wrapper?.level === 'number' ? (wrapper.level as number) : 0
} as HNSWNoun)
)
})
}
const next = await this.verifyLogAuthority()
if (
next.verdict === 'red' &&
next.mismatches.length >= report.mismatches.length &&
!report.mismatchListTruncated
) {
throw new Error(
`adoptLogAuthority(): baseline backfill made no progress ` +
`(${report.mismatches.length}${next.mismatches.length} mismatches; first: ` +
`${next.mismatches[0]?.reason} on ${next.mismatches[0]?.id}) — refusing to loop. ` +
`This is a divergence class the backfill cannot express; investigate.`
)
}
report = next
}
this._logAuthority = await flipToLogAuthority(
this.storage as unknown as LogAuthorityStorage,
report
@ -10795,6 +11047,19 @@ export class Brainy<T = any> implements BrainyInterface<T> {
await this.generationStore.flushPendingSingleOps()
// Flush all components in parallel for performance
// Watermark stamps ride every flush fan-out: stamp each projection with
// the committed generation BEFORE its flush persists (stamp-after-data
// holds inside each owner — the stamp is its LAST write; here we only
// hand the generation over). No committedGeneration capability = no
// stamp = the owner's verdict machinery treats the artifact as legacy.
{
const wmGen = this.storage?.committedGeneration?.() ?? null
if (wmGen !== null) {
this.metadataIndex.stampWatermark(wmGen)
;(this.index as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
;(this.graphIndex as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
}
}
await Promise.all([
// 1. Flush storage adapter counts (entity/verb counts by type)
(async () => {
@ -10974,6 +11239,32 @@ export class Brainy<T = any> implements BrainyInterface<T> {
return this.storage.requestFlushOverFilesystem(timeoutMs)
}
/**
* @description The per-projection catch-up gauges served on
* `getIndexStatus().projections` (both the initialized and the
* pre-init snapshot the numbers are safe to read at any lifecycle
* stage). Semantic reports the live deferred-embed backlog; metadata and
* graph are synchronous today (updated inside the write path);
* aggregation reports its rescan/catch-up backlogs (zero when the
* aggregation engine was never engaged).
*/
private projectionGauges(): {
semantic: { pendingEmbeds: number }
metadata: { synchronous: true }
graph: { synchronous: true }
aggregation: { pendingBackfills: number; pendingCatchUps: number }
} {
return {
semantic: { pendingEmbeds: this._pendingEmbedIds.size },
metadata: { synchronous: true },
graph: { synchronous: true },
aggregation: {
pendingBackfills: this._aggregationIndex?.getPendingBackfills().length ?? 0,
pendingCatchUps: this._aggregationIndex?.getPendingCatchUps().length ?? 0
}
}
}
/**
* Get index loading status (Diagnostic for lazy loading)
*
@ -10986,6 +11277,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* console.log(`HNSW Index: ${status.hnswIndex.size} entities`)
* console.log(`Metadata Index: ${status.metadataIndex.entries} entries`)
* console.log(`Graph Index: ${status.graphIndex.relationships} relationships`)
* console.log(`Pending embeds: ${status.projections.semantic.pendingEmbeds}`)
* console.log(`Lazy rebuild completed: ${status.lazyRebuildCompleted}`)
* ```
*/
@ -10994,6 +11286,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
lazyRebuildCompleted: boolean
/** Deferred embeds not yet landed (MT5) — the eventual-vector-index backlog. */
pendingEmbeds: number
/** Per-projection catch-up gauges the honest numbers behind
* {@link waitForIndexed}. `synchronous: true` marks projections updated
* inside the write path today: their barrier leg resolves immediately by
* design, and the flag becomes a real backlog gauge when the
* log-authority read path makes them asynchronous. */
projections: {
/** The deferred-embedding backlog (same number as the top-level
* `pendingEmbeds`, which stays for compat). */
semantic: { pendingEmbeds: number }
metadata: { synchronous: true }
graph: { synchronous: true }
aggregation: {
/** Aggregates flagged for a full rescan of existing entities
* (drained on the next aggregate query). */
pendingBackfills: number
/** Aggregates adopted behind the watermark, with exact missing
* windows still to reconcile. */
pendingCatchUps: number
}
}
disableAutoRebuild: boolean
/** `true` while a native provider runs the one-time 7.x 8.0 rebuild LOCK.
* A readiness probe should map this to HTTP 503 + Retry-After (transiently
@ -11040,6 +11352,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
initialized: false,
lazyRebuildCompleted: this.lazyRebuildCompleted,
pendingEmbeds: this._pendingEmbedIds.size,
projections: this.projectionGauges(),
disableAutoRebuild: this.config.disableAutoRebuild || false,
migrating: false,
rebuildFailed: this._indexRebuildFailed != null,
@ -11083,6 +11396,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
initialized: this.initialized,
lazyRebuildCompleted: this.lazyRebuildCompleted,
pendingEmbeds: this._pendingEmbedIds.size,
projections: this.projectionGauges(),
disableAutoRebuild: this.config.disableAutoRebuild || false,
// A non-fatal index-rebuild failure recorded at init(), or adopt-forward
// degraded ids, are degraded states (queries may be incomplete) — surface
@ -11525,21 +11839,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// Get total count for pagination UI (O(1) when possible)
count: async (params: Omit<FindParams<T>, 'limit' | 'offset'>) => {
// Match-all normalization (shared with find()): an empty `where: {}`
// carries no predicates. Counting it as a filter would route through
// getIdsForFilter({}) → [] → a silent count of 0 while rows exist.
const constrainingWhere = whereConstrains(params.where) ? params.where : undefined
// For simple type queries, use O(1) index counting
if (params.type && !params.subtype && !params.query && !params.where && !params.connected) {
if (params.type && !params.subtype && !params.query && !constrainingWhere && !params.connected) {
const types = Array.isArray(params.type) ? params.type : [params.type]
return types.reduce((sum, type) => sum + this.metadataIndex.getEntityCountByType(type), 0)
}
// For complex queries, use metadata index for efficient counting
if (params.where || params.subtype || params.service) {
if (constrainingWhere || params.subtype || params.service) {
let filter: any = {}
if (params.where) {
if (constrainingWhere) {
// Where keys pass through UNTOUCHED — the one addressing law
// parses them at the index boundary (bare = user metadata,
// system.* = engine scalars). The old where.type→noun alias is
// dead: a bare 'type' is the user's own field now.
Object.assign(filter, params.where)
Object.assign(filter, constrainingWhere)
}
if (params.service) filter['system.service'] = params.service
if (params.subtype !== undefined) {
@ -11600,13 +11919,18 @@ export class Brainy<T = any> implements BrainyInterface<T> {
return {
// Stream all entities with optional filtering
entities: async function* (this: Brainy<T>, filter?: Partial<FindParams<T>>) {
if (filter?.type || filter?.subtype || filter?.where || filter?.service) {
// Match-all normalization (shared with find()): an empty `where: {}`
// carries no predicates — routing it through getIdsForFilter({})
// would stream NOTHING while storage holds rows. Treat it as absent
// so it falls to the unfiltered storage-paginated walk below.
const constrainingWhere = whereConstrains(filter?.where) ? filter!.where : undefined
if (filter && (filter.type || filter.subtype || constrainingWhere || filter.service)) {
// Use MetadataIndexManager for efficient filtered streaming
let filterObj: any = {}
if (filter.where) {
if (constrainingWhere) {
// Where keys pass through — the addressing law parses them at
// the index boundary; the type→noun alias is dead.
Object.assign(filterObj, filter.where)
Object.assign(filterObj, constrainingWhere)
}
if (filter.service) filterObj['system.service'] = filter.service
if (filter.subtype !== undefined) {
@ -13743,15 +14067,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
service?: string
excludeVFS?: boolean
}): any | null {
if (!(params.where || params.type || params.subtype || params.service || params.excludeVFS)) {
// An empty `where: {}` carries no predicates — it is NOT structured
// criteria (see whereConstrains). Counting it would produce an empty
// filter object, and getIdsForFilter({}) / getIdSetForFilter({}) answer
// the empty set by contract — silently emptying a match-all query.
const constrainingWhere = whereConstrains(params.where) ? params.where : undefined
if (!(constrainingWhere || params.type || params.subtype || params.service || params.excludeVFS)) {
return null
}
let filter: any = {}
if (params.where) {
if (constrainingWhere) {
// Where keys pass through UNTOUCHED — the one addressing law parses
// them at the index boundary (bare = user metadata, system.* = engine
// scalars, typed refusal otherwise). The old type→noun alias is dead.
Object.assign(filter, params.where)
Object.assign(filter, constrainingWhere)
}
if (params.service) filter['system.service'] = params.service
if (params.excludeVFS === true) {
@ -16999,6 +17328,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
/**
* @description Whether a `where` clause actually constrains the result set
* i.e. it is a non-null object carrying at least one predicate key. An empty
* `where: {}` carries ZERO predicates and must behave exactly like an absent
* `where` everywhere it is consulted; treating it as "a filter is present"
* routes the query into the index-filter path, where `getIdsForFilter({})`
* answers `[]` by contract a silent empty on a match-all query (the
* forbidden answer class: served-or-refused, never silently nothing).
* @param where - The raw `where` value from a query/selector params object.
* @returns `true` when `where` holds at least one predicate.
*/
function whereConstrains(where: unknown): where is Record<string, unknown> {
return (
where !== null &&
typeof where === 'object' &&
!Array.isArray(where) &&
Object.keys(where).length > 0
)
}
/**
* @description Extract the entity/relationship id from a canonical storage
* path of the form `entities/(nouns|verbs)/<shard>/<id>/metadata.json`.