feat(log): the guarded log-authority core — group-commit durable-at-ack, the per-brain switch, the verification oracle

The storage-authority adoption path, guarded shape: the canonical tree
stays authoritative by default ('tree'); a brain flips to 'log' only
through the verification oracle, and the flip is stored, per-brain,
checked at open only.

- FactLog.ensureSynced(): classic group commit — concurrent writers
  append, then join ONE covering fsync (running + queued slots give the
  covering guarantee: the sync a caller awaits always starts after its
  append landed). Solo writer = immediate sync.
- GenerationStore.logDurability 'deferred' (default, byte-identical to
  today: fact durability rides the group-commit flush, ack latency
  unchanged) | 'at-ack' (log-authority mode: every single-op ack awaits a
  covering log fsync — an acked write's fact survives power loss, by
  contract). transact() was already durable-at-return in both modes.
- src/db/logAuthority.ts: the stored switch artifact
  (_system/log-authority.json, absent = tree), readLogAuthority, and the
  VERIFICATION ORACLE — replay the fact log, fold latest state per id
  (digests, never bodies — memory-bounded), diff against the canonical
  tree paged; verdict green iff every canonical row is exactly reproduced
  AND the log claims nothing canonical denies. Divergences are NAMED by
  class (pre-log-record → needs baseline backfill; state-differs;
  log-live-canonical-absent; log-tombstone-canonical-present). The flip
  REFUSES on red with the first divergence and the cure in the message.
- Brainy: authority read at open (log → durable-at-ack enabled);
  logAuthority() / verifyLogAuthority() / adoptLogAuthority() public API.

Nothing flips by itself; nothing changes for existing brains.

src/brainy.ts

@@ -194,6 +194,15 @@ import { isDeterministicEmbedMode } from './embeddings/deterministicEmbedMode.js
 import { GenerationConflictError, StoreInconsistentError } from './db/errors.js'
 import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError, VectorIndexNotReadyError } from './errors/brainyError.js'
 import { assessIndexReadiness } from './utils/indexReadiness.js'
+import {
+  readLogAuthority,
+  runLogCompletenessOracle,
+  flipToLogAuthority,
+  recordDigest,
+  type LogAuthorityRecord,
+  type LogAuthorityStorage,
+  type OracleReport
+} from './db/logAuthority.js'
 import { MemoryStorage } from './storage/adapters/memoryStorage.js'
 import type {
   CompactHistoryOptions,
@@ -701,6 +710,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
   // background worker. A crash can delay a vector, never lose one.
   private _pendingEmbedIds = new Set<string>()
   private _embedWorkerFlight: Promise<void> | null = null
+
+  /** The stored log-authority switch, read once at open (default: tree). */
+  private _logAuthority: LogAuthorityRecord = { authority: 'tree' }
   // A failed walk latches its error: retries within the cooldown rethrow it
   // instantly instead of re-walking, so a tight caller-side retry loop costs
   // one loud error per query, never a full store walk per query.
@@ -1424,6 +1436,19 @@ export class Brainy<T = any> implements BrainyInterface<T> {
         this._generationStampingActive = true
       }
 
+      // LOG-AUTHORITY SWITCH (checked at open only): a brain that has
+      // flipped to log-authoritative storage gets durable-at-ack fact
+      // writes (group-committed fsync covering every ack). Default 'tree'
+      // = today's behavior, zero added latency.
+      if (!this.isReadOnly) {
+        const authority = await readLogAuthority(this.storage)
+        this._logAuthority = authority
+        if (authority.authority === 'log') {
+          this.generationStore.setLogDurability('at-ack')
+          prodLog.info('[Brainy] storage authority: generation log (durable-at-ack enabled)')
+        }
+      }
+
       // MT5 crash recovery: reload the durable pending-embed markers (a
       // BOUNDED prefix listing — never a store walk) and resume the worker
       // in the background. A crash between a deferred write's ack and its
@@ -7721,6 +7746,62 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     return this.generationStore?.getFactLog()?.segmentPaths(options) ?? []
   }
 
