From 9e7aef6a6b43f8847d2eb4c097c0323a9304fade Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 5 Sep 2026 09:28:00 +0200 Subject: [PATCH] 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:; 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 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- src/openultrasast/cli.py | 29 +++++++++ src/openultrasast/improve/evolve.py | 99 ++++++++++++++++++++++++++++- tests/test_cli_improve.py | 9 +++ tests/test_improve.py | 83 ++++++++++++++++++++++++ 4 files changed, 217 insertions(+), 3 deletions(-) diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 36608e6..1e0bbb2 100644 --- a/src/openultrasast/cli.py +++ b/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)") diff --git a/src/openultrasast/improve/evolve.py b/src/openultrasast/improve/evolve.py index 4fb2779..a824873 100644 --- a/src/openultrasast/improve/evolve.py +++ b/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 diff --git a/tests/test_cli_improve.py b/tests/test_cli_improve.py index 52bbdac..0a4de47 100644 --- a/tests/test_cli_improve.py +++ b/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")]) diff --git a/tests/test_improve.py b/tests/test_improve.py index d2416c0..afa0bec 100644 --- a/tests/test_improve.py +++ b/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()