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) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

.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')"
 }

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",
+]

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