ladder.py

"""ladder.py — The Acquisition Frontier (Phase 2).

Generalizes learn_null's L0-only acquire() into the cheapest-first SOURCE LADDER from
phase2-acquisition-frontier.md. A null climbs only as far as it must; every acquisition records
WHICH RUNG answered, because trust depends on the rung.

    L0  local corpus     the cart's own parsed docs/impls        (== learn_null.acquire today)
    L1  adjacent local   New Brain documents, plans folder,      (same trust as L0: still ground
                         adjacent code roots not yet indexed      truth or Aaron's own words)
    L2  external         official docs > source repos > community (web; allowlist-gated, tiered)
    L3  Aaron            decision-nulls & personal-domain facts   (batched ask; never for code)

WRITE-BACK ROUTES BY FACT CLASS (Fable ruling — New Brain facts 48/55):
    - L0/L1 code-behavior facts (Class::member) -> learned_facts.jsonl (JudgeChecker reads there)
    - L2 external / L3 / personal facts         -> New Brain via remember() (typed, conflict-checked)
  Wiring the checker to read code evidence FROM New Brain is the deferred "cartridge #3" (fact 55).

TRUST TIERS (design): 1 ground-truth(code/fs) · 2 documented-intent(docs/plans/Aaron) ·
    3 external-authoritative(official docs, re-verify cadence) · 4 external-community(NEVER
    sufficient alone) · 5 model-prior(never evidence). The rung sets the tier; verify() enforces
    the L2 corroboration asymmetry: tier-4-only, uncorroborated -> a DOCUMENT (a lead), not a fact.

SEAMS (honest, not faked): the L2 web-SEARCH provider is left injectable and defaults to none —
    finding candidate URLs for a free-text query needs the orchestrator/dispatch layer, which is
    an explicitly UNDECIDED null (fact 38; fact 43: e.g. Gemini for search). The allowlist-gated
    FETCH is real in-process urllib. Triage TYPICALITY is a seam defaulting to 1.0 — the per-fact
    density signal (0.78 AUROC) is the shelved-JEPA estate, not a live scorer.

STDOUT LAW: diagnostics to stderr (this module can touch the New Brain store / MCP transport).

Run:  .venv/Scripts/python.exe ladder.py        # demo: climb every abstain in field_log.jsonl
"""

from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
import sys
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from fnmatch import fnmatch
from pathlib import Path

from checker import JudgeChecker, _ollama_generate

HERE = Path(__file__).resolve().parent
FIELD_LOGS = [HERE / "field_log.jsonl", HERE / "ue_field_log.jsonl"]
LEARNED = HERE / "learned_facts.jsonl"
ALLOWLIST = HERE / "l2_allowlist.json"
L2_CANDIDATES = HERE / "l2_candidates.jsonl"
L3_ASKS = HERE / "l3_asks.jsonl"
PLANS = Path(r"C:\Users\layth\.claude\plans")

# Budget caps per rung — PLACEHOLDERS (design: "numbers TBD from usage", Aaron: no decision needed).
DEFAULT_BUDGET = {"L1_code": 20, "L1_docs": 3, "L1_plans": 3, "L2_search": 5, "L2_fetch": 3}

# Adjacent-code roots for L1 (not the active L0 domain). Empty by default so L1 never scans the
# giant UE tree unbidden; set L1_CODE_ROOTS or pass ctx.adjacent_roots to widen it.
L1_CODE_ROOTS: list[Path] = []

FACT_RE = re.compile(r"^([A-Z_]+):\s*([\w?]+)::(\w+)\s*(?:=\s*(.+))?$")


# ---------------------------------------------------------------------------
# data shapes
# ---------------------------------------------------------------------------
@dataclass
class Null:
    """An identified gap. Code facts (cls+member) climb L0->L2 then park; personal/decision
    nulls skip straight to L3 (Aaron) — only his call reaches him (design triage rule)."""
    fact: str
    query: str
    cls: str | None = None
    member: str | None = None
    value: str | None = None
    personal: bool = False
    blocking: bool = False           # actively blocking work -> L3 interrupts immediately (fact 20)

    @property
    def is_code_fact(self) -> bool:
        return bool(self.cls and self.member) and not self.personal


@dataclass
class Evidence:
    source: str
    text: str
    trust_tier: int


@dataclass
class Acquisition:
    rung: str
    evidence: list[Evidence]
    note: str = ""


