open-brainy/tests/integration/canonical-count-ledger.test.ts

/**
 * @module tests/integration/canonical-count-ledger
 * @description The canonical count ledger — the denominators a derived-index
 * provider's coverage ledger subtracts from. Laws under test:
 * (1) THE ALL-VISIBILITY SCALAR IS THE UNFILTERED WALK'S TOTAL — the
 *     storage-level `getNouns()` / `getVerbs()` `totalCount` counts EVERY tier
 *     (system, internal, public) because the walk yields every tier; the
 *     user-facing `getNounCount()` / `getVerbCount()` keep skipping hidden
 *     tiers. A ledger built on the user-facing scalar would read "over-posted"
 *     on every store with a VFS — the mismatch this pin makes unbuildable.
 * (2) NEVER CLAMPED — `Math.max(scalar, scanned)` could only move a scalar up,
 *     so an inflated counter hid forever. An inflated scalar is now VISIBLE
 *     (totalCount ≠ walk) and the sanctioned recount heals it, durably.
 * (3) NEVER GUESSED — a delete that cannot prove the record existed marks the
 *     ledger SUSPECT (persisted) instead of decrementing on faith; the recount
 *     clears the flag with proof.
 * (4) LEGACY FILES DERIVE ONCE — a counts.json written before the ledger is
 *     upgraded from the canonical id tree at open, then persisted.
 */
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'
import * as fs from 'node:fs'
import * as os from 'node:os'
import * as path from 'node:path'
import * as zlib from 'node:zlib'
import { Brainy } from '../../src/index.js'

/** Count canonical `<root>/entities/<kind>/<shard>/<id>` directories — every tier. */
function countIdDirs(root: string, kind: 'nouns' | 'verbs'): number {
  const base = path.join(root, 'entities', kind)
  if (!fs.existsSync(base)) return 0
  let n = 0
  for (const shard of fs.readdirSync(base)) {
    const shardDir = path.join(base, shard)
    if (!fs.statSync(shardDir).isDirectory()) continue
    for (const id of fs.readdirSync(shardDir)) {
      if (fs.statSync(path.join(shardDir, id)).isDirectory()) n++
    }
  }
  return n
}

const countsPath = (root: string) => path.join(root, '_system', 'counts.json')

