open-brainy/src/db/logAuthority.ts

/**
 * @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
  }
  /**
   * Recorded when an OPEN-TIME adoption attempt (the 10.0.0 fleet default)
   * was refused — the oracle could not go green. Keeps subsequent opens
   * cheap; an operator re-runs adoptLogAuthority() after resolving it.
   */
  adoptRefusal?: { at: number; reason: string }
}

/** 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' }
}

/**
 * Normalize a canonical noun record to its ENTITY TRUTH before diffing:
 * the canonical vector-file wrapper denormalizes derived index residue
 * (`connections` — HNSW graph edges; `level` — the node's random skip-list
 * level) that the generation log deliberately does NOT carry (projections
 * own their own rebuild paths). Digesting the residue would report false
 * `state-differs` on ~any brain whose HNSW assigned a nonzero level. Both
 * sides of every oracle comparison pass through this normalizer.
 */
export function nounEntityTruth(record: {
  metadata: unknown
  vector: unknown
}): { metadata: unknown; vector: unknown } {
  const v = record.vector
  if (v && typeof v === 'object' && !Array.isArray(v)) {
    const { connections: _c, level: _l, ...entity } = v as Record<string, unknown>
    return { metadata: record.metadata, vector: entity }
  }
  return { metadata: record.metadata, vector: v }
}

/**
 * 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
  /**
   * Verb legs (optional until every owner wires them): the canonical verb
   * digest + the paged verb enumeration. When ABSENT, the oracle counts NO
   * verbs and says so via verbsChecked = 0 — an honest partial verdict,
   * never a silent full-pass claim.
   */
  canonicalVerbDigest?: (id: string) => Promise<string | null>
  getVerbs?: (opts: {
    pagination: { limit: number; offset?: number; cursor?: string }
  }) => Promise<{ items: unknown[]; hasMore?: boolean; nextCursor?: string }>
  /**
   * Cap on the LISTED mismatches (counts are always complete). Defaults to
   * the wire-friendly {@link MISMATCH_LIST_CAP}; the adoption backfill passes
   * `Infinity` so ONE scan yields the ENTIRE curable set — a production
   * brain with a 12.7k-row pre-log baseline once advanced only 800 rows per
   * adoption call because each pass could see (and cure) at most 200.
   */
  mismatchListCap?: number
}): Promise<OracleReport> {
  const listCap = args.mismatchListCap ?? MISMATCH_LIST_CAP
  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 < listCap) 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 }>()
  const verbLogState = 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) {
        const state =
          op.record === null
            ? { tombstoned: true, digest: null }
            : { tombstoned: false, digest: args.factRecordDigest(op.record) }
        if (op.kind === 'noun') logState.set(op.id, state)
        else verbLogState.set(op.id, state)
      }
    }
  }

  // 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' })
    }
  }

  // Verb passes — only when the owner wired the verb legs; otherwise the
  // report says verbsChecked: 0, an honest partial scope, never a claim.
  if (args.canonicalVerbDigest && args.getVerbs) {
    const seenVerbs = new Set<string>()
    let vOffset = 0
    let vCursor: string | undefined
    for (;;) {
      const page = await args.getVerbs({
        pagination: vCursor ? { limit: PAGE, cursor: vCursor } : { limit: PAGE, offset: vOffset }
      })
      for (const item of page.items) {
        const id = (item as { id: string }).id
        seenVerbs.add(id)
        report.verbsChecked++
        const inLog = verbLogState.get(id)
        if (!inLog) {
          addMismatch({ id, kind: 'verb', reason: 'pre-log-record' })
          continue
        }
        if (inLog.tombstoned) {
          addMismatch({ id, kind: 'verb', reason: 'log-tombstone-canonical-present' })
          continue
        }
        const canonical = await args.canonicalVerbDigest(id)
        if (canonical === null) {
          addMismatch({ id, kind: 'verb', reason: 'pre-log-record' })
          continue
        }
        if (canonical === inLog.digest) report.matched++
        else addMismatch({ id, kind: 'verb', reason: 'state-differs' })
      }
      if (!page.hasMore || page.items.length === 0) break
      if (page.nextCursor) vCursor = page.nextCursor
      else vOffset += page.items.length
    }
    for (const [id, state] of verbLogState) {
      if (!state.tombstoned && !seenVerbs.has(id)) {
        addMismatch({ id, kind: 'verb', 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
}