@dataclass
class Outcome:
    null: Null
    rung: str | None
    verdict: str                     # verified | refuted | document | unsure | abstain | batched | blocking
    evidence: list[Evidence]
    trust_tier: int | None
    note: str = ""


@dataclass
class Ctx:
    jc: JudgeChecker
    store: object | None = None                      # NewBrainStore | None (None -> New Brain routing is dry-run)
    allowlist_path: Path = ALLOWLIST
    candidate_path: Path = L2_CANDIDATES
    l3_path: Path = L3_ASKS
    adjacent_roots: list[Path] = field(default_factory=lambda: list(L1_CODE_ROOTS))
    search: object = None                            # (query, k) -> list[str] URLs; SEAM, default none
    fetch: object = None                             # (url) -> str text; default in-process urllib
    now: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    def web_search(self, query: str, k: int) -> list[str]:
        return list(self.search(query, k)) if callable(self.search) else []

    def web_fetch(self, url: str) -> str:
        return (self.fetch or _urllib_fetch)(url)


# ---------------------------------------------------------------------------
# L2 allowlist
# ---------------------------------------------------------------------------
def load_allowlist(path: Path = ALLOWLIST) -> list[dict]:
    if not path.exists():
        return []
    return json.loads(path.read_text(encoding="utf-8")).get("entries", [])


def _strip_scheme(url: str) -> str:
    return re.sub(r"^https?://", "", url.strip())


def allow_tier(url: str, entries: list[dict]) -> int | None:
    """Trust tier for a URL per the allowlist, or None if off-list (-> candidate queue). A path-level
    'exclude' sub-glob drops the URL off the list even if the pattern matches."""
    bare = _strip_scheme(url)
    for e in entries:
        if fnmatch(bare, e["pattern"]):
            if any(fnmatch(bare, ex) for ex in e.get("exclude", [])):
                return None
            return e["tier"]
    return None