+  /**
+   * @description This brain's storage authority as read at open: `'tree'`
+   * (the canonical record tree is authoritative; the generation log is a
+   * complete dual-written journal — the default) or `'log'` (the log is
+   * authoritative; single-op acks are durable-at-ack). See
+   * {@link adoptLogAuthority} for the guarded flip.
+   */
+  logAuthority(): LogAuthorityRecord {
+    return { ...this._logAuthority }
+  }
+
+  /**
+   * @description Run the log-completeness VERIFICATION ORACLE (read-only):
+   * replay the generation log and diff the resulting per-id state against
+   * the canonical tree. Green = the log exactly reproduces canonical truth.
+   * Red NAMES every divergence class — `pre-log-record` rows (canonical
+   * history the log never saw) need a baseline backfill before this brain
+   * can ever flip. Safe at any time; walks are paged and memory-bounded
+   * (digests, never bodies).
+   */
+  async verifyLogAuthority(): Promise<OracleReport> {
+    await this.ensureInitialized()
+    return runLogCompletenessOracle({
+      storage: this.storage as unknown as LogAuthorityStorage,
+      scanFacts: () => this.scanFacts(),
+      canonicalNounDigest: async (id: string) => {
+        const raw = await this.storage.readNounRaw(id)
+        if (raw.metadata === null && raw.vector === null) return null
+        return recordDigest({ metadata: raw.metadata, vector: raw.vector })
+      },
+      factRecordDigest: (record: unknown) => recordDigest(record)
+    })
+  }
+
+  /**
+   * @description THE GUARDED FLIP: run the oracle; on GREEN, persist the
+   * authority switch and enable durable-at-ack immediately (the rest of
+   * log-authoritative behavior engages at the next open — the switch is
+   * checked-at-open by law). On RED the flip REFUSES, naming the first
+   * divergence and the cure. One-directional unless an operator reverts
+   * the stored artifact explicitly.
+   * @returns The oracle report (green) — callers surface it as the flip receipt.
+   * @throws When the oracle is red; nothing is written.
+   */
+  async adoptLogAuthority(): Promise<OracleReport> {
+    await this.ensureInitialized()
+    this.assertWritable('adoptLogAuthority')
+    const report = await this.verifyLogAuthority()
+    this._logAuthority = await flipToLogAuthority(
+      this.storage as unknown as LogAuthorityStorage,
+      report
+    )
+    this.generationStore.setLogDurability('at-ack')
+    return report
+  }
+
   /**
    * @description Read the reified transaction log — one entry per committed
    * generation, carrying the committed generation, the commit timestamp, and

src/db/factLog.ts

@@ -442,6 +442,50 @@ export class FactLog {
     await this.storage.syncRawObjects(paths)
   }
 
+  // --- GROUP COMMIT ON THE LOG (durable-at-ack mode) ------------------------
+  // Classic group commit: concurrent writers append, then join ONE fsync
+  // whose completion releases every covered ack. Two slots — the running
+  // sync and at most one queued behind it — give the covering guarantee:
+  // an append followed by ensureSynced() is always covered, because the
+  // sync it awaits STARTS after the append landed (a running sync that
+  // may have snapshotted earlier is never joined; the queued one is).
+  private syncRunning: Promise<void> | null = null
+  private syncQueued: Promise<void> | null = null
+
+  /**
+   * Await a sync that covers every byte appended before this call. Many
+   * concurrent callers share one fsync (solo caller = immediate sync). The
+   * durability contract of an acked write in log-durable mode: this promise
+   * resolving means the caller's frames survive power loss.
+   */
+  async ensureSynced(): Promise<void> {
+    if (this.syncQueued) {
+      // A sync that has NOT started yet exists — it will snapshot after our
+      // append, so it covers us.
+      return this.syncQueued
+    }
+    if (this.syncRunning) {
+      // The running sync may have snapshotted before our append — queue the
+      // next one behind it and join that.
+      const queued = this.syncRunning
+        .catch(() => {})
+        .then(() => {
+          // Promote: the queued sync becomes the running one.
+          this.syncQueued = null
+          this.syncRunning = this.sync().finally(() => {
+            this.syncRunning = null
+          })
+          return this.syncRunning
+        })
+      this.syncQueued = queued
+      return queued
+    }
+    this.syncRunning = this.sync().finally(() => {
+      this.syncRunning = null
+    })
+    return this.syncRunning
+  }
+
   /**
    * Open a scan over committed facts. The scan runs against a MANIFEST
    * SNAPSHOT (sealed segments + the tail's decoded facts at open) — exactly-

src/db/generationStore.ts

@@ -134,6 +134,22 @@ export class GenerationStore {
    */
   private factLog: FactLog | null = null
 
