From 287384cf1e30a23a88a211de6bf92803407b844e Mon Sep 17 00:00:00 2001 From: David Snelling Date: Wed, 5 Aug 2026 16:26:43 -0700 Subject: [PATCH] =?UTF-8?q?feat(embedding):=20MT5=20=E2=80=94=20deferred?= =?UTF-8?q?=20embedding=20with=20durable=20markers;=20write=20acks=20never?= =?UTF-8?q?=20wait=20on=20a=20neural=20net?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A3 of the service-class pair (BRAINY-PROD-LATENCY-TRIAD): a VFS file write ran the embedder synchronously while the caller waited — 5.6s p50 / 21.4s p95 per small file on a production deployment, the dominant stage of every capture write. - add()/update() gain deferEmbedding: the write acks at durability (data + metadata persisted, a DURABLE pending marker under _system/pending_embeds/ written BEFORE the commit — orphan-safe direction); the single-flight background worker embeds the CURRENT data and swaps the vector in ATOMICALLY (ReplaceInVectorIndex — the row is never absent from search; a deferred UPDATE keeps serving the OLD vector, stale-beats-absent per the flicker law). Typed refusals: defer+vector, defer-without-data. - CRASH-SAFE: markers are recovered at open by a BOUNDED prefix listing (never a store walk) and the worker resumes in the background — a crash can delay a vector, never lose one. A wedged embedder trips a LOUD 60s hang guard and the worker moves on (marker retained for retry). - The honest gauges: getIndexStatus().pendingEmbeds + pendingEmbedCount(); awaitPendingEmbeds() is the eventual-vector-index BARRIER for callers and tests that need searchability before proceeding. - VFS adopts it everywhere a write path could wait on the embedder: writeFile (both branches) and directory creation. Pinned in the strongest form: writeFile resolves while the embedder HANGS FOREVER. Pins: deferred-embedding 5/5 (ack law · stale-beats-absent · crash recovery across sessions · VFS hung-embedder ack · typed refusals). Gates: unit 1928/1928 · integration 765 · conformance 27/27. --- src/brainy.ts | 264 +++++++++++++++++-- src/types/brainy.types.ts | 23 ++ src/utils/paramValidation.ts | 29 ++ src/vfs/VirtualFileSystem.ts | 12 + tests/integration/deferred-embedding.test.ts | 171 ++++++++++++ 5 files changed, 477 insertions(+), 22 deletions(-) create mode 100644 tests/integration/deferred-embedding.test.ts diff --git a/src/brainy.ts b/src/brainy.ts index 3ad8ba31..1ef9dabc 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -695,6 +695,12 @@ export class Brainy implements BrainyInterface { private _persistLastFlushAt = Date.now() private _persistIdleTimer: ReturnType | null = null private _persistBackgroundFlight: Promise | null = null + + // DEFERRED EMBEDDING (MT5): durable pending markers under + // _system/pending_embeds/, mirrored in-memory, drained by ONE + // background worker. A crash can delay a vector, never lose one. + private _pendingEmbedIds = new Set() + private _embedWorkerFlight: Promise | null = null // A failed walk latches its error: retries within the cooldown rethrow it // instantly instead of re-walking, so a tight caller-side retry loop costs // one loud error per query, never a full store walk per query. @@ -1418,6 +1424,33 @@ export class Brainy implements BrainyInterface { this._generationStampingActive = true } + // MT5 crash recovery: reload the durable pending-embed markers (a + // BOUNDED prefix listing — never a store walk) and resume the worker + // in the background. A crash between a deferred write's ack and its + // background embed DELAYED a vector; this is where it lands. + if (!this.isReadOnly) { + try { + const markerPaths = await this.storage.listRawObjects(Brainy.PENDING_EMBED_PREFIX) + for (const path of markerPaths) { + const id = path.slice(path.lastIndexOf('/') + 1) + if (id) this._pendingEmbedIds.add(id) + } + if (this._pendingEmbedIds.size > 0) { + prodLog.info( + `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` + + `session — resuming in the background` + ) + const t = setTimeout(() => this.kickEmbedWorker(), 0) + ;(t as { unref?: () => void }).unref?.() + } + } catch (err) { + prodLog.warn( + `[Brainy] pending-embed recovery listing failed: ${(err as Error).message} — ` + + `markers remain durable; recovery retries next open` + ) + } + } + // Eager embedding initialization. // // Adaptive default (8.0): the WASM embedding engine eagerly initializes @@ -1840,6 +1873,133 @@ export class Brainy implements BrainyInterface { * @param run - The single-op's existing operation batch builder (the * `tx => {…}` body previously passed straight to `executeTransaction`). */ + /** Storage-root-relative prefix of the durable pending-embed markers. */ + private static readonly PENDING_EMBED_PREFIX = '_system/pending_embeds/' + + /** + * @description Persist the durable pending-embed marker (MT5) and mirror + * it in memory. Written BEFORE the write it belongs to commits — an + * orphaned marker (commit failed) is harmless and reaped by the worker; + * the reverse ordering could lose an embed silently on a crash. + */ + private async enqueuePendingEmbed(id: string): Promise { + this._pendingEmbedIds.add(id) + await this.storage.writeRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`, { + id, + enqueuedAt: Date.now() + }) + } + + /** Remove a pending-embed marker (memory + durable), tolerating races. */ + private async clearPendingEmbed(id: string): Promise { + this._pendingEmbedIds.delete(id) + await this.storage + .deleteRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`) + .catch(() => {}) + } + + /** + * @description Start (or skip into) the ONE deferred-embedding worker. + * Never awaited by write paths; failures are LOUD and markers survive for + * the next kick (next deferred write, or the next open's recovery). + */ + private kickEmbedWorker(): void { + if (this._embedWorkerFlight || this._pendingEmbedIds.size === 0 || this.isReadOnly) return + this._embedWorkerFlight = this.runEmbedWorker() + .catch((err) => { + prodLog.error( + `[Brainy] deferred-embed worker failed: ${(err as Error).message} — ` + + `markers retained; retries at the next deferred write or open` + ) + }) + .finally(() => { + this._embedWorkerFlight = null + if (this._pendingEmbedIds.size > 0) { + // New arrivals during the run: schedule (never recurse) the next pass. + const t = setTimeout(() => this.kickEmbedWorker(), 0) + ;(t as { unref?: () => void }).unref?.() + } + }) + } + + /** + * @description Drain the pending-embed set: embed each row's CURRENT data + * (a row updated again before its turn embeds the latest content — the + * marker set is idempotent per id) and swap the vector in ATOMICALLY + * (ReplaceInVectorIndex → the in-place update; the row is never absent + * from search). Orphans (row deleted, or no data) reap their markers. + */ + private async runEmbedWorker(): Promise { + const batch = Array.from(this._pendingEmbedIds) + for (const id of batch) { + try { + const entity = await this.get(id, { includeVectors: true }) + if (!entity || entity.data === undefined || entity.data === null) { + await this.clearPendingEmbed(id) + continue + } + // Hang guard: a wedged embedder must not block every later pending + // embed forever — time out LOUDLY, keep the marker, move on. (A + // failure is retryable; an unbounded silent wait is the outlawed + // shape.) + const newVector = await Promise.race([ + this.embed(entity.data), + new Promise((_, reject) => { + const t = setTimeout( + () => reject(new Error('deferred embed timed out after 60s')), + 60_000 + ) + ;(t as { unref?: () => void }).unref?.() + }) + ]) + if (!this.dimensions) { + this.dimensions = newVector.length + } else if (newVector.length !== this.dimensions) { + throw new Error( + `deferred embed produced ${newVector.length} dimensions, store expects ${this.dimensions}` + ) + } + const oldVector = (entity.vector as number[] | undefined) ?? [] + await this.persistSingleOp({ nouns: [id] }, async (tx) => { + tx.addOperation( + new SaveNounOperation(this.storage, { + id, + vector: newVector, + connections: new Map(), + level: 0 + }) + ) + tx.addOperation( + new ReplaceInVectorIndexOperation(this.index, id, oldVector, newVector) + ) + }) + await this.clearPendingEmbed(id) + } catch (err) { + prodLog.warn( + `[Brainy] deferred embed for ${id} failed: ${(err as Error).message} — marker retained for retry` + ) + } + } + } + + /** + * @description The deferred-embedding BARRIER: resolves when every pending + * embed has landed (vector searchable) or been reaped. The eventual- + * vector-index contract's awaitable edge — tests and "must be searchable + * before I proceed" callers use this; nothing else ever needs to wait. + */ + public async awaitPendingEmbeds(): Promise { + while (this._pendingEmbedIds.size > 0 || this._embedWorkerFlight) { + this.kickEmbedWorker() + await (this._embedWorkerFlight ?? Promise.resolve()) + } + } + + /** The deferred-embedding backlog size (also on getIndexStatus().pendingEmbeds). */ + public pendingEmbedCount(): number { + return this._pendingEmbedIds.size + } + /** * @description The write-side persistence trigger (policy `'auto'`): count * the committed write, kick a single-flight BACKGROUND flush when the @@ -2166,15 +2326,26 @@ export class Brainy implements BrainyInterface { } // Get or compute vector - const vector = params.vector || (await this.embed(params.data)) + // MT5 deferred embedding: ack at durability with a stub vector and a + // DURABLE pending marker (written BEFORE the commit — an orphaned marker + // from a failed commit is harmless and reaped by the worker; a + // marker-less committed row would be a silently missing vector, which is + // the disallowed direction). The background worker embeds + inserts. + const deferringEmbed = params.deferEmbedding === true && !params.vector + const vector = deferringEmbed + ? [] + : params.vector || (await this.embed(params.data)) - // Ensure dimensions are set - if (!this.dimensions) { - this.dimensions = vector.length - } else if (vector.length !== this.dimensions) { - throw new Error( - `Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}` - ) + // Ensure dimensions are set (a deferred-embed stub carries no dimension + // information — the worker's real vector goes through the same guard). + if (!deferringEmbed) { + if (!this.dimensions) { + this.dimensions = vector.length + } else if (vector.length !== this.dimensions) { + throw new Error( + `Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}` + ) + } } // Prepare metadata for storage: a v2 nested-bag record — engine fields @@ -2254,6 +2425,12 @@ export class Brainy implements BrainyInterface { } : undefined + // MT5: the durable marker lands BEFORE the commit (orphan-safe; the + // reverse order could lose an embed silently on a crash). + if (deferringEmbed) { + await this.enqueuePendingEmbed(id) + } + const runInsert: TransactionFunction = async (tx) => { // Operation 1: Save metadata FIRST (TypeAwareStorage caching) // isNew=true: skip pre-read for rollback (entity doesn't exist yet) @@ -2272,10 +2449,14 @@ export class Brainy implements BrainyInterface { }, true) ) - // Operation 3: Add to HNSW index (after entity saved) - tx.addOperation( - new AddToVectorIndexOperation(this.index, id, vector) - ) + // Operation 3: Add to HNSW index (after entity saved). A deferred + // embed has nothing to index yet — the worker's atomic update + // inserts the real vector. + if (!deferringEmbed) { + tx.addOperation( + new AddToVectorIndexOperation(this.index, id, vector) + ) + } // Operation 4: Add to metadata index tx.addOperation( @@ -2343,6 +2524,7 @@ export class Brainy implements BrainyInterface { this._aggregationIndex.onEntityAdded(id, entityForIndexing) } + if (deferringEmbed) this.kickEmbedWorker() return id } @@ -2828,6 +3010,11 @@ export class Brainy implements BrainyInterface { // new `data`); otherwise new `data` re-embeds; otherwise the existing // vector is kept. Any vector change re-indexes HNSW below. let vector = existing.vector + // MT5 deferred re-embedding: the OLD vector keeps serving semantic + // search — stale-but-present, never absent (the flicker law) — until + // the background worker embeds the new data and swaps it atomically. + const deferringEmbed = + params.deferEmbedding === true && Boolean(params.data) && !params.vector if (params.vector) { if (this.dimensions && params.vector.length !== this.dimensions) { throw new Error( @@ -2835,10 +3022,14 @@ export class Brainy implements BrainyInterface { ) } vector = params.vector - } else if (params.data) { + } else if (params.data && !deferringEmbed) { vector = await this.embed(params.data) } - const needsReindexing = Boolean(params.data || params.type || params.vector) + // A deferred data change does NOT reindex now (the vector is unchanged; + // the worker's atomic swap carries the real reindex later). + const needsReindexing = Boolean( + (params.data && !deferringEmbed) || params.type || params.vector + ) // Always update the noun with new metadata const newMetadata = params.merge !== false @@ -2925,6 +3116,11 @@ export class Brainy implements BrainyInterface { updatedMetadata._rev = authoritativeRev + 1 } + // MT5: durable marker BEFORE the commit (orphan-safe direction). + if (deferringEmbed) { + await this.enqueuePendingEmbed(params.id) + } + // Execute atomically with transaction system, generation-stamped as one // immutable Model-B generation (before-image = the entity's prior state). await this.persistSingleOp({ nouns: [params.id] }, async (tx) => { @@ -3026,6 +3222,8 @@ export class Brainy implements BrainyInterface { existing as unknown as Record ) } + + if (deferringEmbed) this.kickEmbedWorker() } /** @@ -9123,13 +9321,23 @@ export class Brainy implements BrainyInterface { } } - const vector = params.vector || (await this.embed(params.data)) - if (!this.dimensions) { - this.dimensions = vector.length - } else if (vector.length !== this.dimensions) { - throw new Error( - `Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}` - ) + // MT5 deferred embedding: ack at durability with a stub vector and a + // DURABLE pending marker (written BEFORE the commit — an orphaned marker + // from a failed commit is harmless and reaped by the worker; a + // marker-less committed row would be a silently missing vector, which is + // the disallowed direction). The background worker embeds + inserts. + const deferringEmbed = params.deferEmbedding === true && !params.vector + const vector = deferringEmbed + ? [] + : params.vector || (await this.embed(params.data)) + if (!deferringEmbed) { + if (!this.dimensions) { + this.dimensions = vector.length + } else if (vector.length !== this.dimensions) { + throw new Error( + `Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}` + ) + } } // isNew controls the operation's rollback strategy: a custom id may @@ -9192,10 +9400,18 @@ export class Brainy implements BrainyInterface { } } + if (deferringEmbed) { + // Durable marker BEFORE the batch commits (orphan-safe direction); + // the worker kicks post-commit via the plan hook. + await this.enqueuePendingEmbed(id) + plan.postCommit.push(() => this.kickEmbedWorker()) + } plan.operations.push( new SaveNounMetadataOperation(this.storage, id, storageMetadata, isNew), new SaveNounOperation(this.storage, { id, vector, connections: new Map(), level: 0 }, isNew), - new AddToVectorIndexOperation(this.index, id, vector), + ...(deferringEmbed + ? [] + : [new AddToVectorIndexOperation(this.index, id, vector)]), new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing) ) plan.touchedNouns.push(id) @@ -10672,6 +10888,8 @@ export class Brainy implements BrainyInterface { async getIndexStatus(): Promise<{ initialized: boolean lazyRebuildCompleted: boolean + /** Deferred embeds not yet landed (MT5) — the eventual-vector-index backlog. */ + pendingEmbeds: 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 @@ -10717,6 +10935,7 @@ export class Brainy implements BrainyInterface { return { initialized: false, lazyRebuildCompleted: this.lazyRebuildCompleted, + pendingEmbeds: this._pendingEmbedIds.size, disableAutoRebuild: this.config.disableAutoRebuild || false, migrating: false, rebuildFailed: this._indexRebuildFailed != null, @@ -10759,6 +10978,7 @@ export class Brainy implements BrainyInterface { return { initialized: this.initialized, lazyRebuildCompleted: this.lazyRebuildCompleted, + pendingEmbeds: this._pendingEmbedIds.size, 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 diff --git a/src/types/brainy.types.ts b/src/types/brainy.types.ts index cfd23d9f..2d4ff5e3 100644 --- a/src/types/brainy.types.ts +++ b/src/types/brainy.types.ts @@ -338,6 +338,20 @@ export interface AddParams { id?: string /** Pre-computed embedding vector (skips auto-embedding when provided) */ vector?: Vector + /** + * DEFER THE EMBEDDING (MT5, the deferred-embedding worker): the write + * acknowledges at durability — data + metadata persisted, a durable + * pending-embed marker written — and the embedding + vector-index insert + * run on the engine's single-flight background worker. HONEST SEMANTICS: + * the row is findable by id/metadata/path IMMEDIATELY; vector/semantic + * search sees it when the background embed completes (eventual vector + * index — `getIndexStatus().pendingEmbeds` counts the backlog, and + * `awaitPendingEmbeds()` is the barrier). CRASH-SAFE: markers persist + * before the ack and are recovered at the next open — a crash can DELAY + * a vector, never lose one. Refused (typed) together with `vector` — + * a supplied vector has nothing to defer. + */ + deferEmbedding?: boolean /** Multi-tenancy service identifier */ service?: string /** Type classification confidence (0-1) */ @@ -379,6 +393,15 @@ export interface AddParams { export interface UpdateParams { id: string // Entity to update data?: any // New content to re-embed + /** + * Defer the re-embedding of new `data` (see `AddParams.deferEmbedding`). + * The write acks at durability; the OLD vector keeps serving semantic + * search — stale-but-present, never absent (the flicker law) — until the + * background worker embeds the new content and swaps it in atomically. + * `data` reads return the NEW content immediately. Refused (typed) with + * an explicit `vector`. + */ + deferEmbedding?: boolean type?: NounType // Change type subtype?: string // Change subtype (set to '' or null-equivalent via dedicated unset is future work) /** diff --git a/src/utils/paramValidation.ts b/src/utils/paramValidation.ts index 359413d7..b8036746 100644 --- a/src/utils/paramValidation.ts +++ b/src/utils/paramValidation.ts @@ -540,6 +540,22 @@ function rejectForgedSystemKeys(metadata: Record | undefined, s export function validateAddParams(params: AddParams): void { rejectForgedSystemKeys(params.metadata as Record | undefined, 'add()') + // MT5 deferred embedding: an explicit vector has nothing to defer, and a + // deferral without data has nothing to embed — both are caller bugs that + // must refuse with the fix, never be silently reinterpreted. + if ((params as AddParams & { deferEmbedding?: boolean }).deferEmbedding === true) { + if (params.vector) { + throw new Error( + `add(): deferEmbedding cannot be combined with an explicit 'vector' — ` + + `the vector is already computed; drop one of the two.` + ) + } + if (!params.data) { + throw new Error( + `add(): deferEmbedding requires 'data' (the content the background worker will embed).` + ) + } + } // Universal truth: must have data or vector if (!params.data && !params.vector) { throw new Error( @@ -581,6 +597,19 @@ export function validateAddParams(params: AddParams): void { */ export function validateUpdateParams(params: UpdateParams): void { rejectForgedSystemKeys(params.metadata as Record | undefined, 'update()') + if ((params as UpdateParams & { deferEmbedding?: boolean }).deferEmbedding === true) { + if (params.vector) { + throw new Error( + `update(): deferEmbedding cannot be combined with an explicit 'vector' — ` + + `the vector is already computed; drop one of the two.` + ) + } + if (!params.data) { + throw new Error( + `update(): deferEmbedding requires new 'data' — without a data change there is nothing to re-embed.` + ) + } + } // Universal truth: must have an ID if (!params.id) { throw new Error('id is required for update') diff --git a/src/vfs/VirtualFileSystem.ts b/src/vfs/VirtualFileSystem.ts index 00bddefb..ed272109 100644 --- a/src/vfs/VirtualFileSystem.ts +++ b/src/vfs/VirtualFileSystem.ts @@ -694,6 +694,12 @@ export class VirtualFileSystem implements IVirtualFileSystem { await this.brain.update({ id: existingId, data: embeddingData, + // MT5: the caller's write acks at durability; the re-embed (a neural + // net — it dominated the measured 5.6s p50 per file write) runs on + // the background worker and swaps in atomically. Content is readable + // and metadata-findable immediately; semantic search converges when + // the embed lands (eventual vector index, the documented contract). + deferEmbedding: true, metadata }) @@ -729,6 +735,9 @@ export class VirtualFileSystem implements IVirtualFileSystem { data: embeddingData, // Always provide string for embeddings type: this.getFileNounType(mimeType), subtype: 'vfs-file', // Standard subtype for VFS file entities (7.30+) + // MT5: ack at durability; embedding backgrounds (see the overwrite + // branch note above). + deferEmbedding: true, metadata }) @@ -1117,6 +1126,9 @@ export class VirtualFileSystem implements IVirtualFileSystem { data: path, // Directory path as string content type: NounType.Collection, subtype: 'vfs-directory', // Standard subtype for VFS directory entities (7.30+) + // MT5: a directory creation on a write path must not wait on the + // embedder either — same ack-at-durability contract as file writes. + deferEmbedding: true, metadata }) diff --git a/tests/integration/deferred-embedding.test.ts b/tests/integration/deferred-embedding.test.ts new file mode 100644 index 00000000..819ddbad --- /dev/null +++ b/tests/integration/deferred-embedding.test.ts @@ -0,0 +1,171 @@ +/** + * @module tests/integration/deferred-embedding + * @description MT5 — THE DEFERRED-EMBEDDING CONTRACT (A3 of the service-class + * pair, BRAINY-PROD-LATENCY-TRIAD). The production disease: a VFS file write + * ran a neural network synchronously while the caller waited (5.6s p50 per + * small file). The contract pinned here: + * + * 1. ACK AT DURABILITY: a deferred write never calls the embedder on the + * caller's path — the row is id/metadata-findable immediately, with a + * durable pending marker and an honest `pendingEmbeds` gauge. + * 2. EVENTUAL VECTOR INDEX: `awaitPendingEmbeds()` is the barrier — after + * it, the vector is real, indexed, and the marker is reaped. + * 3. STALE-BEATS-ABSENT on deferred updates: the OLD vector keeps serving + * until the atomic swap (the flicker law, never a dark window). + * 4. CRASH-SAFE: markers survive a session that dies mid-defer; the next + * open recovers and lands the vector. A crash DELAYS a vector, never + * loses one. + * 5. TYPED REFUSALS: deferEmbedding + vector, and deferEmbedding without + * data, are caller bugs that refuse with the fix in the message. + */ +import { describe, it, expect, afterEach, vi } 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[] = [] + +async function memBrain(): Promise { + const b = new Brainy({ storage: { type: 'memory' }, requireSubtype: false }) + await b.init() + brains.push(b) + return b +} + +afterEach(async () => { + vi.restoreAllMocks() + for (const b of brains.splice(0)) await b.close().catch(() => {}) + for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }) +}) + +describe('MT5 — deferred embedding', () => { + it('ACK LAW: add({deferEmbedding}) never embeds on the caller path; row findable immediately; barrier lands the vector and reaps the marker', async () => { + const brain = await memBrain() + const embedSpy = vi.spyOn(brain, 'embed') + + const id = await brain.add({ + data: 'deferred content', + type: NounType.Document, + deferEmbedding: true, + metadata: { tag: 'deferred' } + }) + + // The caller's path never ran the embedder. + expect(embedSpy, 'no embed on the ack path').not.toHaveBeenCalled() + + // Immediately findable by metadata; vector is the stub; gauge honest. + const found = await brain.find({ where: { tag: 'deferred' }, limit: 5 }) + expect(found.map((r) => r.id)).toContain(id) + expect((await brain.getIndexStatus()).pendingEmbeds).toBeGreaterThanOrEqual(1) + + // The barrier: vector lands, marker reaped, index carries the row. + await brain.awaitPendingEmbeds() + expect(embedSpy).toHaveBeenCalled() + const after = await brain.get(id, { includeVectors: true }) + expect((after!.vector as number[]).length, 'real vector after the barrier').toBeGreaterThan(0) + expect(brain.pendingEmbedCount()).toBe(0) + expect((await brain.getIndexStatus()).pendingEmbeds).toBe(0) + }) + + it('STALE-BEATS-ABSENT: a deferred update serves the OLD vector until the atomic swap; data reads NEW immediately', async () => { + const brain = await memBrain() + const id = await brain.add({ data: 'original content', type: NounType.Document, metadata: {} }) + const before = await brain.get(id, { includeVectors: true }) + const oldVector = [...(before!.vector as number[])] + expect(oldVector.length).toBeGreaterThan(0) + + await brain.update({ id, data: 'completely different content', deferEmbedding: true }) + + // Data is new IMMEDIATELY; the vector is still the old one (present, + // never absent) until the worker swaps it. + const mid = await brain.get(id, { includeVectors: true }) + expect(mid!.data).toBe('completely different content') + expect(mid!.vector as number[], 'old vector keeps serving').toEqual(oldVector) + + await brain.awaitPendingEmbeds() + const after = await brain.get(id, { includeVectors: true }) + expect((after!.vector as number[]).length).toBeGreaterThan(0) + expect(after!.vector as number[], 'vector swapped after the barrier').not.toEqual(oldVector) + }) + + it('CRASH-SAFE: a session dying mid-defer leaves the durable marker; the next open recovers and lands the vector', async () => { + const dir = mkdtempSync(join(tmpdir(), 'brainy-defer-crash-')) + dirs.push(dir) + + // Session 1: the embedder hangs → the worker can never complete; close() + // does not wait for it (crash-equivalent for the embed leg). + let brain = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false }) + await brain.init() + brains.push(brain) + vi.spyOn(brain, 'embed').mockImplementation(() => new Promise(() => {})) + const id = await brain.add({ + data: 'survives the crash', + type: NounType.Document, + deferEmbedding: true, + metadata: { k: 1 } + }) + expect(brain.pendingEmbedCount()).toBe(1) + await brain.close() + brains.pop() + vi.restoreAllMocks() + + // Session 2: recovery lists the marker and resumes in the background. + brain = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false }) + await brain.init() + brains.push(brain) + expect(brain.pendingEmbedCount(), 'marker recovered at open').toBe(1) + + await brain.awaitPendingEmbeds() + const after = await brain.get(id, { includeVectors: true }) + expect((after!.vector as number[]).length, 'the delayed vector landed').toBeGreaterThan(0) + expect(brain.pendingEmbedCount()).toBe(0) + }, 120000) + + it('VFS ACK LAW: writeFile resolves even when the embedder HANGS forever — the ack never depends on a neural net', async () => { + const brain = await memBrain() + // The strongest form of the pin: an embedder that never resolves. If any + // part of the writeFile ack path awaited an embed, this test would hang. + // (The background worker legitimately picks the deferred embeds up later + // — it may even interleave on the event loop during writeFile's other + // awaits — but the CALLER'S promise must never depend on it.) + const hang = vi + .spyOn(brain, 'embed') + .mockImplementation(() => new Promise(() => {})) + + await brain.vfs.writeFile('/notes/today.md', '# The day\nA deferred capture.') + + // Acked with the embedder hung: content + metadata fully readable. + const content = await brain.vfs.readFile('/notes/today.md') + expect(content.toString()).toContain('A deferred capture.') + expect(brain.pendingEmbedCount()).toBeGreaterThanOrEqual(1) + + // Un-hang, abandon the poisoned in-flight 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), drain, verify. + hang.mockRestore() + ;(brain as unknown as { _embedWorkerFlight: Promise | null })._embedWorkerFlight = null + await brain.awaitPendingEmbeds() + expect(brain.pendingEmbedCount()).toBe(0) + }) + + it('TYPED REFUSALS: defer+vector and defer-without-data both refuse with the fix', async () => { + const brain = await memBrain() + await expect( + brain.add({ + data: 'x', + vector: new Array(384).fill(0.1), + type: NounType.Document, + deferEmbedding: true, + metadata: {} + }) + ).rejects.toThrow(/deferEmbedding cannot be combined/) + + const id = await brain.add({ data: 'y', type: NounType.Document, metadata: {} }) + await expect( + brain.update({ id, deferEmbedding: true, metadata: { z: 1 } }) + ).rejects.toThrow(/requires new 'data'/) + }) +})