diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 08da970..64af83f 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -291,7 +291,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 12.1, 12.2, 12.3, 12.6, 12.8_ - _Boundary: Admission Runner_ - _Depends: 11.5, 12.3_ -- [ ] 12.6 Add paired scoring and the admission report +- [x] 12.6 Add paired scoring and the admission report - Score vulnerable/fixed twins, reviewed ordinary changes, indeterminate rate, coverage, runtime and instruments. - Preserve sample counts and uncertainty intervals and return NO-GO on any pre-registered gate failure. - Done when fixture pairs detect vulnerable-only, fixed-side regression and mislabeled-negative cases. diff --git a/benchmarks/valid/admission_scoring.py b/benchmarks/valid/admission_scoring.py new file mode 100644 index 0000000..86d89c3 --- /dev/null +++ b/benchmarks/valid/admission_scoring.py @@ -0,0 +1,157 @@ +"""Paired scoring and the admission report for the access-control prover arm. + +Task 12.6 of the access-control-prover specification (Reqs 5.2, 12.4, 12.5, +12.7, 12.8, 12.9). The scorer consumes a validated admission manifest plus the +judgments measured on its population: it scores paired vulnerable/fixed twins +and reviewed ordinary changes, keeps sample counts and Wilson 95% intervals on +every rate, and returns NO-GO when the budget gate, the instrument gate or any +pre-registered quality criterion fails. Indeterminate pairs are reported as +indeterminate and never converted to clean results; ordinary changes whose +review found a real missing-authorization defect are refused before scoring. +""" + +from dataclasses import dataclass +from typing import Any + +from benchmarks.unseen import score +from benchmarks.valid.prover_admission import AdmissionManifest + +__all__ = ["AdmissionCase", "AdmissionScoringError", "OrdinaryChange", "admission_report"] + +_JUDGMENTS = ("unguarded", "discharged", "indeterminate") + + +class AdmissionScoringError(ValueError): + """Raised when a population cannot be scored; the message names the reason.""" + + +@dataclass(frozen=True) +class AdmissionCase: + """One scored vulnerable/fixed twin pair.""" + + case_id: str + vulnerable_judgment: str + fixed_judgment: str + location_verified: bool + seconds: float + + +@dataclass(frozen=True) +class OrdinaryChange: + """One reviewed ordinary change forming the false-alarm denominator.""" + + change_id: str + judgment: str + mislabeled_defect: bool + + +def _refuse_inputs( + manifest: AdmissionManifest, + cases: tuple[AdmissionCase, ...], + ordinary: tuple[OrdinaryChange, ...], + coverage: float, +) -> None: + """Refuse unscorable populations; check order is fixed for determinism.""" + if manifest.population_role != "admission": + raise AdmissionScoringError(f"population-role: {manifest.population_role} population cannot be used for admission claims") + for case in cases: + for side, judgment in (("vulnerable", case.vulnerable_judgment), ("fixed", case.fixed_judgment)): + if judgment not in _JUDGMENTS: + raise AdmissionScoringError(f"judgment: {judgment!r} at {case.case_id}/{side}") + for change in ordinary: + if change.judgment not in _JUDGMENTS: + raise AdmissionScoringError(f"judgment: {change.judgment!r} at {change.change_id}") + if change.mislabeled_defect: + raise AdmissionScoringError( + f"mislabeled: ordinary change {change.change_id} contains a real " + "missing-authorization defect and must be removed or relabelled before scoring" + ) + if not 0.0 <= coverage <= 1.0: + raise AdmissionScoringError(f"coverage: {coverage} is outside [0, 1]") + + +def admission_report( + manifest: AdmissionManifest, + cases: tuple[AdmissionCase, ...], + ordinary: tuple[OrdinaryChange, ...], + *, + coverage: float, + instrument_failures: int = 0, + budget_seconds: float = 120.0, + criteria: dict[str, float] | None = None, +) -> dict[str, Any]: + """Score paired twins and reviewed ordinary changes; return the admission report. + + The report carries every rate with its sample count and Wilson 95% interval + and a GO/NO-GO decision with the named failed gates. + """ + _refuse_inputs(manifest, cases, ordinary, coverage) + detected = tuple(case.vulnerable_judgment == "unguarded" for case in cases) + pair_correct = tuple(detected[index] and cases[index].fixed_judgment == "discharged" for index in range(len(cases))) + flagged = tuple(change.judgment == "unguarded" for change in ordinary) + per_100 = score.rate(flagged) + scaled = { + "n": per_100["n"], + "count": per_100["count"], + "rate": None if per_100["rate"] is None else per_100["rate"] * 100, + "wilson95": [None if value is None else value * 100 for value in per_100["wilson95"]], + } + side_judgments = [case.vulnerable_judgment for case in cases] + side_judgments += [case.fixed_judgment for case in cases] + side_judgments += [change.judgment for change in ordinary] + pair_correctness = score.rate(list(pair_correct)) + fix_recall = score.rate(list(detected)) + verified_location_recall = score.rate([case.location_verified for case in cases if case.vulnerable_judgment == "unguarded"]) + precision = score.rate([*detected, *(False for _ in range(sum(flagged)))]) + indeterminate_rate = score.rate([judgment == "indeterminate" for judgment in side_judgments]) + report: dict[str, Any] = { + "manifest": { + "population_role": manifest.population_role, + "analyzer_revision": manifest.analyzer_revision, + "fact_set_digest": manifest.fact_set_digest, + "acquisition_record_digest": manifest.acquisition_record_digest, + "configuration_digest": manifest.configuration_digest, + "population_digest": manifest.population_digest, + }, + "pairs": {"n": len(cases), "correct": sum(pair_correct)}, + "pair_correctness": pair_correctness, + "fix_recall": fix_recall, + "verified_location_recall": verified_location_recall, + "flagged_ordinary_per_100": scaled, + "precision": precision, + "indeterminate_rate": indeterminate_rate, + "coverage": coverage, + "runtime": { + "n": len(cases), + "mean_seconds": sum(case.seconds for case in cases) / len(cases) if cases else 0.0, + "max_seconds": max((case.seconds for case in cases), default=0.0), + }, + "instrument_failures": instrument_failures, + } + failed_gates: list[str] = [] + for case in cases: + if case.seconds > budget_seconds: + failed_gates.append(f"budget:{case.case_id}:{case.seconds:g}s>{budget_seconds:g}s") + if instrument_failures: + failed_gates.append(f"instrument:{instrument_failures} failures") + rates: dict[str, dict[str, Any]] = { + "min_fix_recall": fix_recall, + "min_pair_correctness": pair_correctness, + "min_verified_location_recall": verified_location_recall, + "min_precision": precision, + "max_flagged_ordinary_per_100": scaled, + "max_indeterminate_rate": indeterminate_rate, + } + for key, threshold in (criteria or {}).items(): + if key not in rates: + raise AdmissionScoringError(f"criterion: unknown pre-registered criterion {key!r}") + actual = rates[key]["rate"] + if actual is None: + failed_gates.append(f"{key}: not measured (n=0) vs {threshold}") + elif key.startswith("min_") and actual < threshold: + failed_gates.append(f"{key}: {actual} < {threshold}") + elif key.startswith("max_") and actual > threshold: + failed_gates.append(f"{key}: {actual} > {threshold}") + report["failed_gates"] = failed_gates + report["decision"] = "GO" if not failed_gates else "NO-GO" + return report diff --git a/tests/test_admission_scoring.py b/tests/test_admission_scoring.py new file mode 100644 index 0000000..fdc0007 --- /dev/null +++ b/tests/test_admission_scoring.py @@ -0,0 +1,187 @@ +"""Paired scoring and the admission report (Task 12.6, Reqs 5.2, 12.4, 12.5, 12.7-12.9).""" + +import json + +import pytest + +from benchmarks.valid.admission_scoring import ( + AdmissionCase, + AdmissionScoringError, + OrdinaryChange, + admission_report, +) +from benchmarks.valid.prover_admission import AdmissionManifest, validate_admission_manifest + + +def admission_manifest(role: str = "admission") -> AdmissionManifest: + payload = { + "produced_by": "valid-detection-benchmark", + "reviewed": True, + "label_isolated": True, + "spent": False, + "population_role": role, + "analyzer_revision": "analyzer-1", + "fact_set_digest": "facts-1", + "acquisition_record_digest": "acq-1", + "configuration_digest": "cfg-1", + "population_digest": "pop-1", + "labels": ("population-overview",), + } + return validate_admission_manifest(payload) + + +def correct_pair(case_id: str = "p1", *, seconds: float = 1.0) -> AdmissionCase: + return AdmissionCase( + case_id=case_id, + vulnerable_judgment="unguarded", + fixed_judgment="discharged", + location_verified=True, + seconds=seconds, + ) + + +def quiet_ordinary(change_id: str = "o1") -> OrdinaryChange: + return OrdinaryChange(change_id=change_id, judgment="discharged", mislabeled_defect=False) + + +def test_vulnerable_only_pair_scores_correct() -> None: + report = admission_report(admission_manifest(), (correct_pair(),), (quiet_ordinary(),), coverage=1.0) + assert report["pairs"] == {"n": 1, "correct": 1} + assert report["pair_correctness"]["rate"] == 1.0 + assert report["fix_recall"]["rate"] == 1.0 + assert report["decision"] == "GO" + assert report["failed_gates"] == [] + + +def test_fixed_side_regression_counts_pair_incorrect() -> None: + regressed = AdmissionCase( + case_id="p1", + vulnerable_judgment="unguarded", + fixed_judgment="unguarded", + location_verified=True, + seconds=1.0, + ) + report = admission_report(admission_manifest(), (regressed,), (), coverage=1.0) + assert report["pairs"]["correct"] == 0 + assert report["pair_correctness"]["rate"] == 0.0 + assert report["fix_recall"]["rate"] == 1.0 + + +def test_indeterminate_vulnerable_side_is_never_clean() -> None: + indeterminate = AdmissionCase( + case_id="p1", + vulnerable_judgment="indeterminate", + fixed_judgment="discharged", + location_verified=False, + seconds=1.0, + ) + report = admission_report(admission_manifest(), (indeterminate,), (), coverage=1.0) + assert report["fix_recall"]["count"] == 0 + assert report["fix_recall"]["rate"] == 0.0 + assert report["indeterminate_rate"]["count"] == 1 + assert report["indeterminate_rate"]["rate"] > 0.0 + + +def test_mislabeled_negative_is_refused() -> None: + mislabeled = OrdinaryChange(change_id="o1", judgment="discharged", mislabeled_defect=True) + with pytest.raises(AdmissionScoringError, match="mislabeled"): + admission_report(admission_manifest(), (correct_pair(),), (mislabeled,), coverage=1.0) + + +def test_development_population_is_refused() -> None: + with pytest.raises(AdmissionScoringError, match="population-role"): + admission_report(admission_manifest(role="development"), (correct_pair(),), (), coverage=1.0) + + +def test_verified_location_recall_keeps_sample_count() -> None: + unverified = AdmissionCase( + case_id="p2", + vulnerable_judgment="unguarded", + fixed_judgment="discharged", + location_verified=False, + seconds=1.0, + ) + report = admission_report(admission_manifest(), (correct_pair("p1"), unverified), (), coverage=1.0) + assert report["verified_location_recall"]["n"] == 2 + assert report["verified_location_recall"]["count"] == 1 + assert report["verified_location_recall"]["rate"] == 0.5 + + +def test_flagged_ordinary_per_100_and_precision() -> None: + flagged = [OrdinaryChange(f"o{i}", "unguarded", False) for i in range(2)] + quiet = [OrdinaryChange(f"q{i}", "discharged", False) for i in range(2)] + pairs = tuple(correct_pair(f"p{i}") for i in range(3)) + report = admission_report(admission_manifest(), pairs, tuple(flagged + quiet), coverage=1.0) + per_100 = report["flagged_ordinary_per_100"] + assert per_100["n"] == 4 + assert per_100["count"] == 2 + assert per_100["rate"] == 50.0 + assert per_100["wilson95"][0] is not None and per_100["wilson95"][0] <= 50.0 <= per_100["wilson95"][1] + assert report["precision"]["rate"] == pytest.approx(3 / 5) + + +def test_budget_gate_returns_no_go() -> None: + slow = correct_pair("slow", seconds=121.0) + report = admission_report(admission_manifest(), (slow,), (), coverage=1.0) + assert report["decision"] == "NO-GO" + assert any("budget" in gate for gate in report["failed_gates"]) + assert any("slow" in gate for gate in report["failed_gates"]) + + +def test_instrument_gate_returns_no_go() -> None: + report = admission_report(admission_manifest(), (correct_pair(),), (), coverage=1.0, instrument_failures=1) + assert report["decision"] == "NO-GO" + assert any("instrument" in gate for gate in report["failed_gates"]) + + +def test_pre_registered_criterion_failure_returns_no_go() -> None: + missed = AdmissionCase( + case_id="p1", + vulnerable_judgment="discharged", + fixed_judgment="discharged", + location_verified=False, + seconds=1.0, + ) + report = admission_report( + admission_manifest(), + (missed,), + (), + coverage=1.0, + criteria={"min_fix_recall": 1.0}, + ) + assert report["decision"] == "NO-GO" + assert any("min_fix_recall" in gate for gate in report["failed_gates"]) + + +def test_invalid_judgment_is_refused() -> None: + broken = AdmissionCase( + case_id="p1", + vulnerable_judgment="banana", + fixed_judgment="discharged", + location_verified=True, + seconds=1.0, + ) + with pytest.raises(AdmissionScoringError, match="judgment"): + admission_report(admission_manifest(), (broken,), (), coverage=1.0) + + +def test_report_is_deterministic() -> None: + args = ( + admission_manifest(), + (correct_pair("p1"), correct_pair("p2")), + (quiet_ordinary("o1"), OrdinaryChange("o2", "unguarded", False)), + ) + first = admission_report(*args, coverage=0.75) + second = admission_report(*args, coverage=0.75) + assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True) + + +def test_report_carries_provenance_without_labels() -> None: + report = admission_report(admission_manifest(), (correct_pair(),), (), coverage=1.0) + manifest = report["manifest"] + assert manifest["analyzer_revision"] == "analyzer-1" + assert manifest["population_digest"] == "pop-1" + assert manifest["population_role"] == "admission" + text = json.dumps(report, sort_keys=True) + for marker in ("vulnerability", "fix-location", "ordinary-change"): + assert marker not in text