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/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
+}