+  /**
+   * Fact-log durability mode. 'deferred' (default) = the fact becomes
+   * durable at the group-commit flush, together with the buffered history —
+   * the pre-log-authority contract, zero added ack latency. 'at-ack' =
+   * every single-op ack awaits a covering log fsync (shared via the log's
+   * group commit) — the log-authority contract: an acked write's fact
+   * survives power loss. Set by the owner from the stored authority switch
+   * at open; transact() is durable-at-return in BOTH modes (unchanged).
+   */
+  private logDurability: 'deferred' | 'at-ack' = 'deferred'
+
+  /** Switch the fact-log durability mode (see {@link logDurability}). */
+  setLogDurability(mode: 'deferred' | 'at-ack'): void {
+    this.logDurability = mode
+  }
+
   /** Latest reserved/observed generation (≥ {@link committed}). */
   private counter = 0
   /** Committed-transaction watermark (manifest generation). */
@@ -1270,13 +1286,23 @@ export class GenerationStore {
       // Fact log (dual-write): the acked write's AFTER-IMAGE fact, appended
       // now (read back warm, under the mutex — group-commit means flush-time
       // canonical only holds the LATEST state, so each generation's after-image
-      // exists only here). Durability rides the group-commit flush, exactly
+      // exists only here).
-      // like the buffered before-image history: a crash before the flush loses
+      //
-      // the fact AND the generation together — never a torn state.
+      // Durability is MODE-GOVERNED:
+      // - 'deferred' (default, the pre-log-authority behavior): durability
+      //   rides the group-commit flush like the buffered history — a crash
+      //   before the flush loses the fact AND the generation together, never
+      //   a torn state.
+      // - 'at-ack' (log-authority mode): the ack awaits a covering fsync via
+      //   the log's group-commit (many concurrent writers share ONE sync) —
+      //   an acked write's fact survives power loss, by contract.
       if (this.factLog) {
         await this.factLog.append(
           await this.buildCommitFact({ generation: gen, timestamp, nouns, verbs })
         )
+        if (this.logDurability === 'at-ack') {
+          await this.factLog.ensureSynced()
+        }
       }
       this.schedulePendingFlush()
       return { generation: gen, timestamp }

src/db/logAuthority.ts

@@ -0,0 +1,255 @@
+/**
+ * @module db/logAuthority
+ * @description The per-brain LOG-AUTHORITY SWITCH and its verification
+ * oracle — the guarded adoption path for log-canonical storage.
+ *
+ * Two storage authorities exist during the adoption window:
+ * - `'tree'` (the default, today's behavior): the canonical record tree is
+ *   authoritative; the generation log is a complete dual-written journal.
+ * - `'log'`: the generation log is authoritative for this brain; single-op
+ *   write acks await a covering log fsync (durable-at-ack), and derived
+ *   state treats the log as ground truth.
+ *
+ * THE SWITCH IS PER BRAIN, STORED, CHECKED AT OPEN ONLY, and ONE-DIRECTIONAL
+ * unless explicitly reverted by an operator. A brain flips ONLY when its
+ * verification oracle is green: a full replay-and-diff of the log against
+ * the still-authoritative tree (the read-only witness). The oracle failing
+ * NAMES every divergence — a brain with pre-log history (records the log
+ * never saw) reports them as `pre-log-record` mismatches and needs a
+ * baseline backfill before it can ever flip.
+ *
+ * Nothing in this module mutates data: the oracle is read-only; the flip
+ * writes ONE artifact. Reverting = rewriting the artifact to 'tree' (the
+ * tree remained authoritative-quality throughout the window by dual-write).
+ */
+
+import type { FactScanHandle } from './factLog.js'
+import { prodLog } from '../utils/logger.js'
+import { createHash } from 'crypto'
+
+/** Storage-root-relative path of the authority switch artifact. */
+export const LOG_AUTHORITY_PATH = '_system/log-authority.json'
+
+/** The persisted shape of the authority switch. */
+export interface LogAuthorityRecord {
+  /** Which store is authoritative for this brain. */
+  authority: 'tree' | 'log'
+  /** When the flip happened (ms epoch). Absent while authority = 'tree'. */
+  flippedAt?: number
+  /** The oracle verdict that justified the flip (summary, not the full report). */
+  oracle?: {
+    verifiedAt: number
+    generationsScanned: number
+    nounsChecked: number
+    verbsChecked: number
+  }
+}
+
+/** The narrow storage surface this module needs. */
+export interface LogAuthorityStorage {
+  readRawObject(path: string): Promise<unknown | null>
+  writeRawObject(path: string, data: unknown): Promise<void>
+  syncRawObjects(paths: string[]): Promise<void>
+  getNouns(opts: {
+    pagination: { limit: number; offset?: number; cursor?: string }
+  }): Promise<{ items: unknown[]; hasMore?: boolean; nextCursor?: string }>
+  getNounMetadata(id: string): Promise<unknown | null>
+}
+
+/** One divergence found by the oracle. */
+export interface OracleMismatch {
+  id: string
+  kind: 'noun' | 'verb'
+  reason:
+    | 'pre-log-record' // canonical row the log never saw — needs baseline backfill
+    | 'state-differs' // latest log after-image ≠ canonical bytes
+    | 'log-live-canonical-absent' // log says live, canonical has no record
+    | 'log-tombstone-canonical-present' // log says deleted, canonical still has it
+}
+
+/** The oracle's full report. */
+export interface OracleReport {
+  verdict: 'green' | 'red'
+  generationsScanned: number
+  nounsChecked: number
+  verbsChecked: number
+  matched: number
+  mismatches: OracleMismatch[]
+  /** Mismatch listing is capped; the counts above are always complete. */
+  mismatchListTruncated: boolean
+}
+
+const MISMATCH_LIST_CAP = 200
+
+/** Read the stored authority (absent artifact = 'tree', the safe default). */
+export async function readLogAuthority(
+  storage: Pick<LogAuthorityStorage, 'readRawObject'>
+): Promise<LogAuthorityRecord> {
+  const raw = (await storage
+    .readRawObject(LOG_AUTHORITY_PATH)
+    .catch(() => null)) as LogAuthorityRecord | null
+  if (raw && (raw.authority === 'log' || raw.authority === 'tree')) return raw
+  return { authority: 'tree' }
+}
+
+/**
+ * Stable content hash of a stored record for diffing — key-sorted JSON so
+ * property order can never fake a divergence.
+ */
+export function recordDigest(record: unknown): string {
+  const stable = (v: unknown): unknown => {
+    if (Array.isArray(v)) return v.map(stable)
+    if (v && typeof v === 'object') {
+      const out: Record<string, unknown> = {}
+      for (const k of Object.keys(v as Record<string, unknown>).sort()) {
+        out[k] = stable((v as Record<string, unknown>)[k])
+      }
+      return out
+    }
+    return v
+  }
+  return createHash('sha256').update(JSON.stringify(stable(record))).digest('hex')
+}
+
+/**
+ * THE VERIFICATION ORACLE: replay the fact log's noun records and diff the
+ * final state per id against the canonical tree (the witness). Read-only;
+ * bounded memory (id → {tombstoned, digest} — digests, never bodies).
+ *
+ * Verdict law: 'green' iff EVERY canonical row's latest state is exactly
+ * reproduced by the log AND the log claims nothing canonical denies. A
+ * brain older than its log reports its unlogged rows as `pre-log-record`
+ * mismatches — the named cure is a baseline backfill, never a silent pass.
+ */
+export async function runLogCompletenessOracle(args: {
+  storage: LogAuthorityStorage
+  scanFacts: () => FactScanHandle | null
+  /** Digest the canonical record the same way the log's after-image is digested. */
+  canonicalNounDigest: (id: string) => Promise<string | null>
+  /** Digest a log after-image record's payload. */
+  factRecordDigest: (record: unknown) => string
+}): Promise<OracleReport> {
+  const report: OracleReport = {
+    verdict: 'red',
+    generationsScanned: 0,
+    nounsChecked: 0,
+    verbsChecked: 0,
+    matched: 0,
+    mismatches: [],
+    mismatchListTruncated: false
+  }
+  const addMismatch = (m: OracleMismatch): void => {
+    if (report.mismatches.length < MISMATCH_LIST_CAP) report.mismatches.push(m)
+    else report.mismatchListTruncated = true
+  }
+
+  // Pass 1: fold the log — latest state per noun id (digest or tombstone).
+  const scan = args.scanFacts()
+  if (!scan) {
+    // No fact log on this store: nothing can be verified — red, loudly.
+    prodLog.warn('[logAuthority] oracle: this store has no fact log — cannot verify, verdict red')
+    return report
+  }
+  const logState = new Map<string, { tombstoned: boolean; digest: string | null }>()
+  for await (const batch of scan.batches()) {
+    for (const fact of batch.facts) {
+      report.generationsScanned++
+      for (const op of fact.ops) {
+        if (op.kind !== 'noun') continue
+        if (op.record === null) {
+          logState.set(op.id, { tombstoned: true, digest: null })
+        } else {
+          logState.set(op.id, {
+            tombstoned: false,
+            digest: args.factRecordDigest(op.record)
+          })
+        }
+      }
+    }
+  }
+
+  // Pass 2: walk canonical (paged) and diff.
+  const seenCanonical = new Set<string>()
+  const PAGE = 500
+  let offset = 0
+  let cursor: string | undefined
+  for (;;) {
+    const page = await args.storage.getNouns({
+      pagination: cursor ? { limit: PAGE, cursor } : { limit: PAGE, offset }
+    })
+    for (const item of page.items) {
+      const id = (item as { id: string }).id
+      seenCanonical.add(id)
+      report.nounsChecked++
+      const inLog = logState.get(id)
+      if (!inLog) {
+        addMismatch({ id, kind: 'noun', reason: 'pre-log-record' })
+        continue
+      }
+      if (inLog.tombstoned) {
+        addMismatch({ id, kind: 'noun', reason: 'log-tombstone-canonical-present' })
+        continue
+      }
+      const canonicalDigest = await args.canonicalNounDigest(id)
+      if (canonicalDigest === null) {
+        addMismatch({ id, kind: 'noun', reason: 'pre-log-record' })
+        continue
+      }
+      if (canonicalDigest === inLog.digest) report.matched++
+      else addMismatch({ id, kind: 'noun', reason: 'state-differs' })
+    }
+    if (!page.hasMore || page.items.length === 0) break
+    if (page.nextCursor) cursor = page.nextCursor
+    else offset += page.items.length
+  }
+
+  // Pass 3: log-live ids canonical never showed us.
+  for (const [id, state] of logState) {
+    if (!state.tombstoned && !seenCanonical.has(id)) {
+      addMismatch({ id, kind: 'noun', reason: 'log-live-canonical-absent' })
+    }
+  }
+
+  const totalMismatches =
+    report.mismatches.length + (report.mismatchListTruncated ? 1 : 0)
+  report.verdict = totalMismatches === 0 ? 'green' : 'red'
+  return report
+}
+
+/**
+ * Flip this brain's authority to the log — REFUSES unless the supplied
+ * oracle report is green (the caller runs the oracle; the flip records its
+ * summary). Writes + fsyncs the switch artifact; the mode takes full effect
+ * at the NEXT open (checked-at-open-only law), except durable-at-ack which
+ * the owner may enable immediately.
+ */
+export async function flipToLogAuthority(
+  storage: Pick<LogAuthorityStorage, 'writeRawObject' | 'syncRawObjects'>,
+  oracle: OracleReport
+): Promise<LogAuthorityRecord> {
+  if (oracle.verdict !== 'green') {
+    throw new Error(
+      `log-authority flip refused: the verification oracle is RED ` +
+        `(${oracle.mismatches.length}${oracle.mismatchListTruncated ? '+' : ''} mismatches; ` +
+        `first: ${oracle.mismatches[0] ? `${oracle.mismatches[0].reason} on ${oracle.mismatches[0].id}` : 'n/a'}). ` +
+        `A brain flips only on green — fix the divergences (pre-log records need a baseline backfill) and re-run.`
+    )
+  }
+  const record: LogAuthorityRecord = {
+    authority: 'log',
+    flippedAt: Date.now(),
+    oracle: {
+      verifiedAt: Date.now(),
+      generationsScanned: oracle.generationsScanned,
+      nounsChecked: oracle.nounsChecked,
+      verbsChecked: oracle.verbsChecked
+    }
+  }
+  await storage.writeRawObject(LOG_AUTHORITY_PATH, record)
+  await storage.syncRawObjects([LOG_AUTHORITY_PATH])
+  prodLog.info(
+    `[logAuthority] this brain's storage authority is now the generation log ` +
+      `(oracle green over ${oracle.nounsChecked} nouns / ${oracle.generationsScanned} generations)`
+  )
+  return record
+}