ladder_test.py

"""ladder_test.py — deterministic Acquisition Frontier tests.

The judge (local LLM) and the web are STUBBED so every rung's logic is exercised without
Ollama, a network, or the New Brain DB. Covers: allowlist tiering + exclude, off-list candidate
queue, the L2 tier-4-only -> DOCUMENT asymmetry (and corroboration escaping it), climb ordering
(L0 answers before L1/L2 fire), personal -> L3 batching + blocking interrupt, triage score, and
the write-back routing discriminator (code -> learned_facts, external -> New Brain).

Run:  .venv/Scripts/python.exe ladder_test.py           # no DB / Ollama / network needed
"""

from __future__ import annotations

import json
import os
import sys
import tempfile
from pathlib import Path

import ladder
from ladder import (Acquisition, Ctx, Evidence, Null, allow_tier, climb, load_allowlist,
                    triage_score, verify, write_back)


class FakeJC:
    """Stand-in for JudgeChecker: fixed docs/defs, and a scripted judge verdict."""
    model = "stub"

    def __init__(self, docs=None, defs=None):
        self.docs = docs or {}
        self.defs = defs or {}


def with_judge(verdict: str):
    """Patch ladder._judge to a fixed verdict for the duration of one assertion block."""
    ladder._judge = lambda null, evidence, jc: verdict


