From e8e6bafa6137026dd3c49dcdf2e8872e8c9c1160 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 6 Sep 2026 11:28:47 +0200 Subject: [PATCH] feat(learning-harness): round zero and the per-family noise floor Task 3.3. Before anything learns, every family gets the same detector and is measured against itself. Running a pair several times at the same settings does not give the same answer, because the chat provider offers no seed, so the spread between runs is real and unavoidable. That spread is the family's noise floor and it becomes the budget a later round must stay inside before a cross-family change counts as a regression. Choosing a budget instead of measuring one would be a guess dressed as a threshold. A pair the corpus already marked unscorable is never run: spending model calls on a row that cannot be scored buys nothing and would put its noise into a floor where it does not belong. Artifacts are keyed by detector model, so a second model is baselined by changing the model and running again, and the comparison table reads both side by side. A model swap never touches a configuration, which is what makes the two numbers comparable and reconsidering the model cheap. The web slices are the default selection and the memory-safety family is measured with its floor marked as never gating a round. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- src/openultrasast/learning/rounds.py | 227 +++++++++++++++++++++++++++ tests/test_learning_rounds.py | 179 +++++++++++++++++++++ 2 files changed, 406 insertions(+) create mode 100644 src/openultrasast/learning/rounds.py create mode 100644 tests/test_learning_rounds.py diff --git a/src/openultrasast/learning/rounds.py b/src/openultrasast/learning/rounds.py new file mode 100644 index 0000000..2272c62 --- /dev/null +++ b/src/openultrasast/learning/rounds.py @@ -0,0 +1,227 @@ +"""Round zero and the per-family noise floor (learning-harness, Req 8). + +Before anything learns, every family gets the same detector and is measured against itself. Running a +pair several times at the same settings does not give the same answer: the chat provider offers no seed, +so the spread between runs is real and unavoidable. That spread is the family's noise floor, and it +becomes the budget a later round must stay inside before a cross-family change counts as a regression. +Choosing a budget instead of measuring one would be a guess dressed as a threshold. + +Artifacts are keyed by detector model, so a second model is baselined by changing the model and running +again. Configurations are never touched by a model swap, which is what makes the two numbers comparable +and what makes reconsidering the model cheap. +""" + +from __future__ import annotations + +import json +import math +import re +import tempfile +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path + +from ..findings import StaticFinding +from ..pairs import PairCase +from .classify import classify_pair +from .detectors import write_default_configs +from .families import FamilyTaxonomy +from .scoring import FamilyMetrics, PairFamilyScore, aggregate, score_pair_family + +DEFAULT_SLICES = ("vibe-py", "agent-vfc") # the web families, where real-world code and dangerous fruit coincide +NON_GATING_FAMILIES = frozenset({"memory"}) # measured on the vendored slice, never allowed to fail a round +DEFAULT_K_RUNS = 5 +Scan = Callable[[Path], list[StaticFinding]] + + +@dataclass(frozen=True) +class NoiseFloor: + """How much a family disagrees with itself between runs, and the budget that follows from it.""" + + family: str + k_runs: int + pairs: int + flaky_pairs: int + negative_flip_rate: float + budget_flips: int + gates: bool = True + + def to_dict(self) -> dict[str, object]: + return { + "family": self.family, + "k_runs": self.k_runs, + "pairs": self.pairs, + "flaky_pairs": self.flaky_pairs, + "negative_flip_rate": self.negative_flip_rate, + "budget_flips": self.budget_flips, + "gates": self.gates, + } + + +@dataclass(frozen=True) +class BaselineReport: + model: str + k_runs: int + taxonomy_version: str + slices: tuple[str, ...] + floors: dict[str, NoiseFloor] = field(default_factory=dict) + metrics: dict[str, dict[str, object]] = field(default_factory=dict) + scores: tuple[PairFamilyScore, ...] = () + + def to_dict(self) -> dict[str, object]: + return { + "model": self.model, + "k_runs": self.k_runs, + "taxonomy_version": self.taxonomy_version, + "slices": list(self.slices), + "floors": {name: floor.to_dict() for name, floor in sorted(self.floors.items())}, + "metrics": {name: dict(block) for name, block in sorted(self.metrics.items())}, + } + + +def run_baseline( + cases: Sequence[PairCase], + *, + taxonomy: FamilyTaxonomy, + configs_dir: Path, + scan: Scan, + model: str, + out_dir: Path, + slices: Sequence[str] = DEFAULT_SLICES, + k_runs: int = DEFAULT_K_RUNS, + prompt: str | None = None, +) -> BaselineReport: + """Clone one detector into every family, measure each family against itself, and write the artifacts.""" + from ..tool_hunter import _SYSTEM_PROMPT + + if not configs_dir.exists(): + write_default_configs(configs_dir, taxonomy, prompt=prompt or _SYSTEM_PROMPT, version="0") + selected = [case for case in cases if case.slice in set(slices)] + scores = [score_case(case, scan, taxonomy=taxonomy, runs=k_runs) for case in selected] + metrics = aggregate(scores, taxonomy=taxonomy) + floors = {name: _floor(name, block, [item for item in scores if item.family == name], k_runs) for name, block in metrics.items()} + report = BaselineReport( + model=model, + k_runs=k_runs, + taxonomy_version=taxonomy.version, + slices=tuple(slices), + floors=floors, + metrics={name: block.to_dict() for name, block in metrics.items()}, + scores=tuple(scores), + ) + _write(out_dir, report) + return report + + +def score_case(case: PairCase, scan: Scan, *, taxonomy: FamilyTaxonomy, runs: int, family: str | None = None) -> PairFamilyScore: + """Run one detector over both sides of one pair ``runs`` times and score it for its family. + + A pair the corpus already marked unscorable is never run: spending model calls on a row that cannot + be scored buys nothing and would put its noise into a floor it does not belong in. + """ + from ..pairs import _materialize, _overlay_scan + + name = family or _family_of(case, taxonomy) + if case.unscorable: + return PairFamilyScore( + pair=case.name, + family=name, + runs=(), + outcome="unscorable", + slice=case.slice, + unscorable_reason=case.unscorable, + ) + with tempfile.TemporaryDirectory(prefix="ousast-round-") as scratch: + vuln_root = _materialize(Path(scratch) / "vuln", case.vuln_file, case.relpath) + fix_root = _materialize(Path(scratch) / "fixed", case.fixed_file, case.relpath) + ranges = _overlay_scan(vuln_root).ranges + vuln_runs = [scan(vuln_root) for _ in range(max(runs, 1))] + fix_runs = [scan(fix_root) for _ in range(max(runs, 1))] + return score_pair_family(case, name, vuln_runs, fix_runs, ranges=ranges, taxonomy=taxonomy, parse_ok=bool(ranges)) + + +def compare_baselines(out_dir: Path) -> dict[str, dict[str, dict[str, object]]]: + """Every baseline written under ``out_dir``, model by model, so two detectors can be read side by side.""" + root = out_dir / "baseline" + table: dict[str, dict[str, dict[str, object]]] = {} + for directory in sorted(path for path in root.iterdir() if path.is_dir()) if root.is_dir() else []: + report = directory / "report.json" + if not report.is_file(): + continue + payload = json.loads(report.read_text(encoding="utf-8")) + table[str(payload.get("model", directory.name))] = {family: dict(block) for family, block in (payload.get("metrics") or {}).items()} + return table + + +def _floor(family: str, metrics: FamilyMetrics, scores: Sequence[PairFamilyScore], k_runs: int) -> NoiseFloor: + """A family's own disagreement between runs, as a rate and as a whole number of pairs.""" + scorable = [item for item in scores if item.outcome != "unscorable"] + flaky = sum(1 for item in scorable if item.flips) + rate = flaky / len(scorable) if scorable else 0.0 + return NoiseFloor( + family=family, + k_runs=k_runs, + pairs=metrics.scorable, + flaky_pairs=flaky, + negative_flip_rate=rate, + budget_flips=math.ceil(rate * metrics.scorable) if metrics.scorable else 0, + gates=family not in NON_GATING_FAMILIES, + ) + + +def _family_of(case: PairCase, taxonomy: FamilyTaxonomy) -> str: + for row in case.expected: + if row.family: + return str(row.family) + answer = classify_pair(case, taxonomy, client=None) + return answer.families[0] if answer.families else "unknown" + + +def _write(out_dir: Path, report: BaselineReport) -> None: + directory = out_dir / "baseline" / _slug(report.model) + directory.mkdir(parents=True, exist_ok=True) + payload = report.to_dict() + (directory / "noise-floors.json").write_text( + json.dumps({"model": report.model, "k_runs": report.k_runs, "floors": payload["floors"]}, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + (directory / "report.json").write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def _slug(model: str) -> str: + return re.sub(r"[^A-Za-z0-9._-]+", "-", model).strip("-") or "unnamed" + + +def load_noise_floors(out_dir: Path, model: str) -> dict[str, NoiseFloor]: + """The budgets a later round must stay inside, as round zero measured them for this model.""" + path = out_dir / "baseline" / _slug(model) / "noise-floors.json" + if not path.is_file(): + return {} + payload = json.loads(path.read_text(encoding="utf-8")) + floors: dict[str, NoiseFloor] = {} + for name, block in (payload.get("floors") or {}).items(): + if isinstance(block, Mapping): + floors[str(name)] = NoiseFloor( + family=str(block.get("family", name)), + k_runs=int(block.get("k_runs", DEFAULT_K_RUNS) or DEFAULT_K_RUNS), + pairs=int(block.get("pairs", 0) or 0), + flaky_pairs=int(block.get("flaky_pairs", 0) or 0), + negative_flip_rate=float(block.get("negative_flip_rate", 0.0) or 0.0), + budget_flips=int(block.get("budget_flips", 0) or 0), + gates=bool(block.get("gates", True)), + ) + return floors + + +__all__ = [ + "DEFAULT_K_RUNS", + "DEFAULT_SLICES", + "NON_GATING_FAMILIES", + "BaselineReport", + "NoiseFloor", + "Scan", + "compare_baselines", + "load_noise_floors", + "run_baseline", + "score_case", +] diff --git a/tests/test_learning_rounds.py b/tests/test_learning_rounds.py new file mode 100644 index 0000000..92be6ba --- /dev/null +++ b/tests/test_learning_rounds.py @@ -0,0 +1,179 @@ +"""learning-harness task 3.3: round zero and the per-family noise floor (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +from openultrasast.benchmark import ExpectedFinding +from openultrasast.findings import StaticFinding +from openultrasast.learning.families import load_families +from openultrasast.pairs import PairCase + +VULN = "from flask import request\nimport os\n\n\ndef run():\n os.system(request.args.get('cmd'))\n" +FIXED = "from flask import request\nimport subprocess\n\n\ndef run():\n subprocess.run(['echo', 'x'], check=False)\n" + + +def _case(tmp_path: Path, name: str, *, family: str = "injection", slice_name: str = "vibe-py", unscorable: str | None = None) -> PairCase: + (tmp_path / f"{name}-v.py").write_text(VULN) + (tmp_path / f"{name}-f.py").write_text(FIXED) + return PairCase( + name=name, + slice=slice_name, + language="python", + origin="test", + vuln_file=tmp_path / f"{name}-v.py", + fixed_file=tmp_path / f"{name}-f.py", + relpath="app.py", + expected=( + ExpectedFinding( + cwe="CWE-78", vulnerability_class="x", path="app.py", evidence="", function="run", mechanism="other", family=family + ), + ), + min_recall=1.0, + fix_policy="silent", + review_tier="seeded", + ) + + +def _finding(family: str) -> StaticFinding: + return StaticFinding( + finding_id="detector:app.py:6", + path="app.py", + title="t", + severity="high", + confidence="low", + evidence_level="suspicion", + rationale="r", + line=6, + function_name="run", + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=[f"family:{family}", "detector:x@0"], + ranking_priority=0.0, + ) + + +def _steady(root: Path) -> list[StaticFinding]: + return [_finding("injection")] if "-v" in str(root) or root.name == "vuln" else [] + + +def _flaky(counter: list[int]): # type: ignore[no-untyped-def] + def scan(root: Path) -> list[StaticFinding]: + if not ("-v" in str(root) or root.name == "vuln"): + return [] + counter.append(1) + return [_finding("injection")] if len(counter) % 2 else [] + + return scan + + +def test_round_zero_writes_a_configuration_for_every_family(tmp_path: Path) -> None: + from openultrasast.learning.detectors import load_family_configs + from openultrasast.learning.rounds import run_baseline + + taxonomy = load_families() + report = run_baseline( + [_case(tmp_path, "a")], taxonomy=taxonomy, configs_dir=tmp_path / "configs", scan=_steady, model="m", out_dir=tmp_path / "out" + ) + configs = load_family_configs(tmp_path / "configs", taxonomy) + assert set(configs) == {family.id for family in taxonomy.families} + assert all(config.version == "0" for config in configs.values()) + assert report.model == "m" and report.k_runs == 5 + + +def test_a_steady_detector_has_a_zero_noise_floor_and_a_flaky_one_does_not(tmp_path: Path) -> None: + from openultrasast.learning.rounds import run_baseline + + taxonomy = load_families() + steady = run_baseline( + [_case(tmp_path, "a"), _case(tmp_path, "b")], + taxonomy=taxonomy, + configs_dir=tmp_path / "c1", + scan=_steady, + model="m", + out_dir=tmp_path / "o1", + ) + floor = steady.floors["injection"] + assert floor.k_runs == 5 and floor.negative_flip_rate == 0.0 and floor.budget_flips == 0 and floor.gates is True + flaky = run_baseline( + [_case(tmp_path, "a"), _case(tmp_path, "b")], + taxonomy=taxonomy, + configs_dir=tmp_path / "c2", + scan=_flaky([]), + model="m", + out_dir=tmp_path / "o2", + ) + assert flaky.floors["injection"].negative_flip_rate > 0.0 and flaky.floors["injection"].budget_flips >= 1 + + +def test_the_memory_family_is_measured_but_never_gates(tmp_path: Path) -> None: + from openultrasast.learning.rounds import run_baseline + + taxonomy = load_families() + report = run_baseline( + [_case(tmp_path, "m", family="memory", slice_name="vfc")], + taxonomy=taxonomy, + configs_dir=tmp_path / "configs", + scan=_steady, + model="m", + out_dir=tmp_path / "out", + slices=("vfc",), + ) + assert "memory" in report.floors and report.floors["memory"].gates is False + assert report.metrics["memory"]["scorable"] == 1 + + +def test_the_default_selection_is_the_web_slices(tmp_path: Path) -> None: + from openultrasast.learning.rounds import run_baseline + + taxonomy = load_families() + cases = [_case(tmp_path, "web", slice_name="vibe-py"), _case(tmp_path, "c", family="memory", slice_name="vfc")] + report = run_baseline(cases, taxonomy=taxonomy, configs_dir=tmp_path / "configs", scan=_steady, model="m", out_dir=tmp_path / "out") + assert set(report.metrics) == {"injection"} and "memory" not in report.floors + + +def test_artifacts_are_keyed_by_model_so_another_model_can_be_compared(tmp_path: Path) -> None: + from openultrasast.learning.rounds import compare_baselines, run_baseline + + taxonomy = load_families() + out = tmp_path / "out" + for index, model in enumerate(("deepseek-v4-flash", "deepseek-v4-pro")): + run_baseline( + [_case(tmp_path, f"a{index}")], + taxonomy=taxonomy, + configs_dir=tmp_path / "configs", + scan=_steady, + model=model, + out_dir=out, + ) + written = sorted(path.name for path in (out / "baseline").iterdir()) + assert written == ["deepseek-v4-flash", "deepseek-v4-pro"] + floors = json.loads((out / "baseline" / "deepseek-v4-flash" / "noise-floors.json").read_text()) + assert floors["model"] == "deepseek-v4-flash" and floors["floors"]["injection"]["k_runs"] == 5 + report = json.loads((out / "baseline" / "deepseek-v4-flash" / "report.json").read_text()) + assert report["metrics"]["injection"]["recall"] == 1.0 and report["taxonomy_version"] + table = compare_baselines(out) + assert set(table) == {"deepseek-v4-flash", "deepseek-v4-pro"} + assert table["deepseek-v4-pro"]["injection"]["youden"] == 1.0 + # a second baseline never rewrites a configuration + assert (tmp_path / "configs" / "injection" / "family.toml").read_text().count("version") == 1 + + +def test_an_unscorable_pair_is_listed_and_never_run(tmp_path: Path) -> None: + from openultrasast.learning.rounds import run_baseline + + calls: list[Path] = [] + + def counting(root: Path) -> list[StaticFinding]: + calls.append(root) + return _steady(root) + + taxonomy = load_families() + case = _case(tmp_path, "twin") + twin = PairCase(**{**case.__dict__, "unscorable": "identical_twin"}) + report = run_baseline([twin], taxonomy=taxonomy, configs_dir=tmp_path / "configs", scan=counting, model="m", out_dir=tmp_path / "out") + assert report.metrics["injection"]["scorable"] == 0 + assert report.metrics["injection"]["unscorable"] == {"identical_twin": 1} + assert calls == []