def _urllib_fetch(url: str, timeout: int = 15) -> str:
    """In-process fetch (design: 'in-process is fine' — Aaron). Crude tag strip; no JS."""
    req = urllib.request.Request(url, headers={"User-Agent": "ladder/1.0"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        html = r.read().decode("utf-8", errors="ignore")
    text = re.sub(r"(?is)<(script|style)[^>]*>.*?</\1>", " ", html)
    text = re.sub(r"(?s)<[^>]+>", " ", text)
    return re.sub(r"\s+", " ", text).strip()


# ---------------------------------------------------------------------------
# L2 SEARCH PROVIDER — the seam wired (fact 113, supersedes fact 94): candidate-URL search is a
# `claude -p --model haiku` subprocess with the WebSearch tool, SUBSCRIPTION-BILLED. The env is
# scrubbed of paid-API credentials exactly as orchestrator._subscription_env does (facts 109/115),
# so the subprocess uses the Claude Code subscription login or fails loud — it NEVER silently bills
# a paid API. WebSearch is available on the subscription CLI (verified against Claude Code 2.1.198:
# -p, --allowedTools, --permission-mode, --output-format json), so no paid-API path is required.
# ---------------------------------------------------------------------------
CLAUDE_SEARCH_CMD = ["claude", "-p", "--model", "{model}", "--allowedTools", "WebSearch",
                     "--permission-mode", "bypassPermissions", "--output-format", "json", "{prompt}"]
_URL_RE_TEXT = re.compile(r"https?://[^\s)>\]\"'}]+")


def _subscription_env() -> dict:
    """Paid-API credentials scrubbed (facts 109/115) — the claude -p search runs on the subscription
    login or fails loud, never on a silent paid API. Mirrors orchestrator._subscription_env; kept
    local so the acquisition layer does not import the spine that consumes it."""
    env = dict(os.environ)
    env.pop("ANTHROPIC_API_KEY", None)
    env.pop("ANTHROPIC_AUTH_TOKEN", None)
    return env


def make_claude_search(model: str = "haiku", runner: object = None):
    """Build the L2 search provider ((query, k) -> list[str] candidate URLs) that ctx.search wants.
    Spawns `claude -p --model <model>` with the WebSearch tool, subscription-billed. Returns [] on
    an auth failure ('Not logged in') or empty result — the ladder then finds no L2 evidence and
    climbs/parks honestly rather than fabricating a source. `runner` ((cmd, timeout) -> proc with
    .stdout) is an injectable seam so tests exercise command-building + URL parsing with no
    subprocess, network, or spend."""
    def _search(query: str, k: int) -> list[str]:
        prompt = (f"Web-search for authoritative sources answering: {query}\n"
                  f"Return up to {k} candidate source URLs, most authoritative first, ONE per line. "
                  f"Output only the URLs, nothing else.")
        exe = shutil.which(CLAUDE_SEARCH_CMD[0]) or CLAUDE_SEARCH_CMD[0]
        cmd = [exe] + [a.replace("{model}", model).replace("{prompt}", prompt)
                       for a in CLAUDE_SEARCH_CMD[1:]]
        try:
            proc = runner(cmd, 180) if runner is not None else subprocess.run(
                cmd, capture_output=True, text=True, timeout=180, env=_subscription_env())
        except Exception as e:
            _log(f"  L2 claude -p search failed to run: {e}")
            return []
        out = (getattr(proc, "stdout", "") or "").strip()
        text = out
        try:                                                     # --output-format json -> {"result": ...}
            payload = json.loads(out)
            if isinstance(payload, dict):
                if payload.get("is_error"):
                    _log(f"  L2 claude -p search error (not logged in?): {str(payload.get('result'))[:160]}")
                    return []
                text = payload.get("result", out)
        except json.JSONDecodeError:
            pass
        urls, seen = [], set()
        for m in _URL_RE_TEXT.findall(text):
            u = m.rstrip(".,")
            if u not in seen:
                seen.add(u); urls.append(u)
            if len(urls) >= k:
                break
        return urls
    return _search


# ---------------------------------------------------------------------------
# rungs: (null, ctx, budget) -> Acquisition | None
# ---------------------------------------------------------------------------
def acquire_l0(null: Null, ctx: Ctx, budget: dict) -> Acquisition | None:
    """L0 — the cart's own parsed corpus. Relaxed-owner search: every doc/impl for this member
    under ANY owner (identical to learn_null.acquire). Trust tier 1: source code."""
    if not null.member:
        return None
    ev = [Evidence(f"{o}::{m}", f"DOC for {o}::{m}: {doc}", 1)
          for (o, m), doc in ctx.jc.docs.items() if m == null.member]
    ev += [Evidence(f"{o}::{m} (impl)", f"IMPLEMENTATION of {o}::{m}:\n{body}", 1)
           for (o, m), body in ctx.jc.defs.items() if m == null.member]
    return Acquisition("L0", ev) if ev else None


def acquire_l1(null: Null, ctx: Ctx, budget: dict) -> Acquisition | None:
    """L1 — adjacent local: New Brain documents (Aaron's captured words/docs, tier 2), the plans
    folder (documented intent, tier 2), and adjacent code roots not in the active domain (tier 1).
    Same trust as L0 — still ground truth or Aaron's own words, just not indexed yet."""
    ev: list[Evidence] = []
    if null.member and ctx.adjacent_roots:
        ev += _search_adjacent_code(null.member, ctx.adjacent_roots, budget.get("L1_code", 20))
    if ctx.store is not None:
        try:
            from new_brain_store import embed_one
            for d in ctx.store.recall_documents(embed_one(null.query), k=budget.get("L1_docs", 3)):
                ev.append(Evidence(f"newbrain-doc:{d['id']}", (d["content"] or "")[:1500], 2))
        except Exception as e:                                   # store optional — never fail the climb on it
            _log(f"  L1 New Brain docs skipped: {e}")
    ev += _search_plans(null.query, budget.get("L1_plans", 3))
    return Acquisition("L1", ev) if ev else None


def acquire_l2(null: Null, ctx: Ctx, budget: dict) -> Acquisition | None:
    """L2 — external web, allowlist-gated. Candidate URLs come from the injected search SEAM
    (default none); allowed ones are fetched in-process; OFF-LIST sources are neither silently
    used (drift) nor silently dropped (lost knowledge) — they queue for Aaron's batched veto."""
    urls = ctx.web_search(null.query, budget.get("L2_search", 5))
    if not urls:
        return None                                             # no search provider wired -> nothing to fetch
    entries = load_allowlist(ctx.allowlist_path)
    fetched: list[Evidence] = []
    off_list: list[str] = []
    for url in urls[:budget.get("L2_fetch", 3)]:
        tier = allow_tier(url, entries)
        if tier is None:
            off_list.append(url)
            continue
        try:
            text = ctx.web_fetch(url)
        except Exception as e:
            _log(f"  L2 fetch failed {url}: {e}")
            continue
        if text:
            fetched.append(Evidence(url, text[:2000], tier))
    for url in off_list:
        _queue_candidate(url, null, ctx.candidate_path)
    note = f"{len(off_list)} off-list candidate(s) queued" if off_list else ""
    return Acquisition("L2", fetched, note) if (fetched or off_list) else None


def acquire_l3(null: Null, ctx: Ctx, budget: dict) -> Acquisition | None:
    """L3 — Aaron. No automated evidence; his call or his knowledge. Batches by default and
    surfaces at the next touchpoint (facts 19-21); an actively-blocking null interrupts NOW
    (fact 20). Never reached for code facts — those park as honest abstains instead."""
    if null.blocking:
        return Acquisition("L3", [], "INTERRUPT: actively-blocking null — surface immediately")
    queued = _queue_l3_ask(null, ctx.l3_path, ctx.now)
    note = "batched to Aaron (next-touchpoint)" if queued else "already queued/answered — not re-asked"
    return Acquisition("L3", [], note)


# ---------------------------------------------------------------------------
# L1/L2 helpers
# ---------------------------------------------------------------------------
def _search_adjacent_code(member: str, roots: list[Path], cap: int) -> list[Evidence]:
    """Capped grep over adjacent code roots for the member's definition — a line window per hit."""
    out: list[Evidence] = []
    pat = re.compile(rf"\b\w+::{re.escape(member)}\s*\(|\b{re.escape(member)}\s*\(")
    for root in roots:
        for path in Path(root).rglob("*.cpp"):
            if len(out) >= cap:
                return out
            try:
                lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
            except OSError:
                continue
            for i, line in enumerate(lines):
                if pat.search(line):
                    out.append(Evidence(f"{path.name}:{i+1}", "\n".join(lines[i:i + 30]), 1))
                    break
    return out


def _search_plans(query: str, cap: int) -> list[Evidence]:
    """Keyword-overlap scan of the plans folder (documented intent, tier 2)."""
    if not PLANS.exists():
        return []
    terms = {w.lower() for w in re.findall(r"[A-Za-z][A-Za-z0-9_]{3,}", query)}
    if not terms:
        return []
    scored: list[tuple[int, Path]] = []
    for path in PLANS.rglob("*.md"):
        try:
            text = path.read_text(encoding="utf-8", errors="ignore").lower()
        except OSError:
            continue
        hits = sum(1 for t in terms if t in text)
        if hits:
            scored.append((hits, path))
    scored.sort(key=lambda x: -x[0])
    out = []
    for hits, path in scored[:cap]:
        snippet = path.read_text(encoding="utf-8", errors="ignore")[:1200]
        out.append(Evidence(f"plan:{path.name}", snippet, 2))
    return out


def _queue_candidate(url: str, null: Null, path: Path) -> None:
    """Off-list L2 source: record source + what it would have answered (batched approve/veto)."""
    rec = {"ts": datetime.now(timezone.utc).isoformat(), "url": url,
           "would_answer": null.query, "fact": null.fact}
    with path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(rec) + "\n")


def _load_l3_asks(path: Path) -> list[dict]:
    if not path.exists():
        return []
    return [json.loads(l) for l in path.read_text(encoding="utf-8").splitlines() if l.strip()]


def _write_l3_asks(path: Path, asks: list[dict]) -> None:
    with path.open("w", encoding="utf-8") as f:
        for a in asks:
            f.write(json.dumps(a) + "\n")


def _queue_l3_ask(null: Null, path: Path, now: str) -> bool:
    """Queue an L3 ask for Aaron, unless this fact is already queued (pending) or already answered
    (resolved) — an answered question is NEVER re-asked (that is the answer-capture loop's payoff).
    Returns True if a new ask was written."""
    if any(a.get("fact") == null.fact for a in _load_l3_asks(path)):
        return False
    rec = {"ts": now, "fact": null.fact, "query": null.query, "blocking": null.blocking,
           "resolved": False}
    with path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(rec) + "\n")
    return True


