fix(recovery): the fold streams and narrates; the checkpoint chain arms at the flip

A production brain's first process boot after a live authority flip looked
hung and was restarted three times mid-recovery — three defects with one
scene. (1) THE FOLD MATERIALIZED THE LOG: peekFactsAbove(0) decoded every
fact into one array (GBs of after-images on a ~7k-fact log, a GC storm, a
starved write lane). The fold now STREAMS one segment-batch at a time —
memory is one segment at any log size — with structural ordering asserted
loudly. (2) THE FOLD WAS SILENT UNTIL DONE: minutes of boot work with zero
narration is what invited the restarts. It now announces itself BEFORE the
work ('do not restart, the fold is finite') and prints progress every
thousand facts. (3) THE CHAIN COULD ONLY ARM AT A CRASH: a live mid-session
flip left the fold checkpoint unfounded, so the brain's first unclean boot
paid a whole-log fold. Adoption now founds the checkpoint AT THE FLIP — one
paged full canonical barrier (bounded memory), then the stamp — so bounded
recovery holds from minute zero for every store that flips, at any size.

Pinned: a non-fresh flip stamps immediately; the first post-flip unclean
boot folds bounded (an unflushed at-ack fact above the checkpoint is
restored; a barrier-covered row below it is outside the fold). Kill matrix
and both adoption suites green alongside.

src/brainy.ts

@@ -8392,6 +8392,57 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     // Fold-checkpoint chain, phase 2: the flip is recorded — open the stamp
     // gate so the next flush/close barrier writes the first checkpoint.
     this.generationStore.completeFoldCheckpointBootstrap()
+    // ARM-AT-FLIP for the NON-FRESH brain (the chain refused the fresh-brain
+    // arm because committed > 0): run one paged FULL canonical barrier now —
+    // every live row's canonical bytes fsynced, bounded memory — then stamp
+    // the first checkpoint. Without this, the chain could only arm at the
+    // brain's first crash, and that crash paid a WHOLE-LOG fold: a production
+    // brain hit exactly that on its first post-flip boot (a full-log
+    // materializing fold, restarted three times mid-flight). Adoption already
+    // pays O(N) oracle work; one more O(N) barrier founds bounded recovery
+    // from minute zero.
+    if (!this.generationStore.foldCheckpointChainArmed()) {
+      const PAGE = 500
+      let synced = 0
+      prodLog.info(
+        `[Brainy] adoptLogAuthority: founding the fold checkpoint — syncing every ` +
+          `row's canonical bytes (paged; progress every 2000 rows)`
+      )
+      let offset = 0
+      let cursor: string | undefined
+      for (;;) {
+        const page = await this.storage.getNouns({
+          pagination: cursor ? { limit: PAGE, cursor } : { limit: PAGE, offset }
+        })
+        const ids = page.items.map((i) => (i as { id: string }).id)
+        if (ids.length > 0) {
+          await this.storage.syncEntityCanonical?.(ids, [])
+          synced += ids.length
+          if (synced % 2000 < PAGE && synced >= 2000) {
+            prodLog.info(`[Brainy] adoptLogAuthority: checkpoint founding — ${synced} rows synced`)
+          }
+        }
+        if (page.hasMore && page.nextCursor) { cursor = page.nextCursor; offset += ids.length; continue }
+        if (page.hasMore && !page.nextCursor) { offset += PAGE; continue }
+        break
+      }
+      let vOffset = 0
+      let vCursor: string | undefined
+      for (;;) {
+        const page = await this.storage.getVerbs({
+          pagination: vCursor ? { limit: PAGE, cursor: vCursor } : { limit: PAGE, offset: vOffset }
+        })
+        const ids = page.items.map((i) => (i as { id: string }).id)
+        if (ids.length > 0) {
+          await this.storage.syncEntityCanonical?.([], ids)
+          synced += ids.length
+        }
+        if (page.hasMore && page.nextCursor) { vCursor = page.nextCursor; vOffset += ids.length; continue }
+        if (page.hasMore && !page.nextCursor) { vOffset += PAGE; continue }
+        break
+      }
+      await this.generationStore.stampFoldCheckpointAfterFullBarrier()
+    }
     return report
   }
 

