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 == []