describe('canonical count ledger — ALL-visibility scalars, unclamped totals, recount heals', () => {
  let dir: string
  let brain: any

  const open = async () => {
    const b: any = new Brainy({
      requireSubtype: false,
      storage: { type: 'filesystem', path: dir },
      silent: true,
      dimensions: 384
    })
    await b.init()
    return b
  }

  beforeEach(async () => {
    process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-ledger-'))
    brain = await open()
  })
  afterEach(async () => {
    await brain.close?.().catch(() => {})
    fs.rmSync(dir, { recursive: true, force: true })
  })

  it('the unfiltered walk totalCount is the ALL scalar (every tier); the user-facing count stays counted', async () => {
    const a = await brain.add({ data: 'public a', type: 'document' })
    const b = await brain.add({ data: 'internal b', type: 'document', visibility: 'internal' })
    await brain.relate({ from: a, to: b, type: 'relatedTo', visibility: 'internal' })
    await brain.vfs.writeFile('/docs/x.txt', 'hello') // VFS: system-tier nouns + Contains edges
    await brain.flush()

    const ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.suspect).toBe(false)
    expect(ledger.nouns.all).toBe(countIdDirs(dir, 'nouns'))
    expect(ledger.verbs.all).toBe(countIdDirs(dir, 'verbs'))
    expect(ledger.nouns.counted).toBe(await brain.storage.getNounCount())
    expect(ledger.verbs.counted).toBe(await brain.storage.getVerbCount())
    // Hidden tiers exist (the VFS root at minimum, the internal noun, the internal edge):
    expect(ledger.nouns.all).toBeGreaterThan(ledger.nouns.counted)
    expect(ledger.verbs.all).toBeGreaterThan(ledger.verbs.counted)

    // The storage-level unfiltered walks report the ALL scalar, and a full page equals it.
    const nouns = await brain.storage.getNouns({ pagination: { limit: 1000, offset: 0 } })
    expect(nouns.totalCount).toBe(ledger.nouns.all)
    expect(nouns.items.length).toBe(ledger.nouns.all)
    const verbs = await brain.storage.getVerbs({ pagination: { limit: 1000, offset: 0 } })
    expect(verbs.totalCount).toBe(ledger.verbs.all)
    expect(verbs.items.length).toBe(ledger.verbs.all)
  })

  it('proven deletes move the ALL scalar for every tier and the ledger stays exact and unsuspect', async () => {
    const p = await brain.add({ data: 'public p', type: 'document' })
    const q = await brain.add({ data: 'internal q', type: 'document', visibility: 'internal' })
    await brain.relate({ from: p, to: q, type: 'relatedTo' })
    await brain.flush()
    const before = await brain.storage.getCanonicalCounts()

    await brain.remove(q) // cascades the edge
    await brain.remove(p)
    await brain.flush()

    const after = await brain.storage.getCanonicalCounts()
    expect(after.nouns.all).toBe(before.nouns.all - 2)
    expect(after.verbs.all).toBe(before.verbs.all - 1)
    expect(after.nouns.all).toBe(countIdDirs(dir, 'nouns'))
    expect(after.verbs.all).toBe(countIdDirs(dir, 'verbs'))
    expect(after.nouns.counted).toBe(before.nouns.counted - 1)
    expect(after.suspect).toBe(false)
  })

  it('a legacy counts.json without the ALL keys is derived once from the id tree and persisted', async () => {
    await brain.add({ data: 'one', type: 'document' })
    await brain.add({ data: 'two', type: 'document', visibility: 'internal' })
    await brain.vfs.writeFile('/a.txt', 'x')
    await brain.flush()
    await brain.close()

    const raw = JSON.parse(fs.readFileSync(countsPath(dir), 'utf-8'))
    expect(typeof raw.totalNounCountAll).toBe('number')
    delete raw.totalNounCountAll
    delete raw.totalVerbCountAll
    delete raw.allCountsSuspect
    fs.writeFileSync(countsPath(dir), JSON.stringify(raw, null, 2))

    brain = await open()
    const ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.nouns.all).toBe(countIdDirs(dir, 'nouns'))
    expect(ledger.verbs.all).toBe(countIdDirs(dir, 'verbs'))
    expect(ledger.suspect).toBe(false)
    const persisted = JSON.parse(fs.readFileSync(countsPath(dir), 'utf-8'))
    expect(persisted.totalNounCountAll).toBe(ledger.nouns.all)
    expect(persisted.totalVerbCountAll).toBe(ledger.verbs.all)
  })

  it('an inflated ALL scalar is VISIBLE (unclamped) and healed by repairIndex(), surviving reopen', async () => {
    for (let i = 0; i < 3; i++) await brain.add({ data: `real ${i}`, type: 'document' })
    await brain.flush()
    const truth = countIdDirs(dir, 'nouns')

    ;(brain.storage as any).totalNounCountAll = truth + 40
    await (brain.storage as any).persistCounts()
    await brain.close()
    brain = await open()

    // The lie survives reopen AND is observable: totalCount disagrees with the walk.
    const page = await brain.storage.getNouns({ pagination: { limit: 1000, offset: 0 } })
    expect(page.totalCount).toBe(truth + 40)
    expect(page.items.length).toBe(truth)

    await brain.repairIndex()
    expect((await brain.storage.getCanonicalCounts()).nouns.all).toBe(truth)
    expect((await brain.storage.getNouns({ pagination: { limit: 1000, offset: 0 } })).totalCount).toBe(truth)

    await brain.close()
    brain = await open()
    expect((await brain.storage.getCanonicalCounts()).nouns.all).toBe(truth)
  })

  it('an unprovable delete marks the ledger SUSPECT (persisted); the recount clears it with proof', async () => {
    await brain.add({ data: 'anchor', type: 'document' })
    await brain.flush()
    const truth = countIdDirs(dir, 'nouns')

    // A ghost: no canonical record, no prior image — nothing to prove existence with.
    await brain.storage.deleteNounMetadata('00000000-dead-4dea-8dea-000000000000')
    let ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.suspect).toBe(true)
    expect(ledger.nouns.all).toBe(truth) // never decremented on faith

    await brain.close()
    brain = await open()
    expect((await brain.storage.getCanonicalCounts()).suspect).toBe(true) // the flag persists

    await brain.repairIndex()
    ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.suspect).toBe(false)
    expect(ledger.nouns.all).toBe(truth)
  })
})