src/db/factLog.ts

@@ -770,6 +770,43 @@ export class FactLog {
    * segments directly; the torn tail's invalid suffix is ignored exactly
    * like open() would).
    */
+  /**
+   * STREAMING twin of {@link FactLog.peekFactsAbove} for the recovery fold:
+   * yields facts above the bound one SEGMENT at a time, ascending, without
+   * ever materializing the whole log (a production first-boot fold OOM-class
+   * allocation storm came from exactly that — GBs of decoded after-images in
+   * one array while the process looked hung). Memory is one segment's worth.
+   * Works manifest-direct (safe before {@link FactLog.open}). Ordering is
+   * structural (segments rotate in order; appends are ordered within one) and
+   * ASSERTED — a violation aborts loudly, never a silent misordered replay.
+   */
+  async *streamFactsAbove(committedGeneration: number): AsyncGenerator<CommitFact[], void> {
+    const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null
+    if (!stored || typeof stored !== 'object' || !Array.isArray(stored.segments)) return
+    if (stored.formatVersion !== FACTS_FORMAT_VERSION) return
+    const files = [...stored.segments.map((s) => s.file)]
+    if (stored.tailSegment) files.push(stored.tailSegment)
+    let lastGen = committedGeneration
+    for (const file of files) {
+      const bytes = await this.storage.readRawBytes(`${FACTS_PREFIX}/${file}`)
+      if (bytes === null) continue
+      const { facts } = parseSegment(file, bytes)
+      const batch: CommitFact[] = []
+      for (const f of facts) {
+        if (f.generation <= committedGeneration) continue
+        if (f.generation <= lastGen) {
+          throw new Error(
+            `fact log: streamFactsAbove found non-ascending generations ` +
+              `(${f.generation} after ${lastGen} in ${file}) — refusing to replay out of order`
+          )
+        }
+        lastGen = f.generation
+        batch.push(f)
+      }
+      if (batch.length > 0) yield batch
+    }
+  }
+
   async peekFactsAbove(committedGeneration: number): Promise<CommitFact[]> {
     const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null
     if (!stored || typeof stored !== 'object' || !Array.isArray(stored.segments)) return []

src/db/generationStore.ts

@@ -637,33 +637,63 @@ export class GenerationStore {
         this.foldCheckpointChainValid = checkpoint !== null || this.committed === 0
         this.foldCheckpoint = foldBound
         if (uncleanOpen) this.foldCheckpointChainValid = true
-        const factsToReplay = uncleanOpen
+        // THE FOLD STREAMS AND NARRATES. A production first boot after a live
-          ? await this.factLog.peekFactsAbove(foldBound)
+        // flip folded ~7k facts by materializing them all (GBs of decoded
-          : orphans
+        // after-images, a GC storm, a starved write lane) in SILENCE — the
-        if (factsToReplay.length > 0) {
+        // operator restarted the process three times mid-fold, each restart
-          let replayed = 0
+        // making the next boot unclean again. Two laws from that day: the
-          for (const fact of factsToReplay) {
+        // fold consumes the log one segment-batch at a time (memory = one
-            for (const op of fact.ops) {
+        // segment, any log size), and it announces itself BEFORE the work
-              const image =
+        // with progress lines DURING it — an operator who can see a fold
-                op.record === null
+        // converging lets it finish.
-                  ? { metadata: null, vector: null }
+        const foldKind = uncleanOpen
-                  : { metadata: op.record.metadata, vector: op.record.vector }
+          ? foldBound > 0
-              if (op.kind === 'verb') await this.storage.writeVerbRaw(op.id, image)
+            ? `BOUNDED fold above checkpoint ${foldBound}`
-              else await this.storage.writeNounRaw(op.id, image)
+            : 'WHOLE-LOG fold'
-              this.noteCheckpointDirty(op.kind, op.id)
+          : 'above-manifest replay'
-            }
+        let replayed = 0
-            replayed++
+        const replayFact = async (fact: CommitFact): Promise<void> => {
-            if (fact.generation > this.committed) {
+          for (const op of fact.ops) {
-              this.committed = fact.generation
+            const image =
-              this.appendCommittedGen(fact.generation)
+              op.record === null
-              this.setDelta(fact.generation, {
+                ? { metadata: null, vector: null }
-                nouns: new Set(fact.ops.filter((o) => o.kind === 'noun').map((o) => o.id)),
+                : { metadata: op.record.metadata, vector: op.record.vector }
-                verbs: new Set(fact.ops.filter((o) => o.kind === 'verb').map((o) => o.id)),
+            if (op.kind === 'verb') await this.storage.writeVerbRaw(op.id, image)
-                timestamp: fact.timestamp,
+            else await this.storage.writeNounRaw(op.id, image)
-                bytes: 0
+            this.noteCheckpointDirty(op.kind, op.id)
-              })
-            }
           }
+          replayed++
+          if (replayed % 1000 === 0) {
+            prodLog.warn(
+              `[GenerationStore] recovery fold in progress — ${replayed} fact(s) folded ` +
+                `(at generation ${fact.generation}); do not restart, the fold is finite`
+            )
+          }
+          if (fact.generation > this.committed) {
+            this.committed = fact.generation
+            this.appendCommittedGen(fact.generation)
+            this.setDelta(fact.generation, {
+              nouns: new Set(fact.ops.filter((o) => o.kind === 'noun').map((o) => o.id)),
+              verbs: new Set(fact.ops.filter((o) => o.kind === 'verb').map((o) => o.id)),
+              timestamp: fact.timestamp,
+              bytes: 0
+            })
+          }
+        }
+        if (uncleanOpen) {
+          prodLog.warn(
+            `[GenerationStore] log-authority recovery: ${foldKind} beginning ` +
+              `(unclean shutdown detected) — streaming replay, bounded memory, ` +
+              `progress every 1000 facts. Do not restart the process; a restart ` +
+              `re-pays the whole fold.`
+          )
+          for await (const batch of this.factLog.streamFactsAbove(foldBound)) {
+            for (const fact of batch) await replayFact(fact)
+          }
+        } else {
+          for (const fact of orphans) await replayFact(fact)
+        }
+        if (replayed > 0) {
           if (this.counter < this.committed) this.counter = this.committed
           await this.persistCounterUnlocked()
           const manifest: GenerationManifest = {
@@ -676,13 +706,7 @@ export class GenerationStore {
           await this.storage.syncRawObjects([MANIFEST_PATH])
           prodLog.warn(
             `[GenerationStore] log-authority recovery replayed ${replayed} fact(s) into ` +
-              `canonical (${
+              `canonical (${foldKind}; committed at ${this.committed}) — an acked write is never lost`
-                uncleanOpen
-                  ? foldBound > 0
-                    ? `BOUNDED fold above checkpoint ${foldBound} — unclean shutdown`
-                    : 'WHOLE-LOG fold — unclean shutdown'
-                  : 'above-manifest'
-              }; committed at ${this.committed}) — an acked write is never lost`
           )
         }
         // A recovery fold re-applied (and the barrier below re-syncs) every
@@ -897,6 +921,44 @@ export class GenerationStore {
     this.authorityIsLog = true
   }
 
+  /** Whether the fold-checkpoint chain is armed (a bounded fold is possible). */
+  foldCheckpointChainArmed(): boolean {
+    return this.foldCheckpointChainValid
+  }
+
+  /**
+   * @description Stamp the fold checkpoint after the caller has completed a
+   * FULL canonical barrier (every live row's canonical bytes fsynced, paged —
+   * the adoption path does this right after a non-fresh flip). The stamp
+   * asserts total coverage, so it may ONLY be called when the barrier walked
+   * everything; stamp-after-data is the caller's ordering to keep. Arms the
+   * chain: the brain's first unclean boot folds (checkpoint, head] instead of
+   * the whole log — a production first boot after a live flip paid a full-log
+   * fold through three mid-fold restarts because the chain could previously
+   * only arm at a crash.
+   */
+  async stampFoldCheckpointAfterFullBarrier(): Promise<void> {
+    return this.withMutex(async () => {
+      if (!this.authorityIsLog || !this.factLog) {
+        throw new Error(
+          'stampFoldCheckpointAfterFullBarrier: only a log-authority brain stamps a fold checkpoint'
+        )
+      }
+      this.foldCheckpointChainValid = true
+      // The full barrier supersedes any accumulated partial set.
+      this.checkpointDirtyNouns = new Set()
+      this.checkpointDirtyVerbs = new Set()
+      const target = this.committed
+      await this.storage.writeRawObject(FOLD_CHECKPOINT_PATH, { generation: target })
+      await this.storage.syncRawObjects([FOLD_CHECKPOINT_PATH])
+      this.foldCheckpoint = target
+      prodLog.info(
+        `[GenerationStore] fold checkpoint founded at generation ${target} — ` +
+          `crash recovery is bounded from this moment`
+      )
+    })
+  }
+
   /**
    * @description Adoption-time chain bootstrap, abort — called when an
    * adoption attempt throws or refuses after phase 1. Disarms the chain and

tests/integration/fold-checkpoint-bound.test.ts

@@ -161,6 +161,38 @@ describe('fold-checkpoint bound — crash recovery folds (checkpoint, head], nev
     expect(stamped, 'the first whole-log fold is the chain’s base case — it stamps').toBe(committedOf(reopened))
   }, 120000)
 
+  it('ARM-AT-FLIP: a non-fresh adoption founds the checkpoint immediately — the first post-flip boot folds BOUNDED, never whole-log', async () => {
+    const dir = trackDir()
+    // The production shape: a brain with history flips LIVE (no crash ever).
+    const brain = await openBrain(dir, { logAuthority: 'defer' })
+    liveBrains.push(brain)
+    const preFlip = await brain.add({ data: 'pre-flip resident', type: NounType.Document, metadata: { era: 'tree' } })
+    await brain.flush()
+    expect(readCheckpoint(dir), 'no checkpoint before the flip').toBeNull()
+
+    const report = await brain.adoptLogAuthority()
+    expect(report.verdict).toBe('green')
+    // THE PIN: the flip itself founded the checkpoint — no crash required.
+    const founded = readCheckpoint(dir)
+    expect(founded, 'checkpoint founded at flip').toBe(committedOf(brain))
+
+    // First post-flip boot, unclean (the production first-restart shape):
+    // a post-flip write above the checkpoint is restored FROM ITS AT-ACK FACT
+    // (deliberately NOT flushed — a flush would barrier-sync it and advance
+    // the stamp over it, making its loss synthetic); the pre-flip row (its
+    // baseline fact ≤ checkpoint, its bytes barrier-synced at the flip) is
+    // OUTSIDE the fold — vaporizing it synthetically proves the bound.
+    const postFlip = await brain.add({ data: 'post-flip write', type: NounType.Document, metadata: { era: 'log' } })
+    await abandonAsCrashed(liveBrains.pop()!)
+    dropCanonicalNoun(dir, preFlip)
+    dropCanonicalNoun(dir, postFlip)
+
+    const reopened = await openBrain(dir, { logAuthority: 'adopt' })
+    liveBrains.push(reopened)
+    expect(await reopened.get(postFlip), 'above-checkpoint fact re-applied').not.toBeNull()
+    expect(await reopened.get(preFlip), 'below-checkpoint fact skipped — the fold is bounded on the FIRST post-flip boot').toBeNull()
+  }, 240000)
+
   it('a tree-authority brain never stamps a checkpoint', async () => {
     const dir = trackDir()
     const brain = await openBrain(dir, { logAuthority: 'defer' })