feat(engine): the wiring wave — stamps ride every flush, provider generations, waitForIndexed, adopt-backfill, match-all serves
- 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:
parent
b35d87a7ab
commit
b53e6e8987
10 changed files with 1234 additions and 30 deletions
|
|
@ -570,15 +570,35 @@ export class AggregationIndex {
|
|||
}
|
||||
}
|
||||
|
||||
// Restore native provider state from persistence
|
||||
// Restore native provider state from persistence — GATED by the same
|
||||
// adoption verdict as caller-side state (the unconditional adopt was an
|
||||
// asymmetry: a stale native blob restored over a moved store silently
|
||||
// over/under-counted). 'adopt' restores; 'catchup' restores too (the
|
||||
// incremental reconciliation drives the provider through
|
||||
// incrementalUpdate over the exact missing window); 'rescan' SKIPS the
|
||||
// blob — the flagged rebuild repopulates the provider from source.
|
||||
// Legacy unstamped envelopes verdict as rescan, loudly, never silently.
|
||||
if (this.nativeProvider?.restoreState) {
|
||||
const nativeState = await this.storage.getMetadata('__aggregation_native_state__')
|
||||
if (nativeState && typeof nativeState === 'string') {
|
||||
this.nativeProvider.restoreState(nativeState)
|
||||
} else if (nativeState && typeof nativeState === 'object' && nativeState.data) {
|
||||
// flush() persists `{ data: serializeState() }`, so `data` is the
|
||||
// provider's serialized state string.
|
||||
this.nativeProvider.restoreState(nativeState.data as string)
|
||||
const blob =
|
||||
nativeState && typeof nativeState === 'string'
|
||||
? nativeState
|
||||
: nativeState && typeof nativeState === 'object' && nativeState.data
|
||||
? (nativeState.data as string)
|
||||
: null
|
||||
if (blob !== null) {
|
||||
const verdict = this.stateAdoptionVerdict(
|
||||
'__native__',
|
||||
nativeState && typeof nativeState === 'object' ? (nativeState as Record<string, unknown>) : {}
|
||||
)
|
||||
if (verdict === 'adopt' || verdict === 'catchup') {
|
||||
this.nativeProvider.restoreState(blob)
|
||||
} else {
|
||||
prodLog.warn(
|
||||
`[Aggregation] native provider state not adopted (verdict: ${verdict}) — ` +
|
||||
`the flagged rescan repopulates the provider from source`
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -614,12 +634,17 @@ export class AggregationIndex {
|
|||
}
|
||||
}
|
||||
|
||||
// Persist native provider state
|
||||
// Persist native provider state — stamped. noteSourceGeneration lets the
|
||||
// provider bake the committed watermark into its OWN envelope before
|
||||
// serializing (so a native-side reopen can verify honesty without our
|
||||
// wrapper); the wrapper carries the same stamp for OUR adoption verdict.
|
||||
if (this.nativeProvider?.serializeState) {
|
||||
const nativeGen = this.storage.committedGeneration?.() ?? null
|
||||
if (nativeGen !== null) this.nativeProvider.noteSourceGeneration?.(nativeGen)
|
||||
const nativeState = this.nativeProvider.serializeState()
|
||||
await this.storage.saveMetadata(
|
||||
'__aggregation_native_state__',
|
||||
{ data: nativeState }
|
||||
nativeGen === null ? { data: nativeState } : { data: nativeState, sourceGeneration: nativeGen }
|
||||
)
|
||||
}
|
||||
|
||||
|
|
|
|||
373
src/brainy.ts
373
src/brainy.ts
|
|
@ -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`.
|
||||
|
|
|
|||
|
|
@ -128,6 +128,16 @@ export async function runLogCompletenessOracle(args: {
|
|||
canonicalNounDigest: (id: string) => Promise<string | null>
|
||||
/** Digest a log after-image record's payload. */
|
||||
factRecordDigest: (record: unknown) => string
|
||||
/**
|
||||
* Verb legs (optional until every owner wires them): the canonical verb
|
||||
* digest + the paged verb enumeration. When ABSENT, the oracle counts NO
|
||||
* verbs and says so via verbsChecked = 0 — an honest partial verdict,
|
||||
* never a silent full-pass claim.
|
||||
*/
|
||||
canonicalVerbDigest?: (id: string) => Promise<string | null>
|
||||
getVerbs?: (opts: {
|
||||
pagination: { limit: number; offset?: number; cursor?: string }
|
||||
}) => Promise<{ items: unknown[]; hasMore?: boolean; nextCursor?: string }>
|
||||
}): Promise<OracleReport> {
|
||||
const report: OracleReport = {
|
||||
verdict: 'red',
|
||||
|
|
@ -151,19 +161,17 @@ export async function runLogCompletenessOracle(args: {
|
|||
return report
|
||||
}
|
||||
const logState = new Map<string, { tombstoned: boolean; digest: string | null }>()
|
||||
const verbLogState = new Map<string, { tombstoned: boolean; digest: string | null }>()
|
||||
for await (const batch of scan.batches()) {
|
||||
for (const fact of batch.facts) {
|
||||
report.generationsScanned++
|
||||
for (const op of fact.ops) {
|
||||
if (op.kind !== 'noun') continue
|
||||
if (op.record === null) {
|
||||
logState.set(op.id, { tombstoned: true, digest: null })
|
||||
} else {
|
||||
logState.set(op.id, {
|
||||
tombstoned: false,
|
||||
digest: args.factRecordDigest(op.record)
|
||||
})
|
||||
}
|
||||
const state =
|
||||
op.record === null
|
||||
? { tombstoned: true, digest: null }
|
||||
: { tombstoned: false, digest: args.factRecordDigest(op.record) }
|
||||
if (op.kind === 'noun') logState.set(op.id, state)
|
||||
else verbLogState.set(op.id, state)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -210,6 +218,48 @@ export async function runLogCompletenessOracle(args: {
|
|||
}
|
||||
}
|
||||
|
||||
// Verb passes — only when the owner wired the verb legs; otherwise the
|
||||
// report says verbsChecked: 0, an honest partial scope, never a claim.
|
||||
if (args.canonicalVerbDigest && args.getVerbs) {
|
||||
const seenVerbs = new Set<string>()
|
||||
let vOffset = 0
|
||||
let vCursor: string | undefined
|
||||
for (;;) {
|
||||
const page = await args.getVerbs({
|
||||
pagination: vCursor ? { limit: PAGE, cursor: vCursor } : { limit: PAGE, offset: vOffset }
|
||||
})
|
||||
for (const item of page.items) {
|
||||
const id = (item as { id: string }).id
|
||||
seenVerbs.add(id)
|
||||
report.verbsChecked++
|
||||
const inLog = verbLogState.get(id)
|
||||
if (!inLog) {
|
||||
addMismatch({ id, kind: 'verb', reason: 'pre-log-record' })
|
||||
continue
|
||||
}
|
||||
if (inLog.tombstoned) {
|
||||
addMismatch({ id, kind: 'verb', reason: 'log-tombstone-canonical-present' })
|
||||
continue
|
||||
}
|
||||
const canonical = await args.canonicalVerbDigest(id)
|
||||
if (canonical === null) {
|
||||
addMismatch({ id, kind: 'verb', reason: 'pre-log-record' })
|
||||
continue
|
||||
}
|
||||
if (canonical === inLog.digest) report.matched++
|
||||
else addMismatch({ id, kind: 'verb', reason: 'state-differs' })
|
||||
}
|
||||
if (!page.hasMore || page.items.length === 0) break
|
||||
if (page.nextCursor) vCursor = page.nextCursor
|
||||
else vOffset += page.items.length
|
||||
}
|
||||
for (const [id, state] of verbLogState) {
|
||||
if (!state.tombstoned && !seenVerbs.has(id)) {
|
||||
addMismatch({ id, kind: 'verb', reason: 'log-live-canonical-absent' })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const totalMismatches =
|
||||
report.mismatches.length + (report.mismatchListTruncated ? 1 : 0)
|
||||
report.verdict = totalMismatches === 0 ? 'green' : 'red'
|
||||
|
|
|
|||
|
|
@ -83,6 +83,14 @@ export type {
|
|||
AggregationProvider
|
||||
} from './types/brainy.types.js'
|
||||
|
||||
// Read-barrier contract (waitForIndexed): the leg names, the options, and
|
||||
// the typed timeout error (a value export — consumers catch it by instanceof)
|
||||
export type {
|
||||
IndexedProjectionPath,
|
||||
WaitForIndexedOptions
|
||||
} from './types/brainy.types.js'
|
||||
export { WaitForIndexedTimeoutError } from './types/brainy.types.js'
|
||||
|
||||
// Reserved-field contract — the canonical list of Brainy-owned field names
|
||||
// that may never appear inside a `metadata` bag (see docs/concepts/consistency-model.md)
|
||||
export {
|
||||
|
|
|
|||
|
|
@ -1614,6 +1614,15 @@ export interface AggregationProvider {
|
|||
|
||||
/** Serialize internal state for persistence (called during flush) */
|
||||
serializeState?(): string
|
||||
|
||||
/**
|
||||
* Bake the committed generation into the provider's own state envelope
|
||||
* before {@link serializeState} (called during flush, immediately prior).
|
||||
* Lets a native-side reopen verify the envelope's honesty independently of
|
||||
* the host's wrapper stamp. Optional — providers without it rely on the
|
||||
* host wrapper's `sourceGeneration` alone.
|
||||
*/
|
||||
noteSourceGeneration?(generation: number): void
|
||||
}
|
||||
|
||||
// ============= Configuration =============
|
||||
|
|
@ -2244,6 +2253,79 @@ export interface Highlight {
|
|||
contentCategory?: ContentCategory
|
||||
}
|
||||
|
||||
// ============= Read barrier (waitForIndexed) =============
|
||||
|
||||
/**
|
||||
* One projection leg of the read barrier (`brain.waitForIndexed(path)`) — a
|
||||
* derived view of the committed data that queries are served from:
|
||||
*
|
||||
* - `'semantic'` — the vector index (deferred embeds land here asynchronously)
|
||||
* - `'metadata'` — the field/filter index behind `find({ where })`
|
||||
* - `'graph'` — the relationship adjacency index
|
||||
* - `'aggregation'` — the incremental aggregate states
|
||||
*/
|
||||
export type IndexedProjectionPath = 'semantic' | 'metadata' | 'graph' | 'aggregation'
|
||||
|
||||
/**
|
||||
* Options for `brain.waitForIndexed()`.
|
||||
*/
|
||||
export interface WaitForIndexedOptions {
|
||||
/**
|
||||
* Resolve as soon as the projection has caught up to this committed
|
||||
* generation (rather than the current head). Today the pending-embed set
|
||||
* carries no generation stamps, 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.
|
||||
*/
|
||||
generation?: number
|
||||
|
||||
/**
|
||||
* Upper bound on the wait in milliseconds. On expiry the promise REJECTS
|
||||
* with {@link WaitForIndexedTimeoutError} (typed: the leg + the
|
||||
* still-pending count) — never a silent partial wait.
|
||||
*/
|
||||
timeoutMs?: number
|
||||
}
|
||||
|
||||
/**
|
||||
* The typed rejection of `brain.waitForIndexed(path, { timeoutMs })` on
|
||||
* expiry. Carries the projection leg (`path`; `'all'` for the no-argument
|
||||
* barrier) and the deferred-embed backlog size at the moment the timer fired
|
||||
* (`pendingEmbeds` — the same number as
|
||||
* `getIndexStatus().projections.semantic.pendingEmbeds`), so a caller can
|
||||
* log an honest gauge and retry instead of guessing. A timeout means the
|
||||
* projection has NOT caught up — nothing was skipped, nothing partially
|
||||
* waited.
|
||||
*/
|
||||
export class WaitForIndexedTimeoutError extends Error {
|
||||
/** The projection leg that had not caught up (`'all'` = the no-arg barrier). */
|
||||
public readonly path: IndexedProjectionPath | 'all'
|
||||
|
||||
/** The expired timeout, in milliseconds. */
|
||||
public readonly timeoutMs: number
|
||||
|
||||
/** Deferred embeds still pending when the timer fired — the live value of
|
||||
* `getIndexStatus().projections.semantic.pendingEmbeds`. */
|
||||
public readonly pendingEmbeds: number
|
||||
|
||||
constructor(path: IndexedProjectionPath | 'all', timeoutMs: number, pendingEmbeds: number) {
|
||||
super(
|
||||
`waitForIndexed(${path === 'all' ? '' : `'${path}'`}) timed out after ${timeoutMs}ms — ` +
|
||||
`${pendingEmbeds} deferred embed${pendingEmbeds === 1 ? '' : 's'} still pending; the projection has ` +
|
||||
`NOT caught up. Check getIndexStatus().projections.semantic.pendingEmbeds, then retry with a ` +
|
||||
`larger timeoutMs or use awaitPendingEmbeds() for an unbounded drain.`
|
||||
)
|
||||
this.name = 'WaitForIndexedTimeoutError'
|
||||
this.path = path
|
||||
this.timeoutMs = timeoutMs
|
||||
this.pendingEmbeds = pendingEmbeds
|
||||
if (Error.captureStackTrace) {
|
||||
Error.captureStackTrace(this, WaitForIndexedTimeoutError)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ============= Export all types =============
|
||||
|
||||
export * from './graphTypes.js' // Re-export NounType, VerbType, etc.
|
||||
108
tests/integration/brain-relocation.test.ts
Normal file
108
tests/integration/brain-relocation.test.ts
Normal file
|
|
@ -0,0 +1,108 @@
|
|||
/**
|
||||
* @module tests/integration/brain-relocation
|
||||
* @description LC8 — RELOCATABLE BRAIN DIRECTORY. A brain's directory moved
|
||||
* wholesale to a new path (rename/copy — backup-restore, disk migration,
|
||||
* container re-mount) must open and serve IDENTICALLY: no absolute paths may
|
||||
* hide in any persisted artifact. Pinned across every intelligence: point
|
||||
* reads, metadata find, semantic find, graph traversal, aggregation — plus
|
||||
* continued writes with monotonic generations and time-travel reads over
|
||||
* pre-move history.
|
||||
*/
|
||||
import { describe, it, expect, afterEach } from 'vitest'
|
||||
import { mkdtempSync, rmSync, renameSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { Brainy } from '../../src/index.js'
|
||||
import { NounType, VerbType } from '../../src/types/graphTypes.js'
|
||||
|
||||
const dirs: string[] = []
|
||||
const brains: Brainy[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
for (const b of brains.splice(0)) await b.close().catch(() => {})
|
||||
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
const AGG = {
|
||||
name: 'by_kind',
|
||||
source: { type: NounType.Document },
|
||||
groupBy: ['kind'] as string[],
|
||||
metrics: { count: { op: 'count' as const } }
|
||||
}
|
||||
|
||||
describe('LC8 — a moved brain directory opens and serves identically', () => {
|
||||
it('rename the directory: all three intelligences serve, writes continue, history travels', async () => {
|
||||
const home = mkdtempSync(join(tmpdir(), 'brainy-reloc-'))
|
||||
dirs.push(home)
|
||||
const oldPath = join(home, 'brain-old')
|
||||
const newPath = join(home, 'brain-new')
|
||||
|
||||
// Season a brain: rows, a relation, an aggregate, then flush + close.
|
||||
let brain = new Brainy({ storage: { type: 'filesystem', path: oldPath }, requireSubtype: false })
|
||||
await brain.init()
|
||||
brains.push(brain)
|
||||
brain.defineAggregate(AGG)
|
||||
const alpha = await brain.add({
|
||||
data: 'alpha document about mountain geology',
|
||||
type: NounType.Document,
|
||||
metadata: { kind: 'report', n: 1 }
|
||||
})
|
||||
const beta = await brain.add({
|
||||
data: 'beta document about coastal erosion',
|
||||
type: NounType.Document,
|
||||
metadata: { kind: 'report', n: 2 }
|
||||
})
|
||||
await brain.relate({ from: alpha, to: beta, verb: VerbType.RelatedTo })
|
||||
await brain.queryAggregate(AGG.name) // settle backfill
|
||||
const preMoveGen = brain.generation()
|
||||
await brain.flush()
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
// The move: wholesale directory rename.
|
||||
renameSync(oldPath, newPath)
|
||||
|
||||
// Reopen at the NEW path — everything serves.
|
||||
brain = new Brainy({ storage: { type: 'filesystem', path: newPath }, requireSubtype: false })
|
||||
await brain.init()
|
||||
brains.push(brain)
|
||||
brain.defineAggregate(AGG)
|
||||
|
||||
// Point read + metadata find.
|
||||
expect((await brain.get(alpha))!.data).toContain('mountain geology')
|
||||
const found = await brain.find({ where: { kind: 'report' }, limit: 10 })
|
||||
expect(found.map((r) => r.id).sort()).toEqual([alpha, beta].sort())
|
||||
|
||||
// Semantic find.
|
||||
const sem = await brain.find({ query: 'alpha document about mountain geology', limit: 3 })
|
||||
expect(sem.map((r) => r.id)).toContain(alpha)
|
||||
|
||||
// Graph traversal.
|
||||
const related = await brain.related(alpha)
|
||||
expect(related.map((r) => r.to)).toContain(beta)
|
||||
|
||||
// Aggregation.
|
||||
const agg = (await brain.queryAggregate(AGG.name)) as Array<{
|
||||
groupKey: Record<string, unknown>
|
||||
metrics: Record<string, unknown>
|
||||
}>
|
||||
const reportRow = agg.find((g) => g.groupKey['kind'] === 'report')
|
||||
expect(Number(reportRow?.metrics.count)).toBe(2)
|
||||
|
||||
// Writes continue with monotonic generations.
|
||||
const gamma = await brain.add({
|
||||
data: 'gamma addendum after the move',
|
||||
type: NounType.Document,
|
||||
metadata: { kind: 'report', n: 3 }
|
||||
})
|
||||
expect(brain.generation()).toBeGreaterThan(preMoveGen)
|
||||
expect((await brain.get(gamma))!.data).toContain('addendum')
|
||||
|
||||
// Time travel across the move boundary: the pre-move pin sees exactly
|
||||
// the pre-move world (no gamma), served from relocated history.
|
||||
const dbPast = await brain.asOf(preMoveGen)
|
||||
expect(await dbPast.get(gamma)).toBeNull()
|
||||
expect((await dbPast.get(alpha))!.data).toContain('mountain geology')
|
||||
await dbPast.release()
|
||||
}, 120000)
|
||||
})
|
||||
184
tests/integration/find-matchall-cold.test.ts
Normal file
184
tests/integration/find-matchall-cold.test.ts
Normal file
|
|
@ -0,0 +1,184 @@
|
|||
/**
|
||||
* @module tests/integration/find-matchall-cold
|
||||
* @description THE MATCH-ALL SILENT-EMPTY PIN: `find({ where: {} })` is a
|
||||
* match-all query — zero predicates constrain nothing — yet it used to route
|
||||
* through the index-filter branch, where `getIdsForFilter({})` answers `[]`
|
||||
* by contract. Result: 0 rows while storage held rows (worst on a freshly
|
||||
* reopened brain, where it masqueraded as data loss), the forbidden answer
|
||||
* class — a silent empty instead of served-or-refused. These tests pin the
|
||||
* law: an empty `where` routes exactly like an absent `where`, serving from
|
||||
* truth-complete sources (a storage page bounded to the offset+limit window,
|
||||
* or the column store's top-K sort under orderBy) — warm AND cold, on the
|
||||
* live brain, the Db pin path, pagination.count, streaming.entities, and the
|
||||
* semantic path (`{ query, where: {} }` must not short-circuit to `[]`).
|
||||
* The one deliberate refusal: `removeMany({ where: {} })` throws — a
|
||||
* match-all BULK DELETE must be asked for explicitly, never inherited.
|
||||
*/
|
||||
import { describe, it, expect, afterEach } from 'vitest'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { Brainy } from '../../src/index.js'
|
||||
import { NounType } from '../../src/types/graphTypes.js'
|
||||
|
||||
const dirs: string[] = []
|
||||
const brains: Brainy[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
for (const b of brains.splice(0)) await b.close().catch(() => {})
|
||||
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function open(dir: string): Promise<Brainy> {
|
||||
const b = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
|
||||
await b.init()
|
||||
brains.push(b)
|
||||
return b
|
||||
}
|
||||
|
||||
/** Seed three plain documents with a sortable numeric field. */
|
||||
async function seed(brain: Brainy): Promise<string[]> {
|
||||
const ids: string[] = []
|
||||
ids.push(await brain.add({ data: 'alpha row', type: NounType.Document, metadata: { n: 1 } }))
|
||||
ids.push(await brain.add({ data: 'beta row', type: NounType.Document, metadata: { n: 2 } }))
|
||||
ids.push(await brain.add({ data: 'gamma row', type: NounType.Document, metadata: { n: 3 } }))
|
||||
await brain.flush()
|
||||
return ids
|
||||
}
|
||||
|
||||
describe('find({ where: {} }) — match-all serves, warm and cold', () => {
|
||||
it('the repro: a freshly reopened filesystem brain serves match-all (not a silent 0)', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-matchall-cold-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
await seed(brain)
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
const reopened = await open(dir)
|
||||
const rows = await reopened.find({ where: {}, limit: 10 })
|
||||
expect(rows.length, 'match-all serves every stored row on the cold brain').toBe(3)
|
||||
|
||||
// The predicate paths that always worked cold stay working — same brain.
|
||||
expect((await reopened.find({ where: { n: 1 }, limit: 10 })).length).toBe(1)
|
||||
expect((await reopened.find({ where: { 'system.type': 'document' }, limit: 10 })).length).toBe(3)
|
||||
}, 120000)
|
||||
|
||||
it('match-all + orderBy on a metadata field serves sorted after reopen', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-matchall-order-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
await seed(brain)
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
const reopened = await open(dir)
|
||||
const rows = await reopened.find({ where: {}, orderBy: 'n', order: 'desc', limit: 10 })
|
||||
expect(rows.length, 'sorted match-all serves every stored row cold').toBe(3)
|
||||
expect(
|
||||
rows.map((r) => (r.metadata as { n: number }).n),
|
||||
'orderBy is honored on the cold match-all page'
|
||||
).toEqual([3, 2, 1])
|
||||
}, 120000)
|
||||
|
||||
it('warm brain unchanged: match-all, sorted match-all, and predicates all serve in-session', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-matchall-warm-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
await seed(brain)
|
||||
|
||||
expect((await brain.find({ where: {}, limit: 10 })).length).toBe(3)
|
||||
const sorted = await brain.find({ where: {}, orderBy: 'n', order: 'asc', limit: 2 })
|
||||
expect(sorted.map((r) => (r.metadata as { n: number }).n)).toEqual([1, 2])
|
||||
expect((await brain.find({ where: { n: 2 }, limit: 10 })).length).toBe(1)
|
||||
// Pagination window respected: match-all never over-serves the page.
|
||||
expect((await brain.find({ where: {}, limit: 2, offset: 2 })).length).toBe(1)
|
||||
}, 120000)
|
||||
|
||||
it('the semantic path: find({ query, where: {} }) must not short-circuit to []', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-matchall-query-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
await seed(brain)
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
const reopened = await open(dir)
|
||||
// Before the fix, the pre-resolved empty filter matched nothing and the
|
||||
// vector search was skipped entirely — a silent [] for every such query.
|
||||
const rows = await reopened.find({ query: 'alpha row', where: {}, limit: 10 })
|
||||
expect(rows.length, 'an unconstraining where must not empty a semantic query').toBeGreaterThan(0)
|
||||
}, 120000)
|
||||
|
||||
it('the Db pin path: asOf(g).find({ where: {} }) serves at the pinned generation after reopen', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-matchall-asof-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
await brain.add({ data: 'first', type: NounType.Document, metadata: { n: 1 } })
|
||||
await brain.add({ data: 'second', type: NounType.Document, metadata: { n: 2 } })
|
||||
await brain.flush()
|
||||
const gTwo = brain.generation()
|
||||
await brain.add({ data: 'third', type: NounType.Document, metadata: { n: 3 } })
|
||||
await brain.flush()
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
const reopened = await open(dir)
|
||||
// Current-generation pin (delegates to the live find fast path).
|
||||
const now = reopened.now()
|
||||
expect((await now.find({ where: {}, limit: 10 })).length).toBe(3)
|
||||
|
||||
// Historical pin: the record-overlay path must serve match-all too.
|
||||
const past = await reopened.asOf(gTwo)
|
||||
try {
|
||||
const rows = await past.find({ where: {}, limit: 10 })
|
||||
expect(rows.length, 'match-all at the pinned generation sees exactly the rows of that generation').toBe(2)
|
||||
} finally {
|
||||
await past.release()
|
||||
}
|
||||
}, 120000)
|
||||
|
||||
it('pagination.count({ where: {} }) counts every row instead of a silent 0', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-matchall-count-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
await seed(brain)
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
const reopened = await open(dir)
|
||||
// The law: an empty where counts exactly like an absent where (the
|
||||
// unfiltered total — which by long-standing count semantics includes
|
||||
// system entities such as the VFS root, hence >= the 3 user rows).
|
||||
const emptyWhere = await reopened.pagination.count({ where: {} })
|
||||
expect(emptyWhere).toBe(await reopened.pagination.count({}))
|
||||
expect(emptyWhere).toBeGreaterThanOrEqual(3)
|
||||
}, 120000)
|
||||
|
||||
it('streaming.entities({ where: {} }) streams every row instead of nothing', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-matchall-stream-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
await seed(brain)
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
const reopened = await open(dir)
|
||||
const streamed: string[] = []
|
||||
for await (const entity of reopened.streaming.entities({ where: {} })) {
|
||||
streamed.push(entity.id)
|
||||
}
|
||||
expect(streamed.length, 'an unconstraining where streams the full store').toBeGreaterThanOrEqual(3)
|
||||
}, 120000)
|
||||
|
||||
it('removeMany({ where: {} }) refuses loudly — match-all bulk delete is never implicit', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-matchall-remove-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
await seed(brain)
|
||||
|
||||
await expect(brain.removeMany({ where: {} })).rejects.toThrow(/matches EVERYTHING/)
|
||||
// Nothing was deleted by the refused call.
|
||||
expect((await brain.find({ where: {}, limit: 10 })).length).toBe(3)
|
||||
}, 120000)
|
||||
})
|
||||
83
tests/integration/log-authority-adopt.test.ts
Normal file
83
tests/integration/log-authority-adopt.test.ts
Normal file
|
|
@ -0,0 +1,83 @@
|
|||
/**
|
||||
* @module tests/integration/log-authority-adopt
|
||||
* @description THE SANCTIONED FLIP, END TO END: adoptLogAuthority() cures
|
||||
* its own curable divergences by baseline backfill — a FRESH brain (whose
|
||||
* generation-0 VFS root never entered the log) flips WITHOUT any manual
|
||||
* white-box backfill. Before this, no fresh brain could ever flip: the
|
||||
* oracle reported the bootstrap row as pre-log-record and the flip refused.
|
||||
* Log-AHEAD divergences stay incurable and refuse loudly (witness wins).
|
||||
*/
|
||||
import { describe, it, expect, afterEach } from 'vitest'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { Brainy } from '../../src/index.js'
|
||||
import { NounType } from '../../src/types/graphTypes.js'
|
||||
|
||||
const dirs: string[] = []
|
||||
const brains: Brainy[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
for (const b of brains.splice(0)) await b.close().catch(() => {})
|
||||
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function open(dir: string): Promise<Brainy> {
|
||||
const b = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
|
||||
await b.init()
|
||||
brains.push(b)
|
||||
return b
|
||||
}
|
||||
|
||||
describe('adoptLogAuthority — the sanctioned flip with self-backfill', () => {
|
||||
it('a fresh brain flips directly: the backfill cures the generation-0 baseline', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-adopt-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
const idA = await brain.add({ data: 'first row', type: NounType.Document, metadata: { n: 1 } })
|
||||
await brain.add({ data: 'second row', type: NounType.Document, metadata: { n: 2 } })
|
||||
await brain.flush()
|
||||
|
||||
const report = await brain.adoptLogAuthority()
|
||||
expect(report.verdict, 'the flip receipt is a green oracle').toBe('green')
|
||||
expect(brain.logAuthority().authority).toBe('log')
|
||||
|
||||
// The switch survives reopen; the brain keeps serving identically.
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
const reopened = await open(dir)
|
||||
expect(reopened.logAuthority().authority).toBe('log')
|
||||
expect(await reopened.get(idA), 'records serve at reopen').toBeTruthy()
|
||||
const rows = await reopened.find({ where: {}, limit: 10 })
|
||||
expect(rows.length, 'match-all serves on the reopened flipped brain').toBeGreaterThanOrEqual(2)
|
||||
// And a fresh oracle run on the flipped brain stays green.
|
||||
expect((await reopened.verifyLogAuthority()).verdict).toBe('green')
|
||||
}, 120000)
|
||||
|
||||
it('witness drift (out-of-generation canonical rewrite) is cured by the backfill, then flips', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-adopt-drift-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir)
|
||||
const id = await brain.add({ data: 'drifter', type: NounType.Document, metadata: { v: 1 } })
|
||||
await brain.flush()
|
||||
|
||||
// Simulate maintenance rewriting canonical OUTSIDE a generation (the
|
||||
// witness-drift class): mutate the stored record directly.
|
||||
const storage = (brain as unknown as {
|
||||
storage: {
|
||||
readNounRaw(id: string): Promise<{ metadata: unknown; vector: unknown }>
|
||||
writeNounRaw(id: string, r: { metadata: unknown; vector: unknown }): Promise<void>
|
||||
}
|
||||
}).storage
|
||||
const raw = await storage.readNounRaw(id)
|
||||
await storage.writeNounRaw(id, {
|
||||
metadata: { ...(raw.metadata as Record<string, unknown>), drifted: true },
|
||||
vector: raw.vector
|
||||
})
|
||||
expect((await brain.verifyLogAuthority()).verdict, 'drift detected').toBe('red')
|
||||
|
||||
const report = await brain.adoptLogAuthority()
|
||||
expect(report.verdict).toBe('green')
|
||||
expect(brain.logAuthority().authority).toBe('log')
|
||||
}, 120000)
|
||||
})
|
||||
219
tests/integration/wait-for-indexed.test.ts
Normal file
219
tests/integration/wait-for-indexed.test.ts
Normal file
|
|
@ -0,0 +1,219 @@
|
|||
/**
|
||||
* @module tests/integration/wait-for-indexed
|
||||
* @description THE READ BARRIER — `brain.waitForIndexed(path?, opts?)`. A
|
||||
* consumer that writes and then semantically recalls gets ONE honest barrier
|
||||
* instead of guessing. The contract pinned here:
|
||||
*
|
||||
* 1. SEMANTIC LEG: a deferred add followed by `waitForIndexed('semantic')`
|
||||
* resolves only after the vector landed — the row is vector-searchable
|
||||
* the moment the barrier returns.
|
||||
* 2. TYPED TIMEOUT: `timeoutMs` expiry REJECTS with
|
||||
* WaitForIndexedTimeoutError carrying the leg + the pending count and
|
||||
* naming the gauge — never a silent partial wait.
|
||||
* 3. NO-ARG: every projection at the head; today that means the deferred
|
||||
* embed backlog is drained.
|
||||
* 4. SYNCHRONOUS LEGS: metadata/graph/aggregation resolve immediately by
|
||||
* design today (they update inside the write path) — even while the
|
||||
* semantic backlog is wedged.
|
||||
* 5. GAUGES: getIndexStatus().projections carries the per-leg numbers, and
|
||||
* the top-level pendingEmbeds compat field agrees with the semantic one.
|
||||
* 6. GENERATION REFINEMENT: an empty backlog satisfies any generation
|
||||
* immediately; a non-empty one falls back to the full drain.
|
||||
*/
|
||||
import { describe, it, expect, afterEach, vi } from 'vitest'
|
||||
import { Brainy, WaitForIndexedTimeoutError } from '../../src/index.js'
|
||||
import { NounType } from '../../src/types/graphTypes.js'
|
||||
|
||||
const brains: Brainy[] = []
|
||||
|
||||
async function memBrain(): Promise<Brainy> {
|
||||
const b = new Brainy({ storage: { type: 'memory' }, requireSubtype: false })
|
||||
await b.init()
|
||||
brains.push(b)
|
||||
return b
|
||||
}
|
||||
|
||||
/**
|
||||
* Abandon a poisoned in-flight embed run (its embed promise never resolves —
|
||||
* production is covered by the worker's 60s hang guard; the test takes the
|
||||
* white-box shortcut for speed), then drain so teardown never wedges.
|
||||
*/
|
||||
async function unwedge(brain: Brainy): Promise<void> {
|
||||
;(brain as unknown as { _embedWorkerFlight: Promise<void> | null })._embedWorkerFlight = null
|
||||
await brain.awaitPendingEmbeds()
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks()
|
||||
for (const b of brains.splice(0)) await b.close().catch(() => {})
|
||||
})
|
||||
|
||||
describe('waitForIndexed — the read barrier', () => {
|
||||
it("SEMANTIC LEG: deferred add → waitForIndexed('semantic') resolves and the row is vector-searchable after", async () => {
|
||||
const brain = await memBrain()
|
||||
const embedSpy = vi.spyOn(brain, 'embed')
|
||||
|
||||
const id = await brain.add({
|
||||
data: 'the quarterly revenue report for the northern region',
|
||||
type: NounType.Document,
|
||||
deferEmbedding: true,
|
||||
metadata: { kind: 'report' }
|
||||
})
|
||||
expect(embedSpy, 'no embed on the ack path').not.toHaveBeenCalled()
|
||||
expect(brain.pendingEmbedCount()).toBeGreaterThanOrEqual(1)
|
||||
|
||||
await brain.waitForIndexed('semantic')
|
||||
|
||||
// The barrier's meaning: backlog drained, vector real, row searchable.
|
||||
expect(brain.pendingEmbedCount(), 'barrier means drained').toBe(0)
|
||||
const after = await brain.get(id, { includeVectors: true })
|
||||
expect((after!.vector as number[]).length, 'real vector after the barrier').toBeGreaterThan(0)
|
||||
const hits = await brain.find({
|
||||
query: 'the quarterly revenue report for the northern region',
|
||||
searchMode: 'semantic',
|
||||
limit: 5
|
||||
})
|
||||
expect(hits.map((r) => r.id), 'vector-searchable after the barrier').toContain(id)
|
||||
})
|
||||
|
||||
it('TYPED TIMEOUT: a hung embedder + timeoutMs rejects with the typed error naming the pending count and the gauge', async () => {
|
||||
const brain = await memBrain()
|
||||
const hang = vi
|
||||
.spyOn(brain, 'embed')
|
||||
.mockImplementation(() => new Promise<number[]>(() => {}))
|
||||
|
||||
await brain.add({
|
||||
data: 'never lands while the embedder hangs',
|
||||
type: NounType.Document,
|
||||
deferEmbedding: true,
|
||||
metadata: {}
|
||||
})
|
||||
expect(brain.pendingEmbedCount()).toBe(1)
|
||||
|
||||
let caught: unknown
|
||||
try {
|
||||
await brain.waitForIndexed('semantic', { timeoutMs: 200 })
|
||||
} catch (e) {
|
||||
caught = e
|
||||
}
|
||||
|
||||
expect(caught, 'expiry REJECTS — never a silent partial wait').toBeInstanceOf(
|
||||
WaitForIndexedTimeoutError
|
||||
)
|
||||
const err = caught as WaitForIndexedTimeoutError
|
||||
expect(err.path).toBe('semantic')
|
||||
expect(err.timeoutMs).toBe(200)
|
||||
expect(err.pendingEmbeds).toBeGreaterThanOrEqual(1)
|
||||
// The message names what was still pending and the gauge to check.
|
||||
expect(err.message).toContain(`${err.pendingEmbeds} deferred embed`)
|
||||
expect(err.message).toContain('getIndexStatus().projections.semantic.pendingEmbeds')
|
||||
|
||||
hang.mockRestore()
|
||||
await unwedge(brain)
|
||||
expect(brain.pendingEmbedCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('NO-ARG: waitForIndexed() waits on the pending-embed drain (every projection at the head)', async () => {
|
||||
const brain = await memBrain()
|
||||
await brain.add({
|
||||
data: 'a deferred capture that the bare barrier must cover',
|
||||
type: NounType.Document,
|
||||
deferEmbedding: true,
|
||||
metadata: {}
|
||||
})
|
||||
expect(brain.pendingEmbedCount()).toBeGreaterThanOrEqual(1)
|
||||
|
||||
await brain.waitForIndexed()
|
||||
|
||||
expect(
|
||||
brain.pendingEmbedCount(),
|
||||
'the bare barrier drained the only asynchronous projection'
|
||||
).toBe(0)
|
||||
})
|
||||
|
||||
it('SYNCHRONOUS LEGS: metadata/graph/aggregation resolve immediately — even while the semantic backlog is wedged', async () => {
|
||||
const brain = await memBrain()
|
||||
|
||||
// Quiet brain first: all three legs resolve on a brain with no backlog.
|
||||
await brain.add({ data: 'quiet row', type: NounType.Document, metadata: { q: 1 } })
|
||||
await brain.awaitPendingEmbeds()
|
||||
await brain.waitForIndexed('metadata')
|
||||
await brain.waitForIndexed('graph')
|
||||
await brain.waitForIndexed('aggregation')
|
||||
|
||||
// The stronger pin: these projections update inside the write path today,
|
||||
// so their leg resolves immediately BY DESIGN — independent of a wedged
|
||||
// semantic backlog. (If any of them incorrectly delegated to the embed
|
||||
// drain, this test would hang.)
|
||||
const hang = vi
|
||||
.spyOn(brain, 'embed')
|
||||
.mockImplementation(() => new Promise<number[]>(() => {}))
|
||||
await brain.add({
|
||||
data: 'wedged deferred row',
|
||||
type: NounType.Document,
|
||||
deferEmbedding: true,
|
||||
metadata: {}
|
||||
})
|
||||
expect(brain.pendingEmbedCount()).toBe(1)
|
||||
|
||||
await brain.waitForIndexed('metadata')
|
||||
await brain.waitForIndexed('graph')
|
||||
await brain.waitForIndexed('aggregation')
|
||||
|
||||
hang.mockRestore()
|
||||
await unwedge(brain)
|
||||
})
|
||||
|
||||
it('GAUGES: getIndexStatus().projections carries the per-leg shape, and the compat field agrees', async () => {
|
||||
const brain = await memBrain()
|
||||
await brain.add({ data: 'gauge row', type: NounType.Document, metadata: { g: 1 } })
|
||||
await brain.awaitPendingEmbeds()
|
||||
|
||||
const status = await brain.getIndexStatus()
|
||||
expect(status.projections).toEqual({
|
||||
semantic: { pendingEmbeds: 0 },
|
||||
metadata: { synchronous: true },
|
||||
graph: { synchronous: true },
|
||||
aggregation: { pendingBackfills: 0, pendingCatchUps: 0 }
|
||||
})
|
||||
// Compat: the existing top-level gauge stays and agrees.
|
||||
expect(status.pendingEmbeds).toBe(0)
|
||||
|
||||
// The semantic gauge is honest while a backlog exists.
|
||||
const hang = vi
|
||||
.spyOn(brain, 'embed')
|
||||
.mockImplementation(() => new Promise<number[]>(() => {}))
|
||||
await brain.add({
|
||||
data: 'backlogged row',
|
||||
type: NounType.Document,
|
||||
deferEmbedding: true,
|
||||
metadata: {}
|
||||
})
|
||||
const busy = await brain.getIndexStatus()
|
||||
expect(busy.projections.semantic.pendingEmbeds).toBeGreaterThanOrEqual(1)
|
||||
expect(busy.pendingEmbeds).toBe(busy.projections.semantic.pendingEmbeds)
|
||||
|
||||
hang.mockRestore()
|
||||
await unwedge(brain)
|
||||
})
|
||||
|
||||
it('GENERATION REFINEMENT: an empty backlog satisfies any generation immediately; a non-empty one falls back to the full drain', async () => {
|
||||
const brain = await memBrain()
|
||||
await brain.add({ data: 'generation row', type: NounType.Document, metadata: {} })
|
||||
await brain.awaitPendingEmbeds()
|
||||
|
||||
// Empty backlog: the semantic watermark is at the head — >= any committed G.
|
||||
await brain.waitForIndexed('semantic', { generation: 1 })
|
||||
|
||||
// Non-empty backlog: the conservative full drain (a superset of the
|
||||
// requested wait, never a partial one).
|
||||
await brain.add({
|
||||
data: 'second generation row',
|
||||
type: NounType.Document,
|
||||
deferEmbedding: true,
|
||||
metadata: {}
|
||||
})
|
||||
await brain.waitForIndexed('semantic', { generation: 1 })
|
||||
expect(brain.pendingEmbedCount(), 'the fallback is the full drain').toBe(0)
|
||||
})
|
||||
})
|
||||
96
tests/unit/db/log-authority-oracle-verbs.test.ts
Normal file
96
tests/unit/db/log-authority-oracle-verbs.test.ts
Normal file
|
|
@ -0,0 +1,96 @@
|
|||
/**
|
||||
* @module tests/unit/db/log-authority-oracle-verbs
|
||||
* @description The verification oracle's VERB legs — module-level pins with
|
||||
* doubles (the brain-level wiring rides the owner's call site):
|
||||
* 1. Wired verb legs diff verbs exactly like nouns (pre-log / state-differs /
|
||||
* tombstone-vs-present / log-live-absent).
|
||||
* 2. UNWIRED verb legs = an HONEST PARTIAL verdict: verbsChecked stays 0 —
|
||||
* the oracle never claims scope it did not scan.
|
||||
*/
|
||||
import { describe, it, expect } from 'vitest'
|
||||
import { runLogCompletenessOracle, recordDigest } from '../../../src/db/logAuthority.js'
|
||||
import type { FactScanHandle } from '../../../src/db/factLog.js'
|
||||
|
||||
type Op = { kind: 'noun' | 'verb'; id: string; record: { metadata: unknown; vector: unknown } | null }
|
||||
|
||||
function scanOf(facts: Array<{ generation: number; ops: Op[] }>): () => FactScanHandle | null {
|
||||
return () =>
|
||||
({
|
||||
batches: async function* () {
|
||||
yield { facts: facts.map((f) => ({ ...f, timestamp: 0 })) }
|
||||
}
|
||||
}) as unknown as FactScanHandle
|
||||
}
|
||||
|
||||
function pagedList(rows: string[]) {
|
||||
return async ({ pagination }: { pagination: { limit: number; offset?: number } }) => {
|
||||
const start = pagination.offset ?? 0
|
||||
const items = rows.slice(start, start + pagination.limit).map((id) => ({ id }))
|
||||
return { items, hasMore: start + pagination.limit < rows.length }
|
||||
}
|
||||
}
|
||||
|
||||
const rec = (v: number) => ({ metadata: { v }, vector: null })
|
||||
|
||||
describe('oracle verb legs', () => {
|
||||
it('wired: verbs diff by digest — clean log goes green over nouns AND verbs', async () => {
|
||||
const report = await runLogCompletenessOracle({
|
||||
storage: { getNouns: pagedList(['n1']) } as never,
|
||||
scanFacts: scanOf([
|
||||
{ generation: 1, ops: [{ kind: 'noun', id: 'n1', record: rec(1) }] },
|
||||
{ generation: 2, ops: [{ kind: 'verb', id: 'v1', record: rec(7) }] }
|
||||
]),
|
||||
canonicalNounDigest: async () => recordDigest(rec(1)),
|
||||
factRecordDigest: recordDigest,
|
||||
canonicalVerbDigest: async () => recordDigest(rec(7)),
|
||||
getVerbs: pagedList(['v1'])
|
||||
})
|
||||
expect(report.verdict).toBe('green')
|
||||
expect(report.nounsChecked).toBe(1)
|
||||
expect(report.verbsChecked).toBe(1)
|
||||
expect(report.matched).toBe(2)
|
||||
})
|
||||
|
||||
it('wired: every verb divergence class is NAMED', async () => {
|
||||
const report = await runLogCompletenessOracle({
|
||||
storage: { getNouns: pagedList([]) } as never,
|
||||
scanFacts: scanOf([
|
||||
{
|
||||
generation: 1,
|
||||
ops: [
|
||||
{ kind: 'verb', id: 'v-differs', record: rec(1) },
|
||||
{ kind: 'verb', id: 'v-tomb', record: null },
|
||||
{ kind: 'verb', id: 'v-orphan', record: rec(3) }
|
||||
]
|
||||
}
|
||||
]),
|
||||
canonicalNounDigest: async () => null,
|
||||
factRecordDigest: recordDigest,
|
||||
canonicalVerbDigest: async (id) =>
|
||||
id === 'v-differs' ? recordDigest(rec(999)) : id === 'v-tomb' ? recordDigest(rec(2)) : null,
|
||||
// canonical enumerates: v-differs (drifted), v-tomb (log says deleted),
|
||||
// v-prelog (never logged); v-orphan is log-live but canonical-absent.
|
||||
getVerbs: pagedList(['v-differs', 'v-tomb', 'v-prelog'])
|
||||
})
|
||||
expect(report.verdict).toBe('red')
|
||||
const by = (id: string) => report.mismatches.find((m) => m.id === id)
|
||||
expect(by('v-differs')).toMatchObject({ kind: 'verb', reason: 'state-differs' })
|
||||
expect(by('v-tomb')).toMatchObject({ kind: 'verb', reason: 'log-tombstone-canonical-present' })
|
||||
expect(by('v-prelog')).toMatchObject({ kind: 'verb', reason: 'pre-log-record' })
|
||||
expect(by('v-orphan')).toMatchObject({ kind: 'verb', reason: 'log-live-canonical-absent' })
|
||||
})
|
||||
|
||||
it('unwired: verbsChecked stays 0 — honest partial scope, never a silent claim', async () => {
|
||||
const report = await runLogCompletenessOracle({
|
||||
storage: { getNouns: pagedList(['n1']) } as never,
|
||||
scanFacts: scanOf([
|
||||
{ generation: 1, ops: [{ kind: 'noun', id: 'n1', record: rec(1) }] },
|
||||
{ generation: 2, ops: [{ kind: 'verb', id: 'v1', record: rec(7) }] }
|
||||
]),
|
||||
canonicalNounDigest: async () => recordDigest(rec(1)),
|
||||
factRecordDigest: recordDigest
|
||||
})
|
||||
expect(report.verbsChecked).toBe(0)
|
||||
expect(report.nounsChecked).toBe(1)
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue