open-brainy/tests/integration/embed-markers-in-log.test.ts

/**
 * @module tests/integration/embed-markers-in-log
 * @description DEFERRED-EMBED MARKERS ARE LOG RECORDS — the sidecar is dead.
 * The pending-embed lifecycle lives IN the generation log as first-class v2
 * records: `embed.pending` rides the deferred write's OWN commit fact (same
 * generation, one atomic append — a marker can never be orphaned from its
 * write nor the write from its marker) and `embed.landed` rides the
 * background worker's landing commit. Recovery is REPLAY, NOT LISTING: the
 * open-time fold arms every pending without a matching landed (minus rows
 * the log later tombstoned). The pins:
 *
 * (a) SAME-FACT ATOMICITY: a deferred add's commit fact carries the
 *     embed.pending record BESIDE its noun after-image — one generation,
 *     one frame — and no sidecar file is ever written.
 * (b) LANDING: after the barrier, the log carries embed.landed (inline
 *     vector, per the v2 format) riding the landing commit's own fact, and
 *     a fresh fold of the whole log nets ZERO pending.
 * (c) CRASH RECOVERY VIA THE LOG: kill mid-defer (hung embedder, flushed
 *     durability, crash-style abandon), reopen — the fold re-arms exactly
 *     one pending with NO sidecar file existing anywhere, and the vector
 *     then lands.
 * (d) LEGACY BRIDGE: a sidecar marker file left by a pre-log build is
 *     folded in at open, migrated into the log as an embed.pending record,
 *     and the file is deleted — one-time, durable, idempotent.
 * (e) VFS ACK LAW (unchanged contract, new mechanism): writeFile acks
 *     under a forever-hung embedder while its pending marker sits durably
 *     in the log.
 */
import { describe, it, expect, afterEach, vi } from 'vitest'
import * as fs from 'node:fs'
import * as path from 'node:path'
import * as zlib from 'node:zlib'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import type { CommitFact } from '../../src/db/factLog.js'
import {
  makeTempDir,
  openBrain,
  abandonAsCrashed,
  vec,
  uid
} from '../helpers/durabilityKillMatrix.js'

/** The retired sidecar prefix — asserted ABSENT (or bridged away) on disk. */
const SIDECAR_DIR = ['_system', 'pending_embeds'] as const

const sidecarDir = (dir: string): string => path.join(dir, ...SIDECAR_DIR)

/** Every committed fact in the brain's log, generation-ascending. */
async function allFacts(brain: Brainy): Promise<CommitFact[]> {
  const scan = (
    brain as unknown as {
      scanFacts(o?: { fromGeneration?: number }): {
        batches(): AsyncGenerator<{ facts: CommitFact[] }>
      } | null
    }
  ).scanFacts({ fromGeneration: 1 })
  expect(scan, 'filesystem storage hosts a fact log').not.toBeNull()
  const facts: CommitFact[] = []
  for await (const batch of scan!.batches()) facts.push(...batch.facts)
  return facts
}

/** The recovery fold, reimplemented independently: pending arms, landed
 *  disarms, a noun tombstone disarms (a deleted row owes no vector). */
function foldPending(facts: CommitFact[]): Set<string> {
  const pending = new Set<string>()
  for (const fact of facts) {
    for (const record of fact.records ?? []) {
      if (record.type === 'embed.pending') pending.add(record.id)
      else if (record.type === 'embed.landed') pending.delete(record.id)
    }
    for (const op of fact.ops) {
      if (op.kind === 'noun' && op.record === null) pending.delete(op.id)
    }
  }
  return pending
}

/** Hang the embedder forever (the ack-law adversary). */
function hangEmbedder(brain: Brainy): ReturnType<typeof vi.spyOn> {
  return vi
    .spyOn(brain as unknown as { embed(d: unknown): Promise<number[]> }, 'embed')
    .mockImplementation(() => new Promise<number[]>(() => {}))
}

/** Abandon a hung worker pass (its embed promise never resolves; production
 *  is covered by the worker's 60s hang guard — the test takes the white-box
 *  shortcut for speed, same idiom as the deferred-embedding suite). */
function abandonHungWorker(brain: Brainy): void {
  ;(brain as unknown as { _embedWorkerFlight: Promise<void> | null })._embedWorkerFlight = null
}

