"""Curriculum-serve lane — scores curriculum-grounded exam answering. The Phase-2 lane of the generalization arc. For each committed case, the raw question text runs through the exact production path ``chat/curriculum_surface.py::decide_curriculum_question`` — question reader → subject routing → ratified-curriculum premise compilation → the argument bands → the ROBDD engine — and the resulting verdict is compared to gold authored INDEPENDENTLY by ``evals.curriculum_serve.oracle`` (plan §4.4). Three guards run before any case is scored, and any of them failing is a lane failure, not a case miss: 1. **Provenance** (§4.3) — every chain id a case pins must resolve in the subject's ratified curriculum. A case whose curriculum moved under it breaks loudly instead of quietly answering from what is left. 2. **Corpus soundness** (§4.4) — the independent oracle must agree with every committed gold. A gold nobody can re-derive from the curriculum is not gold. 3. **Anti-recall coverage** (§4.7) — the split MUST contain cases whose answer is true in the world but absent from the curriculum. Without them the lane cannot show the system decodes rather than recalls, and it must not ship. Counts mirror the deduction-serve lane exactly: ``wrong`` (a committed verdict that disagrees with gold) MUST stay 0; a decline where gold expected a verdict is a coverage miss, tracked but never conflated with a confabulation. """ from __future__ import annotations import argparse import json from collections import Counter from pathlib import Path from chat.curriculum_surface import decide_curriculum_question, read_curriculum_question from evals.curriculum_serve.oracle import oracle_answer from teaching.curriculum_premises import load_curriculum, resolve_pinned _ROOT = Path(__file__).resolve().parent _SPLITS: tuple[tuple[str, str, Path], ...] = ( # (split name, subject domain, cases) ("physics", "physics", _ROOT / "physics" / "cases.jsonl"), ) #: A split must carry at least this many anti-recall probes (§4.7). The class #: prefix is the contract: a probe is a case whose answer is true in the world #: and absent from the curriculum, so the honest verdict is a non-commitment. _ANTI_RECALL_PREFIX = "anti_recall" _MIN_ANTI_RECALL = 3 class LaneContractError(AssertionError): """A guard failed — the lane itself is unsound, not merely under-covered.""" def _load(path: Path) -> list[dict]: with path.open(encoding="utf-8") as fh: return [json.loads(line) for line in fh if line.strip()] def assert_provenance(domain: str, cases: list[dict]) -> None: """Every pinned chain id must resolve in the ratified curriculum (§4.3).""" curriculum = load_curriculum(domain) for case in cases: pinned = tuple(case.get("chains", ())) if not pinned: continue resolve_pinned(curriculum, pinned) # raises UnratifiedChain def assert_corpus_sound(domain: str, cases: list[dict]) -> None: """The independent oracle must re-derive every committed gold (§4.4).""" for case in cases: query = read_curriculum_question(case["text"]) gold = case["gold"] if not hasattr(query, "subject"): # Unreadable question shapes are gold-``declined`` by construction; # the oracle has no opinion on a text it cannot parse either. if gold != "declined": raise LaneContractError( f"{case['id']}: unreadable question with gold={gold}" ) continue verdict = oracle_answer(domain, query.subject, query.verb, query.obj) if verdict.verdict != gold: raise LaneContractError( f"{case['id']}: oracle={verdict.verdict} gold={gold} " f"(family={verdict.family} depth={verdict.depth}) text={case['text']!r}" ) def assert_anti_recall_coverage(cases: list[dict]) -> None: """The split must probe recall, or it must not ship (§4.7).""" probes = [c for c in cases if str(c.get("class", "")).startswith(_ANTI_RECALL_PREFIX)] if len(probes) < _MIN_ANTI_RECALL: raise LaneContractError( f"split has {len(probes)} anti-recall probes, needs >= {_MIN_ANTI_RECALL}" ) for probe in probes: if probe["gold"] in ("entailed", "refuted"): raise LaneContractError( f"{probe['id']}: an anti-recall probe cannot have a committed gold" ) def build_report(domain: str, cases: list[dict]) -> dict: assert_provenance(domain, cases) assert_corpus_sound(domain, cases) assert_anti_recall_coverage(cases) counts = Counter({"correct": 0, "wrong": 0, "declined": 0}) by_gold: Counter[str] = Counter() correct_by_gold: Counter[str] = Counter() by_band: Counter[str] = Counter() mismatches: list[dict] = [] for case in cases: gold = case["gold"] by_gold[gold] += 1 decision = decide_curriculum_question(case["text"]) if decision.band: by_band[decision.band] += 1 if decision.verdict == gold: counts["correct"] += 1 correct_by_gold[gold] += 1 elif decision.verdict == "declined": counts["declined"] += 1 mismatches.append( {"id": case["id"], "gold": gold, "got": decision.verdict, "reason": decision.reason, "text": case["text"]} ) else: counts["wrong"] += 1 mismatches.append( {"id": case["id"], "gold": gold, "got": decision.verdict, "reason": decision.reason, "text": case["text"]} ) return { "n": len(cases), "counts": dict(counts), "by_gold": dict(by_gold), "correct_by_gold": dict(correct_by_gold), "by_band": dict(by_band), "anti_recall_probes": sum( 1 for c in cases if str(c.get("class", "")).startswith(_ANTI_RECALL_PREFIX) ), "all_cases_correct": counts["correct"] == len(cases), "mismatch_examples": mismatches[:10], } def build_combined_report() -> dict: """Deterministic per-split + aggregate report — safe to SHA-pin.""" splits: dict[str, dict] = {} aggregate = {"n": 0, "correct": 0, "wrong": 0, "declined": 0} for name, domain, path in _SPLITS: report = build_report(domain, _load(path)) splits[name] = { "n": report["n"], "domain": domain, "counts": report["counts"], "by_gold": report["by_gold"], "correct_by_gold": report["correct_by_gold"], "by_band": report["by_band"], "anti_recall_probes": report["anti_recall_probes"], "all_cases_correct": report["all_cases_correct"], } aggregate["n"] += report["n"] for key in ("correct", "wrong", "declined"): aggregate[key] += report["counts"][key] return { "schema_version": 1, "lane": "curriculum_serve", "arc": "generalization", "splits": splits, "aggregate": aggregate, "wrong_is_zero": aggregate["wrong"] == 0, "all_correct": all(s["all_cases_correct"] for s in splits.values()), } def write_combined_report(path: Path) -> dict: report = build_combined_report() path.write_text( json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) return report def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--report", type=Path, default=None) args = parser.parse_args(argv) if args.report is not None: report = write_combined_report(args.report) return 0 if (report["wrong_is_zero"] and report["all_correct"]) else 1 all_ok = True for name, domain, path in _SPLITS: report = build_report(domain, _load(path)) c = report["counts"] print( f"[{name}] n={report['n']} correct={c['correct']} wrong={c['wrong']} " f"declined_mismatch={c['declined']} anti_recall={report['anti_recall_probes']}" ) for m in report["mismatch_examples"]: print(f" {m['id']}: gold={m['gold']} got={m['got']} " f"reason={m['reason']} text={m['text']!r}") all_ok = all_ok and report["all_cases_correct"] return 0 if all_ok else 1 if __name__ == "__main__": raise SystemExit(main())