fix(aggregation): the lifecycle cluster — flush stamps, behind-stamp catches up incrementally, the native rebuild finally gets invoked, deletes are never silently skipped

SELF-ENGINE-LIFECYCLE-SPRINT + BRAINY-PROD-LATENCY-TRIAD, the four asks:

(a) brain.flush() persists aggregation state stamped at the committed
    generation. The stamp used to advance only at close(), so a long-lived
    writer that flushes but never closes — the primary production shape —
    left every write window behind the stamp, and ANY unclean exit forced a
    whole-store backfill walk (per-entity work, measured >60s and
    door-starving on a 9k-row production brain) on the first stats call.

(b) BEHIND-stamp adoption becomes adopt + INCREMENTAL CATCH-UP: the exact
    missing window (stamp, committed] resolves its affected-id set from the
    fact log and reconciles each entity with time-travel before/after reads
    (asOf at both window bounds) through the same delta algebra the live
    hooks use — cost bounded by writes since the last flush, never store
    size, and exact under interleaving because reconciliation targets the
    FIXED window end while later writes chain through hooks. Oversized
    windows (>5000 affected) and unreadable windows demote to the announced
    rescan — never a silent partial serve.

(c) The native provider's parallel rebuildAggregate — on the contract since
    8.x but never invoked anywhere — is now the backfill walk's preferred
    door: one call per aggregate with source-matched entities, replacing
    the per-entity FFI stream.

(d) A delete whose before-image is unavailable can no longer SKIP the
    aggregation hook silently (counts drifted upward forever): both delete
    paths (remove() and transact) flag an exact rescan, loudly.

Pins: integration (flush stamp; unclean-exit reopen → exact counts through
an add + group-move + delete window with the walk spy proving ZERO
whole-store walks) + unit (provider rebuild invoked once with filtered
entities; flagAllForRescan; reconcile delta algebra). Gates: unit 1913/1913
· integration 760 · conformance 27/27.

tests/integration/aggregation-lifecycle-catchup.test.ts

