feat(reprojection): the one doors-open machinery — budget-capped, yielding, foreground-preempted, atomic-swap; poison records quarantine typed

The generic reprojection engine (pure TS; the twin of the native
implementation — same frozen contract, one shared conformance intent):
register any ProjectionAdapter; advance(family, {budgetMs}) folds facts
from the adapter's own watermark to the head in installments ≤50ms with
real macrotask yields; foreground door traffic bumps the DoorSignal and
an in-flight advance yields within one installment ('preempted');
advanceAll round-robins families fairly. swap(family, buildAdapter) is
the doors-open migration primitive: the OLD projection keeps serving
while the new one builds beside it, the flip is atomic at parity, and a
concurrent second swap refuses typed. A fact the fold cannot apply
(typed ProjectionApplyError) is QUARANTINED — skipped, ledgered,
narrated per-doubling, exposed for refuse-affected-reads — the service
class law's fourth answer: never a wedged rebuild, never a silent skip.
The engine never writes stamps: each adapter owns its durability and its
stamp-after-data discipline. Upgrade, heal, and rebuild are now the same
machinery behind open doors.

FactLogSource wires any host's fact scan in one line
(factSourceFromHost(brain)); window-contract violations are loud.

Pins: 23 unit (budget resume without refold · preemption within one
installment · round-robin fairness under a skewed backlog · build-beside
visibility mid-swap · atomic flip · single-flight refusal · quarantine
skip/ledger/doubling · non-typed throw aborts · losing adapter
discarded) + 3 integration on a real brain (fold matches ground truth ·
doors answer mid-fold with the preemption path exercised · crash
mid-fold resumes from the stamp, never refolds).

Gates: unit 2054/2054 (157 files) · integration 820 (93 files) ·
conformance 31/31.

tests/integration/reprojection-doors-open.test.ts