def answer_l3_ask(fact: str, answer: str, ctx: Ctx) -> dict:
    """Fold Aaron's answer to a queued L3 ask back INTO the ladder — the return path _queue_l3_ask
    lacked. His answer is written to the New Brain via remember() with L3/utterance provenance (his
    own words are tier-2 ground truth), and the matching ask is marked resolved in l3_asks.jsonl so
    the loop never re-asks it. `answer` is the text remembered; a no-op if no pending ask matches."""
    asks = _load_l3_asks(ctx.l3_path)
    match = next((a for a in asks if a.get("fact") == fact and not a.get("resolved")), None)
    if match is None:
        return {"resolved": False, "reason": f"no pending L3 ask for fact {fact!r}"}
    if ctx.store is None:
        wb = {"routed": "new_brain", "dry_run": True, "text": answer}   # no store -> dry-run
    else:
        from new_brain_write import remember
        r = remember(ctx.store, answer, source_type="utterance",
                     source_ref=f"L3 answer: {match.get('query', fact)}"[:400], captured_by="ladder")
        wb = {"routed": "new_brain", "result_keys": list(r)}
    match["resolved"] = True
    match["answer"] = answer
    match["resolved_ts"] = datetime.now(timezone.utc).isoformat()
    _write_l3_asks(ctx.l3_path, asks)
    return {"resolved": True, "fact": fact, "write_back": wb}


