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) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
2026-09-06 11:28:47 +02:00
co-authored by Claude Opus 5
parent fcda1ee46b
commit e8e6bafa61
2 changed files with 406 additions and 0 deletions
+179
View File
@@ -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 == []