From ea33222bf64a6e548d91d51a9bbd7e2ddf84aae0 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:44:27 +0100 Subject: [PATCH] feat(access-control-prover): 12.4 four-arm development comparison --- .kiro/specs/access-control-prover/tasks.md | 2 +- benchmarks/valid/development_runner.py | 57 ++++++++++++ tests/test_four_arm.py | 102 +++++++++++++++++++++ 3 files changed, 160 insertions(+), 1 deletion(-) create mode 100644 tests/test_four_arm.py diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 942496b..74bd501 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -277,7 +277,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 5.6, 6.1, 6.3_ - _Boundary: Runner Gates_ - _Depends: 12.2_ -- [ ] 12.4 Register the four-arm development comparison +- [x] 12.4 Register the four-arm development comparison - Run baseline/candidate analyzer crossed with baseline/candidate facts over identical inputs. - Report metric deltas, stage-funnel deltas and judgment transitions without admission language. - Done when fixture arms isolate a fact-only lift from an analyzer-only lift. diff --git a/benchmarks/valid/development_runner.py b/benchmarks/valid/development_runner.py index da8b4e6..f3f7947 100644 --- a/benchmarks/valid/development_runner.py +++ b/benchmarks/valid/development_runner.py @@ -182,12 +182,69 @@ def summarize_development(records: tuple[DevelopmentRecord, ...], *, budget_seco } +_ARM_NAMES = ( + "baseline-analyzer/baseline-facts", + "baseline-analyzer/candidate-facts", + "candidate-analyzer/baseline-facts", + "candidate-analyzer/candidate-facts", +) + + +def _judgment_totals(records: tuple[DevelopmentRecord, ...]) -> dict[str, int]: + """Sum unguarded/discharged/indeterminate counts across a run's records.""" + totals = {"unguarded": 0, "discharged": 0, "indeterminate": 0} + for record in records: + judgment = record.stage_funnel.get("judgment", {}) + for kind in totals: + totals[kind] += int(judgment.get(kind, 0)) + return totals + + +def _judgment_delta(after: dict[str, int], before: dict[str, int]) -> dict[str, int]: + """Per-kind difference between two judgment totals.""" + return {kind: after[kind] - before[kind] for kind in after} + + +def _mean_seconds(records: tuple[DevelopmentRecord, ...]) -> float: + """Mean per-change seconds of a run.""" + if not records: + return 0.0 + return sum(record.seconds for record in records) / len(records) + + +def four_arm_report( + bab: tuple[DevelopmentRecord, ...], + bacf: tuple[DevelopmentRecord, ...], + cabf: tuple[DevelopmentRecord, ...], + cacfon: tuple[DevelopmentRecord, ...], +) -> dict[str, object]: + """Attribute judgment movement to fact-only and analyzer-only variation. + + The four arms share identical inputs by construction; the report never uses + admission vocabulary. + """ + baseline_totals = _judgment_totals(bab) + fact_delta = _judgment_delta(_judgment_totals(bacf), baseline_totals) + analyzer_delta = _judgment_delta(_judgment_totals(cabf), baseline_totals) + transitions = sorted({record.judgment_transition for record in (*bab, *bacf, *cabf, *cacfon)}) + return { + "arms": _ARM_NAMES, + "judgment_delta_fact_only": fact_delta, + "judgment_delta_analyzer_only": analyzer_delta, + "fact_only_lift": dict(fact_delta), + "analyzer_only_lift": dict(analyzer_delta), + "transitions": transitions, + "funnel_delta_fact_only": {"seconds": _mean_seconds(bacf) - _mean_seconds(bab)}, + } + + __all__ = [ "DevelopmentRecord", "GateFailure", "InstrumentFailure", "deadline_record", "enforce_gates", + "four_arm_report", "judgment_transition", "run_development", "stage_funnel", diff --git a/tests/test_four_arm.py b/tests/test_four_arm.py new file mode 100644 index 0000000..0d5cf86 --- /dev/null +++ b/tests/test_four_arm.py @@ -0,0 +1,102 @@ +"""Task 12.4 -- four-arm development comparison (Reqs 5.2, 5.5, 5.7, 11.8). + +Baseline/candidate analyzer crossed with baseline/candidate facts over identical +inputs; fixture arms isolate fact-only lift from analyzer-only lift. No admission +language in any report field. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import cast + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from benchmarks.valid import development_runner # noqa: E402 + + +def _record( + *, + change_id: str = "change-1", + unguarded: int = 1, + indeterminate: int = 0, + seconds: float = 1.0, + judgment_transition: str = "same", +) -> development_runner.DevelopmentRecord: + return development_runner.DevelopmentRecord( + change_id=change_id, + analyzer_revision="a" * 12, + fact_set_digest="b" * 12, + acquisition_record_digest="none", + configuration_digest="c" * 12, + population_digest="d" * 12, + input_digest="e" * 12, + stage_funnel={"judgment": {"unguarded": unguarded, "discharged": 0, "indeterminate": indeterminate}}, + judgment_transition=judgment_transition, + seconds=seconds, + ) + + +def test_four_arm_report_contains_all_arms() -> None: + baseline_analyzer_baseline_facts = (_record(),) + baseline_analyzer_candidate_facts = (_record(unguarded=0, indeterminate=1),) + candidate_analyzer_baseline_facts = (_record(unguarded=0, indeterminate=1),) + candidate_analyzer_candidate_facts = (_record(unguarded=0, indeterminate=1),) + report = development_runner.four_arm_report( + baseline_analyzer_baseline_facts, + baseline_analyzer_candidate_facts, + candidate_analyzer_baseline_facts, + candidate_analyzer_candidate_facts, + ) + assert report["arms"] == ( + "baseline-analyzer/baseline-facts", + "baseline-analyzer/candidate-facts", + "candidate-analyzer/baseline-facts", + "candidate-analyzer/candidate-facts", + ) + + +def test_four_arm_report_isolates_fact_only_and_analyzer_only_lift() -> None: + """Done-when: fixture arms isolate fact-only lift from analyzer-only lift.""" + # baseline/baseline: unguarded; only-facts-changed: discharged; only-analyzer-changed: discharged. + bab = (_record(unguarded=1),) + bacf = (_record(unguarded=0, indeterminate=0),) + cabf = (_record(unguarded=0, indeterminate=0),) + cacfon = (_record(unguarded=0, indeterminate=0),) + report = development_runner.four_arm_report(bab, bacf, cabf, cacfon) + assert report["fact_only_lift"] == {"unguarded": -1, "discharged": 0, "indeterminate": 0} + assert report["analyzer_only_lift"] == {"unguarded": -1, "discharged": 0, "indeterminate": 0} + + +def test_four_arm_report_includes_judgment_transitions() -> None: + bab = (_record(judgment_transition="same"),) + bacf = (_record(judgment_transition="unguarded->indeterminate"),) + cabf = (_record(judgment_transition="same"),) + cacfon = (_record(judgment_transition="unguarded->discharged"),) + report = development_runner.four_arm_report(bab, bacf, cabf, cacfon) + transitions = cast("tuple[str, ...]", report["transitions"]) + assert "unguarded->indeterminate" in transitions + assert "unguarded->discharged" in transitions + + +def test_four_arm_report_contains_no_admission_language() -> None: + bab = bacf = cabf = cacfon = (_record(),) + report = development_runner.four_arm_report(bab, bacf, cabf, cacfon) + text = repr(report) + for word in ("admission", "vulnerable", "ordinary", "label"): + assert word not in text.lower() + + +def test_four_arm_report_includes_metric_and_funnel_deltas() -> None: + bab = (_record(unguarded=1, seconds=2.0),) + bacf = (_record(unguarded=0, seconds=3.0),) + cabf = (_record(unguarded=0, seconds=2.5),) + cacfon = (_record(unguarded=0, seconds=3.5),) + report = development_runner.four_arm_report(bab, bacf, cabf, cacfon) + fact_delta = cast("dict[str, int]", report["judgment_delta_fact_only"]) + analyzer_delta = cast("dict[str, int]", report["judgment_delta_analyzer_only"]) + funnel_delta = cast("dict[str, float]", report["funnel_delta_fact_only"]) + assert fact_delta["unguarded"] == -1 + assert analyzer_delta["unguarded"] == -1 + assert "seconds" in funnel_delta