From 6e5d060de64a0a992a6b191466af8d916509eb7a Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 13 Oct 2026 18:49:58 +0200 Subject: [PATCH] feat(hunt): registered paired ablation, repository folds, leak audit, importance table and the keep/drop rule (4.2-4.4) ablation.py registers the manifest (table digest, arms as masks, folds, metric, sample, budget, rule, seed, program id) before any run and refuses a run or an analysis that does not match it. Arms run over the same changes through runner.run, paired by change; effects on recall and on suspicions per ordinary push are reported pooled and per family with the repository-clustered bootstrap and exact McNemar, plus the recall-against-suspicions curve at every reported confidence (never a tuned cut). The importance table and decisions read held-out effects only; a neutral annotation survives only when a kept one needs it. 30 fixture tests with a scripted model. Implemented by Codex (gpt-6-astra), gated here. Co-Authored-By: Claude Fable 5.1 --- .kiro/specs/hunter-learning-loop/tasks.md | 6 +- src/openultrasast/hunt/ablation.py | 478 ++++++++++++++++++++++ tests/test_hunt_ablation.py | 335 +++++++++++++++ 3 files changed, 816 insertions(+), 3 deletions(-) create mode 100644 src/openultrasast/hunt/ablation.py create mode 100644 tests/test_hunt_ablation.py diff --git a/.kiro/specs/hunter-learning-loop/tasks.md b/.kiro/specs/hunter-learning-loop/tasks.md index f1a7240..87c08a3 100644 --- a/.kiro/specs/hunter-learning-loop/tasks.md +++ b/.kiro/specs/hunter-learning-loop/tasks.md @@ -59,16 +59,16 @@ - About one hundred advisories through the fix-site catalog and verifier, pilot repositories excluded, at least 40 accepted positives across two languages, 20 ordinary pushes per positive from the same repositories; the sourcing run is recorded - Done when the frozen training manifest lists its positives, negatives, languages and the pilot exclusion check - _Requirements: 2.1, 2.2, 2.3_ -- [ ] 4.2 Fold by repository and audit every annotation for leaks +- [x] 4.2 Fold by repository and audit every annotation for leaks - Compile folds and held-out folds by repository group; the leak audit runs over every feature column and reports any that separates positives from negatives on its own - Done when the folds are disjoint by repository and the audit report lists each column with its separation score - _Requirements: 2.5, 2.6_ -- [ ] 4.3 Register and run a paired ablation +- [x] 4.3 Register and run a paired ablation - A manifest fixes the table digest, the masks, the folds, the metric, the sample size, the budget and the decision rule before anything runs; each arm runs over the same changes with the same program and sampling - Results per arm carry the paired effect on search recall and suspicions per ordinary push, per family and pooled, with intervals from the repository-clustered bootstrap and a paired test - Done when an ablation over a fixture manifest with the scripted model yields two arms, the paired statistics, and the full recall-against-suspicions curve - _Requirements: 4.1, 4.2, 4.3, 4.7, 4.8_ -- [ ] 4.4 Build the importance table and apply the keep/drop rule +- [x] 4.4 Build the importance table and apply the keep/drop rule - For every annotation, its effect when removed from the full set and when added to the baseline, with intervals and counts; an annotation stays only on a held-out effect whose interval excludes zero or when a kept annotation needs it; one that raises suspicions without raising recall is dropped and recorded - Done when the table is written for a fixture ablation and the rule's decisions match the fixture's expectations - _Requirements: 4.4, 4.5, 4.6_ diff --git a/src/openultrasast/hunt/ablation.py b/src/openultrasast/hunt/ablation.py new file mode 100644 index 0000000..f7b0b3f --- /dev/null +++ b/src/openultrasast/hunt/ablation.py @@ -0,0 +1,478 @@ +"""Register paired annotation experiments and report effects without choosing a confidence cut. + +Each unit is one change in one repeat. Intervals resample whole repositories. Failed and missing +records remain in the population. Only held-out effects feed the annotation keep/drop rule. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +from .. import stats +from . import runner +from .features import ENGINEERED, annotations_of, feature_frame +from .metric import search_outcome +from .table import FEATURE_NAMES, FeatureTable, canonical_json + + +@dataclass(frozen=True) +class Manifest: + """The choices fixed before a round runs. + + Arms map names to masked input or producer names; full must have an empty mask. Sample holds + the ordered change IDs and the repeat count. Metric is 'hunt.metric.search_outcome(tolerance=N)'. + Rule holds version 'keep-drop-v1', optional bootstrap resamples (default 2000), and annotation + dependencies under needs. Budget (usd, calls) is passed unchanged to each runner invocation. + Compile and holdout folds must be disjoint. Program ID names the common program before masking. + """ + + table_digest: str + arms: Mapping[str, list[str]] + compile_folds: list[int] + holdout_folds: list[int] + metric: str + sample: Mapping[str, Any] + budget: Mapping[str, float] + rule: Mapping[str, Any] + seed: int + program_id: str + + +def assign_folds(table: FeatureTable, *, k: int, seed: int) -> FeatureTable: + """Return a copy with repository-grouped folds; every row of a change stays together.""" + frame = table.to_pandas() + if frame.empty or frame["repo"].isna().any() or frame["change_id"].isna().any(): + raise ValueError("folds require rows with repository and change IDs") + if frame.groupby("change_id")["repo"].nunique().gt(1).any(): + raise ValueError("a change cannot belong to multiple repositories") + result = FeatureTable(table.rows, masked=table.masked) + for fold, (_, indices) in enumerate(stats.group_folds(frame, "repo", k=k, seed=seed)): + for index in indices: + result.rows[index]["fold"] = fold + return result + + +def leak_audit(table: FeatureTable) -> Any: + """Report every engineered column's label separation, scored row count, annotation and note. + + Unscorable columns remain in the report with a null AUC and the reason. This audit drops nothing + and does not decide whether separation is useful evidence or a leak. + """ + frame = feature_frame(table) + if "label" not in frame: + frame["label"] = None + report = stats.separation(frame, "label", ENGINEERED) + report["annotation"] = [ENGINEERED[column][1] for column in report["column"]] + return report + + +def manifest_id(manifest: Manifest) -> str: + """Hash the manifest's canonical JSON, including every registered choice.""" + return hashlib.sha256(canonical_json(asdict(manifest)).encode("utf-8")).hexdigest() + + +def _validate(manifest: Manifest) -> int: + from .program import INPUTS + + match = re.fullmatch(r"hunt\.metric\.search_outcome\(tolerance=(\d+)\)", manifest.metric) + if match is None: + raise ValueError("metric must name hunt.metric.search_outcome and its non-negative tolerance") + if "full" not in manifest.arms or manifest.arms["full"]: + raise ValueError("the full arm must have an empty mask") + allowed = set(FEATURE_NAMES) | {name for name, _, _ in INPUTS} + for arm, mask in manifest.arms.items(): + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]*", arm): + raise ValueError("arm names must be safe directory names") + if not isinstance(mask, list) or any(name not in allowed for name in mask): + raise ValueError("mask must contain only annotation inputs or producers") + folds = manifest.compile_folds + manifest.holdout_folds + if not manifest.holdout_folds or any(type(fold) is not int or fold < 0 for fold in folds) or len(set(folds)) != len(folds): + raise ValueError("compile and holdout folds must be disjoint, with a nonempty holdout") + ids = manifest.sample.get("changes") + if not isinstance(ids, list) or not ids or any(not isinstance(item, str) or not item for item in ids) or len(set(ids)) != len(ids): + raise ValueError("sample changes must be an ordered list of unique change IDs") + if type(manifest.sample.get("repeats")) is not int or manifest.sample["repeats"] < 1: + raise ValueError("sample repeats must be a positive integer") + if set(manifest.budget) != {"usd", "calls"} or any(not math.isfinite(v) or v < 0 for v in manifest.budget.values()): + raise ValueError("budget must have finite non-negative usd and calls bounds") + if manifest.rule.get("version") != "keep-drop-v1": + raise ValueError("unsupported decision rule version") + resamples = manifest.rule.get("resamples", stats.RESAMPLES) + if type(resamples) is not int or resamples < 2: + raise ValueError("bootstrap resamples must be an integer of at least two") + return int(match[1]) + + +def _write(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".partial") + temporary.write_text(canonical_json(value) + "\n", encoding="utf-8") + temporary.replace(path) + + +def _registration(manifest: Manifest, out: Path) -> None: + path = out / "ablation.json" + if not path.is_file(): + raise ValueError("ablation registration is missing") + saved = json.loads(path.read_text(encoding="utf-8")) + if saved != {"manifest": asdict(manifest), "manifest_id": manifest_id(manifest)}: + raise ValueError("ablation registration does not match the manifest") + + +def register(manifest: Manifest, out_dir: Path | str) -> Path: + """Write ablation.json before any run; refuse to replace a different registration or adopt old runs.""" + _validate(manifest) + out = Path(out_dir) + path = out / "ablation.json" + if path.exists(): + _registration(manifest, out) + else: + if out.exists() and any(out.iterdir()): + raise ValueError("registration requires an empty output directory") + _write(path, {"manifest": asdict(manifest), "manifest_id": manifest_id(manifest)}) + return path + + +def _population(manifest: Manifest, changes: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + indexed = {change["id"]: dict(change) for change in changes} + if len(indexed) != len(changes) or set(indexed) != set(manifest.sample["changes"]): + raise ValueError("changes must match the registered sample exactly") + selected = [indexed[item] for item in manifest.sample["changes"]] + repo_folds: dict[str, int] = {} + for change in selected: + fold = change.get("fold") + if type(fold) is not int or fold not in manifest.compile_folds + manifest.holdout_folds: + raise ValueError("each change needs a registered fold") + if change["repo"] in repo_folds and repo_folds[change["repo"]] != fold: + raise ValueError("a repository cannot cross folds") + repo_folds[change["repo"]] = fold + return selected + + +def run( + manifest: Manifest, + pool_manifest_path: Path | str, + repos: Mapping[str, str], + out_dir: Path | str, + *, + program_factory: Any, + lm: Any, + repeats: int, + table: FeatureTable, + load_changes: Any = None, + slice_id: str = "1", + tools_for: Any = None, +) -> dict[str, Any]: + """Run each arm and repeat on the same registered changes using the existing runner. + + The factory is called with mask= for each arm and repeat. LM is the common DSPy model, or a + mapping of arm names to identically configured scripted models for deterministic experiments. + Repeats are numbered from zero. Table supplies the registered digest and change folds; labels + and verification stay with the runner's scoring data and never become mask names or model inputs. + The runner owns budget accounting and resumability; its budget applies per arm and repeat. + """ + import dspy + + from benchmarks.unseen.replay import load_slice + + out = Path(out_dir) + _registration(manifest, out) + _validate(manifest) + if repeats != manifest.sample["repeats"]: + raise ValueError("repeats do not match registration") + if table.digest() != manifest.table_digest: + raise ValueError("table digest does not match registration") + runner.refuse_pilot(Path(pool_manifest_path), pilot_report=False) + folds: dict[str, tuple[str, int]] = {} + for row in table.rows: + key, value = row["change_id"], (row["repo"], row.get("fold")) + if key in folds and folds[key] != value: + raise ValueError("rows of a change must share repository and fold") + folds[key] = value + loaded = (load_changes or load_slice)(Path(pool_manifest_path), slice_id) + indexed = {change["id"]: change for change in loaded} + selected = [] + for item in manifest.sample["changes"]: + if item not in indexed or item not in folds or indexed[item]["repo"] != folds[item][0]: + raise ValueError("registered changes must exist in the pool and feature table") + selected.append({**indexed[item], "fold": folds[item][1]}) + selected = _population(manifest, selected) + # Freeze revisions and folds too: resuming a changed pool must not silently replace paired records. + study = { + "manifest_id": manifest_id(manifest), + "pool_digest": hashlib.sha256(Path(pool_manifest_path).read_bytes()).hexdigest(), + "changes": selected, + } + study_path = out / "study.json" + if study_path.exists() and json.loads(study_path.read_text()) != study: + raise ValueError("pool or changes differ from the registered run") + _write(study_path, study) + _write(out / "leak-audit.json", json.loads(leak_audit(table).to_json(orient="records"))) + if isinstance(lm, Mapping) and set(lm) != set(manifest.arms): + raise ValueError("scripted models must cover exactly the registered arms") + for arm, mask in manifest.arms.items(): + for repeat in range(repeats): + program = program_factory(mask=mask) + with dspy.context(lm=lm[arm] if isinstance(lm, Mapping) else lm, adapter=dspy.ChatAdapter()): + runner.run( + Path(pool_manifest_path), + repos, + out / f"{arm}-{repeat}", + program=program, + program_id=manifest.program_id, + mask=mask, + budget=manifest.budget, + slice_id=slice_id, + tools_for=tools_for, + load_changes=lambda _path, _slice: selected, + versions={"dspy": dspy.__version__}, + ) + return analyse(out, manifest, selected) + + +def _findings(defects: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + return [{**defect, "path": path, "line": line} for defect in defects for path, line in defect.get("locations", [])] + + +def _paired(units: list[dict[str, Any]], a: str, b: str, manifest: Manifest, family: str | None) -> dict[str, Any]: + import pandas as pd + + positives = [u for u in units if u["change"]["kind"] == "vulnerability" and (family is None or u["change"].get("family") == family)] + ordinary = [u for u in units if u["change"]["kind"] == "ordinary"] + + def estimate(part: list[dict[str, Any]], metric: str) -> dict[str, Any]: + def value(unit: dict[str, Any], arm: str) -> int: + if metric == "hit": + return int(unit[arm]["hit"]) + return sum(family is None or d.get("family") == family for d in unit[arm]["defects"]) + + frame = pd.DataFrame([{"repo": u["change"]["repo"], "a": value(u, a), "b": value(u, b)} for u in part], columns=["repo", "a", "b"]) + return asdict( + stats.paired_bootstrap( + frame, "repo", "a", "b", seed=manifest.seed, resamples=manifest.rule.get("resamples", stats.RESAMPLES), alpha=0.05 + ) + ) + + return { + "recall": estimate(positives, "hit"), + "suspicions_per_push": estimate(ordinary, "suspicions"), + "mcnemar": asdict(stats.mcnemar_exact([u[a]["hit"] for u in positives], [u[b]["hit"] for u in positives])), + } + + +def _arm_summary(units: list[dict[str, Any]], arm: str, tolerance: int) -> dict[str, Any]: + positives = [u for u in units if u["change"]["kind"] == "vulnerability"] + ordinary = [u for u in units if u["change"]["kind"] == "ordinary"] + cuts = sorted( + { + d["confidence"] + for u in units + for d in u[arm]["defects"] + if isinstance(d.get("confidence"), int | float) and math.isfinite(d["confidence"]) + }, + reverse=True, + ) + curve = [] + for cut in cuts: + + def above(unit: dict[str, Any], cut: float = cut) -> list[dict[str, Any]]: + return [ + d + for d in unit[arm]["defects"] + if isinstance(d.get("confidence"), int | float) and math.isfinite(d["confidence"]) and d["confidence"] >= cut + ] + + hits = sum(search_outcome(u["change"], _findings(above(u)), tolerance=tolerance).hit for u in positives) + curve.append( + { + "confidence": cut, + "recall": hits / len(positives) if positives else None, + "suspicions_per_push": sum(len(above(u)) for u in ordinary) / len(ordinary) if ordinary else None, + "positive_units": len(positives), + "ordinary_units": len(ordinary), + } + ) + return { + "positive_units": len(positives), + "ordinary_units": len(ordinary), + "recall": sum(u[arm]["hit"] for u in positives) / len(positives) if positives else None, + "suspicions_per_push": sum(len(u[arm]["defects"]) for u in ordinary) / len(ordinary) if ordinary else None, + **{key: sum(u[arm][key] for u in units) for key in ("instrument_failures", "parse_failures", "missing_records")}, + "unranked_findings": sum( + not isinstance(d.get("confidence"), int | float) or not math.isfinite(d["confidence"]) for u in units for d in u[arm]["defects"] + ), + "curve": curve, + } + + +def analyse(out_dir: Path | str, manifest: Manifest, changes: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + """Report arm-minus-full effects, additions versus baseline, and every reported confidence cut. + + Results contain all, compile and heldout sections, each with pooled and per-family comparisons. + Ordinary denominators include all ordinary pushes; family reports count only that family's findings. + Missing, instrument-failed and answer-parse-failed records are misses. An incomplete tool step with + a valid final answer is still scored. Changes must carry folds, or obtain them from the saved study. + Analysis, importance and decisions are written beside the registration for the next round. + """ + out = Path(out_dir) + _registration(manifest, out) + tolerance = _validate(manifest) + study_path = out / "study.json" + if study_path.exists(): + study = json.loads(study_path.read_text()) + saved = {c["id"]: c for c in study["changes"]} + changes = [{**c, "fold": c.get("fold", saved.get(c["id"], {}).get("fold"))} for c in changes] + if study["manifest_id"] != manifest_id(manifest) or {c["id"]: c for c in changes} != saved: + raise ValueError("analysis changes do not match the saved study") + changes = _population(manifest, changes) + units = [] + for change in changes: + for repeat in range(manifest.sample["repeats"]): + unit: dict[str, Any] = {"change": change, "repeat": repeat} + for arm, mask in manifest.arms.items(): + path = runner.record_path(out / f"{arm}-{repeat}", change["id"]) + record = json.loads(path.read_text()) if path.exists() else {} + if record and ( + record.get("id") != change["id"] + or record.get("program_id") != manifest.program_id + or set(record.get("mask", [])) != set(mask) + or record.get("input_digest") != runner.input_digest(change, manifest.program_id, mask) + ): + raise ValueError("record does not match the registered change, program or mask") + failed = record.get("status") == "instrument_failure" + parsed = bool(record.get("parse_failed")) + defects = record.get("push", {}).get("admission", {}).get("defects", []) + if not record.get("done") or failed or parsed: + defects = [] + unit[arm] = { + "defects": defects, + "hit": search_outcome(change, _findings(defects), tolerance=tolerance).hit, + "instrument_failures": int(failed), + "parse_failures": int(parsed), + "missing_records": int(not record), + } + units.append(unit) + families = sorted( + {c["family"] for c in changes if c.get("family")} + | {d["family"] for u in units for arm in manifest.arms for d in u[arm]["defects"] if d.get("family")} + ) + result: dict[str, Any] = {"manifest_id": manifest_id(manifest), "out_dir": str(out.resolve()), "unit": "change, repeat"} + for scope, folds in ( + ("all", manifest.compile_folds + manifest.holdout_folds), + ("compile", manifest.compile_folds), + ("heldout", manifest.holdout_folds), + ): + part = [u for u in units if u["change"]["fold"] in folds] + + def comparison(a: str, b: str, part: list[dict[str, Any]] = part) -> dict[str, Any]: + return { + "pooled": _paired(part, a, b, manifest, None), + "families": {family: _paired(part, a, b, manifest, family) for family in families}, + } + + result[scope] = { + "units": len(part), + "arms": {arm: _arm_summary(part, arm, tolerance) for arm in manifest.arms}, + "comparisons": {arm: comparison("full", arm) for arm in manifest.arms if arm != "full"}, + "additions": { + arm: comparison("baseline", arm) for arm in manifest.arms if arm.startswith("with-") and "baseline" in manifest.arms + }, + } + from .program import INPUTS + + result["annotations_removed"] = { + arm: sorted( + annotations_of(mask) | {name for name, column, _ in INPUTS if name in mask or (column and column.split(".")[1] in mask)} + ) + for arm, mask in manifest.arms.items() + } + result["dry_run"] = sum(c["kind"] == "vulnerability" for c in changes) < 40 + _write(out / "analysis.json", result) + table = importance(result, annotations=[name for name, _, _ in INPUTS]) + _write(out / "importance.json", json.loads(table.to_json(orient="records"))) + decide(table, needs={name: set(deps) for name, deps in manifest.rule.get("needs", {}).items()}) + return result + + +def importance(analysis: Mapping[str, Any], *, annotations: Sequence[str]) -> Any: + """One row per annotation, with held-out removal and addition effects, intervals and unit counts. + + Removal compares without-annotation to full; addition compares with-annotation to baseline. + Missing arms produce null cells. The output directory is carried as dataframe metadata so decide + can persist the next round's decisions without requiring another output argument. + """ + import pandas as pd + + heldout = analysis.get("heldout", {}) + rows = [] + for annotation in annotations: + row: dict[str, Any] = {"annotation": annotation} + for prefix, key, arm in (("remove", "comparisons", f"without-{annotation}"), ("add", "additions", f"with-{annotation}")): + pooled = heldout.get(key, {}).get(arm, {}).get("pooled", {}) + for metric in ("recall", "suspicions_per_push"): + effect = pooled.get(metric, {}) + column = f"{prefix}_{metric}" + row[column] = effect.get("diff") + row[column + "_ci"] = effect.get("ci") + row[column + "_units"] = effect.get("units") + row[column + "_groups"] = effect.get("groups") + rows.append(row) + table = pd.DataFrame(rows) + table.attrs.update(scope="heldout", out_dir=analysis.get("out_dir")) + return table + + +def decide(importance_table: Any, *, needs: Mapping[str, set[str]]) -> dict[str, dict[str, str]]: + """Keep held-out benefits and their neutral dependencies; drop unsupported or noisy annotations. + + Removal must lower recall or raise suspicions with an interval excluding zero to show benefit. + Removal that lowers suspicions while recall includes zero shows noise. Neutral dependencies are + kept transitively from a kept annotation; dependency cycles cannot keep themselves alive. + Missing evidence never establishes benefit. Decisions are saved when the table carries out_dir. + """ + if importance_table.attrs.get("scope") != "heldout": + raise ValueError("decisions require held-out importance effects") + decisions: dict[str, dict[str, str]] = {} + neutral = set() + + def interval(value: Any) -> tuple[float, float] | None: + if isinstance(value, (tuple, list)) and len(value) == 2 and all(isinstance(v, int | float) and math.isfinite(v) for v in value): + return tuple(value) + return None + + for row in importance_table.to_dict(orient="records"): + name = row["annotation"] + recall, suspicions = interval(row["remove_recall_ci"]), interval(row["remove_suspicions_per_push_ci"]) + if recall and recall[1] < 0: + decisions[name] = {"decision": "keep", "reason": "removal lowers recall"} + elif suspicions and suspicions[0] > 0: + decisions[name] = {"decision": "keep", "reason": "removal raises suspicions"} + elif suspicions and suspicions[1] < 0 and recall and recall[0] <= 0 <= recall[1]: + decisions[name] = {"decision": "drop", "reason": "raises suspicions without recall"} + else: + decisions[name] = { + "decision": "drop", + "reason": "no measurable effect" if recall and suspicions else "missing held-out effects", + } + if recall and suspicions and recall[0] <= 0 <= recall[1] and suspicions[0] <= 0 <= suspicions[1]: + neutral.add(name) + changed = True + while changed: + changed = False + for name, decision in decisions.items(): + if decision["decision"] != "keep": + continue + for dependency in sorted(needs.get(name, set()) & neutral): + if decisions[dependency]["decision"] != "keep": + decisions[dependency] = {"decision": "keep", "reason": f"needed by {name}"} + changed = True + if importance_table.attrs.get("out_dir"): + _write(Path(importance_table.attrs["out_dir"]) / "decisions.json", decisions) + return decisions diff --git a/tests/test_hunt_ablation.py b/tests/test_hunt_ablation.py new file mode 100644 index 0000000..5de1af6 --- /dev/null +++ b/tests/test_hunt_ablation.py @@ -0,0 +1,335 @@ +"""A registered ablation on readable git changes, with deterministic model answers.""" + +from __future__ import annotations + +import json +import subprocess +from dataclasses import replace + +import pytest + +dspy = pytest.importorskip("dspy") +pd = pytest.importorskip("pandas") +pytest.importorskip("statsmodels") +pytest.importorskip("sklearn") +from dspy.utils import DummyLM # noqa: E402 + +from openultrasast.hunt import ablation as ab # noqa: E402 +from openultrasast.hunt import program as hp # noqa: E402 +from openultrasast.hunt import runner # noqa: E402 +from openultrasast.hunt.features import ENGINEERED # noqa: E402 +from openultrasast.hunt.table import FeatureTable # noqa: E402 + + +@pytest.fixture +def population(tmp_path): + repos, changes, rows = {}, [], [] + for number in range(3): + root = tmp_path / f"repo-{number}" + root.mkdir() + + def git(*args, root=root): + return subprocess.run(["git", "-C", str(root), *args], check=True, capture_output=True, text=True).stdout.strip() + + git("init", "-q") + git("config", "user.name", "Fixture") + git("config", "user.email", "fixture@example.invalid") + (root / "app.py").write_text("def run():\n return input()\n") + git("add", "app.py") + git("commit", "-qm", "initial") + base = git("rev-parse", "HEAD") + (root / "app.py").write_text("def run():\n return eval(input())\n") + git("add", "app.py") + git("commit", "-qm", "change") + head = git("rev-parse", "HEAD") + repo = f"fixture/repo-{number}" + repos[repo] = str(root) + for item in range(2): + positive = number < 2 and item == 0 + change = { + "id": f"c{number}-{item}", + "repo": repo, + "base": base, + "head": head, + "kind": "vulnerability" if positive else "ordinary", + "case_id": f"case-{number}-{item}", + "path": "app.py", + "function": "run", + "function_lines": [1, 2], + "family": "injection", + "files": [{"path": "app.py", "head_lines": [[2, 2]]}], + } + changes.append(change) + for line in (1, 2): + rows.append( + { + "change_id": change["id"], + "repo": repo, + "head": head, + "path": "app.py", + "line": line, + "kind": change["kind"], + "label": int(positive), + "feature.diff.distance": number, + "provenance.diff.ok": True, + "provenance.diff.reason": None, + "provenance.diff.version": 1, + "provenance.diff.bytes_read": 36, + } + ) + table = ab.assign_folds(FeatureTable(rows), k=3, seed=13) + folds = {row["change_id"]: row["fold"] for row in table.rows} + changes = [{**change, "fold": folds[change["id"]]} for change in changes] + pool = tmp_path / "pool-fixture.toml" + pool.write_text("status = 'fixture'\n") + manifest = ab.Manifest( + table_digest=table.digest(), + arms={"full": [], "without-roles": ["roles"]}, + compile_folds=[], + holdout_folds=[0, 1, 2], + metric="hunt.metric.search_outcome(tolerance=10)", + sample={"changes": [c["id"] for c in changes], "repeats": 2}, + budget={"usd": 1.0, "calls": 100}, + rule={"version": "keep-drop-v1", "resamples": 100, "needs": {}}, + seed=13, + program_id="fixture-program", + ) + return pool, repos, changes, table, manifest + + +def test_folds_and_audit_keep_every_row_and_column(population): + _, _, _, table, _ = population + frame = table.to_pandas() + assert frame.groupby("repo")["fold"].nunique().eq(1).all() + assert frame.groupby("change_id")["fold"].nunique().eq(1).all() + assert set(frame.fold) == {0, 1, 2} + assert ab.assign_folds(table, k=3, seed=13).digest() == table.digest() + audit = ab.leak_audit(table) + assert set(audit.column) == set(ENGINEERED) + assert audit.set_index("column").loc["diff.distance", "n"] == 12 + missing = audit.set_index("column").loc["semantic.sources"] + assert pd.isna(missing.auc) and missing.n == 0 and missing.note + assert dict(zip(audit.column, audit.annotation, strict=True)) == {key: value[1] for key, value in ENGINEERED.items()} + assert len(ab.leak_audit(FeatureTable([]))) == len(ENGINEERED) + + +def finding(confidence=0.9, family="injection"): + return {"path": "app.py", "line": 2, "family": family, "confidence": confidence, "rationale": "eval", "cited_lines": [2]} + + +def model(answers): + return DummyLM([{"findings": json.dumps(answer)} for answer in answers]) + + +def test_registration_paired_runs_statistics_and_curve(population, tmp_path): + pool, repos, changes, table, manifest = population + out = tmp_path / "ablation" + # Each arm has its own deterministic sequence, but uses the same model class and sampling. + models = {"full": model([[finding()]] * 12), "without-roles": model([[]] * 12)} + kwargs = dict(program_factory=hp.HuntProgram, lm=models, repeats=2, table=table, load_changes=lambda m, s: changes) + with pytest.raises(ValueError, match="registration"): + ab.run(manifest, pool, repos, out, **kwargs) + ab.register(manifest, out) + registration = json.loads((out / "ablation.json").read_text()) + assert registration["manifest_id"] == ab.manifest_id(manifest) + with pytest.raises(ValueError, match="registration"): + ab.run(replace(manifest, seed=22), pool, repos, out, **kwargs) + ab.run(manifest, pool, repos, out, **kwargs) + for arm in manifest.arms: + for repeat in range(2): + records = [json.loads(runner.record_path(out / f"{arm}-{repeat}", c["id"]).read_text()) for c in changes] + assert {r["id"] for r in records} == {c["id"] for c in changes} + assert all(r["instrument"]["hook_bytes_read"] > 0 and r["instrument"]["head_verified"] for r in records) + assert all(("roles" in r["inputs_present"]) == (arm == "full") for r in records) + analysis = ab.analyse(out, manifest, changes) + pooled = analysis["heldout"]["comparisons"]["without-roles"]["pooled"] + assert pooled["recall"]["diff"] == -1 and pooled["recall"]["ci"] == (-1, -1) + assert pooled["recall"]["units"] == 4 and pooled["recall"]["groups"] == 2 + assert pooled["suspicions_per_push"]["diff"] == -1 and pooled["suspicions_per_push"]["units"] == 8 + assert pooled["mcnemar"]["b"] == 4 and pooled["mcnemar"]["c"] == 0 + assert "injection" in analysis["heldout"]["comparisons"]["without-roles"]["families"] + assert analysis["heldout"]["arms"]["full"]["curve"] == [ + {"confidence": 0.9, "recall": 1.0, "suspicions_per_push": 1.0, "positive_units": 4, "ordinary_units": 8} + ] + assert json.loads((out / "decisions.json").read_text())["roles"]["decision"] == "keep" + # The saved registration also protects against changed folds and repeat counts. + with pytest.raises(ValueError, match="repeats"): + ab.run(manifest, pool, repos, out, **{**kwargs, "repeats": 1}) + with pytest.raises(ValueError, match="digest"): + ab.run(manifest, pool, repos, out, **{**kwargs, "table": FeatureTable(table.rows[:-1])}) + + +def test_parse_and_instrument_failures_stay_in_denominator(population, tmp_path): + pool, repos, changes, table, original = population + manifest = replace(original, sample={**original.sample, "repeats": 1}) + out = tmp_path / "failed" + ab.register(manifest, out) + # ChatAdapter retries a malformed answer through JSONAdapter once. + bad = DummyLM([{"findings": "not json"}] * 2 + [{"findings": "[]"}] * 5) + ab.run( + manifest, + pool, + repos, + out, + program_factory=hp.HuntProgram, + lm={"full": bad, "without-roles": model([[]] * 6)}, + repeats=1, + table=table, + load_changes=lambda m, s: changes, + ) + record = json.loads(runner.record_path(out / "full-0", changes[0]["id"]).read_text()) + assert record["parse_failed"] and record["instrument"]["hook_bytes_read"] > 0 + # An absent record is also a miss, not a reason to shrink the paired population. + runner.record_path(out / "without-roles-0", changes[2]["id"]).unlink() + analysis = ab.analyse(out, manifest, changes) + full = analysis["heldout"]["arms"]["full"] + assert full["parse_failures"] == 1 and full["positive_units"] == 2 and full["recall"] == 0 + assert analysis["heldout"]["arms"]["without-roles"]["missing_records"] == 1 + assert analysis["heldout"]["comparisons"]["without-roles"]["pooled"]["recall"]["units"] == 2 + + +def effect(diff, ci): + return {"diff": diff, "ci": ci, "units": 12, "groups": 3} + + +def test_importance_and_decisions_use_heldout_effects_and_dependencies(tmp_path): + neutral = {"recall": effect(0, (-0.1, 0.1)), "suspicions_per_push": effect(0, (-0.1, 0.1))} + helpful = {**neutral, "recall": effect(-0.5, (-0.8, -0.2))} + noisy = {**neutral, "suspicions_per_push": effect(-1, (-2, -0.5))} + comparisons = { + f"without-{name}": {"pooled": value} + for name, value in (("roles", helpful), ("callers", noisy), ("def_use", neutral), ("obligations", neutral)) + } + analysis = { + "out_dir": str(tmp_path), + "heldout": {"comparisons": comparisons, "additions": {"with-roles": {"pooled": {**neutral, "recall": effect(0.5, (0.2, 0.8))}}}}, + "compile": {"comparisons": {"without-callers": {"pooled": helpful}}}, + } + table = ab.importance(analysis, annotations=["roles", "callers", "def_use", "obligations", "families"]) + assert len(table) == 5 + roles = table.set_index("annotation").loc["roles"] + assert roles.remove_recall == -0.5 and roles.add_recall == 0.5 and roles.remove_recall_units == 12 + assert pd.isna(table.set_index("annotation").loc["families", "remove_recall"]) + decisions = ab.decide(table, needs={"roles": {"def_use"}, "def_use": {"obligations"}}) + assert decisions["roles"]["decision"] == "keep" + assert decisions["callers"] == {"decision": "drop", "reason": "raises suspicions without recall"} + assert decisions["def_use"]["decision"] == decisions["obligations"]["decision"] == "keep" + assert ab.decide(table, needs={})["def_use"] == {"decision": "drop", "reason": "no measurable effect"} + assert json.loads((tmp_path / "decisions.json").read_text())["def_use"]["decision"] == "drop" + + +@pytest.mark.parametrize("mask", [["label"], ["verification"], ["advisory"], ["provenance.source.ok"]]) +def test_registration_rejects_non_feature_masks(population, tmp_path, mask): + manifest = replace(population[-1], arms={"full": [], "bad": mask}) + with pytest.raises(ValueError, match="mask"): + ab.register(manifest, tmp_path / "bad") + + +def write_records(out, manifest, changes, answers): + """Save runner-shaped records to isolate analysis of fixed, known outcomes.""" + ab.register(manifest, out) + for arm, per_change in answers.items(): + directory = out / f"{arm}-0" + directory.mkdir() + for change, findings in zip(changes, per_change, strict=True): + record = { + "id": change["id"], + "done": True, + "status": "hunt_covered", + "program_id": manifest.program_id, + "mask": manifest.arms[arm], + "input_digest": runner.input_digest(change, manifest.program_id, manifest.arms[arm]), + "push": { + "admission": { + "defects": [runner.defect_from(f, program_id=manifest.program_id, mask=tuple(manifest.arms[arm])) for f in findings] + } + }, + } + runner.record_path(directory, change["id"]).write_text(json.dumps(record)) + + +def test_curve_counts_ties_all_suspicions_and_families(population, tmp_path): + _, _, changes, _, original = population + manifest = replace(original, sample={**original.sample, "repeats": 1}) + out = tmp_path / "curve" + full = [[finding(0.9)], [finding(0.9), finding(0.4, "path")], [finding(0.4)], [], [], []] + write_records(out, manifest, changes, {"full": full, "without-roles": [[]] * 6}) + result = ab.analyse(out, manifest, changes) + arm = result["all"]["arms"]["full"] + assert arm["curve"] == [ + {"confidence": 0.9, "recall": 0.5, "suspicions_per_push": 0.25, "positive_units": 2, "ordinary_units": 4}, + {"confidence": 0.4, "recall": 1.0, "suspicions_per_push": 0.5, "positive_units": 2, "ordinary_units": 4}, + ] + families = result["all"]["comparisons"]["without-roles"]["families"] + assert families["injection"]["recall"]["diff"] == -1 + assert families["path"]["recall"]["units"] == 0 and families["path"]["recall"]["ci"] is None + for family in ("injection", "path"): + assert families[family]["suspicions_per_push"]["diff"] == -0.25 + assert families[family]["suspicions_per_push"]["units"] == 4 + # A flagged failure with stale findings must be a miss in both the summary and the curve. + path = runner.record_path(out / "full-0", changes[0]["id"]) + record = json.loads(path.read_text()) + record["status"] = "instrument_failure" + path.write_text(json.dumps(record)) + failed = ab.analyse(out, manifest, changes)["all"]["arms"]["full"] + assert failed["instrument_failures"] == 1 and failed["positive_units"] == 2 and failed["recall"] == 0.5 + assert failed["curve"][0]["recall"] == 0 + + +def test_holdout_and_baseline_addition_are_separate(population, tmp_path): + _, _, changes, _, original = population + compile_fold = changes[0]["fold"] + manifest = replace( + original, + sample={**original.sample, "repeats": 1}, + compile_folds=[compile_fold], + holdout_folds=[f for f in original.holdout_folds if f != compile_fold], + arms={"full": [], "without-roles": ["roles"], "baseline": ["roles", "callers"], "with-roles": ["callers"]}, + ) + out = tmp_path / "split" + write_records( + out, + manifest, + changes, + { + "full": [[], [], [finding()], [], [], []], + "without-roles": [[finding()], [], [], [], [], []], + "baseline": [[]] * 6, + "with-roles": [[], [], [finding()], [], [], []], + }, + ) + result = ab.analyse(out, manifest, changes) + assert result["compile"]["comparisons"]["without-roles"]["pooled"]["recall"]["diff"] == 1 + assert result["heldout"]["comparisons"]["without-roles"]["pooled"]["recall"]["diff"] == -1 + importance = ab.importance(result, annotations=["roles"]) + assert importance.iloc[0].add_recall == 1 and importance.iloc[0].remove_recall == -1 + assert ab.decide(importance, needs={})["roles"]["decision"] == "keep" + assert result["annotations_removed"]["without-roles"] == ["roles"] + + +def test_decisions_keep_suspicion_benefits_but_not_neutral_cycles(): + neutral = {"recall": effect(0, (0, 0)), "suspicions_per_push": effect(0, (0, 0))} + effects = {"roles": {**neutral, "suspicions_per_push": effect(1, (0.5, 2))}, "callers": neutral, "def_use": neutral} + table = ab.importance( + {"heldout": {"comparisons": {f"without-{name}": {"pooled": value} for name, value in effects.items()}}}, annotations=list(effects) + ) + decisions = ab.decide(table, needs={"callers": {"def_use"}, "def_use": {"callers"}}) + assert decisions["roles"] == {"decision": "keep", "reason": "removal raises suspicions"} + assert decisions["callers"]["decision"] == decisions["def_use"]["decision"] == "drop" + table.attrs["scope"] = "compile" + with pytest.raises(ValueError, match="held-out"): + ab.decide(table, needs={}) + + +def test_registration_cannot_be_rewritten_or_have_its_contents_changed(population, tmp_path): + manifest = population[-1] + out = tmp_path / "registered" + ab.register(manifest, out) + assert ab.manifest_id(replace(manifest, arms=dict(reversed(list(manifest.arms.items()))))) == ab.manifest_id(manifest) + with pytest.raises(ValueError, match="registration"): + ab.register(replace(manifest, seed=99), out) + payload = json.loads((out / "ablation.json").read_text()) + payload["manifest"]["seed"] = 99 + (out / "ablation.json").write_text(json.dumps(payload)) + with pytest.raises(ValueError, match="registration"): + ab.analyse(out, manifest, population[2])