merge(plane): scoring parity with the reference -- sites, ranges, triage cost

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

tests/test_plane_increment.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import json
 import re
 import shutil
+import subprocess
 import tomllib
 from pathlib import Path
 
@@ -15,13 +16,14 @@ from test_plane_reconciler import AX_FIELDS, non_ax_fields
 from openultrasast.model.endpoint import price_of
 from openultrasast.plane import reconciler
 from openultrasast.plane.budget import prices_from
-from openultrasast.plane.generate import TEMPLATES, increment, read_cases, render
+from openultrasast.plane.generate import TEMPLATES, fix_ranges, increment, read_cases, render, repin_templates
 from openultrasast.plane.manifests import load_manifests
 
 PLANE = Path("plane")
 RUN = PLANE / "runs" / "validation-46.yaml"
 POPULATION = Path("benchmarks/independent/population-v2.toml")
 RESULTS = Path.home() / "ousast-results"
+REPOS = Path.home() / ".cache" / "openultrasast" / "independent"
 PIN = "0123456789abcdef0123456789abcdef01234567"
 
 
@@ -81,14 +83,30 @@ def test_validation_run_resolves_and_renders_every_task(tmp_path: Path) -> None:
 def test_inputs_carry_the_validation_set() -> None:
     manifests = load_manifests(sorted((PLANE / "workspaces").glob("*-inputs.yaml")))
     assert len(manifests.workspaces) == 15
-    candidates = kept = 0
+    candidates = kept = in_set = before = 0
+    usd = triage = 0.0
     for workspace in manifests.workspaces.values():
         files = {f.path: json.loads(f.content) for f in workspace.files}
         assert set(files) == {"candidates.json", "functions.json", "case.json", "triage.json"}
         candidates += len(files["triage.json"]["candidates"])
         kept += len(files["candidates.json"]["candidates"])
         assert [[r["path"], r["function"], r["line"]] for r in files["functions.json"]] == files["candidates.json"]["candidates"]
-    assert (candidates, kept) == (46, 43)
+        case = files["case.json"]
+        assert case["ranges"] and all(first <= last for spans in case["ranges"].values() for first, last in spans), case["id"]
+        assert set(case["sites_in_set"]) <= set(case["sites"])
+        in_set += len(case["sites_in_set"])
+        before += case["cost"]["candidates_before_triage"]
+        usd += case["cost"]["recorded_usd"]
+        triage += case["cost"]["recorded_triage_usd"]
+        assert 0 < case["cost"]["recorded_triage_usd"] <= case["cost"]["recorded_usd"], case["id"]
+    assert (candidates, kept, in_set, before) == (46, 43, 20, 46)
+    # the reference's $1.01 for 46 candidates, two passes, triage included; triage was about 9% of it
+    assert round(usd / before, 3) == 0.022 and 0.05 < triage / usd < 0.15
+
+
+def test_templates_are_pinned_to_one_runner_image() -> None:
+    images = {task.image for task in load_manifests(sorted((PLANE / "tasks").glob("*.yaml"))).tasks.values()}
+    assert len(images) == 1 and re.fullmatch(r"\S+@sha256:[0-9a-f]{64}", images.pop() or "")
 
 
 def _committed_command() -> str:
@@ -98,7 +116,9 @@ def _committed_command() -> str:
     return found.group(1)
 
 
-@pytest.mark.skipif(not (RESULTS / "independent-v2-batched-check" / "scans").is_dir(), reason="recorded results not on this host")
+@pytest.mark.skipif(
+    not (RESULTS / "independent-v2-batched-check" / "scans").is_dir() or not REPOS.is_dir(), reason="recorded results not on this host"
+)
 def test_committed_manifests_match_the_generator() -> None:
     templates = load_manifests([PLANE / "tasks" / f"{name}.yaml" for name in TEMPLATES]).tasks
     cases = read_cases(
@@ -106,6 +126,7 @@ def test_committed_manifests_match_the_generator() -> None:
         RESULTS / "independent-v2-modelsinks",
         RESULTS / "independent-v2-batched-check" / "scans",
         POPULATION,
+        REPOS,
     )
     for relative, text in render(cases, templates, "validation-46", _committed_command()).items():
         assert (PLANE / relative).read_text() == text, relative