@@ -0,0 +1,257 @@
+/**
+ * @module tests/integration/reprojection-doors-open
+ * @description The reprojection engine against a REAL brain on filesystem
+ * storage: a toy secondary projection (bucket counts with its own watermark
+ * artifact, stamp-after-data per src/utils/projectionWatermark.ts) folds the
+ * brain's committed facts through the engine, wired with the callback-form
+ * {@link FactLogSource} over `brain.scanFacts`.
+ *
+ * Proves the three doors-open rows:
+ * (i)   folding to caught-up matches ground-truth counts;
+ * (ii)  mid-fold, `find()` and `get()` still answer, and a door bump
+ *       preempts the advance at the next boundary (mechanism-pinned via
+ *       batch counts, not wall-clock);
+ * (iii) a crash mid-fold (abandon; reopen; re-advance) resumes from the
+ *       durable stamp — never refolds from zero.
+ */
+import { describe, it, expect, beforeAll, afterAll } from 'vitest'
+import { mkdtempSync, mkdirSync, readFileSync, writeFileSync, rmSync, existsSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { Brainy } from '../../src/index.js'
+import {
+  ReprojectionEngine,
+  type ProjectionAdapter
+} from '../../src/reprojection/reprojectionEngine.js'
+import { FactLogSource } from '../../src/reprojection/factLogSource.js'
+import { makeProjectionStamp, readStampedWatermark } from '../../src/utils/projectionWatermark.js'
+import type { CommitFact } from '../../src/db/factLog.js'
+
+/** 50 rows, 5 buckets, 10 each. */
+const ROWS = 50
+const BUCKETS = 5
+const GROUND_TRUTH: Record<string, number> = { b0: 10, b1: 10, b2: 10, b3: 10, b4: 10 }
+
+/**
+ * The toy secondary projection: latest bucket per entity id, persisted as a
+ * data file plus a SEPARATE stamp artifact written stamp-after-data via the
+ * shared projectionWatermark helpers. Idempotent by construction (latest-
+ * state per id), so at-least-once redelivery on resume is harmless.
+ */
+class BucketCountProjection implements ProjectionAdapter {
+  readonly family = 'bucket-counts'
+  /** Every generation this INSTANCE applied — the refold detector for (iii). */
+  readonly appliedGenerations: number[] = []
+  private latest: Map<string, string | null>
+  private wm: number | null
+
+  private constructor(
+    private readonly dir: string,
+    wm: number | null,
+    latest: Map<string, string | null>
+  ) {
+    this.wm = wm
+    this.latest = latest
+  }
+
+  /** Load from the artifact dir — data is trusted only under a valid stamp. */
+  static async open(dir: string): Promise<BucketCountProjection> {
+    mkdirSync(dir, { recursive: true })
+    const stampPath = join(dir, 'stamp.json')
+    const dataPath = join(dir, 'data.json')
+    let wm: number | null = null
+    if (existsSync(stampPath)) {
+      wm = readStampedWatermark(JSON.parse(readFileSync(stampPath, 'utf8')))
+    }
+    const latest = new Map<string, string | null>(
+      wm !== null && existsSync(dataPath)
+        ? (JSON.parse(readFileSync(dataPath, 'utf8')) as Array<[string, string | null]>)
+        : []
+    )
+    return new BucketCountProjection(dir, wm, latest)
+  }
+
+  /** Non-null bucket tallies from the latest-state map. */
+  counts(): Record<string, number> {
+    const out: Record<string, number> = {}
+    for (const bucket of this.latest.values()) {
+      if (bucket !== null) out[bucket] = (out[bucket] ?? 0) + 1
+    }
+    return out
+  }
+
+  watermark(): number | null {
+    return this.wm
+  }
+
+  async applyBatch(facts: CommitFact[], upTo: number): Promise<void> {
+    for (const fact of facts) {
+      this.appliedGenerations.push(fact.generation)
+      for (const op of fact.ops) {
+        if (op.kind !== 'noun') continue
+        if (op.record === null) {
+          this.latest.set(op.id, null) // tombstone
+          continue
+        }
+        // The stored noun record nests user metadata under `.metadata`.
+        const stored = op.record.metadata as Record<string, unknown> | null
+        const user = (stored?.metadata ?? stored) as Record<string, unknown> | null
+        const bucket = typeof user?.bucket === 'string' ? user.bucket : null
+        this.latest.set(op.id, bucket)
+      }
+    }
+    // Durability THEN stamp — the projectionWatermark law.
+    writeFileSync(join(this.dir, 'data.json'), JSON.stringify([...this.latest]))
+    writeFileSync(join(this.dir, 'stamp.json'), JSON.stringify(makeProjectionStamp(upTo)))
+    this.wm = upTo
+  }
+
+  async discard(): Promise<void> {
+    rmSync(this.dir, { recursive: true, force: true })
+  }
+}
+
+describe('reprojection doors-open — a real brain, a toy secondary projection', () => {
+  let brainDir: string
+  let projRoot: string
+  let brain: Brainy
+  const ids: string[] = []
+
+  const openBrain = async (dir: string): Promise<Brainy> => {
+    const b = new Brainy({
+      storage: { type: 'filesystem', path: dir },
+      requireSubtype: false,
+      silent: true,
+      dimensions: 384
+    })
+    await b.init()
+    return b
+  }
+
+  /**
+   * The production wiring, callback form: the engine's `from` is an EXCLUSIVE
+   * lower bound, `scanFacts` bounds are inclusive — hence `from + 1`; the
+   * first batch is returned and the handle closed (short batches at segment
+   * boundaries are legal — only EMPTY means caught up).
+   */
+  const sourceFor = (b: Brainy): FactLogSource =>
+    new FactLogSource(async (from, limit) => {
+      const scan = b.scanFacts({ fromGeneration: from + 1, batchSize: limit })
+      if (!scan) throw new Error('this brain hosts no fact log — cannot reproject')
+      const iterator = scan.batches()
+      try {
+        const first = await iterator.next()
+        return first.done ? [] : first.value.facts
+      } finally {
+        if (typeof iterator.return === 'function') await iterator.return(undefined)
+      }
+    })
+
+  beforeAll(async () => {
+    brainDir = mkdtempSync(join(tmpdir(), 'brainy-reproj-'))
+    projRoot = mkdtempSync(join(tmpdir(), 'brainy-reproj-artifacts-'))
+    brain = await openBrain(brainDir)
+    for (let i = 0; i < ROWS; i++) {
+      ids.push(
+        await brain.add({
+          data: `record ${i} filed in bucket ${i % BUCKETS}`,
+          type: 'document',
+          metadata: { bucket: `b${i % BUCKETS}` }
+        })
+      )
+    }
+  }, 240_000)
+
+  afterAll(async () => {
+    await brain?.close().catch(() => {})
+    rmSync(brainDir, { recursive: true, force: true })
+    rmSync(projRoot, { recursive: true, force: true })
+  })
+
+  it('(i) folds to caught-up through the engine and matches ground-truth counts', async () => {
+    const projection = await BucketCountProjection.open(join(projRoot, 'i'))
+    const engine = new ReprojectionEngine({ source: sourceFor(brain), batchSize: 8 })
+    engine.register(projection)
+
+    const result = await engine.advance(projection.family, { budgetMs: 60_000 })
+
+    expect(result.status).toBe('caught-up')
+    expect(result.watermark).toBeGreaterThanOrEqual(ROWS) // one generation per add, at least
+    expect(result.applied).toBeGreaterThanOrEqual(ROWS)
+    expect(engine.quarantined(projection.family)).toEqual([])
+    expect(projection.counts()).toEqual(GROUND_TRUTH)
+    // The stamp on disk is the adapter's own — stamped exactly at the fold head.
+    const reloaded = await BucketCountProjection.open(join(projRoot, 'i'))
+    expect(reloaded.watermark()).toBe(result.watermark)
+    expect(reloaded.counts()).toEqual(GROUND_TRUTH)
+  })
+
+  it('(ii) doors stay open mid-fold: find() and get() answer, and a bump preempts the advance', async () => {
+    const projection = await BucketCountProjection.open(join(projRoot, 'ii'))
+    const engine = new ReprojectionEngine({ source: sourceFor(brain), batchSize: 4 })
+    engine.register(projection)
+    const head = brain.scanFacts()!.headGeneration
+
+    const inFlight = engine.advance(projection.family, { budgetMs: 60_000 })
+    // The read hook: foreground door traffic announces itself, then reads —
+    // both interleave with the running fold on the same event loop.
+    engine.doorSignal.bump()
+    const found = await brain.find({ query: 'record filed in bucket', limit: 3 })
+    const got = await brain.get(ids[0])
+    const result = await inFlight
+
+    // The doors answered mid-fold.
+    expect(found.length).toBeGreaterThan(0)
+    expect(got).toBeTruthy()
+    const gotMeta = got!.metadata as Record<string, unknown> | undefined
+    expect((gotMeta?.bucket ?? (gotMeta?.metadata as Record<string, unknown>)?.bucket)).toBe('b0')
+
+    // THE PREEMPTION PIN — mechanism, not wall-clock: the bump landed before
+    // the first installment boundary, so the advance yielded after exactly
+    // one batch (≤ batchSize facts), far short of the head.
+    expect(result.status).toBe('preempted')
+    expect(result.applied).toBeGreaterThan(0)
+    expect(result.applied).toBeLessThanOrEqual(4)
+    expect(projection.appliedGenerations.length).toBe(result.applied)
+    expect(projection.watermark()).not.toBeNull()
+    expect(projection.watermark()!).toBeLessThan(head)
+
+    // Resuming folds the remainder; nothing was lost to the preemption.
+    const resumed = await engine.advance(projection.family, { budgetMs: 60_000 })
+    expect(resumed.status).toBe('caught-up')
+    expect(projection.counts()).toEqual(GROUND_TRUTH)
+  })
+
+  it('(iii) crash mid-fold: reopen and re-advance resumes from the stamp, never refolds from zero', async () => {
+    const projDir = join(projRoot, 'iii')
+    const before = await BucketCountProjection.open(projDir)
+    const engine1 = new ReprojectionEngine({ source: sourceFor(brain), batchSize: 4 })
+    engine1.register(before)
+
+    // A zero budget folds exactly one guaranteed batch, then stops.
+    const partial = await engine1.advance(before.family, { budgetMs: 0 })
+    expect(partial.status).toBe('budget-exhausted')
+    const stamped = before.watermark()
+    expect(stamped).not.toBeNull()
+    expect(stamped!).toBeGreaterThan(0)
+
+    // CRASH: abandon the engine and adapter mid-fold; reopen the brain cold.
+    await brain.close()
+    brain = await openBrain(brainDir)
+
+    const after = await BucketCountProjection.open(projDir)
+    expect(after.watermark()).toBe(stamped) // the stamp survived the crash
+
+    const engine2 = new ReprojectionEngine({ source: sourceFor(brain), batchSize: 4 })
+    engine2.register(after)
+    const resumed = await engine2.advance(after.family, { budgetMs: 60_000 })
+    expect(resumed.status).toBe('caught-up')
+
+    // NEVER REFOLDS FROM ZERO: every generation the resumed instance applied
+    // sits strictly above the crash stamp.
+    expect(after.appliedGenerations.length).toBeGreaterThan(0)
+    expect(Math.min(...after.appliedGenerations)).toBeGreaterThan(stamped!)
+    // And the combined state — durable prefix plus resumed fold — is exact.
+    expect(after.counts()).toEqual(GROUND_TRUTH)
+  })
+})

tests/unit/reprojection/reprojection-engine.test.ts

@@ -0,0 +1,590 @@
+/**
+ * @module tests/unit/reprojection/reprojection-engine
+ * @description Spec-by-example for the pure-TS reprojection engine — the
+ * frozen contract mirrored from the native twin (a shared conformance suite
+ * runs against both, so the shapes pinned here are load-bearing):
+ *
+ * (a) register + advance folds a scripted source to caught-up with exact
+ *     watermark/applied counts and adapter-owned stamping;
+ * (b) budget exhaustion answers mid-stream and a second advance RESUMES from
+ *     the watermark — never a refold;
+ * (c) a door bump mid-advance preempts within one installment — pinned by
+ *     MECHANISM (no further applyBatch after the bumping step), with only a
+ *     generous wall-clock sanity bound;
+ * (d) advanceAll round-robins families at batch granularity — no starvation;
+ * (e) swap builds beside (the old adapter serves throughout), flips
+ *     atomically at parity, refuses a concurrent swap with a typed error;
+ * (f) quarantine: a typed poison fact is skipped + ledgered, narration
+ *     doubles, a NON-typed throw aborts loudly;
+ * (g) discard() lands on the LOSING adapter after a swap.
+ */
+import { describe, it, expect, vi, afterEach } from 'vitest'
+import {
+  ReprojectionEngine,
+  DoorSignal,
+  ProjectionApplyError,
+  SwapInFlightError,
+  MAX_INSTALLMENT_MS,
+  type ProjectionAdapter,
+  type FactSource
+} from '../../../src/reprojection/reprojectionEngine.js'
+import { FactLogSource } from '../../../src/reprojection/factLogSource.js'
+import type { CommitFact } from '../../../src/db/factLog.js'
+import { prodLog } from '../../../src/utils/logger.js'
+
+/** Build one committed fact for a generation. */
+function fact(generation: number): CommitFact {
+  return {
+    generation,
+    timestamp: 1_700_000_000_000 + generation,
+    ops: [
+      {
+        kind: 'noun',
+        id: `id-${generation}`,
+        record: { metadata: { n: generation }, vector: null }
+      }
+    ]
+  }
+}
+
+/** A scripted FactSource over a (possibly mutable) list of generations. */
+function scriptedSource(gens: () => number[]): FactSource {
+  return {
+    async scan(from: number, limit: number): Promise<CommitFact[]> {
+      return gens()
+        .filter((g) => g > from)
+        .sort((x, y) => x - y)
+        .slice(0, limit)
+        .map(fact)
+    }
+  }
+}
+
+/**
+ * A recording in-memory adapter: stamps after data (the watermark advances
+ * only after a successful apply), applies idempotently (a Map keyed by
+ * generation), and can be scripted to poison (typed) or hard-fail (untyped)
+ * specific generations, or to run a hook inside applyBatch.
+ */
+class RecordingAdapter implements ProjectionAdapter {
+  readonly family: string
+  /** Generations per applyBatch call, in call order (empty arrays included). */
+  readonly batches: number[][] = []
+  /** The upTo passed to each applyBatch call, in call order. */
+  readonly upTos: number[] = []
+  /** Latest state per generation — idempotent under at-least-once delivery. */
+  readonly state = new Map<number, unknown>()
+  /** Generations that throw a typed ProjectionApplyError. */
+  readonly poison = new Set<number>()
+  /** Generations that throw a plain (untyped) Error. */
+  readonly hardFail = new Set<number>()
+  /** Runs inside applyBatch after validation, before the stamp. */
+  onApply?: (gens: number[]) => void | Promise<void>
+  discarded = 0
+  private wm: number | null
+
+  constructor(family: string, watermark: number | null = null) {
+    this.family = family
+    this.wm = watermark
+  }
+
+  watermark(): number | null {
+    return this.wm
+  }
+
+  async applyBatch(facts: CommitFact[], upTo: number): Promise<void> {
+    for (const [i, f] of facts.entries()) {
+      if (this.hardFail.has(f.generation)) {
+        throw new Error(`disk exploded at generation ${f.generation}`)
+      }
+      if (this.poison.has(f.generation)) {
+        throw new ProjectionApplyError({
+          generation: f.generation,
+          recordIndex: i,
+          cause: new Error(`unfoldable payload at ${f.generation}`)
+        })
+      }
+    }
+    for (const f of facts) this.state.set(f.generation, f.ops)
+    const gens = facts.map((f) => f.generation)
+    this.batches.push(gens)
+    this.upTos.push(upTo)
+    if (this.onApply) await this.onApply(gens)
+    this.wm = upTo // stamp-after-data
+  }
+
+  async discard(): Promise<void> {
+    this.discarded++
+  }
+}
+
+const range = (from: number, to: number): number[] =>
+  Array.from({ length: to - from + 1 }, (_, i) => from + i)
+
+afterEach(() => {
+  vi.restoreAllMocks()
+})
+
+describe('reprojection engine — (a) register + advance to caught-up', () => {
+  it('folds a scripted source in order, adapter-stamped, with exact counts', async () => {
+    const source = scriptedSource(() => range(1, 7))
+    const engine = new ReprojectionEngine({ source, batchSize: 3 })
+    const adapter = new RecordingAdapter('a')
+    engine.register(adapter)
+
+    const result = await engine.advance('a', { budgetMs: 10_000 })
+
+    expect(result.status).toBe('caught-up')
+    expect(result.watermark).toBe(7)
+    expect(result.applied).toBe(7)
+    // Batch shape and the upTo handed to the adapter's own stamp.
+    expect(adapter.batches).toEqual([[1, 2, 3], [4, 5, 6], [7]])
+    expect(adapter.upTos).toEqual([3, 6, 7])
+    // The watermark is the ADAPTER's stamp — the engine never wrote one.
+    expect(adapter.watermark()).toBe(7)
+    expect(engine.getAdapter('a')).toBe(adapter)
+  })
+
+  it('honors upTo as an inclusive cap and answers caught-up at the cap', async () => {
+    const source = scriptedSource(() => range(1, 9))
+    const engine = new ReprojectionEngine({ source, batchSize: 3 })
+    const adapter = new RecordingAdapter('a')
+    engine.register(adapter)
+
+    const result = await engine.advance('a', { budgetMs: 10_000, upTo: 5 })
+
+    expect(result.status).toBe('caught-up')
+    expect(result.watermark).toBe(5)
+    expect(result.applied).toBe(5)
+    expect(adapter.batches.flat()).toEqual([1, 2, 3, 4, 5])
+  })
+
+  it('a caught-up family answers immediately with zero applied', async () => {
+    const source = scriptedSource(() => range(1, 4))
+    const engine = new ReprojectionEngine({ source, batchSize: 10 })
+    const adapter = new RecordingAdapter('a', 4) // already stamped to the head
+    engine.register(adapter)
+
+    const result = await engine.advance('a', { budgetMs: 10_000 })
+
+    expect(result).toEqual({ status: 'caught-up', watermark: 4, applied: 0 })
+    expect(adapter.batches).toEqual([])
+  })
+
+  it('refuses duplicate registration and unregistered families loudly', async () => {
+    const engine = new ReprojectionEngine({ source: scriptedSource(() => []) })
+    engine.register(new RecordingAdapter('a'))
+    expect(() => engine.register(new RecordingAdapter('a'))).toThrow(/already registered/)
+    await expect(engine.advance('ghost', { budgetMs: 0 })).rejects.toThrow(/not registered/)
+  })
+})
+
+describe('reprojection engine — (b) budget exhaustion resumes, never refolds', () => {
+  it('returns budget-exhausted mid-stream; the next advance resumes from the watermark', async () => {
+    const source = scriptedSource(() => range(1, 10))
+    const engine = new ReprojectionEngine({ source, batchSize: 2 })
+    const adapter = new RecordingAdapter('b')
+    engine.register(adapter)
+
+    // Zero budget: exactly ONE step of guaranteed progress, then the answer.
+    const first = await engine.advance('b', { budgetMs: 0 })
+    expect(first.status).toBe('budget-exhausted')
+    expect(first.watermark).toBe(2)
+    expect(first.applied).toBe(2)
+    expect(adapter.batches).toEqual([[1, 2]])
+
+    // The second advance RESUMES from the stamp — its first batch starts at 3.
+    const second = await engine.advance('b', { budgetMs: 10_000 })
+    expect(second.status).toBe('caught-up')
+    expect(second.watermark).toBe(10)
+    expect(second.applied).toBe(8)
+    expect(adapter.batches[1]).toEqual([3, 4])
+    // No refold: every generation delivered exactly once across both calls.
+    expect(adapter.batches.flat()).toEqual(range(1, 10))
+  })
+})
+
+describe('reprojection engine — (c) door bump preempts within one installment', () => {
+  it('a bump during a step yields preempted at that step boundary — no further applyBatch', async () => {
+    const source = scriptedSource(() => range(1, 12))
+    const engine = new ReprojectionEngine({ source, batchSize: 2 })
+    const adapter = new RecordingAdapter('c')
+    adapter.onApply = (gens) => {
+      if (gens[0] === 3) engine.doorSignal.bump() // door traffic mid-second-batch
+    }
+    engine.register(adapter)
+
+    const started = Date.now()
+    const result = await engine.advance('c', { budgetMs: 60_000 })
+    const elapsed = Date.now() - started
+
+    expect(result.status).toBe('preempted')
+    expect(result.watermark).toBe(4)
+    expect(result.applied).toBe(4)
+    // THE MECHANISM PIN: the batch that observed the bump was the LAST batch —
+    // preemption landed at the very next boundary, not after more work.
+    expect(adapter.batches).toEqual([[1, 2], [3, 4]])
+    // Generous wall-clock sanity only (the pin above carries the contract):
+    // two tiny batches plus one installment boundary sit far under 5s.
+    expect(elapsed).toBeLessThan(5_000)
+    expect(MAX_INSTALLMENT_MS).toBe(50)
+
+    // Resuming folds the rest — preemption lost nothing.
+    const resumed = await engine.advance('c', { budgetMs: 60_000 })
+    expect(resumed.status).toBe('caught-up')
+    expect(resumed.watermark).toBe(12)
+    expect(adapter.batches.flat()).toEqual(range(1, 12))
+  })
+
+  it('bumps are edge-triggered per advance: a stale bump never preempts', async () => {
+    const source = scriptedSource(() => range(1, 4))
+    const doorSignal = new DoorSignal()
+    const engine = new ReprojectionEngine({ source, doorSignal, batchSize: 2 })
+    const adapter = new RecordingAdapter('c2')
+    engine.register(adapter)
+
+    doorSignal.bump() // BEFORE the advance — belongs to earlier traffic
+    const result = await engine.advance('c2', { budgetMs: 10_000 })
+    expect(result.status).toBe('caught-up')
+    expect(result.watermark).toBe(4)
+  })
+})
+
+describe('reprojection engine — (d) advanceAll round-robin fairness', () => {
+  it('a one-batch family is served on the first round despite a huge backlog next to it', async () => {
+    const source = scriptedSource(() => range(1, 40))
+    const engine = new ReprojectionEngine({ source, batchSize: 5 })
+    const callOrder: string[] = []
+    const big = new RecordingAdapter('big') // 8 batches behind
+    const small = new RecordingAdapter('small', 35) // 1 batch behind
+    big.onApply = () => {
+      callOrder.push('big')
+    }
+    small.onApply = () => {
+      callOrder.push('small')
+    }
+    engine.register(big)
+    engine.register(small)
+
+    const results = await engine.advanceAll({ budgetMs: 10_000 })
+
+    expect(results.big).toEqual({ status: 'caught-up', watermark: 40, applied: 40 })
+    expect(results.small).toEqual({ status: 'caught-up', watermark: 40, applied: 5 })
+    // Fairness pin: 'small' folded its single batch on round ONE — it never
+    // waited behind 'big''s backlog.
+    expect(callOrder[1]).toBe('small')
+    expect(callOrder.filter((f) => f === 'small')).toHaveLength(1)
+  })
+
+  it('two full-backlog families interleave strictly, one batch each per round', async () => {
+    const source = scriptedSource(() => range(1, 40))
+    const engine = new ReprojectionEngine({ source, batchSize: 5 })
+    const callOrder: string[] = []
+    const first = new RecordingAdapter('first')
+    const second = new RecordingAdapter('second')
+    first.onApply = () => {
+      callOrder.push('first')
+    }
+    second.onApply = () => {
+      callOrder.push('second')
+    }
+    engine.register(first)
+    engine.register(second)
+
+    const results = await engine.advanceAll({ budgetMs: 10_000 })
+
+    expect(results.first.status).toBe('caught-up')
+    expect(results.second.status).toBe('caught-up')
+    // 8 rounds × (first, second): strict alternation — neither ever ran twice
+    // while the other waited.
+    expect(callOrder).toHaveLength(16)
+    for (let i = 0; i < callOrder.length; i += 2) {
+      expect(callOrder.slice(i, i + 2)).toEqual(['first', 'second'])
+    }
+  })
+
+  it('budget exhaustion mid-round reports every unfinished family at its own watermark', async () => {
+    const source = scriptedSource(() => range(1, 40))
+    const engine = new ReprojectionEngine({ source, batchSize: 5 })
+    const a = new RecordingAdapter('a')
+    const b = new RecordingAdapter('b')
+    engine.register(a)
+    engine.register(b)
+
+    const results = await engine.advanceAll({ budgetMs: 0 })
+
+    // Zero budget: the leading family gets its one guaranteed step, then the
+    // budget answer lands for everyone still mid-stream.
+    expect(results.a.status).toBe('budget-exhausted')
+    expect(results.b.status).toBe('budget-exhausted')
+    expect(results.a.applied + results.b.applied).toBeGreaterThanOrEqual(5)
+    // A later advanceAll resumes both to the head.
+    const finished = await engine.advanceAll({ budgetMs: 10_000 })
+    expect(finished.a.status).toBe('caught-up')
+    expect(finished.b.status).toBe('caught-up')
+    expect(a.batches.flat()).toEqual(range(1, 40))
+    expect(b.batches.flat()).toEqual(range(1, 40))
+  })
+})
+
+describe('reprojection engine — (e) swap: build-beside, atomic flip, single-flight', () => {
+  it('the old adapter serves at its own watermark throughout the build; the flip is atomic at parity', async () => {
+    const log = range(1, 20)
+    const source = scriptedSource(() => log)
+    const engine = new ReprojectionEngine({ source, batchSize: 4 })
+    const oldAdapter = new RecordingAdapter('e')
+    engine.register(oldAdapter)
+    await engine.advance('e', { budgetMs: 10_000 })
+    expect(oldAdapter.watermark()).toBe(20)
+
+    // The log grows after the old adapter stamped — the build must reach the
+    // HEAD (24), not merely the old watermark (20), before the flip.
+    log.push(21, 22, 23, 24)
+
+    const servingDuringBuild: Array<{ adapter: ProjectionAdapter | undefined; watermark: number | null }> = []
+    let replacement!: RecordingAdapter
+    const result = await engine.swap('e', async () => {
+      replacement = new RecordingAdapter('e')
+      replacement.onApply = () => {
+        servingDuringBuild.push({
+          adapter: engine.getAdapter('e'),
+          watermark: engine.getAdapter('e')!.watermark()
+        })
+      }
+      return replacement
+    })
+
+    // Build-beside pin: EVERY mid-build observation saw the OLD adapter,
+    // still serving, still at its own stamp.
+    expect(servingDuringBuild.length).toBeGreaterThan(0)
+    for (const seen of servingDuringBuild) {
+      expect(seen.adapter).toBe(oldAdapter)
+      expect(seen.watermark).toBe(20)
+    }
+    // The flip: the registry now serves the replacement, at parity with head.
+    expect(engine.getAdapter('e')).toBe(replacement)
+    expect(result.watermark).toBe(24)
+    expect(result.applied).toBe(24)
+    expect(replacement.batches.flat()).toEqual(range(1, 24))
+  })
+
+  it('a second concurrent swap on the same family refuses with the typed single-flight error', async () => {
+    const source = scriptedSource(() => range(1, 8))
+    const engine = new ReprojectionEngine({ source, batchSize: 4 })
+    engine.register(new RecordingAdapter('e2'))
+
+    let release!: () => void
+    const gate = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    const inFlight = engine.swap('e2', async () => {
+      const building = new RecordingAdapter('e2')
+      building.onApply = () => gate // the build parks mid-fold
+      return building
+    })
+
+    // While the first swap builds, a second one is refused — typed.
+    const refusal = await engine.swap('e2', async () => new RecordingAdapter('e2')).catch((e) => e)
+    expect(refusal).toBeInstanceOf(SwapInFlightError)
+    expect((refusal as SwapInFlightError).family).toBe('e2')
+
+    release()
+    const done = await inFlight
+    expect(done.watermark).toBe(8)
+    // Single-flight released: a follow-up swap is admitted again.
+    const again = await engine.swap('e2', async () => new RecordingAdapter('e2'))
+    expect(again.watermark).toBe(8)
+  })
+
+  it('a failed build discards the partial replacement and leaves the old adapter serving', async () => {
+    const source = scriptedSource(() => range(1, 8))
+    const engine = new ReprojectionEngine({ source, batchSize: 4 })
+    const oldAdapter = new RecordingAdapter('e3')
+    engine.register(oldAdapter)
+    await engine.advance('e3', { budgetMs: 10_000 })
+
+    let failed!: RecordingAdapter
+    await expect(
+      engine.swap('e3', async () => {
+        failed = new RecordingAdapter('e3')
+        failed.hardFail.add(5) // an UNTYPED failure mid-build
+        return failed
+      })
+    ).rejects.toThrow(/disk exploded/)
+
+    expect(failed.discarded).toBe(1) // the partial build was cleaned up
+    expect(oldAdapter.discarded).toBe(0)
+    expect(engine.getAdapter('e3')).toBe(oldAdapter) // still serving, untouched
+    expect(oldAdapter.watermark()).toBe(8)
+  })
+})
+
+describe('reprojection engine — (f) quarantine: the fourth answer class', () => {
+  it('a typed poison fact is skipped, ledgered, and the rest folds to quarantined', async () => {
+    const source = scriptedSource(() => range(1, 10))
+    const engine = new ReprojectionEngine({ source, batchSize: 4 })
+    const adapter = new RecordingAdapter('f')
+    adapter.poison.add(6)
+    engine.register(adapter)
+
+    const result = await engine.advance('f', { budgetMs: 10_000 })
+
+    expect(result.status).toBe('quarantined')
+    expect(result.watermark).toBe(10)
+    expect(result.applied).toBe(9) // every generation but the poison
+    expect(adapter.batches.flat().sort((x, y) => x - y)).toEqual([1, 2, 3, 4, 5, 7, 8, 9, 10])
+    expect(adapter.state.has(6)).toBe(false)
+
+    const ledger = engine.quarantined('f')
+    expect(ledger).toHaveLength(1)
+    expect(ledger[0].generation).toBe(6)
+    expect(ledger[0].error).toBeInstanceOf(ProjectionApplyError)
+    expect(ledger[0].error.recordIndex).toBe(1) // 6 sat at index 1 of [5..8]
+    expect(typeof ledger[0].at).toBe('number')
+  })
+
+  it('narration doubles: warns on the 1st, 2nd, and 4th quarantine — not the 3rd', async () => {
+    const warnSpy = vi.spyOn(prodLog, 'warn').mockImplementation(() => {})
+    const source = scriptedSource(() => range(1, 10))
+    const engine = new ReprojectionEngine({ source, batchSize: 10 })
+    const adapter = new RecordingAdapter('f2')
+    for (const g of [2, 4, 6, 8]) adapter.poison.add(g)
+    engine.register(adapter)
+
+    const result = await engine.advance('f2', { budgetMs: 10_000 })
+
+    expect(result.status).toBe('quarantined')
+    expect(result.watermark).toBe(10)
+    expect(result.applied).toBe(6)
+    expect(engine.quarantined('f2').map((q) => q.generation)).toEqual([2, 4, 6, 8])
+    const quarantineWarns = warnSpy.mock.calls.filter((c) => String(c[0]).includes('quarantined generation'))
+    // 4 entries, narrated at counts 1, 2, and 4 — the 3rd stayed quiet.
+    expect(quarantineWarns).toHaveLength(3)
+    expect(quarantineWarns.map((c) => String(c[0]))).toEqual([
+      expect.stringContaining('(1 quarantined total)'),
+      expect.stringContaining('(2 quarantined total)'),
+      expect.stringContaining('(4 quarantined total)')
+    ])
+  })
+
+  it('an all-poison window still advances the stamp via an empty applyBatch', async () => {
+    const source = scriptedSource(() => range(1, 3))
+    const engine = new ReprojectionEngine({ source, batchSize: 3 })
+    const adapter = new RecordingAdapter('f3')
+    for (const g of [1, 2, 3]) adapter.poison.add(g)
+    engine.register(adapter)
+
+    const result = await engine.advance('f3', { budgetMs: 10_000 })
+
+    expect(result.status).toBe('quarantined')
+    expect(result.watermark).toBe(3)
+    expect(result.applied).toBe(0)
+    // The final call carried NO facts but a real upTo — the pure watermark
+    // advance past poison, stamped by the adapter itself.
+    expect(adapter.batches).toEqual([[]])
+    expect(adapter.upTos).toEqual([3])
+    expect(engine.quarantined('f3').map((q) => q.generation)).toEqual([1, 2, 3])
+  })
+
+  it('a NON-typed throw aborts the advance loudly — unknown failure is never poison', async () => {
+    const source = scriptedSource(() => range(1, 8))
+    const engine = new ReprojectionEngine({ source, batchSize: 4 })
+    const adapter = new RecordingAdapter('f4')
+    adapter.hardFail.add(5)
+    engine.register(adapter)
+
+    await expect(engine.advance('f4', { budgetMs: 10_000 })).rejects.toThrow(/disk exploded at generation 5/)
+
+    expect(adapter.watermark()).toBe(4) // the clean first batch landed; nothing after
+    expect(engine.quarantined('f4')).toEqual([]) // no ledger entry for an unknown failure
+  })
+
+  it('an adapter re-condemning an already-quarantined generation is refused loudly', async () => {
+    const source = scriptedSource(() => range(1, 4))
+    const engine = new ReprojectionEngine({ source, batchSize: 4 })
+    // A misbehaving adapter: always blames generation 3, even once it is
+    // filtered out of its batches.
+    const adapter: ProjectionAdapter = {
+      family: 'f5',
+      watermark: () => null,
+      applyBatch: async () => {
+        throw new ProjectionApplyError({ generation: 3, cause: new Error('always 3') })
+      },
+      discard: async () => {}
+    }
+    engine.register(adapter)
+
+    await expect(engine.advance('f5', { budgetMs: 10_000 })).rejects.toThrow(/ALREADY quarantined/)
+    expect(engine.quarantined('f5').map((q) => q.generation)).toEqual([3])
+  })
+
+  it('an adapter that never stamps is refused loudly instead of spinning', async () => {
+    const source = scriptedSource(() => range(1, 4))
+    const engine = new ReprojectionEngine({ source, batchSize: 2 })
+    const adapter: ProjectionAdapter = {
+      family: 'f6',
+      watermark: () => null, // never advances
+      applyBatch: async () => {},
+      discard: async () => {}
+    }
+    engine.register(adapter)
+
+    await expect(engine.advance('f6', { budgetMs: 10_000 })).rejects.toThrow(/not stamping/)
+  })
+})
+
+describe('reprojection engine — (g) discard lands on the losing adapter after a swap', () => {
+  it('the OLD adapter is discarded exactly once, after the flip; the winner is never discarded', async () => {
+    const source = scriptedSource(() => range(1, 6))
+    const engine = new ReprojectionEngine({ source, batchSize: 3 })
+    const losing = new RecordingAdapter('g')
+    engine.register(losing)
+    await engine.advance('g', { budgetMs: 10_000 })
+    expect(losing.discarded).toBe(0) // serving adapters are never discarded
+
+    let winner!: RecordingAdapter
+    await engine.swap('g', async () => {
+      winner = new RecordingAdapter('g')
+      winner.onApply = () => {
+        // Mid-build the loser still serves and is still intact.
+        expect(losing.discarded).toBe(0)
+      }
+      return winner
+    })
+
+    expect(losing.discarded).toBe(1)
+    expect(winner.discarded).toBe(0)
+    expect(engine.getAdapter('g')).toBe(winner)
+  })
+})
+
+describe('FactLogSource — the production source enforces the window contract', () => {
+  it('delegates to the injected callback and passes clean windows through', async () => {
+    const calls: Array<[number, number]> = []
+    const source = new FactLogSource(async (from, limit) => {
+      calls.push([from, limit])
+      return range(from + 1, Math.min(from + limit, 5)).map(fact)
+    })
+    const facts = await source.scan(2, 2)
+    expect(facts.map((f) => f.generation)).toEqual([3, 4])
+    expect(calls).toEqual([[2, 2]])
+    expect(await source.scan(5, 3)).toEqual([])
+  })
+
+  it('refuses out-of-contract callbacks loudly: oversize, non-ascending, at-or-below from', async () => {
+    const oversize = new FactLogSource(async () => range(1, 5).map(fact))
+    await expect(oversize.scan(0, 2)).rejects.toThrow(/contract violation/)
+
+    const unsorted = new FactLogSource(async () => [fact(3), fact(2)])
+    await expect(unsorted.scan(0, 10)).rejects.toThrow(/strictly ascending/)
+
+    const stale = new FactLogSource(async () => [fact(2)])
+    await expect(stale.scan(2, 10)).rejects.toThrow(/strictly ascending/)
+  })
+
+  it('validates its own window arguments', async () => {
+    const source = new FactLogSource(async () => [])
+    await expect(source.scan(-1, 5)).rejects.toThrow(/non-negative integer/)
+    await expect(source.scan(0, 0)).rejects.toThrow(/positive integer/)
+  })
+})