describe('deferred-embed markers in the log — the sidecar is dead', () => {
  const dirs: string[] = []
  const brains: Brainy[] = []

  const trackDir = (): string => {
    const dir = makeTempDir()
    dirs.push(dir)
    return dir
  }
  const track = (brain: Brainy): Brainy => {
    brains.push(brain)
    return brain
  }

  afterEach(async () => {
    vi.restoreAllMocks()
    for (const b of brains.splice(0)) {
      abandonHungWorker(b)
      await b.close().catch(() => {})
    }
    for (const d of dirs.splice(0)) fs.rmSync(d, { recursive: true, force: true })
  })

  it('(a) SAME-FACT ATOMICITY: the deferred add\'s ONE commit fact carries embed.pending beside its after-image; no sidecar file exists', async () => {
    const dir = trackDir()
    const brain = track(await openBrain(dir))
    hangEmbedder(brain) // hold the pending state open for the scan

    const id = await brain.add({
      data: 'deferred content whose marker rides the fact',
      type: NounType.Document,
      deferEmbedding: true,
      metadata: { pin: 'a' }
    })
    expect(brain.pendingEmbedCount()).toBe(1)

    const facts = await allFacts(brain)
    const carrying = facts.filter((f) =>
      (f.records ?? []).some((r) => r.type === 'embed.pending' && r.id === id)
    )
    expect(carrying, 'exactly ONE fact carries the pending marker').toHaveLength(1)
    const fact = carrying[0]
    // The SAME fact (same generation, one atomic append) carries the write's
    // own after-image — marker and write are inseparable by construction.
    const afterImage = fact.ops.find((op) => op.kind === 'noun' && op.id === id)
    expect(afterImage, 'the marker rides the write\'s own fact').toBeDefined()
    expect(afterImage!.record, 'an after-image, not a tombstone').not.toBeNull()
    const marker = (fact.records ?? []).find((r) => r.type === 'embed.pending' && r.id === id)
    expect(marker && marker.type === 'embed.pending' && marker.enqueuedAt).toBeGreaterThan(0)

    // The sidecar is dead: nothing under the retired prefix, ever.
    expect(fs.existsSync(sidecarDir(dir)), 'no sidecar directory is created').toBe(false)
  })

  it('(b) LANDING: after the barrier the log carries embed.landed (inline vector) on the landing commit\'s own fact, and a fresh fold nets zero pending', async () => {
    const dir = trackDir()
    const brain = track(await openBrain(dir))

    const id = await brain.add({
      data: 'content that lands in the background',
      type: NounType.Document,
      deferEmbedding: true,
      metadata: { pin: 'b' }
    })
    await brain.awaitPendingEmbeds()
    expect(brain.pendingEmbedCount()).toBe(0)

    const facts = await allFacts(brain)
    const landingFacts = facts.filter((f) =>
      (f.records ?? []).some((r) => r.type === 'embed.landed' && r.id === id)
    )
    expect(landingFacts, 'exactly ONE landing fact').toHaveLength(1)
    const landed = (landingFacts[0].records ?? []).find(
      (r) => r.type === 'embed.landed' && r.id === id
    )
    expect(landed && landed.type === 'embed.landed' && landed.vector.length).toBeGreaterThan(0)
    // The landing commit's own after-image rides the same fact — the worker's
    // vector swap and its durable "pending consumed" are one atomic append.
    const landingAfterImage = landingFacts[0].ops.find((op) => op.kind === 'noun' && op.id === id)
    expect(landingAfterImage, 'the landed marker rides the swap\'s own fact').toBeDefined()
    expect(landingAfterImage!.record).not.toBeNull()

    // A fresh fold of the WHOLE log — the exact recovery computation — nets zero.
    expect(foldPending(facts).size).toBe(0)
    expect(fs.existsSync(sidecarDir(dir))).toBe(false)
  })

  it('(c) CRASH RECOVERY VIA THE LOG: kill mid-defer, reopen — one pending re-armed from the fold, NO sidecar file anywhere, and the vector then lands', async () => {
    const dir = trackDir()

    // Session 1: embedder hung, deferred add acked, durability flushed, then
    // a crash-style abandon (RAM gone, no close, no background machinery).
    const first = await openBrain(dir)
    brains.push(first)
    hangEmbedder(first)
    const id = await first.add({
      data: 'survives the kill through the log',
      type: NounType.Document,
      deferEmbedding: true,
      metadata: { pin: 'c' }
    })
    expect(first.pendingEmbedCount()).toBe(1)
    await first.flush() // the durability barrier: fact (with marker) + manifest
    expect(fs.existsSync(sidecarDir(dir)), 'no sidecar before the kill').toBe(false)
    await abandonAsCrashed(first)
    brains.splice(brains.indexOf(first), 1)
    vi.restoreAllMocks()

    // Session 2: recovery folds the log — embedder hung BEFORE init so the
    // re-armed pending is observable, not raced away by the fast worker.
    const second = new Brainy({
      requireSubtype: false,
      storage: { type: 'filesystem', path: dir },
      silent: true,
      persistence: { policy: 'manual' }
    })
    const hang = hangEmbedder(second)
    await second.init()
    track(second)
    expect(second.pendingEmbedCount(), 'the fold re-armed the pending').toBe(1)
    expect(fs.existsSync(sidecarDir(dir)), 'recovery used the LOG, not files').toBe(false)

    // Un-hang and drain: a crash DELAYED the vector, never lost it.
    hang.mockRestore()
    abandonHungWorker(second)
    await second.awaitPendingEmbeds()
    expect(second.pendingEmbedCount()).toBe(0)
    const after = await second.get(id, { includeVectors: true })
    expect(after, 'the deferred row survived the crash').toBeTruthy()
    expect((after!.vector as number[]).length, 'the delayed vector landed').toBeGreaterThan(0)
    expect(foldPending(await allFacts(second)).size, 'the landing is durable in the log').toBe(0)
  })

  it('(d) LEGACY BRIDGE: a pre-log sidecar marker folds in at open, migrates into the log, and the file dies — one-time and durable', async () => {
    const dir = trackDir()

    // Session 1: a normal committed row (the entity the legacy marker names).
    const first = await openBrain(dir)
    brains.push(first)
    const id = uid('legacy-defer')
    await first.add({
      id,
      data: 'legacy deferred content',
      type: NounType.Document,
      vector: vec(9),
      metadata: { pin: 'd' }
    })
    await first.flush()
    await first.close()
    brains.splice(brains.indexOf(first), 1)

    // A pre-log build's sidecar marker, hand-written exactly as the old
    // writeRawObject persisted it (the filesystem adapter compresses raw
    // objects by default: gzipped JSON at `<path>.gz`).
    fs.mkdirSync(sidecarDir(dir), { recursive: true })
    const sidecarFile = path.join(sidecarDir(dir), id)
    fs.writeFileSync(
      `${sidecarFile}.gz`,
      zlib.gzipSync(JSON.stringify({ id, enqueuedAt: 1234567890 }, null, 2))
    )

    // Session 2: the bridge fires at open. Embedder hung BEFORE init so the
    // folded pending is observable.
    const second = new Brainy({
      requireSubtype: false,
      storage: { type: 'filesystem', path: dir },
      silent: true,
      persistence: { policy: 'manual' }
    })
    const hang = hangEmbedder(second)
    await second.init()
    track(second)
    expect(second.pendingEmbedCount(), 'the legacy marker folded in').toBe(1)
    expect(fs.existsSync(sidecarFile), 'the sidecar file was deleted').toBe(false)
    expect(fs.existsSync(`${sidecarFile}.gz`), 'the compressed variant too').toBe(false)
    const migrated = await allFacts(second)
    expect(
      migrated.some((f) => (f.records ?? []).some((r) => r.type === 'embed.pending' && r.id === id)),
      'the marker now lives IN the log'
    ).toBe(true)

    // Drain: the bridged pending embeds and lands like any other.
    hang.mockRestore()
    abandonHungWorker(second)
    await second.awaitPendingEmbeds()
    expect(second.pendingEmbedCount()).toBe(0)
    const facts = await allFacts(second)
    expect(
      facts.some((f) => (f.records ?? []).some((r) => r.type === 'embed.landed' && r.id === id)),
      'the bridged pending landed durably'
    ).toBe(true)
    expect(foldPending(facts).size).toBe(0)
    await second.flush()
    await second.close()
    brains.splice(brains.indexOf(second), 1)

    // Session 3: nothing resurrects — the bridge was one-time, the clear durable.
    const third = track(await openBrain(dir))
    expect(third.pendingEmbedCount(), 'no zombie pending on the next open').toBe(0)
    expect(fs.existsSync(sidecarDir(dir)) && fs.readdirSync(sidecarDir(dir)).length > 0).toBe(false)
  })

  it('(e) VFS ACK LAW: writeFile acks under a forever-hung embedder while its pending marker sits durably in the log', async () => {
    const dir = trackDir()
    const brain = track(await openBrain(dir))
    const hang = hangEmbedder(brain)

    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)

    // The marker is already durable IN the log while the embedder hangs —
    // the exact state a crash here would recover from.
    expect(foldPending(await allFacts(brain)).size).toBeGreaterThanOrEqual(1)
    expect(fs.existsSync(sidecarDir(dir))).toBe(false)

    // Un-hang, abandon the poisoned pass, drain, verify.
    hang.mockRestore()
    abandonHungWorker(brain)
    await brain.awaitPendingEmbeds()
    expect(brain.pendingEmbedCount()).toBe(0)
    expect(foldPending(await allFacts(brain)).size).toBe(0)
  })
})