@@ -173,3 +194,43 @@ def test_generator_refuses_a_set_the_recorded_scans_did_not_ask(tmp_path: Path)
     (paths["scans"] / "alpha--vulnerable_a.jsonl").unlink()
     with pytest.raises(ValueError, match="no recorded triage"):
         _generate(paths, paths["plane"])
+
+
+def _git(repo: Path, *args: str) -> str:
+    done = subprocess.run(
+        ["git", "-C", str(repo), "-c", "user.name=t", "-c", "user.email=t@t", *args], capture_output=True, text=True, check=True
+    )
+    return done.stdout.strip()
+
+
+def test_fix_ranges_are_the_old_side_of_the_fix_diff(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    _git(repo, "init", "-q")
+    (repo / "a.py").write_text("".join(f"line {i}\n" for i in range(1, 31)))
+    _git(repo, "add", "-A")
+    _git(repo, "commit", "-qm", "vulnerable")
+    vulnerable = _git(repo, "rev-parse", "HEAD")
+    (repo / "a.py").write_text("".join(f"line {i}\n" if i not in (3, 4, 20) else "fixed\n" for i in range(1, 31)) + "added\n")
+    _git(repo, "commit", "-qam", "fixed")
+    assert fix_ranges(repo, vulnerable, _git(repo, "rev-parse", "HEAD")) == {"a.py": [(3, 4), (20, 20), (30, 30)]}
+    with pytest.raises(ValueError, match="no repository"):
+        fix_ranges(tmp_path / "absent", vulnerable, vulnerable)
+
+
+def test_case_record_carries_sites_in_set_and_cost_and_the_image_is_repinned(tmp_path: Path) -> None:
+    paths = _fixture(tmp_path)
+    digest = "localhost:5001/ousast-runner@sha256:" + "ab" * 32
+    (tmp_path / "runner-image").write_text(digest + "\n")
+    increment(
+        paths["population"], paths["set"], paths["candidates"], paths["scans"], plane=paths["plane"], run_name="check",
+        command="ousast plane test", runner_image=tmp_path / "runner-image",
+    )  # fmt: skip
+    manifests = load_manifests(sorted(paths["plane"].rglob("*.yaml")))
+    assert {t.image for t in manifests.tasks.values()} == {digest}
+    case = json.loads(next(f.content for f in manifests.workspaces["alpha-inputs"].files if f.path == "case.json"))
+    assert "ranges" not in case and case["sites_in_set"] == ["a.py::run"]  # no --repos: agree reports not assessed
+    assert case["cost"]["candidates_before_triage"] == 2 and case["cost"]["recorded_triage_usd"] == 0.5  # no hunt usage recorded
+    (tmp_path / "runner-image").write_text("localhost:5001/ousast-runner:dev\n")
+    with pytest.raises(ValueError, match="digest-pinned"):
+        repin_templates(paths["plane"], tmp_path / "runner-image")

tests/test_plane_scoring_parity.py

@@ -0,0 +1,144 @@
+"""`agree` scores exactly as the reference measurement did (ai-service-plane Requirement 6).
+
+The reference is `benchmarks/independent/score_batched.py` over the batched-check scans: sites anchored by
+`score_batched.anchored`, matched by `evaluate.matches_v2` against `evaluate.hunks(case, "old")`. The plane cannot
+ship the benchmark tree, so the generator resolves the ranges into `case.json` and `agree.matches_declared`
+reproduces the rule. These tests run both on the recorded reference outputs of the 15 validation cases and
+require identical verdicts, identical ranges, and the reference's own numbers (16 of 20, $0.022) back from `agree`.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+import tomllib
+from pathlib import Path
+from types import ModuleType
+from typing import Any
+
+import pytest
+
+from openultrasast.plane.manifests import load_manifests
+from openultrasast.plane.tasks import agree
+
+BENCH = Path("benchmarks/independent")
+POPULATION = BENCH / "population-v2.toml"
+SCANS = Path.home() / "ousast-results" / "independent-v2-batched-check" / "scans"
+REPOS = Path.home() / ".cache" / "openultrasast" / "independent"
+INPUTS = sorted(Path("plane/workspaces").glob("*-inputs.yaml"))
+
+pytestmark = pytest.mark.skipif(
+    not SCANS.is_dir() or not REPOS.is_dir(), reason="recorded batched-check scans or case clones not on this host"
+)
+
+
+def _load(name: str) -> ModuleType:
+    sys.path.insert(0, str(BENCH.resolve()))  # score_batched imports evaluate by name
+    try:
+        spec = importlib.util.spec_from_file_location(f"parity_{name}", BENCH / f"{name}.py")
+        assert spec is not None and spec.loader is not None
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        sys.path.remove(str(BENCH.resolve()))
+
+
+@pytest.fixture(scope="module")
+def reference() -> tuple[ModuleType, ModuleType]:
+    evaluate, score_batched = _load("evaluate"), _load("score_batched")
+    evaluate.use(POPULATION.resolve())
+    return evaluate, score_batched
+
+
+def _inputs() -> dict[str, dict[str, Any]]:
+    out: dict[str, dict[str, Any]] = {}
+    for workspace in load_manifests(INPUTS).workspaces.values():
+        files = {f.path: json.loads(f.content) for f in workspace.files}
+        out[files["case.json"]["id"]] = files
+    assert len(out) == 15, sorted(out)
+    return out
+
+
+def _records() -> dict[str, dict[str, Any]]:
+    return {c["id"]: c for c in tomllib.loads(POPULATION.read_text())["case"]}
+
+
+def test_committed_ranges_are_the_reference_hunks(reference: tuple[ModuleType, ModuleType]) -> None:
+    evaluate, _ = reference
+    records = _records()
+    for case_id, files in _inputs().items():
+        hunks = {path: [list(span) for span in spans] for path, spans in evaluate.hunks(records[case_id], "old").items()}
+        assert files["case.json"]["ranges"] == hunks, case_id
+
+
+def test_matching_verdicts_are_identical_on_the_recorded_outputs(reference: tuple[ModuleType, ModuleType]) -> None:
+    evaluate, score_batched = reference
+    records = _records()
+    compared = declared = matched = 0
+    for case_id, files in _inputs().items():
+        record, mine = records[case_id], files["case.json"]
+        old = evaluate.hunks(record, "old")
+        recorded = json.loads((SCANS / f"{case_id}--vulnerable_a.json").read_text())
+        raw = [dict(f) for key in ("findings", "disputed") for f in recorded.get(key, [])]
+        anchored = score_batched.anchored(json.loads(json.dumps({"findings": raw})))["findings"]
+        assert [agree.anchor_site(f["candidate"], f["site"]) for f in raw] == [f["site"] for f in anchored], case_id
+        family = record["family"]
+        probes = [{"site": f["site"], "family": f["family"]} for f in anchored]
+        probes += [{"site": f"{p}:{ln}:{fn}", "family": family} for p, fn, ln in files["triage.json"]["candidates"]]
+        sites = [{"site": f"{s.partition('::')[0]}:0:{s.partition('::')[2]}", "family": family} for s in record.get("sites", [])]
+        declared += len(sites)
+        probes += sites
+        probes += [dict(p, family="not-" + family) for p in probes]
+        for probe in probes:
+            theirs = evaluate.matches_v2(probe, record, old)
+            assert agree.matches_declared(probe, mine, mine["ranges"]) == theirs, (case_id, probe)
+            compared += 1
+            matched += theirs
+    assert declared == 22 and compared > 2 * (declared + 46)
+    assert matched > 0
+
+
+def _units(case_id: str, kept: list[list[Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]], float]:
+    """The recorded per-file hunts as two `verify` passes: a hunt's usd is the file's spend minus its triage."""
+    from openultrasast.model.endpoint import price_of
+    from openultrasast.plane.budget import cost_of
+
+    prices = price_of("deepseek-flash")
+    assert prices is not None
+    passes: tuple[list[dict[str, Any]], list[dict[str, Any]]] = ([], [])
+    triage = 0.0
+    for text in (SCANS / f"{case_id}--vulnerable_a.jsonl").read_text().splitlines():
+        row = json.loads(text)
+        hunts = cost_of(row.get("usage") or {}, prices)
+        triage += max(float(row["usd"]) - hunts, 0.0)
+        asked = [c for c in kept if c[0] == row["path"]]
+        if not asked:
+            continue
+        for index, flagged in enumerate(row["passes"]):
+            passes[index].append({"path": row["path"], "candidates": asked, "flagged": flagged, "usd": hunts / 2, "turns": 0})
+    return passes[0], passes[1], triage
+
+
+def test_agree_on_the_reference_outputs_reproduces_the_reference_numbers(tmp_path: Path) -> None:
+    score = json.loads((SCANS.parent / "score.json").read_text())
+    detected = {c["id"]: len(c["detected_sites"]) for c in score["cases"] if "detected_sites" in c and c["detected_sites"] is not None}
+    agreed = in_set = before = 0
+    spend = with_triage = 0.0
+    for case_id, files in _inputs().items():
+        case = files["case.json"]
+        pass_a, pass_b, triage = _units(case_id, files["candidates.json"]["candidates"])
+        assert case["cost"]["recorded_triage_usd"] == pytest.approx(triage, abs=1e-6)
+        summary = agree.run(tmp_path / case_id, files["candidates.json"], pass_a, pass_b, sites=case)
+        metrics = summary["metrics"]
+        assert metrics["site_matching"] == "assessed"
+        assert metrics["declared_sites_matched_agreed"] == detected[case_id], case_id
+        agreed += metrics["declared_sites_matched_agreed"]
+        in_set += metrics["declared_sites_in_set"]
+        before += metrics["candidates_before_triage"]
+        spend += metrics["usd_total"]
+        with_triage += metrics["usd_total_with_recorded_triage"]
+    assert (agreed, in_set, before) == (16, 20, 46)
+    assert with_triage == pytest.approx(1.0123, abs=1e-3)  # the recorded spend: the reference's $1.01
+    assert round(with_triage / before, 3) == 0.022 and spend < with_triage

tests/test_plane_verify.py

@@ -230,8 +230,8 @@ def _two_passes(tmp_path: Path) -> tuple[Path, Path, Path]:
 def test_agree_over_two_scripted_passes_computes_agreement_and_metrics(tmp_path: Path) -> None:
     _, a, b = _two_passes(tmp_path)
     out = tmp_path / "agree"
-    sites = {"id": "case", "family": "injection", "sites": ["app.py::run", "db.py::query"]}
-    summary = agree.run(out, CANDIDATES, agree.load_pass(a), agree.load_pass(b), sites=sites, matcher=agree.load_matcher())
+    sites = {"id": "case", "family": "injection", "sites": ["app.py::run", "db.py::query"], "ranges": {"app.py": [[40, 41]]}}
+    summary = agree.run(out, CANDIDATES, agree.load_pass(a), agree.load_pass(b), sites=sites)
     assert summary["status"] == "done" and summary["usd"] == 0 and summary["calls"] == 0 and summary["model"] is None
     report = json.loads((out / "agreed.json").read_text())
     by_name = {r["candidate"]: r for r in report["candidates"]}
@@ -255,6 +255,34 @@ def test_agree_over_two_scripted_passes_computes_agreement_and_metrics(tmp_path:
     assert by_name["app.py::run"]["usd"] == pytest.approx(2 * per_call, abs=1e-6)
 
 
+def test_agree_reports_cost_with_the_recorded_triage_and_needs_fix_ranges_to_assess_sites(tmp_path: Path) -> None:
+    _, a, b = _two_passes(tmp_path)
+    cost = {"candidates_before_triage": 4, "recorded_usd": 1.0, "recorded_triage_usd": 0.01, "recorded_triage_basis": "derived"}
+    sites = {"id": "case", "family": "injection", "sites": ["app.py::run"], "sites_in_set": ["app.py::run"], "cost": cost}
+    metrics = agree.run(tmp_path / "agree", CANDIDATES, agree.load_pass(a), agree.load_pass(b), sites=sites)["metrics"]
+    spend = metrics["usd_total"]
+    assert metrics["candidates"] == 3 and metrics["candidates_before_triage"] == 4 and metrics["declared_sites_in_set"] == 1
+    assert metrics["cost_per_candidate"] == pytest.approx(spend / 4, abs=1e-6)
+    assert metrics["usd_total_with_recorded_triage"] == pytest.approx(spend + 0.01, abs=1e-6)
+    assert metrics["cost_per_candidate_with_recorded_triage"] == pytest.approx((spend + 0.01) / 4, abs=1e-6)
+    # without the fix ranges a miss cannot be told from a match: not assessed, never a silent zero
+    assert metrics["site_matching"] == "not assessed: no fix ranges given" and metrics["declared_sites_matched"] == 0
+
+
+def test_matches_declared_is_the_protocol_v2_rule() -> None:
+    case = {"family": "injection", "sites": ["a.py::run", "b.py::<global>"]}
+    ranges = {"a.py": [[50, 52]], "c.py": [[5, 5]]}
+    hit = lambda site, family="injection": agree.matches_declared({"site": site, "family": family}, case, ranges)  # noqa: E731
+    assert hit("a.py:0:run") and hit("b.py:0:<global>")  # a declared path::function, whether or not the fix touched it
+    assert not hit("a.py:0:run", "ssrf")  # another family never matches
+    assert hit("a.py:62:helper") and not hit("a.py:63:helper") and hit("a.py:40:x") and not hit("a.py:39:x")  # 10-line window
+    assert hit("c.py:900:run") and not hit("d.py:1:run")  # a declared function name in a changed file only
+    assert (
+        not hit("a.py:x:helper") and hit("c.py:x:helper") and agree.anchor_site("a.py::run", "other.py:7:run") == "a.py:0:run"
+    )  # a non-numeric line is -1, as in the reference
+    assert agree.anchor_site("a.py::run", "a.py:7:renamed") == "a.py:7:run"
+
+
 def test_agree_without_sites_leaves_site_matching_unassessed_and_flags_unasked_candidates(tmp_path: Path) -> None:
     _, a, b = _two_passes(tmp_path)
     rows_b = [r for r in agree.load_pass(b) if r["path"] != "db.py"]
@@ -269,7 +297,8 @@ def test_agree_main_reads_the_env_contract(tmp_path: Path, monkeypatch: pytest.M
     _, a, b = _two_passes(tmp_path)
     (tmp_path / "candidates.json").write_text(json.dumps(CANDIDATES))
     (tmp_path / "case.toml").write_text(
-        '[[case]]\nid = "case"\nfamily = "injection"\nsites = ["app.py::run"]\n\n[[case]]\nid = "x"\nfamily = "injection"\nsites = []\n'
+        '[[case]]\nid = "case"\nfamily = "injection"\nsites = ["app.py::run"]\nranges = { "app.py" = [[40, 41]] }\n\n'
+        '[[case]]\nid = "x"\nfamily = "injection"\nsites = []\n'
     )
     env = {
         "OUSAST_OUTPUT_DIR": str(tmp_path / "agree"), "OUSAST_INPUT_PASS_A": str(a), "OUSAST_INPUT_PASS_B": str(b / "units.jsonl"),