@@ -0,0 +1,143 @@
+/**
+ * @module tests/integration/aggregation-lifecycle-catchup
+ * @description THE AGGREGATION LIFECYCLE PINS (SELF-ENGINE-LIFECYCLE-SPRINT /
+ * BRAINY-PROD-LATENCY-TRIAD asks (a)+(b)). The production disease: the
+ * aggregation stamp persisted ONLY at close(), so a long-lived writer that
+ * flushes but never closes left its stamp behind after every write window —
+ * and the exact-match adoption rule then forced a WHOLE-STORE backfill walk
+ * (per-entity work, measured >60s and door-starving on a 9k-row production
+ * brain) on the first stats call after any unclean exit.
+ *
+ * The cures pinned here:
+ * (a) `brain.flush()` persists aggregation state, stamped at the committed
+ *     generation — the stamp tracks every flush, not just close().
+ * (b) BEHIND-stamp state is ADOPTED and reconciled INCREMENTALLY over its
+ *     exact missing window (fact-log affected ids + time-travel before/after
+ *     reads) — the full walk never runs for an unclean exit. Pinned by call
+ *     shape (the walk spy), not by latency.
+ */
+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 AGG = {
+  name: 'by_subtype',
+  source: { type: NounType.Document },
+  groupBy: ['system.subtype'] as string[],
+  metrics: { count: { op: 'count' as const } }
+}
+
+const dirs: string[] = []
+const brains: Brainy[] = []
+
+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
+}
+
+function countFor(results: Array<{ groupKey: Record<string, unknown>; metrics: Record<string, unknown> }>, subtype: string): number {
+  const row = results.find(r => r.groupKey['system.subtype'] === subtype)
+  return row ? Number(row.metrics.count) : 0
+}
+
+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 })
+})
+
+describe('aggregation lifecycle — flush stamps, behind-stamp catches up incrementally', () => {
+  it('(a) brain.flush() persists aggregation state stamped at the committed generation', async () => {
+    const dir = mkdtempSync(join(tmpdir(), 'brainy-agg-flush-'))
+    dirs.push(dir)
+    const brain = await open(dir)
+    brain.defineAggregate(AGG)
+    await brain.add({ data: 'a', type: NounType.Document, subtype: 'invoice', metadata: {} })
+    await brain.add({ data: 'b', type: NounType.Document, subtype: 'invoice', metadata: {} })
+    await brain.queryAggregate(AGG.name) // settle backfill-on-define
+
+    await brain.flush()
+
+    const internals = brain as unknown as {
+      storage: {
+        getMetadata(k: string): Promise<{ sourceGeneration?: number } | null>
+        committedGeneration?(): number
+      }
+    }
+    const persisted = await internals.storage.getMetadata('__aggregation_state_by_subtype__')
+    expect(persisted, 'state persisted by flush(), not only close()').toBeTruthy()
+    expect(
+      persisted!.sourceGeneration,
+      'stamp equals the committed generation at flush time'
+    ).toBe(internals.storage.committedGeneration?.())
+  })
+
+  it('(b) an unclean exit reconciles incrementally — exact counts, ZERO full-store walks', async () => {
+    const dir = mkdtempSync(join(tmpdir(), 'brainy-agg-catchup-'))
+    dirs.push(dir)
+
+    // Session 1: define + write + flush (stamps at G), then MORE writes of
+    // every kind (add / update-that-moves-groups / delete) and a clean close
+    // — but we then REWIND the persisted aggregation artifact to its at-G
+    // bytes, which is byte-for-byte the unclean-exit state: stamp G, store
+    // committed at G+k.
+    let brain = await open(dir)
+    brain.defineAggregate(AGG)
+    await brain.add({ data: 'a', type: NounType.Document, subtype: 'invoice', metadata: {} })
+    await brain.add({ data: 'b', type: NounType.Document, subtype: 'invoice', metadata: {} })
+    const moving = await brain.add({ data: 'c', type: NounType.Document, subtype: 'draft', metadata: {} })
+    const doomed = await brain.add({ data: 'd', type: NounType.Document, subtype: 'draft', metadata: {} })
+    await brain.queryAggregate(AGG.name)
+    await brain.flush()
+
+    const internals = brain as unknown as {
+      storage: {
+        getMetadata(k: string): Promise<Record<string, unknown> | null>
+        saveMetadata(k: string, v: Record<string, unknown>): Promise<void>
+      }
+    }
+    const stateAtG = JSON.parse(
+      JSON.stringify(await internals.storage.getMetadata('__aggregation_state_by_subtype__'))
+    )
+
+    // The missing window: one add, one group-moving update, one delete.
+    await brain.add({ data: 'e', type: NounType.Document, subtype: 'invoice', metadata: {} })
+    await brain.update({ id: moving, subtype: 'invoice' })
+    await brain.remove(doomed)
+    await brain.close()
+    brains.pop()
+
+    // Rewind the aggregation artifact to the at-G bytes (the unclean exit).
+    {
+      const reopenForRewind = await open(dir)
+      const rw = reopenForRewind as unknown as typeof internals
+      await rw.storage.saveMetadata('__aggregation_state_by_subtype__', stateAtG)
+      await reopenForRewind.close()
+      brains.pop()
+    }
+
+    // Session 2: reopen — adoption must see BEHIND and reconcile, never walk.
+    brain = await open(dir)
+    brain.defineAggregate(AGG)
+    const walkSpy = vi.spyOn(
+      brain as unknown as { runAggregationBackfillWalk(): Promise<void> },
+      'runAggregationBackfillWalk'
+    )
+
+    const results = await brain.queryAggregate(AGG.name)
+
+    // Ground truth after the window: invoice = a,b,e + moved c = 4; draft = 0
+    // (c moved out, d deleted).
+    expect(countFor(results as never, 'invoice'), 'invoice count exact after catch-up').toBe(4)
+    expect(countFor(results as never, 'draft'), 'draft count exact after catch-up').toBe(0)
+
+    // THE CALL-SHAPE PIN: the whole-store walk never ran.
+    expect(walkSpy, 'full backfill walk must not run for a behind-stamp reopen').not.toHaveBeenCalled()
+
+    vi.restoreAllMocks()
+  }, 120000)
+})