/** Count `<root>/entities/nouns/<shard>/<id>/vectors.json[.gz]` files holding a non-empty `vector`. */
function countVectoredNouns(root: string): number {
  const base = path.join(root, 'entities', 'nouns')
  if (!fs.existsSync(base)) return 0
  let n = 0
  for (const shard of fs.readdirSync(base)) {
    const shardDir = path.join(base, shard)
    if (!fs.statSync(shardDir).isDirectory()) continue
    for (const id of fs.readdirSync(shardDir)) {
      const idDir = path.join(shardDir, id)
      if (!fs.statSync(idDir).isDirectory()) continue
      const plainPath = path.join(idDir, 'vectors.json')
      const gzPath = `${plainPath}.gz`
      let record: any = null
      if (fs.existsSync(plainPath)) {
        record = JSON.parse(fs.readFileSync(plainPath, 'utf-8'))
      } else if (fs.existsSync(gzPath)) {
        record = JSON.parse(zlib.gunzipSync(fs.readFileSync(gzPath)).toString('utf-8'))
      } else {
        continue
      }
      if (Array.isArray(record.vector) && record.vector.length > 0) n++
    }
  }
  return n
}

describe('canonical count ledger — the vectored-noun scalar (the vector leg\'s coverage denominator)', () => {
  let dir: string
  let brain: any

  const open = async () => {
    const b: any = new Brainy({
      requireSubtype: false,
      storage: { type: 'filesystem', path: dir },
      silent: true,
      dimensions: 384
    })
    await b.init()
    return b
  }

  /** Baseline vectored count right after a fresh open() — init() creates a
   *  hidden system VFS-root noun that itself carries a real vector, so a
   *  brand-new store's `vectors.all` is 1, not 0. Tests assert DELTAS off
   *  this baseline rather than hardcoding it away. */
  let baseline: number

  beforeEach(async () => {
    process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-vectored-ledger-'))
    brain = await open()
    baseline = (await brain.storage.getCanonicalCounts()).vectors.all
  })
  afterEach(async () => {
    vi.restoreAllMocks()
    await brain.close?.().catch(() => {})
    fs.rmSync(dir, { recursive: true, force: true })
  })

  it('an explicit-vector add counts immediately; the ledger matches the on-disk vectors.json content', async () => {
    await brain.add({ data: 'a', type: 'document', vector: Array(384).fill(0).map((_, i) => Math.sin(i)) })
    await brain.add({ data: 'b', type: 'document' }) // embedded (non-deferred) — also a real vector
    await brain.flush()

    const ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.vectors.all).toBe(baseline + 2)
    expect(ledger.vectors.all).toBe(countVectoredNouns(dir))
    expect(ledger.suspect).toBe(false)
  })

  it('a deferred-embed add does NOT count until its embed LANDS', async () => {
    // Hold the background worker's embed call open under manual control — a
    // deterministic embedder is fast enough that the landing could otherwise
    // race ahead of the "still unlanded" assertion below.
    let resolveEmbed: ((v: number[]) => void) | undefined
    vi.spyOn(brain, 'embed').mockImplementation(
      () => new Promise<number[]>((resolve) => { resolveEmbed = resolve })
    )

    const id = await brain.add({ data: 'deferred content', type: 'document', deferEmbedding: true })
    await brain.flush()

    // Landed nothing yet — the ledger must not count the stub.
    let ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.vectors.all).toBe(baseline)
    expect(ledger.vectors.all).toBe(countVectoredNouns(dir))

    // Release the held embed, then cross the barrier: the vector lands
    // (system:embed-landing).
    resolveEmbed!(Array(384).fill(0).map((_, i) => Math.cos(i)))
    await brain.awaitPendingEmbeds()
    const landed = await brain.get(id, { includeVectors: true })
    expect((landed!.vector as number[]).length).toBeGreaterThan(0)

    ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.vectors.all).toBe(baseline + 1)
    expect(ledger.vectors.all).toBe(countVectoredNouns(dir))
    expect(ledger.suspect).toBe(false)
  })

  it('a proven delete of a vectored noun decrements; a non-vectored (unlanded) delete does not', async () => {
    const vectoredId = await brain.add({ data: 'v', type: 'document' }) // real embed, unmocked
    // Block the embed worker AFTER the real add above — a deterministic
    // embedder is fast enough that the deferred noun below could otherwise
    // land before this test observes its "still unlanded" state.
    vi.spyOn(brain, 'embed').mockImplementation(() => new Promise(() => {}))
    const deferredId = await brain.add({ data: 'd', type: 'document', deferEmbedding: true })
    await brain.flush()
    expect((await brain.storage.getCanonicalCounts()).vectors.all).toBe(baseline + 1)

    await brain.remove(vectoredId)
    await brain.flush()
    let ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.vectors.all).toBe(baseline)
    expect(ledger.suspect).toBe(false)

    await brain.remove(deferredId) // never had a real vector — no decrement, still unsuspect
    await brain.flush()
    ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.vectors.all).toBe(baseline)
    expect(ledger.suspect).toBe(false)
    expect(ledger.vectors.all).toBe(countVectoredNouns(dir))
  })

  it('the recount corrects a tampered vectors.all scalar, surviving reopen', async () => {
    await brain.add({ data: 'real 1', type: 'document' })
    await brain.add({ data: 'real 2', type: 'document' })
    await brain.flush()
    const truth = countVectoredNouns(dir)
    expect(truth).toBe(baseline + 2)

    ;(brain.storage as any).totalVectoredNounCount = truth + 40
    await (brain.storage as any).persistCounts()
    await brain.close()
    brain = await open()

    // The lie survives reopen (never clamped).
    expect((await brain.storage.getCanonicalCounts()).vectors.all).toBe(truth + 40)

    await brain.repairIndex()
    expect((await brain.storage.getCanonicalCounts()).vectors.all).toBe(truth)

    await brain.close()
    brain = await open()
    expect((await brain.storage.getCanonicalCounts()).vectors.all).toBe(truth)
  })

  it('a legacy counts.json without totalVectoredNounCount is derived once from vectors.json content and persisted', async () => {
    await brain.add({ data: 'one', type: 'document' }) // real embed, unmocked
    // Block the embed worker AFTER the real add above — a deterministic
    // embedder is fast enough that the deferred noun below could otherwise
    // land before close(), which would inflate this test's expected count.
    vi.spyOn(brain, 'embed').mockImplementation(() => new Promise(() => {}))
    await brain.add({ data: 'two deferred', type: 'document', deferEmbedding: true })
    await brain.flush()
    await brain.close()

    const countsPath = path.join(dir, '_system', 'counts.json')
    const raw = JSON.parse(fs.readFileSync(countsPath, 'utf-8'))
    expect(typeof raw.totalVectoredNounCount).toBe('number')
    delete raw.totalVectoredNounCount
    fs.writeFileSync(countsPath, JSON.stringify(raw, null, 2))

    brain = await open()
    const ledger = await brain.storage.getCanonicalCounts()
    expect(ledger.vectors.all).toBe(baseline + 1) // the root + the one non-deferred noun
    expect(ledger.vectors.all).toBe(countVectoredNouns(dir))
    const persisted = JSON.parse(fs.readFileSync(countsPath, 'utf-8'))
    expect(persisted.totalVectoredNounCount).toBe(baseline + 1)
  })
})