Merge pull request #410 from AssetOverflow/feat/teaching-coverage-cli

feat(D): core teaching coverage — per-shape admission histogram

core/cli.py

@@ -69,6 +69,7 @@ _TEST_SUITES: dict[str, tuple[str, ...]] = {
         "tests/test_adr_0172_w5_inference_proposal.py",
         "tests/test_math_frame_ratification.py",
         "tests/test_math_composition_ratification.py",
+        "tests/test_teaching_coverage_cli.py",
     ),
     "packs": (
         "tests/test_core_semantic_seed_pack.py",
@@ -1994,6 +1995,118 @@ def cmd_teaching_seed_recognizer(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_teaching_coverage(args: argparse.Namespace) -> int:
+    """Brief D — per-shape admission histogram with deltas vs committed baseline.
+
+    Reads (or runs, if ``--run``) the lane's ``report.json`` and emits
+    a clean histogram of counts + refusal taxonomy. Pure read by default.
+    Useful for measuring the effect of ratifications + matcher
+    extensions without re-eyeballing report.json.
+    """
+    from pathlib import Path
+
+    from teaching.coverage import (
+        build_coverage_report,
+        fetch_committed_baseline,
+    )
+
+    lane = args.lane or "gsm8k_math"
+    split = args.split or "train_sample"
+    version = args.version or "v1"
+
+    # Validate inputs against a strict whitelist before any path
+    # construction or subprocess invocation. The runner module name is
+    # built from these tokens (``f"evals.{lane}.{split}.{version}.runner"``)
+    # and passed to ``python -m``. Python's module loader would reject
+    # most malicious payloads, but a strict whitelist is the defense-in-
+    # depth response to the Sourcery security advisory: reject
+    # everything except ``[a-z0-9_]+``.
+    import re as _re
+    _safe_token_re = _re.compile(r"^[a-z0-9_]+$")
+    for label, value in (("lane", lane), ("split", split), ("version", version)):
+        if not _safe_token_re.match(value):
+            print(
+                f"ERROR: {label}={value!r} must match ^[a-z0-9_]+$",
+                file=sys.stderr,
+            )
+            return 1
+
+    repo_root = Path(__file__).resolve().parent.parent
+    lane_dir = repo_root / "evals" / lane / split / version
+    if not lane_dir.is_dir():
+        print(f"ERROR: lane directory not found: {lane_dir}", file=sys.stderr)
+        return 1
+    report_path = lane_dir / "report.json"
+
+    use_reader = bool(args.use_reader)
+
+    if args.run or not report_path.exists():
+        import subprocess
+        runner_module = f"evals.{lane}.{split}.{version}.runner"
+        runner_args = [sys.executable, "-m", runner_module]
+        if use_reader:
+            runner_args.append("--use-reader")
+        try:
+            subprocess.run(
+                runner_args,
+                cwd=repo_root,
+                check=True,
+                capture_output=True,
+            )
+        except (subprocess.CalledProcessError, FileNotFoundError) as exc:
+            print(f"ERROR: runner failed: {exc}", file=sys.stderr)
+            return 1
+
+    baseline_path = None
+    if args.delta:
+        report_relpath = (
+            f"evals/{lane}/{split}/{version}/report.json"
+        )
+        baseline_path = fetch_committed_baseline(report_relpath, repo_root)
+
+    report = build_coverage_report(
+        report_path,
+        lane=lane,
+        split=split,
+        version=version,
+        use_reader=use_reader,
+        baseline_path=baseline_path,
+    )
+
+    if args.json:
+        print(json.dumps(report.as_dict(), indent=2, sort_keys=True))
+    else:
+        print(f"Lane: {report.lane}/{report.split}/{report.version} (use_reader={report.use_reader})")
+        if report.delta:
+            print(
+                f"Counts: correct={report.counts.correct} "
+                f"refused={report.counts.refused} "
+                f"wrong={report.counts.wrong}  "
+                f"(Δ from HEAD: correct={report.delta['correct']:+d} "
+                f"refused={report.delta['refused']:+d} "
+                f"wrong={report.delta['wrong']:+d})"
+            )
+        else:
+            print(
+                f"Counts: correct={report.counts.correct} "
+                f"refused={report.counts.refused} "
+                f"wrong={report.counts.wrong}"
+            )
+        print()
+        if report.refusal_taxonomy:
+            print("Refusal taxonomy:")
+            for bucket, n in report.refusal_taxonomy.items():
+                print(f"  {n:>3}  {bucket}")
+            print()
+        wrong_ok = "✓" if report.counts.wrong == 0 else "✗"
+        hazard = report.case_0050_verdict
+        print(f"Wrong=0: {wrong_ok}")
+        if hazard is not None:
+            hazard_ok = "✓" if hazard == "refused" else "✗"
+            print(f"Case 0050 hazard pin: {hazard} {hazard_ok}")
+    return 0
+
+
 def cmd_teaching_refusal_taxonomy(args: argparse.Namespace) -> int:
     """ADR-0163 Phase A — categorise refused statements by shape.
 
@@ -4499,6 +4612,36 @@ def build_parser() -> argparse.ArgumentParser:
     )
     teaching_seed_recognizer.set_defaults(func=cmd_teaching_seed_recognizer)
 
+    teaching_coverage = teaching_sub.add_parser(
+        "coverage",
+        help="Brief D — per-shape admission histogram with optional deltas vs HEAD",
+    )
+    teaching_coverage.add_argument(
+        "--lane", default="gsm8k_math", help="eval lane (default: gsm8k_math)",
+    )
+    teaching_coverage.add_argument(
+        "--split", default="train_sample", help="lane split (default: train_sample)",
+    )
+    teaching_coverage.add_argument(
+        "--version", default="v1", help="lane version (default: v1)",
+    )
+    teaching_coverage.add_argument(
+        "--use-reader", action="store_true",
+        help="pass --use-reader to the runner (matches train_sample default)",
+    )
+    teaching_coverage.add_argument(
+        "--run", action="store_true",
+        help="re-run the lane's runner even if report.json exists",
+    )
+    teaching_coverage.add_argument(
+        "--delta", action="store_true",
+        help="compute delta vs the report.json committed at HEAD",
+    )
+    teaching_coverage.add_argument(
+        "--json", action="store_true", help="emit machine-readable JSON",
+    )
+    teaching_coverage.set_defaults(func=cmd_teaching_coverage)
+
     teaching_refusal_taxonomy = teaching_sub.add_parser(
         "refusal-taxonomy",
         help="ADR-0163 Phase A — categorise refused statements by shape",

teaching/coverage.py

@@ -0,0 +1,203 @@
+"""Brief D — coverage report aggregator for math eval lanes.
+
+Reads the lane's ``report.json`` and emits a per-ShapeCategory
+refusal histogram with optional delta-vs-committed-baseline. Pure
+read; no side effects on lane state. Used by
+``core teaching coverage``.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import tempfile
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Mapping
+
+
+_REASON_CATEGORY_RE: re.Pattern[str] = re.compile(r"\(category=([a-z_]+)\)")
+
+
+@dataclass(frozen=True, slots=True)
+class CoverageCounts:
+    correct: int
+    refused: int
+    wrong: int
+
+    def total(self) -> int:
+        return self.correct + self.refused + self.wrong
+
+
+@dataclass(frozen=True, slots=True)
+class CoverageReport:
+    lane: str
+    split: str
+    version: str
+    counts: CoverageCounts
+    refusal_taxonomy: Mapping[str, int]
+    case_0050_verdict: str | None
+    use_reader: bool
+    delta: Mapping[str, int] = field(default_factory=dict)
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "lane": self.lane,
+            "split": self.split,
+            "version": self.version,
+            "counts": {
+                "correct": self.counts.correct,
+                "refused": self.counts.refused,
+                "wrong": self.counts.wrong,
+                "total": self.counts.total(),
+            },
+            "refusal_taxonomy": dict(self.refusal_taxonomy),
+            "case_0050_verdict": self.case_0050_verdict,
+            "use_reader": self.use_reader,
+            "delta": dict(self.delta),
+        }
+
+
+def _classify_refusal(reason: str) -> str:
+    """Map a per-case refusal reason string to a stable category bucket.
+
+    Matching is case-insensitive throughout — the raw runner output is
+    consistently lowercase prose today, but normalizing once avoids
+    drift if upstream casing changes.
+
+    Buckets:
+    - ``recognizer_empty_injection(<ShapeCategory>)`` — recognizer
+      matched but the per-category injector returned empty
+    - ``no_admissible_question`` — statement(s) admitted; question
+      parser refused
+    - ``no_admissible_statement`` — neither parser nor recognizer
+      admitted any statement
+    - ``unexpected_question_count`` — !=1 question sentence
+    - ``other`` — any unmatched reason text
+    """
+    if not reason:
+        return "other"
+    lower = reason.lower()
+    if "recognizer matched but produced no injection" in lower:
+        m = _REASON_CATEGORY_RE.search(lower)
+        cat = m.group(1) if m else "unknown"
+        return f"recognizer_empty_injection({cat})"
+    if "no admissible candidate for question" in lower:
+        return "no_admissible_question"
+    if "no admissible candidate for statement" in lower:
+        return "no_admissible_statement"
+    if "expected exactly one question sentence" in lower:
+        return "unexpected_question_count"
+    return "other"
+
+
+def build_coverage_report(
+    report_path: Path,
+    *,
+    lane: str,
+    split: str,
+    version: str,
+    use_reader: bool,
+    baseline_path: Path | None = None,
+) -> CoverageReport:
+    """Build a :class:`CoverageReport` from a runner-emitted report.json.
+
+    Optional ``baseline_path`` enables a delta computation.
+    """
+    if not report_path.exists():
+        raise FileNotFoundError(f"report.json not found at {report_path}")
+    data = json.loads(report_path.read_text(encoding="utf-8"))
+    counts_raw = data.get("counts") or {}
+    counts = CoverageCounts(
+        correct=int(counts_raw.get("correct", 0)),
+        refused=int(counts_raw.get("refused", 0)),
+        wrong=int(counts_raw.get("wrong", 0)),
+    )
+
+    per_case = data.get("per_case") or []
+    taxonomy: dict[str, int] = {}
+    case_0050_verdict: str | None = None
+    for case in per_case:
+        verdict = case.get("verdict")
+        cid = case.get("case_id", "")
+        if cid.endswith("-0050"):
+            case_0050_verdict = verdict
+        if verdict != "refused":
+            continue
+        bucket = _classify_refusal(case.get("reason") or "")
+        taxonomy[bucket] = taxonomy.get(bucket, 0) + 1
+
+    # Sort taxonomy by count desc, then alpha for stable output.
+    sorted_taxonomy = dict(
+        sorted(taxonomy.items(), key=lambda kv: (-kv[1], kv[0]))
+    )
+
+    delta: dict[str, int] = {}
+    if baseline_path is not None and baseline_path.exists():
+        try:
+            base_data = json.loads(baseline_path.read_text(encoding="utf-8"))
+            base_counts = base_data.get("counts") or {}
+            delta = {
+                "correct": counts.correct - int(base_counts.get("correct", 0)),
+                "refused": counts.refused - int(base_counts.get("refused", 0)),
+                "wrong": counts.wrong - int(base_counts.get("wrong", 0)),
+            }
+        except (json.JSONDecodeError, KeyError, TypeError):
+            delta = {}
+
+    return CoverageReport(
+        lane=lane,
+        split=split,
+        version=version,
+        counts=counts,
+        refusal_taxonomy=sorted_taxonomy,
+        case_0050_verdict=case_0050_verdict,
+        use_reader=use_reader,
+        delta=delta,
+    )
+
+
+def fetch_committed_baseline(
+    report_relpath: str,
+    repo_root: Path,
+) -> Path | None:
+    """Return a temp path containing HEAD's committed report.json, or None.
+
+    Uses ``git show HEAD:<relpath>``. Falls back to None on any git
+    error so the CLI doesn't depend on git availability.
+    """
+    try:
+        result = subprocess.run(
+            ["git", "-C", str(repo_root), "show", f"HEAD:{report_relpath}"],
+            capture_output=True,
+            check=True,
+            text=True,
+        )
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return None
+    if not result.stdout.strip():
+        return None
+    # Use the system temp dir with a unique filename to avoid:
+    # (a) failures in non-git checkouts or worktrees where .git is a
+    #     file pointing elsewhere
+    # (b) concurrent-access collisions if two operators run
+    #     ``core teaching coverage --delta`` simultaneously
+    fd, tmp_path = tempfile.mkstemp(
+        prefix="core_coverage_baseline_", suffix=".json"
+    )
+    try:
+        with open(fd, "w", encoding="utf-8") as fh:
+            fh.write(result.stdout)
+    except Exception:
+        Path(tmp_path).unlink(missing_ok=True)
+        return None
+    return Path(tmp_path)
+
+
+__all__ = [
+    "CoverageCounts",
+    "CoverageReport",
+    "build_coverage_report",
+    "fetch_committed_baseline",
+]

tests/test_teaching_coverage_cli.py

@@ -0,0 +1,209 @@
+"""Brief D — coverage report aggregator tests."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from teaching.coverage import (
+    CoverageCounts,
+    _classify_refusal,
+    build_coverage_report,
+)
+
+
+def _write_report(path: Path, *, correct: int, refused: int, wrong: int, per_case: list[dict]) -> None:
+    path.write_text(
+        json.dumps({
+            "counts": {"correct": correct, "refused": refused, "wrong": wrong},
+            "per_case": per_case,
+        }),
+        encoding="utf-8",
+    )
+
+
+def test_classify_recognizer_empty_injection():
+    r = ("candidate_graph: recognizer matched but produced no injection "
+         "for statement: 'X.' (category=multiplicative_aggregation)")
+    assert _classify_refusal(r) == "recognizer_empty_injection(multiplicative_aggregation)"
+
+
+def test_classify_no_admissible_question():
+    assert _classify_refusal("candidate_graph: no admissible candidate for question: 'X?'") == "no_admissible_question"
+
+
+def test_classify_no_admissible_statement():
+    assert _classify_refusal("candidate_graph: no admissible candidate for statement: 'X.'") == "no_admissible_statement"
+
+
+def test_classify_unknown_falls_back_to_other():
+    assert _classify_refusal("some new failure mode") == "other"
+
+
+def test_classify_empty_returns_other():
+    assert _classify_refusal("") == "other"
+
+
+def test_build_coverage_report_basic(tmp_path: Path):
+    report_path = tmp_path / "report.json"
+    _write_report(
+        report_path,
+        correct=2,
+        refused=3,
+        wrong=0,
+        per_case=[
+            {"case_id": "x-0001", "verdict": "correct"},
+            {"case_id": "x-0002", "verdict": "correct"},
+            {"case_id": "x-0003", "verdict": "refused", "reason": "candidate_graph: recognizer matched but produced no injection for statement: 'A.' (category=currency_amount)"},
+            {"case_id": "x-0004", "verdict": "refused", "reason": "candidate_graph: recognizer matched but produced no injection for statement: 'B.' (category=currency_amount)"},
+            {"case_id": "x-0005", "verdict": "refused", "reason": "candidate_graph: no admissible candidate for question: 'How?'"},
+        ],
+    )
+    r = build_coverage_report(
+        report_path, lane="t", split="s", version="v1", use_reader=True,
+    )
+    assert r.counts == CoverageCounts(correct=2, refused=3, wrong=0)
+    assert r.counts.total() == 5
+    assert dict(r.refusal_taxonomy) == {
+        "recognizer_empty_injection(currency_amount)": 2,
+        "no_admissible_question": 1,
+    }
+    assert r.case_0050_verdict is None
+
+
+def test_case_0050_verdict_captured(tmp_path: Path):
+    report_path = tmp_path / "report.json"
+    _write_report(
+        report_path,
+        correct=0,
+        refused=1,
+        wrong=0,
+        per_case=[
+            {"case_id": "gsm8k-train-sample-v1-0050", "verdict": "refused", "reason": "x"},
+        ],
+    )
+    r = build_coverage_report(
+        report_path, lane="gsm8k_math", split="train_sample", version="v1", use_reader=True,
+    )
+    assert r.case_0050_verdict == "refused"
+
+
+def test_delta_computation(tmp_path: Path):
+    report_path = tmp_path / "report.json"
+    baseline_path = tmp_path / "baseline.json"
+    _write_report(report_path, correct=4, refused=46, wrong=0, per_case=[])
+    _write_report(baseline_path, correct=3, refused=47, wrong=0, per_case=[])
+    r = build_coverage_report(
+        report_path,
+        lane="t", split="s", version="v1", use_reader=True,
+        baseline_path=baseline_path,
+    )
+    assert r.delta == {"correct": 1, "refused": -1, "wrong": 0}
+
+
+def test_no_baseline_means_empty_delta(tmp_path: Path):
+    report_path = tmp_path / "report.json"
+    _write_report(report_path, correct=3, refused=47, wrong=0, per_case=[])
+    r = build_coverage_report(
+        report_path,
+        lane="t", split="s", version="v1", use_reader=False,
+        baseline_path=None,
+    )
+    assert r.delta == {}
+
+
+def test_missing_report_raises(tmp_path: Path):
+    with pytest.raises(FileNotFoundError):
+        build_coverage_report(
+            tmp_path / "nope.json",
+            lane="t", split="s", version="v1", use_reader=False,
+        )
+
+
+def test_as_dict_round_trip(tmp_path: Path):
+    report_path = tmp_path / "report.json"
+    _write_report(report_path, correct=1, refused=1, wrong=0, per_case=[
+        {"case_id": "x-0001", "verdict": "correct"},
+        {"case_id": "x-0002", "verdict": "refused", "reason": "candidate_graph: recognizer matched but produced no injection for statement: 'X.' (category=discrete_count_statement)"},
+    ])
+    r = build_coverage_report(report_path, lane="t", split="s", version="v1", use_reader=True)
+    d = r.as_dict()
+    assert d["counts"]["total"] == 2
+    assert "recognizer_empty_injection(discrete_count_statement)" in d["refusal_taxonomy"]
+    assert d["use_reader"] is True
+
+
+def test_taxonomy_sorted_by_count_desc(tmp_path: Path):
+    report_path = tmp_path / "report.json"
+    _write_report(report_path, correct=0, refused=4, wrong=0, per_case=[
+        {"case_id": f"x-{i:04d}", "verdict": "refused", "reason": "candidate_graph: no admissible candidate for question: '?'" if i < 1 else "candidate_graph: recognizer matched but produced no injection for statement: 'X.' (category=multiplicative_aggregation)"}
+        for i in range(4)
+    ])
+    r = build_coverage_report(report_path, lane="t", split="s", version="v1", use_reader=False)
+    keys = list(r.refusal_taxonomy.keys())
+    assert keys[0] == "recognizer_empty_injection(multiplicative_aggregation)"  # 3 > 1
+    assert keys[1] == "no_admissible_question"
+
+
+def test_wrong_zero_invariant_visible_via_as_dict(tmp_path: Path):
+    report_path = tmp_path / "report.json"
+    _write_report(report_path, correct=0, refused=0, wrong=2, per_case=[])
+    r = build_coverage_report(report_path, lane="t", split="s", version="v1", use_reader=False)
+    assert r.as_dict()["counts"]["wrong"] == 2
+
+
+# ---------------------------------------------------------------------------
+# Sourcery review follow-ups (PR #410 round 1)
+# ---------------------------------------------------------------------------
+
+
+def test_classify_refusal_is_case_insensitive():
+    """Mixed-case reason text classifies into the same bucket."""
+    mixed = (
+        "Candidate_Graph: Recognizer Matched But Produced No Injection "
+        "for statement: 'X.' (category=currency_amount)"
+    )
+    assert _classify_refusal(mixed) == "recognizer_empty_injection(currency_amount)"
+
+
+def test_classify_docstring_matches_implementation_buckets():
+    """The docstring's bucket list matches the values _classify_refusal returns."""
+    expected_buckets = {
+        "recognizer_empty_injection",
+        "no_admissible_question",
+        "no_admissible_statement",
+        "unexpected_question_count",
+        "other",
+    }
+    doc = _classify_refusal.__doc__ or ""
+    for bucket in expected_buckets:
+        assert bucket in doc, f"docstring missing bucket: {bucket}"
+
+
+def test_fetch_committed_baseline_uses_system_temp(tmp_path, monkeypatch):
+    """The baseline temp file lives in tempfile.gettempdir(), not .git/."""
+    import tempfile as _tempfile
+    from teaching.coverage import fetch_committed_baseline
+
+    class _FakeCompleted:
+        def __init__(self, stdout: str) -> None:
+            self.stdout = stdout
+
+    import subprocess as _sp
+
+    fake_baseline = '{"counts": {"correct": 1, "refused": 1, "wrong": 0}}'
+
+    def _fake_run(*_args, **_kwargs):
+        return _FakeCompleted(fake_baseline)
+
+    monkeypatch.setattr(_sp, "run", _fake_run)
+
+    out = fetch_committed_baseline("evals/x/report.json", tmp_path)
+    assert out is not None
+    sys_temp = Path(_tempfile.gettempdir()).resolve()
+    assert str(out.resolve()).startswith(str(sys_temp)), (
+        f"baseline temp must be under {sys_temp}, got {out!r}"
+    )
+    out.unlink(missing_ok=True)