# ---------------------------------------------------------------------------
# verify — trust-tier aware
# ---------------------------------------------------------------------------
def _judge(null: Null, evidence: list[Evidence], jc: JudgeChecker) -> str:
    joined = "\n---\n".join(e.text for e in evidence)
    claim = f"{null.cls}::{null.member}: {null.value}" if null.is_code_fact else null.query
    prompt = (
        "You are checking a claim against real evidence (source code, docs, or captured notes). "
        "IMPLEMENTATION, when present, shows what the code actually does.\n"
        f"EVIDENCE:\n{joined}\n\n"
        f'CLAIM: "{claim}"\n\n'
        "Answer with exactly one word:\n"
        "- CONTRADICTED  if the evidence shows the claim is false.\n"
        "- CONSISTENT    if the evidence confirms the claim.\n"
        "- UNSURE        only if the evidence genuinely does not show enough to decide."
    )
    u = _ollama_generate(jc.model, prompt).upper()
    if "CONTRADICT" in u:
        return "refuted"
    if "CONSISTENT" in u:
        return "verified"
    return "unsure"


def verify(null: Null, acq: Acquisition, ctx: Ctx) -> tuple[str, int]:
    """Trust-tier rules. L0/L1 verify trinary as the loop already does. L2 adds the asymmetry:
    tier-4-only evidence that is NOT corroborated across independent sources returns 'document'
    (a lead, not a fact) — the fact layer's purity outranks coverage."""
    best_tier = min(e.trust_tier for e in acq.evidence)
    verdict = _judge(null, acq.evidence, ctx.jc)
    if acq.rung == "L2":
        tiers = {e.trust_tier for e in acq.evidence}
        independent = len({e.source for e in acq.evidence}) >= 2
        ground_truth = any(t <= 1 for t in tiers)
        if tiers <= {4} and not independent and not ground_truth and verdict == "verified":
            return "document", best_tier
    return verdict, best_tier


# ---------------------------------------------------------------------------
# climb — cheapest-first, only as far as it must
# ---------------------------------------------------------------------------
def _rungs_for(null: Null):
    if null.personal:
        return [("L3", acquire_l3)]                             # only Aaron's call reaches Aaron
    return [("L0", acquire_l0), ("L1", acquire_l1), ("L2", acquire_l2)]


def climb(null: Null, ctx: Ctx, budget: dict | None = None) -> Outcome:
    budget = budget or DEFAULT_BUDGET
    # ownership/location claims are INHERENTLY NULL: a spawnable component can live on ANY actor,
    # so "where it lives" is a design decision the source cannot answer. PARK — never judge it
    # (the judge would false-refute legitimate design choices). Mirrors learn_null's guard, which
    # exists precisely because a direct judge call bypasses JudgeChecker.check's own OWNERSHIP guard.
    if null.is_code_fact and null.value and JudgeChecker.OWNERSHIP_RE.search(null.value):
        return Outcome(null, None, "abstain", [], None,
                       "inherently null — ownership/location is a design decision, not source-verifiable")
    for name, rung in _rungs_for(null):
        acq = rung(null, ctx, budget)
        if acq is None:
            continue
        if name == "L3":
            vd = "blocking" if acq.note.startswith("INTERRUPT") else "batched"
            return Outcome(null, "L3", vd, [], None, acq.note)
        if not acq.evidence:
            continue
        verdict, tier = verify(null, acq, ctx)
        if verdict in ("verified", "refuted", "document"):
            return Outcome(null, acq.rung, verdict, acq.evidence, tier, acq.note)
        # unsure -> climb higher (a costlier rung may hold conclusive evidence)
    return Outcome(null, None, "abstain", [], None, "honest null — no rung resolved it")