def main() -> int:
    fails = []
    entries = load_allowlist()

    # --- allowlist: tiering + path-level exclude + off-list ---
    cases = [
        ("https://dev.epicgames.com/documentation/en-us/unreal-engine/foo", 3),
        ("https://dev.epicgames.com/community/snippets/xyz", 4),
        ("https://forums.unrealengine.com/t/some-thread/123", 4),
        ("https://forums.unrealengine.com/c/off-topic/random", None),   # excluded sub-glob
        ("https://raw.githubusercontent.com/lib/repo/main/src.py", 1),
        ("https://malware.example.com/whatever", None),                 # off-list
    ]
    for url, want in cases:
        got = allow_tier(url, entries)
        if got != want:
            fails.append(f"allow_tier({url}) = {got}, want {want}")

    # --- L2 asymmetry: tier-4-only + uncorroborated + judge 'verified' -> DOCUMENT, not fact ---
    with_judge("verified")
    ctx = Ctx(jc=FakeJC())
    n = Null("BEHAVIOR: UFoo::Bar = does X", "does UFoo::Bar do X?", cls="UFoo", member="Bar", value="does X")
    acq_t4 = Acquisition("L2", [Evidence("forums.unrealengine.com/t/1", "some forum text", 4)])
    vd, tier = verify(n, acq_t4, ctx)
    if vd != "document":
        fails.append(f"L2 tier-4-only should be 'document', got {vd}")
    # corroborated across two independent tier-4 sources -> escapes to a real verdict
    acq_corrob = Acquisition("L2", [Evidence("forums.unrealengine.com/t/1", "x", 4),
                                    Evidence("stackoverflow.com/q/2", "x", 4)])
    vd2, _ = verify(n, acq_corrob, ctx)
    if vd2 != "verified":
        fails.append(f"corroborated L2 should verify, got {vd2}")

    # --- climb ordering: L0 has the answer, so L1/L2 must NOT fire ---
    with_judge("verified")
    jc = FakeJC(docs={("UFoo", "Bar"): "Bar does X"}, defs={})
    fired = []
    orig = (ladder.acquire_l1, ladder.acquire_l2)
    ladder.acquire_l1 = lambda *a: fired.append("L1")
    ladder.acquire_l2 = lambda *a: fired.append("L2")
    o = climb(n, Ctx(jc=jc))
    ladder.acquire_l1, ladder.acquire_l2 = orig
    if not (o.rung == "L0" and o.verdict == "verified" and not fired):
        fails.append(f"climb should stop at L0, got rung={o.rung} verdict={o.verdict} fired={fired}")

    # --- personal null -> L3 batched (writes an ask, no code rung) ---
    with tempfile.TemporaryDirectory() as td:
        l3 = Path(td) / "l3.jsonl"
        pctx = Ctx(jc=FakeJC(), l3_path=l3)
        pnull = Null("Should we use Postgres or SQLite?", "Postgres or SQLite?", personal=True)
        po = climb(pnull, pctx)
        if not (po.rung == "L3" and po.verdict == "batched" and l3.exists()):
            fails.append(f"personal null should batch to L3, got {po}")
        # blocking personal null -> interrupt, NOT batched
        bnull = Null("blocking decision", "blocking?", personal=True, blocking=True)
        bo = climb(bnull, pctx)
        if not (bo.verdict == "blocking" and bo.note.startswith("INTERRUPT")):
            fails.append(f"blocking null should interrupt, got {bo}")

    # --- off-list L2 source -> candidate queue (never silently used/dropped) ---
    with tempfile.TemporaryDirectory() as td:
        cand = Path(td) / "cand.jsonl"
        sctx = Ctx(jc=FakeJC(), candidate_path=cand,
                   search=lambda q, k: ["https://malware.example.com/x"],
                   fetch=lambda u: "should not be fetched")
        acq = ladder.acquire_l2(n, sctx, ladder.DEFAULT_BUDGET)
        if not (cand.exists() and "off-list" in (acq.note if acq else "")):
            fails.append(f"off-list source should queue as candidate, got {acq}")

    # --- triage score = usage x typicality (seam defaults to 1.0) ---
    counts = {"BEHAVIOR: UFoo::Bar = does X": 3}
    if triage_score(n, counts) != 3:
        fails.append(f"triage usage-only should be 3, got {triage_score(n, counts)}")
    if triage_score(n, counts, typicality=0.5) != 1.5:
        fails.append(f"triage with typicality 0.5 should be 1.5, got {triage_score(n, counts, 0.5)}")

    # --- routing discriminator: code L0 -> learned_facts; external L2 -> new_brain (dry-run) ---
    code_out = ladder.Outcome(n, "L0", "verified", [Evidence("UFoo::Bar", "x", 1)], 1)
    ext_null = Null("some external claim", "external claim?")
    ext_out = ladder.Outcome(ext_null, "L2", "verified", [Evidence("docs.python.org/x", "x", 3)], 3)
    saved = ladder._write_learned
    wrote = []
    ladder._write_learned = lambda o: wrote.append(o)
    r_code = write_back(code_out, Ctx(jc=FakeJC()))
    r_ext = write_back(ext_out, Ctx(jc=FakeJC(), store=None))
    ladder._write_learned = saved
    if not (r_code.get("routed") == "learned_facts.jsonl" and wrote):
        fails.append(f"code fact should route to learned_facts, got {r_code}")
    if not (r_ext.get("routed") == "new_brain" and r_ext.get("dry_run")):
        fails.append(f"external fact should route to new_brain, got {r_ext}")

    # --- ownership/location claim is INHERENTLY NULL: parked, never judged/written ---
    with_judge("refuted")   # judge would refute if reached — it must NOT be reached
    own = Null("BEHAVIOR: ?::InventoryComponent = lives on the Pawn", "does it live on the Pawn?",
               cls="?", member="InventoryComponent", value="lives on the Pawn, rebuilt each spawn")
    oo = climb(own, Ctx(jc=FakeJC(docs={("X", "InventoryComponent"): "d"})))
    if not (oo.verdict == "abstain" and "ownership" in oo.note):
        fails.append(f"ownership claim should park as inherently null, got {oo}")

    # --- value-less code null (EXISTS) writes NOTHING (no behavior statement) ---
    exists_out = ladder.Outcome(Null("EXISTS: UBar::Baz", "does UBar::Baz exist?", cls="UBar",
                                      member="Baz", value=None), "L0", "refuted",
                                [Evidence("UBar::Baz", "x", 1)], 1)
    r_exists = write_back(exists_out, Ctx(jc=FakeJC()))
    if r_exists.get("routed") is not None:
        fails.append(f"value-less code null should write nothing, got {r_exists}")

    # --- L2 search provider (task 10): claude -p + WebSearch, subscription-scrubbed, URL parsing ---
    class _P:
        def __init__(self, stdout): self.stdout = stdout
    seen_cmd = {}
    def fake_claude(cmd, timeout):
        seen_cmd["cmd"] = cmd
        return _P(json.dumps({"result": "Sources:\nhttps://docs.example.com/a.\nhttps://forums.example.com/b\n"}))
    search = ladder.make_claude_search(model="haiku", runner=fake_claude)
    urls = search("how does X work?", 5)
    if urls != ["https://docs.example.com/a", "https://forums.example.com/b"]:
        fails.append(f"claude -p search should parse candidate URLs (trailing punct stripped), got {urls}")
    cmd = seen_cmd.get("cmd", [])
    if not ("claude" in (cmd[0] or "").lower() and "-p" in cmd and "haiku" in cmd
            and "WebSearch" in cmd and "--output-format" in cmd):
        fails.append(f"claude -p search cmd must carry -p, model, WebSearch, output-format, got {cmd}")
    # auth failure (is_error) -> [] (ladder finds no L2 evidence; NEVER fabricates a source)
    err_search = ladder.make_claude_search(runner=lambda c, t: _P(json.dumps({"is_error": True, "result": "Not logged in"})))
    if err_search("q", 3) != []:
        fails.append("claude -p search must return [] on auth error, not fabricate URLs")
    # k cap honored
    many = ladder.make_claude_search(runner=lambda c, t: _P("https://a.com https://b.com https://c.com"))
    if len(many("q", 2)) != 2:
        fails.append("claude -p search must honor the k cap")
    # subscription env scrubs the paid-API key (facts 109/115 — no silent paid API)
    os.environ["ANTHROPIC_API_KEY"] = "sk-scrub-me"
    try:
        if "ANTHROPIC_API_KEY" in ladder._subscription_env():
            fails.append("ladder _subscription_env must scrub ANTHROPIC_API_KEY (no silent paid API)")
    finally:
        os.environ.pop("ANTHROPIC_API_KEY", None)
    # the injected search feeds acquire_l2's allowlist gate unchanged (off-list URL -> candidate queue)
    with tempfile.TemporaryDirectory() as td:
        cand = Path(td) / "c.jsonl"
        n2 = Null("BEHAVIOR: UFoo::Bar = x", "does UFoo::Bar do x?", cls="UFoo", member="Bar", value="x")
        sctx = Ctx(jc=FakeJC(), candidate_path=cand,
                   search=ladder.make_claude_search(runner=lambda c, t: _P("https://malware.example.com/x")))
        acq = ladder.acquire_l2(n2, sctx, ladder.DEFAULT_BUDGET)
        if not (cand.exists() and acq and "off-list" in acq.note):
            fails.append(f"claude -p search output should flow into acquire_l2's allowlist gate, got {acq}")

    # --- L3 answer-capture: dedup on re-ask, Aaron's answer -> New Brain write-back + resolved ---
    with tempfile.TemporaryDirectory() as td:
        l3 = Path(td) / "l3.jsonl"
        actx = Ctx(jc=FakeJC(), l3_path=l3, store=None)
        pn = Null("Which DB — Postgres or SQLite?", "Postgres or SQLite?", personal=True)
        climb(pn, actx)
        climb(pn, actx)                                          # re-climb must NOT duplicate the ask
        if len(ladder._load_l3_asks(l3)) != 1:
            fails.append(f"L3 ask not deduped on re-climb: {ladder._load_l3_asks(l3)}")
        res = ladder.answer_l3_ask(pn.fact, "Aaron chose Postgres for pgvector support.", actx)
        asks = ladder._load_l3_asks(l3)
        ok = (res.get("resolved") and res["write_back"].get("dry_run")
              and asks and asks[0].get("resolved") is True and asks[0].get("answer"))
        if not ok:
            fails.append(f"L3 answer-capture failed: res={res} asks={asks}")
        climb(pn, actx)                                          # answered ask must NOT be re-asked
        if len(ladder._load_l3_asks(l3)) != 1:
            fails.append("answered L3 ask was re-asked")
        if ladder.answer_l3_ask("no such fact", "x", actx).get("resolved"):
            fails.append("answering a non-existent ask should be a no-op")

    # --- trust-external stamping (task 14): L2 write-back stamps trust:external + tier + cadence ---
    l2_out = ladder.Outcome(Null("ext claim", "ext?"), "L2", "verified",
                            [Evidence("dev.epicgames.com/documentation/x", "x", 3)], 3)
    r_l2 = write_back(l2_out, Ctx(jc=FakeJC(), store=None))
    if not (r_l2.get("trust") == "external" and r_l2.get("source_tier") == 3
            and r_l2.get("reverify_cadence") == "30d"):
        fails.append(f"L2 write-back should stamp trust:external tier3 cadence30d, got {r_l2}")
    # L3/personal write-back is Aaron's own words — NOT external, no stamp
    l3_out = ladder.Outcome(Null("a decision", "decide?", personal=True), "L3", "verified",
                            [Evidence("aaron", "x", 2)], 2)
    if write_back(l3_out, Ctx(jc=FakeJC(), store=None)).get("trust") == "external":
        fails.append("L3/personal write-back must NOT be stamped external")
    # tier-4-only lead: 'document' verdict -> DOCUMENTS layer, never a fact
    doc_out = ladder.Outcome(Null("lead claim", "lead?"), "L2", "document",
                             [Evidence("forums.unrealengine.com/t/9", "x", 4)], 4)
    r_doc = write_back(doc_out, Ctx(jc=FakeJC(), store=None))
    if not (r_doc.get("routed") == "new_brain_document" and r_doc.get("trust") == "external"
            and r_doc.get("reverify_cadence") == "before-relying"):
        fails.append(f"tier-4 document lead should route to documents layer w/ external stamp, got {r_doc}")

    if fails:
        print("FAIL", file=sys.stderr)
        for f in fails:
            print("  -", f, file=sys.stderr)
        return 1
    print("PASS -- ladder: allowlist, L2 asymmetry, climb order, L3 batching + answer-capture, "
          "candidate queue, triage, routing, L2 claude -p search provider, trust-external stamping",
          file=sys.stderr)
    return 0


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