feat(pair-corpus-honesty): reject improve rounds that regress any provenance profile on holdout

Task 6.3.

- improve/evolve.py: evaluate_profiles and profile_regressions score the
  holdout pair split per provenance profile every round; a drop in pair
  pass rate or Youden beyond --profile-tolerance rejects the round with
  reason profile_regression:<profile>; profiles under
  --min-holdout-pairs are reported, not gated; journal records both.
- cli improve: --pair-catalog, --no-pair-gate, --profile-tolerance,
  --min-holdout-pairs; a missing catalog fails loud instead of skipping
  the gate.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

src/openultrasast/cli.py

@@ -130,6 +130,19 @@ def main(argv: list[str] | None = None) -> int:
         "--ruleset-dir", type=Path, default=DEFAULT_RULESET_DIR, help="ruleset directory to improve (default: the bundled ruleset)"
     )
     improve.add_argument("--dry-run", action="store_true", help="run rounds against a throwaway ledger; never touch the target's ledger")
+    improve.add_argument(
+        "--pair-catalog",
+        type=Path,
+        default=DEFAULT_CATALOG,
+        help="pair catalog whose holdout split gates every round per provenance profile",
+    )
+    improve.add_argument(
+        "--no-pair-gate", action="store_true", help="skip the per-profile holdout clause (faster; not for accepted ledgers)"
+    )
+    improve.add_argument(
+        "--profile-tolerance", type=float, default=0.0, help="allowed per-profile drop in pair_correct/Youden before rejecting"
+    )
+    improve.add_argument("--min-holdout-pairs", type=int, default=5, help="profiles with fewer holdout pairs are reported, not gated")
 
     pairs = subparsers.add_parser(
         "pairs",
@@ -164,6 +177,9 @@ def main(argv: list[str] | None = None) -> int:
             journal=args.journal,
             ruleset_dir=args.ruleset_dir,
             dry_run=args.dry_run,
+            pair_catalog=None if args.no_pair_gate else args.pair_catalog,
+            profile_tolerance=args.profile_tolerance,
+            min_holdout_pairs=args.min_holdout_pairs,
         )
     if args.command == "pairs":
         return _pairs(
@@ -718,9 +734,17 @@ def _improve(
     journal: Path | None,
     ruleset_dir: Path,
     dry_run: bool,
+    pair_catalog: Path | None = None,
+    profile_tolerance: float = 0.0,
+    min_holdout_pairs: int = 5,
 ) -> int:
     if not manifest_path.exists() or not manifest_path.is_file():
         raise SystemExit(f"benchmark manifest is not a file: {manifest_path}")
+    pair_cases: tuple[object, ...] = ()
+    if pair_catalog is not None:
+        if not pair_catalog.is_file():
+            raise SystemExit(f"pair catalog is not a file: {pair_catalog} (pass --no-pair-gate to skip the per-profile holdout clause)")
+        pair_cases = select_split(load_pair_catalog(pair_catalog), "holdout")
     manifest = load_benchmark_manifest(manifest_path)
     target = resolve_benchmark_source(manifest_path, manifest)
     policy = load_policy()
@@ -746,6 +770,9 @@ def _improve(
             max_rounds=max_rounds,
             recall_floor=recall_floor,
             fp_ceiling=fp_ceiling,
+            pair_cases=pair_cases,
+            profile_tolerance=profile_tolerance,
+            min_holdout_pairs=min_holdout_pairs,
         )
         _print_improve_outcomes(outcomes, manifest_path, target, ledger_path, dry_run=dry_run)
     return 0
@@ -793,6 +820,8 @@ def _print_improve_outcomes(outcomes: list[RoundOutcome], manifest_path: Path, t
             f"fp {outcome.fp_before:.2%}->{outcome.fp_after:.2%} "
             f"score {outcome.score_before}->{outcome.score_after} | edits: {edits}"
         )
+        if outcome.profile_regressions or outcome.profiles_under_minimum:
+            print(f"  profiles: regressed={outcome.profile_regressions or '-'} under_minimum={outcome.profiles_under_minimum or '-'}")
     print(f"rounds={len(outcomes)} accepted={len(accepted)}")
     if dry_run:
         print("dry_run=true (no ledger written)")

src/openultrasast/improve/evolve.py

@@ -14,7 +14,7 @@ feasibility constraint that is never folded into the reward.
 from __future__ import annotations
 
 import tempfile
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
 
@@ -44,6 +44,62 @@ class RoundOutcome:
     score_after: int = 0
     matched_before: int = 0
     matched_after: int = 0
+    profile_regressions: list[str] = field(default_factory=list)
+    profiles_under_minimum: list[str] = field(default_factory=list)
+    per_profile_before: dict[str, dict[str, float]] = field(default_factory=dict)
+    per_profile_after: dict[str, dict[str, float]] = field(default_factory=dict)
+
+
+MAX_PROFILES = 4
+
+
+def evaluate_profiles(
+    pair_cases: Sequence[object],
+    rules: tuple[PatternRule, ...],
+) -> dict[str, dict[str, float]]:
+    """Pair-corpus metrics per provenance profile under ``rules`` (holdout split, no inventory sidecar)."""
+    from ..pairs import PairCase, evaluate_catalog
+
+    cases = [case for case in pair_cases if isinstance(case, PairCase)]
+    if not cases:
+        return {}
+    result = evaluate_catalog(cases, ruleset=rules, inventory_sidecar=False)
+    return {
+        profile: {
+            "pairs": float(metrics.pairs),
+            "pair_correct": float(metrics.pair_correct),
+            "pair_pass_rate": metrics.pair_pass_rate,
+            "youden": metrics.youden,
+        }
+        for profile, metrics in result.per_profile.items()
+    }
+
+
+def profile_regressions(
+    before: dict[str, dict[str, float]],
+    after: dict[str, dict[str, float]],
+    *,
+    tolerance: float = 0.0,
+    min_pairs: int = 5,
+) -> tuple[list[str], list[str]]:
+    """(regressed profiles, profiles under the minimum).
+
+    A profile regresses when its pair pass rate or Youden drops by more than ``tolerance``
+    (both are rates in [-1, 1], so one tolerance applies to both).
+    """
+    regressed: list[str] = []
+    small: list[str] = []
+    for profile in sorted(set(before) | set(after)):
+        old = before.get(profile, {"pairs": 0.0, "pair_correct": 0.0, "pair_pass_rate": 0.0, "youden": 0.0})
+        new = after.get(profile, old)
+        if old["pairs"] < min_pairs:
+            small.append(profile)
+            continue
+        old_rate = old.get("pair_pass_rate", old["pair_correct"] / old["pairs"] if old["pairs"] else 0.0)
+        new_rate = new.get("pair_pass_rate", new["pair_correct"] / new["pairs"] if new["pairs"] else 0.0)
+        if new_rate < old_rate - tolerance or new["youden"] < old["youden"] - tolerance:
+            regressed.append(profile)
+    return regressed, small
 
 
 def build_rule_signals(result: BenchmarkResult) -> list[dict[str, object]]:
@@ -90,6 +146,9 @@ def run_round(
     policy: dict[str, CwePolicy] | None = None,
     recall_floor: float = 0.9,
     fp_ceiling: float = 0.1,
+    pair_cases: Sequence[object] | None = None,
+    profile_tolerance: float = 0.0,
+    min_holdout_pairs: int = 5,
 ) -> RoundOutcome:
     policy = policy if policy is not None else load_policy()
     rounds = load_journal(journal_path)
@@ -128,13 +187,37 @@ def run_round(
     accepted = (
         after.recall >= recall_floor and after.fp_rate < fp_ceiling and after.score >= before.score and after.matched >= before.matched
     )
+    profiles_before: dict[str, dict[str, float]] = {}
+    profiles_after: dict[str, dict[str, float]] = {}
+    regressed: list[str] = []
+    small: list[str] = []
+    if accepted and pair_cases:
+        # pair-corpus-honesty Req 7.2: a change may not help one provenance profile by hurting another.
+        profiles_before = evaluate_profiles(pair_cases, before_rules)
+        profiles_after = evaluate_profiles(pair_cases, after_rules)
+        if len(profiles_before) > MAX_PROFILES:
+            raise ValueError(f"at most {MAX_PROFILES} provenance profiles are supported; got {sorted(profiles_before)}")
+        regressed, small = profile_regressions(profiles_before, profiles_after, tolerance=profile_tolerance, min_pairs=min_holdout_pairs)
+        if regressed:
+            accepted = False
     if accepted:
         write_rule_ledger(ledger_path, candidate)
         reason = "accepted"
+    elif regressed:
+        reason = f"profile_regression:{regressed[0]}"  # every regressed profile is listed in profile_regressions
     else:
         reason = "reverted"  # ledger_path untouched -> byte-for-byte revert
-    _record(journal_path, round_index, edits, reason, before, after)
-    return _outcome(round_index, accepted=accepted, reason=reason, edits=edits, before=before, after=after)
+    _record(journal_path, round_index, edits, "accepted" if accepted else "reverted", before, after, reason, regressed, small)
+    outcome = _outcome(round_index, accepted=accepted, reason=reason, edits=edits, before=before, after=after)
+    return RoundOutcome(
+        **{
+            **outcome.__dict__,
+            "profile_regressions": regressed,
+            "profiles_under_minimum": small,
+            "per_profile_before": profiles_before,
+            "per_profile_after": profiles_after,
+        }
+    )
 
 
 def run_improvement(
@@ -148,6 +231,9 @@ def run_improvement(
     max_rounds: int = 5,
     recall_floor: float = 0.9,
     fp_ceiling: float = 0.1,
+    pair_cases: Sequence[object] | None = None,
+    profile_tolerance: float = 0.0,
+    min_holdout_pairs: int = 5,
 ) -> list[RoundOutcome]:
     """Run improvement rounds until convergence (no new proposals) or ``max_rounds``."""
     outcomes: list[RoundOutcome] = []
@@ -161,6 +247,9 @@ def run_improvement(
             policy=policy,
             recall_floor=recall_floor,
             fp_ceiling=fp_ceiling,
+            pair_cases=pair_cases,
+            profile_tolerance=profile_tolerance,
+            min_holdout_pairs=min_holdout_pairs,
         )
         outcomes.append(outcome)
         if outcome.reason == "no_proposals":
@@ -246,6 +335,8 @@ def _record(
     before: _Eval,
     after: _Eval,
     reason: str = "",
+    profile_regressions: list[str] | None = None,
+    profiles_under_minimum: list[str] | None = None,
 ) -> None:
     append_round(
         journal_path,
@@ -253,6 +344,8 @@ def _record(
             "round": round_index,
             "outcome": outcome,
             "reason": reason or outcome,
+            "profile_regressions": list(profile_regressions or []),
+            "profiles_under_minimum": list(profiles_under_minimum or []),
             "edits": [
                 {"key": e.key(), "lever": e.lever, "rule_id": e.rule_id, "from": e.from_status, "to": e.to_status, "rationale": e.rationale}
                 for e in edits

tests/test_cli_improve.py

@@ -58,3 +58,12 @@ def test_improve_rejects_missing_manifest(tmp_path: Path) -> None:
 
     with pytest.raises(SystemExit):
         main(["improve", str(tmp_path / "nope.toml")])
+
+
+def test_improve_fails_loud_when_pair_catalog_is_missing(tmp_path: Path) -> None:
+    import pytest
+
+    _, ruleset_dir = _scaffold(tmp_path)
+    manifest = tmp_path / "bench.toml"
+    with pytest.raises(SystemExit, match="pair catalog"):
+        main(["improve", str(manifest), "--ruleset-dir", str(ruleset_dir), "--dry-run", "--pair-catalog", str(tmp_path / "missing.toml")])

tests/test_improve.py

@@ -168,3 +168,86 @@ def test_build_rule_signals_separates_miss_and_fp() -> None:
     signals = build_rule_signals(result)
     assert {s["signal"] for s in signals} == {"miss", "fp"}
     assert {s["rule_id"] for s in signals} == {"sqli", "noisy"}
+
+
+# ---- pair-corpus-honesty: per-profile holdout clause (Req 7.2-7.4) ----------
+
+
+def _pair(tmp_path: Path, name: str, vuln_src: str, fixed_src: str, rule_id: str, sink: str, *, provenance: str):
+    from openultrasast.benchmark import ExpectedFinding
+    from openultrasast.pairs import PairCase
+
+    vuln = tmp_path / f"{name}-vuln.py"
+    fixed = tmp_path / f"{name}-fixed.py"
+    vuln.write_text(vuln_src)
+    fixed.write_text(fixed_src)
+    return PairCase(
+        name=name,
+        slice="github",
+        language="python",
+        origin="test",
+        vuln_file=vuln,
+        fixed_file=fixed,
+        relpath="app.py",
+        expected=(
+            ExpectedFinding(
+                cwe="CWE-95",
+                vulnerability_class="x",
+                path="app.py",
+                evidence="",
+                rule_id=rule_id,
+                sink=sink,
+                mechanism="source_reaches_sink",
+            ),
+        ),
+        min_recall=1.0,
+        fix_policy="silent",
+        provenance=provenance,
+        split="holdout",
+    )
+
+
+def test_round_rejects_when_a_profile_regresses_on_holdout(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text("def f(x):\n    return eval(x)\ndef g():\n    print('hello')\n")
+    ruleset_dir = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")])
+    _, manifest = _manifest(tmp_path, _expected("good-eval"))
+    ledger = tmp_path / "rule_policy.json"
+    journal = tmp_path / "journal.json"
+    # The agent profile's holdout pair is only detected by the rule the round wants to shadow.
+    agent_pairs = [
+        _pair(tmp_path, f"agent{i}", "def h(y):\n    print(y)\n", "def h(y):\n    return y\n", "noisy-print", "print", provenance="agent")
+        for i in range(5)
+    ]
+    outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir, pair_cases=agent_pairs)
+    assert not outcome.accepted
+    assert outcome.reason == "profile_regression:agent"
+    assert outcome.profile_regressions == ["agent"]
+    assert outcome.per_profile_before["agent"]["pair_correct"] == 5.0 and outcome.per_profile_after["agent"]["pair_correct"] == 0.0
+    assert not ledger.exists()
+    assert json.loads(journal.read_text())[0]["profile_regressions"] == ["agent"]
+
+
+def test_round_accepts_when_profiles_hold_and_reports_small_profiles(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text("def f(x):\n    return eval(x)\ndef g():\n    print('hello')\n")
+    ruleset_dir = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")])
+    _, manifest = _manifest(tmp_path, _expected("good-eval"))
+    ledger = tmp_path / "rule_policy.json"
+    journal = tmp_path / "journal.json"
+    human_pairs = [
+        _pair(tmp_path, f"h{i}", "def h(y):\n    return eval(y)\n", "def h(y):\n    return y\n", "good-eval", "eval", provenance="human")
+        for i in range(5)
+    ]
+    tiny_agent = [
+        _pair(tmp_path, "a0", "def h(y):\n    print(y)\n", "def h(y):\n    return y\n", "noisy-print", "print", provenance="agent")
+    ]
+    outcome = run_round(
+        repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir, pair_cases=human_pairs + tiny_agent
+    )
+    assert outcome.accepted and outcome.reason == "accepted"
+    assert outcome.profile_regressions == []
+    assert outcome.profiles_under_minimum == ["agent"]  # one pair: reported, not gated
+    assert ledger.exists()