# ---------------------------------------------------------------------------
# write-back — routes BY FACT CLASS (Fable ruling)
# ---------------------------------------------------------------------------
def _already_learned() -> set:
    if not LEARNED.exists():
        return set()
    lines = [l for l in LEARNED.read_text(encoding="utf-8").splitlines() if l.strip()]
    return {(e["cls"], e["member"], e["statement"]) for e in map(json.loads, lines)}


def _write_learned(o: Outcome) -> None:
    entry = {"ts": datetime.now(timezone.utc).isoformat(), "cls": o.null.cls, "member": o.null.member,
             "statement": o.null.value, "verdict": o.verdict, "rung": o.rung, "trust_tier": o.trust_tier,
             "evidence_source": "; ".join(e.source for e in o.evidence)}
    with LEARNED.open("a", encoding="utf-8") as f:
        f.write(json.dumps(entry) + "\n")


def _newbrain_text(o: Outcome) -> str:
    src = "; ".join(e.source for e in o.evidence)
    claim = f"{o.null.cls}::{o.null.member}: {o.null.value}" if o.null.member else o.null.query
    stance = {"verified": "is confirmed", "refuted": "is contradicted",
              "document": "is suggested (uncorroborated lead)"}[o.verdict]
    return f"{claim} — {stance} by {o.rung} evidence (trust tier {o.trust_tier}): {src}."


def _reverify_cadence(tier: int | None) -> str:
    """Re-verify cadence hint by trust tier: external-authoritative (3) ages and should be
    rechecked periodically; external-community (4) must be re-verified before it's relied on.
    Ground-truth / documented-intent (<=2) carries no external cadence."""
    return {3: "30d", 4: "before-relying"}.get(tier or 0, "none")


def write_back(o: Outcome, ctx: Ctx) -> dict:
    """L0/L1 code-behavior -> learned_facts.jsonl (the checker's evidence store). Everything else
    that produced an answer -> New Brain (typed, conflict-checked, external-trust provenance). A
    batched/blocking/unsure/abstain outcome writes NOTHING (no answer yet)."""
    if o.verdict not in ("verified", "refuted", "document"):
        return {"routed": None, "reason": o.verdict}

    if o.null.is_code_fact:
        # a value-less code null (e.g. EXISTS existence question) has no behavior statement to
        # persist — the checker serves learned facts by their statement, and None is not one
        if not o.null.value:
            return {"routed": None, "reason": "value-less code null (no behavior statement)"}
        # answered LOCALLY (L0/L1) -> learned_facts.jsonl, the checker's evidence store
        if o.rung in ("L0", "L1") and o.verdict in ("verified", "refuted"):
            if (o.null.cls, o.null.member, o.null.value) in _already_learned():
                return {"routed": "learned_facts.jsonl", "skipped": "already learned"}
            _write_learned(o)
            return {"routed": "learned_facts.jsonl", "verdict": o.verdict, "rung": o.rung}
        # answered EXTERNALLY (L2) or a 'document' lead -> New Brain, carrying external-trust
        # provenance the learned store can't model (the checker reading it back is deferred, fact 55)

    # New Brain path (L2 external, L3, personal). trust:external meta + re-verify cadence for L2.
    text = _newbrain_text(o)
    src_ref = "; ".join(e.source for e in o.evidence)[:400]
    external = o.rung == "L2"                          # L2 = web evidence; L3/personal = Aaron's words
    stamp = ({"trust": "external", "source_tier": o.trust_tier,
              "reverify_cadence": _reverify_cadence(o.trust_tier)} if external else None)

    # A 'document' verdict is verify()'s tier-4-only, uncorroborated LEAD: stamped into the
    # DOCUMENTS layer, NEVER minted as a fact — the fact layer's purity outranks coverage.
    if o.verdict == "document":
        if ctx.store is None:
            return {"routed": "new_brain_document", "dry_run": True, **(stamp or {}), "text": text}
        from new_brain_store import embed_one
        doc_id = ctx.store.insert_document(text, src_ref, embed_one(text), stamp or {})
        return {"routed": "new_brain_document", "document_id": doc_id, **(stamp or {})}

    if ctx.store is None:
        return {"routed": "new_brain", "dry_run": True, **(stamp or {}), "text": text}
    from new_brain_write import remember
    src_type = "url" if external else "utterance"
    r = remember(ctx.store, text, source_type=src_type, source_ref=src_ref, captured_by="ladder",
                 extra_meta=stamp)
    return {"routed": "new_brain", "verdict": o.verdict, "rung": o.rung,
            **({"trust": "external", "source_tier": o.trust_tier} if external else {}),
            "result_keys": list(r)}


