From 95ffd901e4089ab8ecd4ea7c15d71f8bdeb9d336 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 6 Sep 2026 09:33:08 +0200 Subject: [PATCH] feat(learning-harness): aggregation, noise handling and honest denominators Task 2.4. Per family, and per slice when asked: outcome counts, recall, silence, Youden, a directional-bias index and recall at a fixed false-positive ceiling. Rows that cannot be scored are listed by reason and leave every denominator, and a family with nothing scorable reports no rate rather than a perfect one. Two numbers exist because Youden cannot tell apart a detector that flags everything from one that says everything is safe: the directional-bias index separates them, and recall at a fixed ceiling is forfeited entirely once a family leaks more than the budget allows. Change against a baseline is not a difference of means. Flips are counted per pair in both directions and a two-sided exact sign test decides whether the change is bigger than the noise, so one flip in ten reads as noise and six reads as a real regression. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- .kiro/specs/learning-harness/spec.json | 10 +- src/openultrasast/learning/scoring.py | 131 ++++++++++++++++++++++++- tests/test_learning_aggregation.py | 98 ++++++++++++++++++ 3 files changed, 230 insertions(+), 9 deletions(-) create mode 100644 tests/test_learning_aggregation.py diff --git a/.kiro/specs/learning-harness/spec.json b/.kiro/specs/learning-harness/spec.json index f46fb77..8bec1d5 100644 --- a/.kiro/specs/learning-harness/spec.json +++ b/.kiro/specs/learning-harness/spec.json @@ -1,9 +1,9 @@ { "feature_name": "learning-harness", "created_at": "2026-09-06T06:32:41Z", - "updated_at": "2026-09-06T06:57:33Z", + "updated_at": "2026-09-06T07:00:52Z", "language": "en", - "phase": "tasks-generated", + "phase": "implementation", "approvals": { "requirements": { "generated": true, @@ -15,10 +15,10 @@ }, "tasks": { "generated": true, - "approved": false + "approved": true } }, - "ready_for_implementation": false, + "ready_for_implementation": true, "depends_on": [ "pair-corpus-honesty", "corpus-seeded-mechanisms", @@ -26,5 +26,5 @@ "tree-sitter-overlay-extra", "harnessx-self-improving-rulesets" ], - "approval_note": "2026-09-06 requirements approved by the maintainer in session ('approve the requirements and run the design phase'); 2026-09-06 design approved by the maintainer in session ('approve the design and generate the tasks')" + "approval_note": "2026-09-06 requirements approved by the maintainer in session ('approve the requirements and run the design phase'); 2026-09-06 design approved by the maintainer in session ('approve the design and generate the tasks'); 2026-09-06 tasks approved by the maintainer in session ('approve the tasks and go on with the implementation')" } diff --git a/src/openultrasast/learning/scoring.py b/src/openultrasast/learning/scoring.py index ed2cd46..7463849 100644 --- a/src/openultrasast/learning/scoring.py +++ b/src/openultrasast/learning/scoring.py @@ -12,9 +12,10 @@ outcome is the majority of the runs, and the runs that disagree with it are the from __future__ import annotations +import math from collections import Counter from collections.abc import Mapping, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Literal from ..findings import StaticFinding @@ -34,6 +35,7 @@ class PairFamilyScore: family: str runs: tuple[FamilyOutcome, ...] outcome: FamilyOutcome + slice: str = "" other_findings_vuln: int = 0 # right place, another family: not a detection, never a leak other_findings_fixed: int = 0 fabricated_vuln: int = 0 # a family id outside the taxonomy: noise, and worth naming @@ -54,9 +56,10 @@ def score_pair_family( ) -> PairFamilyScore: """Score one pair for one family over K runs of the detector on each side.""" name = str(getattr(case, "name", "?")) + slice_name = str(getattr(case, "slice", "") or "") reason = _unscorable(case, ranges) if reason is not None: - return PairFamilyScore(pair=name, family=family, runs=(), outcome="unscorable", unscorable_reason=reason) + return PairFamilyScore(pair=name, family=family, slice=slice_name, runs=(), outcome="unscorable", unscorable_reason=reason) spans = _spans(case, ranges) outcomes: list[FamilyOutcome] = [] other_vuln = other_fixed = fabricated_vuln = fabricated_fixed = 0 @@ -73,11 +76,12 @@ def score_pair_family( rung = _best_rung(rung, hits_vuln) outcomes.append(_outcome(bool(hits_vuln), bool(hits_fixed))) if not outcomes: - return PairFamilyScore(pair=name, family=family, runs=(), outcome="unscorable", unscorable_reason="no_runs") + return PairFamilyScore(pair=name, family=family, slice=slice_name, runs=(), outcome="unscorable", unscorable_reason="no_runs") majority = Counter(outcomes).most_common(1)[0][0] return PairFamilyScore( pair=name, family=family, + slice=slice_name, runs=tuple(outcomes), outcome=majority, other_findings_vuln=other_vuln, @@ -158,4 +162,123 @@ def _best_rung(current: Rung, hits: Sequence[StaticFinding]) -> Rung: return best -__all__ = ["FamilyOutcome", "FunctionRanges", "PairFamilyScore", "Rung", "score_pair_family"] +@dataclass(frozen=True) +class FamilyMetrics: + """One family's numbers, with the denominator they were computed over and why rows fell out of it.""" + + family: str + taxonomy_version: str + scorable: int + unscorable: dict[str, int] = field(default_factory=dict) + outcomes: dict[str, int] = field(default_factory=dict) + recall: float = 0.0 # flagged the vulnerable side + silence: float = 0.0 # said nothing on the fixed side + youden: float = 0.0 + directional_bias: float = 0.0 # positive: flags both sides; negative: silent on both. Youden cannot tell them apart. + leak_rate: float = 0.0 + fixed_fpr_recall: float = 0.0 # recall, forfeited entirely when the leak rate exceeds the ceiling + negative_flips: int = 0 # pairs that were correct in the baseline and are not now + positive_flips: int = 0 + p_value: float | None = None # two-sided sign test over the discordant pairs + reliable_change: bool = False + + def to_dict(self) -> dict[str, object]: + return { + "family": self.family, + "taxonomy_version": self.taxonomy_version, + "scorable": self.scorable, + "unscorable": dict(sorted(self.unscorable.items())), + "outcomes": dict(sorted(self.outcomes.items())), + "recall": self.recall, + "silence": self.silence, + "youden": self.youden, + "directional_bias": self.directional_bias, + "leak_rate": self.leak_rate, + "fixed_fpr_recall": self.fixed_fpr_recall, + "negative_flips": self.negative_flips, + "positive_flips": self.positive_flips, + "p_value": self.p_value, + "reliable_change": self.reliable_change, + } + + +def aggregate( + scores: Sequence[PairFamilyScore], + *, + taxonomy: FamilyTaxonomy, + baseline: Sequence[PairFamilyScore] | None = None, + fpr_ceiling: float = 0.1, + per_slice: bool = False, + alpha: float = 0.05, +) -> dict[str, FamilyMetrics]: + """Group scores into per-family numbers. Unscorable rows are listed by reason and leave every denominator.""" + grouped: dict[str, list[PairFamilyScore]] = {} + for score in scores: + grouped.setdefault(f"{score.slice}/{score.family}" if per_slice else score.family, []).append(score) + was_correct = {(item.slice if per_slice else "", item.family, item.pair): item.outcome for item in baseline or ()} + out: dict[str, FamilyMetrics] = {} + for key, group in sorted(grouped.items()): + scorable = [item for item in group if item.outcome != "unscorable"] + reasons: dict[str, int] = {} + for item in group: + if item.outcome == "unscorable": + reason = item.unscorable_reason or "unknown" + reasons[reason] = reasons.get(reason, 0) + 1 + counts = Counter(item.outcome for item in scorable) + total = len(scorable) + detected = counts["pair_correct"] + counts["both_flagged"] + silent = counts["pair_correct"] + counts["both_silent"] + leaks = counts["both_flagged"] + counts["reversed"] + recall = detected / total if total else 0.0 + silence = silent / total if total else 0.0 + leak_rate = leaks / total if total else 0.0 + negative = positive = 0 + for item in scorable: + before = was_correct.get((item.slice if per_slice else "", item.family, item.pair)) + if before is None or before == "unscorable": + continue + if before == "pair_correct" and item.outcome != "pair_correct": + negative += 1 + elif before != "pair_correct" and item.outcome == "pair_correct": + positive += 1 + p_value = sign_test(positive, negative) if baseline is not None else None + out[key] = FamilyMetrics( + family=group[0].family, + taxonomy_version=taxonomy.version, + scorable=total, + unscorable=reasons, + outcomes={name: count for name, count in sorted(counts.items())}, + recall=recall, + silence=silence, + youden=(recall + silence - 1.0) if total else 0.0, + directional_bias=((counts["both_flagged"] - counts["both_silent"]) / total) if total else 0.0, + leak_rate=leak_rate, + fixed_fpr_recall=recall if leak_rate <= fpr_ceiling else 0.0, + negative_flips=negative, + positive_flips=positive, + p_value=p_value, + reliable_change=bool(p_value is not None and p_value < alpha and positive != negative), + ) + return out + + +def sign_test(better: int, worse: int) -> float: + """Two-sided exact sign test over the pairs that changed. 1.0 when nothing changed, so noise never reads as progress.""" + total = better + worse + if total == 0: + return 1.0 + smaller = min(better, worse) + tail = sum(math.comb(total, index) for index in range(smaller + 1)) / float(2**total) + return float(min(1.0, 2.0 * tail)) + + +__all__ = [ + "FamilyMetrics", + "FamilyOutcome", + "FunctionRanges", + "PairFamilyScore", + "Rung", + "aggregate", + "score_pair_family", + "sign_test", +] diff --git a/tests/test_learning_aggregation.py b/tests/test_learning_aggregation.py new file mode 100644 index 0000000..7f70ed7 --- /dev/null +++ b/tests/test_learning_aggregation.py @@ -0,0 +1,98 @@ +"""learning-harness task 2.4: aggregation, noise handling and honest denominators (offline).""" + +from __future__ import annotations + +import json + +from openultrasast.learning.families import load_families +from openultrasast.learning.scoring import FamilyOutcome, PairFamilyScore + + +def _score( + pair: str, outcome: FamilyOutcome, *, family: str = "access_control", slice_name: str = "vibe-py", **extra: object +) -> PairFamilyScore: + runs = () if outcome == "unscorable" else (outcome,) + return PairFamilyScore(pair=pair, family=family, slice=slice_name, runs=runs, outcome=outcome, **extra) # type: ignore[arg-type] + + +def test_rates_use_only_the_scorable_rows_and_the_reasons_are_kept() -> None: + from openultrasast.learning.scoring import aggregate + + taxonomy = load_families() + scores = [ + _score("a", "pair_correct"), + _score("b", "both_silent"), + _score("c", "unscorable", unscorable_reason="identical_twin"), + _score("d", "unscorable", unscorable_reason="identical_twin"), + _score("e", "unscorable", unscorable_reason="unsupported_language"), + ] + metrics = aggregate(scores, taxonomy=taxonomy)["access_control"] + assert metrics.scorable == 2 and metrics.unscorable == {"identical_twin": 2, "unsupported_language": 1} + assert metrics.recall == 0.5 and metrics.silence == 1.0 and metrics.youden == 0.5 + assert metrics.outcomes == {"pair_correct": 1, "both_silent": 1} + assert metrics.taxonomy_version == taxonomy.version + json.dumps(metrics.to_dict()) + + +def test_a_family_with_nothing_scorable_reports_no_rate_rather_than_a_perfect_one() -> None: + from openultrasast.learning.scoring import aggregate + + metrics = aggregate([_score("a", "unscorable", unscorable_reason="identical_twin")], taxonomy=load_families())["access_control"] + assert metrics.scorable == 0 and metrics.recall == 0.0 and metrics.silence == 0.0 and metrics.youden == 0.0 + + +def test_directional_bias_separates_flagging_everything_from_saying_everything_is_safe() -> None: + from openultrasast.learning.scoring import aggregate + + taxonomy = load_families() + loud = aggregate([_score("a", "both_flagged"), _score("b", "both_flagged")], taxonomy=taxonomy)["access_control"] + quiet = aggregate([_score("a", "both_silent"), _score("b", "both_silent")], taxonomy=taxonomy)["access_control"] + assert loud.directional_bias == 1.0 and quiet.directional_bias == -1.0 + assert loud.youden == quiet.youden == 0.0 # the same Youden, opposite failures: the index is what tells them apart + + +def test_recall_at_a_fixed_false_positive_ceiling_is_forfeited_when_the_family_leaks() -> None: + from openultrasast.learning.scoring import aggregate + + taxonomy = load_families() + clean = [_score(str(index), "pair_correct") for index in range(9)] + [_score("x", "both_silent")] + assert aggregate(clean, taxonomy=taxonomy, fpr_ceiling=0.1)["access_control"].fixed_fpr_recall == 0.9 + leaky = [_score(str(index), "pair_correct") for index in range(8)] + [_score("y", "both_flagged"), _score("z", "reversed")] + metrics = aggregate(leaky, taxonomy=taxonomy, fpr_ceiling=0.1)["access_control"] + assert metrics.recall == 0.9 and metrics.fixed_fpr_recall == 0.0 # a 20% leak rate buys no credit at a 10% ceiling + + +def test_negative_flips_and_the_reliable_change_test_compare_against_a_baseline() -> None: + from openultrasast.learning.scoring import aggregate + + taxonomy = load_families() + baseline = [_score(str(index), "pair_correct") for index in range(10)] + same = aggregate(baseline, taxonomy=taxonomy, baseline=baseline)["access_control"] + assert same.negative_flips == 0 and same.reliable_change is False + broke_one = [_score("0", "both_silent")] + [_score(str(index), "pair_correct") for index in range(1, 10)] + one = aggregate(broke_one, taxonomy=taxonomy, baseline=baseline)["access_control"] + assert one.negative_flips == 1 and one.reliable_change is False # one flip in ten is inside the noise + broke_six = [_score(str(index), "both_silent") for index in range(6)] + [_score(str(index), "pair_correct") for index in range(6, 10)] + six = aggregate(broke_six, taxonomy=taxonomy, baseline=baseline)["access_control"] + assert six.negative_flips == 6 and six.reliable_change is True + assert six.p_value is not None and six.p_value < 0.05 + + +def test_metrics_are_reported_per_slice_when_asked() -> None: + from openultrasast.learning.scoring import aggregate + + taxonomy = load_families() + scores = [_score("a", "pair_correct", slice_name="vibe-py"), _score("b", "both_silent", slice_name="agent-vfc")] + overall = aggregate(scores, taxonomy=taxonomy) + assert set(overall) == {"access_control"} and overall["access_control"].scorable == 2 + per_slice = aggregate(scores, taxonomy=taxonomy, per_slice=True) + assert set(per_slice) == {"vibe-py/access_control", "agent-vfc/access_control"} + assert per_slice["vibe-py/access_control"].recall == 1.0 and per_slice["agent-vfc/access_control"].recall == 0.0 + + +def test_the_sign_test_is_two_sided_and_symmetric() -> None: + from openultrasast.learning.scoring import sign_test + + assert sign_test(0, 0) == 1.0 + assert sign_test(6, 0) == sign_test(0, 6) < 0.05 + assert sign_test(1, 0) > 0.05 and sign_test(3, 3) == 1.0