# ---------------------------------------------------------------------------
# triage — usage frequency x typicality (typicality is a SEAM)
# ---------------------------------------------------------------------------
TRIAGE_FLOOR = 1                     # placeholder; below this a null parks as an honest abstain


def usage_counts(logs: list[Path] = FIELD_LOGS) -> dict[str, int]:
    counts: dict[str, int] = {}
    for log in logs:
        if not log.exists():
            continue
        for line in log.read_text(encoding="utf-8").splitlines():
            for r in json.loads(line).get("results", []):
                if r.get("verdict") == "abstain":
                    counts[r["fact"]] = counts.get(r["fact"], 0) + 1
    return counts


def triage_score(null: Null, counts: dict[str, int], typicality: float | None = None) -> float:
    """score = usage_frequency x typicality. usage_frequency is live (field-log abstain counts).
    typicality is the SEAM: the per-fact density signal (0.78 AUROC, shelved-JEPA) — defaults to
    1.0 until wired, so scoring degrades to pure usage-frequency, never fabricates a signal."""
    usage = counts.get(null.fact, 1)
    return usage * (typicality if typicality is not None else 1.0)


# ---------------------------------------------------------------------------
# detect — abstains from the field log, as Nulls
# ---------------------------------------------------------------------------
def abstained_nulls(logs: list[Path] = FIELD_LOGS) -> list[Null]:
    out, seen = [], set()
    for log in logs:
        if not log.exists():
            continue
        for line in log.read_text(encoding="utf-8").splitlines():
            for r in json.loads(line).get("results", []):
                if r.get("verdict") != "abstain" or r["fact"] in seen:
                    continue
                seen.add(r["fact"])
                m = FACT_RE.match(r["fact"])
                if m:
                    _, cls, member, value = m.groups()
                    q = f"What does {cls}::{member} do? {value or ''}".strip()
                    out.append(Null(r["fact"], q, cls=cls, member=member, value=value))
                else:
                    out.append(Null(r["fact"], r["fact"]))
    return out


def _log(msg: str) -> None:
    print(msg, file=sys.stderr, flush=True)


# ---------------------------------------------------------------------------
# demo — climb every abstain, triaged, and route write-back
# ---------------------------------------------------------------------------
def main() -> int:
    store = None
    try:                                                        # New Brain is optional for the demo
        from new_brain_store import NewBrainStore
        store = NewBrainStore()
    except Exception as e:
        _log(f"(New Brain store unavailable — L2/personal write-back dry-runs: {e})")

    jc = JudgeChecker()
    ctx = Ctx(jc=jc, store=store, search=make_claude_search())   # L2 = claude -p + WebSearch (fact 113)
    counts = usage_counts()
    nulls = sorted(abstained_nulls(), key=lambda n: -triage_score(n, counts))

    climbed = parked = 0
    for null in nulls:
        score = triage_score(null, counts)
        if score < TRIAGE_FLOOR:
            parked += 1
            _log(f"  PARK (score {score}): {null.fact[:70]}")
            continue
        o = climb(null, ctx)
        wb = write_back(o, ctx)
        climbed += 1
        _log(f"  {o.verdict.upper():9} @{o.rung or '-':3} (score {score}) "
             f"-> {wb.get('routed')}: {null.fact[:60]}")
    _log(f"\n{climbed} climbed | {parked} parked below triage floor | {len(nulls)} abstains total")
    if store is not None:
        store.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())