diff --git a/.kiro/specs/hunter-learning-loop/tasks.md b/.kiro/specs/hunter-learning-loop/tasks.md index 3436889..39f57ac 100644 --- a/.kiro/specs/hunter-learning-loop/tasks.md +++ b/.kiro/specs/hunter-learning-loop/tasks.md @@ -88,7 +88,7 @@ - _Requirements: 5.6, 6.1, 6.2, 6.3, 6.4, 6.5, 6.7_ - [ ] 6. Validation: the pilot gain -- [ ] 6.1 Run an accepted program once over the held-out pilot and report the gain +- [x] 6.1 Run an accepted program once over the held-out pilot and report the gain - Same scorer and metrics as the pilot; before-and-after of search recall, suspicions per push, push-level recall, false alarms per 100 and precision, per family, with intervals; the proof arm's workload per 100 pushes; the gate distance read off the curve; the run recorded next to the pilot's measurement, at most once per accepted program - Done when the measurement record exists with the before-and-after table and the instrument proof for every change - _Requirements: 7.3, 7.4, 7.5, 8.1, 8.2, 8.3, 8.6_ diff --git a/src/openultrasast/hunt/report.py b/src/openultrasast/hunt/report.py new file mode 100644 index 0000000..d70a50a --- /dev/null +++ b/src/openultrasast/hunt/report.py @@ -0,0 +1,472 @@ +"""Report one accepted program on the held-out pilot, with instrument proof before scores.""" + +from __future__ import annotations + +import fcntl +import json +import math +import os +from collections.abc import Mapping, Sequence +from copy import deepcopy +from dataclasses import asdict +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from .. import stats +from . import accept, contamination, runner +from .ablation import _arm_summary, _findings, _write +from .metric import search_outcome +from .propose import Candidate +from .rounds import _candidate_program, _estimate, _families, _outcome, _paired + + +def _reserve(root: Path, program_id: str) -> Path: + """Reserve a program under a file lock and flush the ledger before starting its run.""" + date = datetime.now(UTC).date().isoformat() + if not program_id or any(c not in "0123456789abcdef" for c in program_id): + raise ValueError("program ID must be a hexadecimal content address") + root.mkdir(parents=True, exist_ok=True) + directory = root / f"{date}-hunt-gain-{program_id[:12]}" + with (root / "hunt-pilot-runs.jsonl").open("a+", encoding="utf-8") as stream: + fcntl.flock(stream, fcntl.LOCK_EX) + stream.seek(0) + if any(json.loads(line)["program_id"] == program_id for line in stream if line.strip()): + raise ValueError("this program ID already has a pilot run") + directory.mkdir(exist_ok=False) + stream.write(json.dumps(dict(program_id=program_id, date=date, measurement_directory=str(directory.resolve()))) + "\n") + stream.flush() + os.fsync(stream.fileno()) + return directory + + +def _records(directory: Path, changes: Sequence[Mapping]) -> list[dict]: + """Load benchmark records and retain every expected change, even after a budget stop.""" + from benchmarks.unseen import score + + records = score.load_records(Path(directory)) if (Path(directory) / "inputs.json").exists() else [] + indexed = {r["id"]: r for r in records} + if len(indexed) != len(records) or set(indexed) - {c["id"] for c in changes}: + raise ValueError("records do not match the pilot population") + result = [] + for change in changes: + record = indexed.get(change["id"], dict(change)) + if any(record.get(key) != change.get(key) for key in ("repo", "base", "head", "kind", "family", "files", "targets")): + raise ValueError("record describes a different pilot change") + result.append(record) + return result + + +def _instrument(records: Sequence[Mapping]) -> dict: + """Record changed-file bytes and verified revisions; unreadable and absent changes fail.""" + from benchmarks.unseen import score + + proof = [] + for record in records: + instrument = record.get("instrument", {}) + sizes = instrument.get("changed_bytes", {}) + failed = not record.get("done") or record.get("status") == "instrument_failure" or score.coverage(record) == "instrument_failure" + proof.append( + dict( + id=record["id"], + repo=record["repo"], + base=record["base"], + head=record["head"], + base_verified=bool(instrument.get("base_verified")), + head_verified=bool(instrument.get("head_verified")), + changed_bytes={f["path"]: sizes.get(f["path"], 0) for f in record.get("files", [])}, + instrument=dict(instrument), + instrument_failure=failed, + reason=instrument.get("failure") or ("missing or unreadable record" if failed else None), + ) + ) + failures = sum(row["instrument_failure"] for row in proof) + return dict( + changes=len(proof), + failures=failures, + failure_fraction=failures / len(proof) if proof else None, + valid=bool(proof) and failures / len(proof) <= 0.05, + records=proof, + ) + + +def _clean(records: Sequence[Mapping]) -> list[dict]: + """Remove stale findings on failures while preserving the scorer's coverage checks.""" + from benchmarks.unseen import score + + result = [] + for original in records: + record = deepcopy(original) + if not record.get("done") or record.get("status") == "instrument_failure" or score.coverage(record) == "instrument_failure": + record["status"] = "instrument_failure" + record.setdefault("instrument", {})["failure"] = "instrument_failure" + outcome = _outcome(record, record, 10) + if not record.get("done") or record.get("status") == "instrument_failure" or record.get("parse_failed"): + record["push"] = {"admission": {"defects": []}, "quick_tier": []} + else: + record.setdefault("push", {}).setdefault("admission", {})["defects"] = outcome["defects"] + result.append(record) + return result + + +def _rate(values: list[float], repos: list[str], *, binary: bool = False, scale: float = 1) -> dict: + """Report a mean with its count and a repository-clustered interval, plus Wilson for proportions.""" + from benchmarks.unseen import score + + interval = score.cluster_interval(list(zip(repos, values, strict=True)), seed=0) + result = dict( + rate=scale * sum(values) / len(values) if values else None, + count=sum(values), + n=len(values), + groups=len(set(repos)), + ci=[scale * v for v in interval] if values else None, + ) + if binary: + wilson = stats.wilson(int(sum(values)), len(values)) + result["wilson95"] = [scale * v for v in wilson] if wilson else None + return result + + +def _family_records(records: Sequence[Mapping], family: str | None) -> list[dict]: + """Keep a family's positives and count only its findings over all ordinary pushes.""" + result = [] + for original in records: + if family is not None and original["kind"] == "vulnerability" and original.get("family") != family: + continue + record = deepcopy(original) + admission = record.setdefault("push", {}).setdefault("admission", {}) + admission["defects"] = [d for d in admission.get("defects", []) if family is None or d.get("family") == family] + result.append(record) + return result + + +def _summaries(records: list[dict], families: list[str]) -> dict: + """Use the pilot functions for push metrics and the hunt metric for search outcomes.""" + from benchmarks.unseen import score + from benchmarks.valid import metrics + + summaries = {} + for family in ["pooled", *families]: + part = _family_records(records, None if family == "pooled" else family) + positives = [r for r in part if r["kind"] == "vulnerability"] + ordinary = [r for r in part if r["kind"] == "ordinary"] + outcomes = {r["id"]: search_outcome(r, _findings(r.get("push", {}).get("admission", {}).get("defects", []))) for r in part} + search = dict( + recall=_rate([int(outcomes[r["id"]].hit) for r in positives], [r["repo"] for r in positives], binary=True), + suspicions_per_push=_rate([outcomes[r["id"]].suspicions for r in ordinary], [r["repo"] for r in ordinary]), + ) + # The whole run already passed its instrument gate. The pilot scorer excludes these failures + # from push denominators, while the search summaries above keep them as misses. + usable = [r for r in part if score.coverage(r) != "instrument_failure"] + if usable: + scored = score.score(metrics.scoring_records(usable)) + push = metrics.summarize(usable, scored=scored) + else: + push = {key: score.rate([]) for key in ("recall", "precision", "false_alarms_per_100_pushes")} + summaries[family] = dict(search=search, push=push) + return summaries + + +def _curve(records: list[dict]) -> tuple[list[dict], dict]: + """Extend ablation confidence cuts with false alarms, intervals and the two gate bounds.""" + from benchmarks.unseen import score + + units = [dict(change=r, after=_outcome(r, r, 10)) for r in records] + curve = _arm_summary(units, "after", 10)["curve"] + for point in curve: + cut = point["confidence"] + filtered = deepcopy(records) + for record in filtered: + defects = record.get("push", {}).get("admission", {}).get("defects", []) + record.setdefault("push", {}).setdefault("admission", {})["defects"] = [ + d + for d in defects + if isinstance(d.get("confidence"), int | float) and math.isfinite(d["confidence"]) and d["confidence"] >= cut + ] + positives = [r for r in filtered if r["kind"] == "vulnerability"] + ordinary = [r for r in filtered if r["kind"] == "ordinary"] + hits = [int(search_outcome(r, _findings(r["push"]["admission"]["defects"])).hit) for r in positives] + point["estimates"] = dict( + recall=_rate(hits, [r["repo"] for r in positives], binary=True), + suspicions_per_push=_rate([len(r["push"]["admission"]["defects"]) for r in ordinary], [r["repo"] for r in ordinary]), + false_alarms_per_100_pushes=_rate( + [int(score.false_alarm(r)) for r in ordinary], [r["repo"] for r in ordinary], binary=True, scale=100 + ), + ) + point["false_alarms_per_100_pushes"] = point["estimates"]["false_alarms_per_100_pushes"]["rate"] + high_recall = [p for p in curve if p["recall"] is not None and p["recall"] >= 0.9 and p["false_alarms_per_100_pushes"] is not None] + low_alarm = [ + p + for p in curve + if p["false_alarms_per_100_pushes"] is not None and p["false_alarms_per_100_pushes"] < 10 and p["recall"] is not None + ] + return curve, dict( + false_alarms_at_recall_90=min(high_recall, key=lambda p: p["false_alarms_per_100_pushes"]) if high_recall else None, + recall_under_10_false_alarms=max(low_alarm, key=lambda p: p["recall"]) if low_alarm else None, + ) + + +def _precision(column: Any) -> float | None: + """Reduce paired (correctly flagged, flagged) counts to push precision.""" + flagged = sum(value[1] for value in column) + return sum(value[0] for value in column) / flagged if flagged else None + + +def _baseline_comparison(baselines: list[list[dict]], records: list[dict], families: list[str]) -> dict: + """Pair every baseline repeat with the one accepted run, without inventing new repetitions.""" + from benchmarks.unseen import score + from benchmarks.valid import metrics + + units = [ + dict(change=after, repeat=repeat, a=_outcome(before, before, 10), b=_outcome(after, after, 10)) + for repeat, baseline in enumerate(baselines) + for before, after in zip(baseline, records, strict=True) + ] + result = {} + for name in ["pooled", *families]: + family = None if name == "pooled" else name + recall, suspicions = _paired(units, family, seed=0, resamples=stats.RESAMPLES) + positives = [u for u in units if u["change"]["kind"] == "vulnerability" and (family is None or u["change"].get("family") == family)] + paired = dict( + search_recall=asdict(recall), + suspicions_per_push=asdict(suspicions), + mcnemar=asdict(stats.mcnemar_exact([u["a"]["hit"] for u in positives], [u["b"]["hit"] for u in positives])), + ) + rows = {"recall": [], "false_alarms_per_100_pushes": [], "precision": []} + for baseline in baselines: + before = metrics.scoring_records(_family_records(baseline, family)) + after = metrics.scoring_records(_family_records(records, family)) + for a, b in zip(before, after, strict=True): + flags = [int(score.false_alarm(r)) for r in (a, b)] + hits = [int(bool(score.catches(r))) if r["kind"] == "vulnerability" else 0 for r in (a, b)] + key = "recall" if a["kind"] == "vulnerability" else "false_alarms_per_100_pushes" + values = hits if key == "recall" else [100 * v for v in flags] + rows[key].append(dict(repo=a["repo"], a=values[0], b=values[1])) + rows["precision"].append(dict(repo=a["repo"], a=(hits[0], flags[0]), b=(hits[1], flags[1]))) + for metric, frame in rows.items(): + paired[metric] = asdict(_estimate(frame, seed=0, resamples=stats.RESAMPLES, stat=_precision if metric == "precision" else None)) + for metric in ("recall", "false_alarms_per_100_pushes"): + paired[metric + "_mcnemar"] = asdict( + stats.mcnemar_exact([bool(r["a"]) for r in rows[metric]], [bool(r["b"]) for r in rows[metric]]) + ) + result[name] = paired + return result + + +def _before(pilot_metrics: Mapping[str, Path]) -> dict: + """Copy the two recorded metric files unchanged, with their source paths.""" + if set(pilot_metrics) != {"raw", "filtered"}: + raise ValueError("pilot_metrics must map raw and filtered to their recorded JSON files") + return {name: dict(source=str(Path(path)), metrics=json.loads(Path(path).read_text())) for name, path in pilot_metrics.items()} + + +def _format(metric: Mapping | None) -> str: + """Show the estimate, interval and count together, leaving missing measurements explicit.""" + if not metric or metric.get("rate") is None: + return "not recorded" if not metric else f"not measured (n={metric.get('n', 0)})" + intervals = [] + for key, label in (("ci", "cluster 95%"), ("cluster95", "cluster 95%"), ("wilson95", "Wilson 95%")): + ci = metric.get(key) + if ci and all(v is not None for v in ci): + intervals.append(f"{label} {ci[0]:.6g} to {ci[1]:.6g}") + return f"{metric['rate']:.6g}; {'; '.join(intervals)}; count={metric['count']}, n={metric['n']}" + + +def _notes(analysis: dict) -> str: + """Write readable before-and-after tables and the evidence needed to interpret them.""" + lines = ["# Hunter pilot gain", "", f"Program: `{analysis['program_id']}`", ""] + proof = analysis["instrument"] + lines += [ + "## Instrument", + "", + f"{proof['failures']} instrument failures across {proof['changes']} changes.", + "Bytes per changed file and verified base and head are in instrument.json.", + "", + ] + if not analysis["valid"]: + lines += [f"Run invalid: {analysis['reason']}. No performance numbers are quoted.", ""] + else: + lines += [ + "## Before and after", + "", + "Before values are quoted as recorded; unavailable family or search values are not inferred.", + "Recall and precision are proportions. False alarms are per 100 ordinary pushes.", + "Push metrics exclude instrument failures as in the pilot. Search metrics and paired comparisons retain failures as misses.", + "The confidence curve describes the search output; it does not choose a threshold or change the classifier.", + "", + "| Family | Metric | Before raw | Before filtered | After |", + "|---|---|---|---|---|", + ] + for family, summary in analysis["after"].items(): + for stage, key, label in ( + ("search", "recall", "Search recall"), + ("search", "suspicions_per_push", "Suspicions per ordinary push"), + ("push", "recall", "Push recall"), + ("push", "false_alarms_per_100_pushes", "False alarms per 100"), + ("push", "precision", "Precision"), + ): + before = [ + analysis["before"][name]["metrics"].get(key) if family == "pooled" and stage == "push" else None + for name in ("raw", "filtered") + ] + lines.append(f"| {family} | {label} | {_format(before[0])} | {_format(before[1])} | {_format(summary[stage][key])} |") + for name, recorded in analysis["before"].items(): + lines += ["", f"Recorded {name} source: `{recorded['source']}`."] + if analysis.get("baseline"): + lines += [ + "", + "## Baseline repeats", + "", + "Each repeat is paired with the single accepted run; intervals resample repositories.", + "", + "| Repeat | Family | Metric | Baseline | After |", + "|---|---|---|---|---|", + ] + for repeat, baseline in enumerate(analysis["baseline"]): + if not baseline["valid"]: + lines += [f"Baseline repeat {repeat} is invalid; its performance numbers and paired gain are not quoted."] + continue + for family, summary in baseline["summary"].items(): + for stage, keys in ( + ("search", ("recall", "suspicions_per_push")), + ("push", ("recall", "false_alarms_per_100_pushes", "precision")), + ): + for key in keys: + lines.append( + f"| {repeat} | {family} | {stage} {key} | {_format(summary[stage][key])} | " + f"{_format(analysis['after'][family][stage][key])} |" + ) + if "paired" in analysis: + lines += [ + "", + "Paired differences and exact McNemar results:", + "", + "```json", + json.dumps(analysis["paired"], indent=2), + "```", + ] + lines += ["", "## Workload", ""] + for name, value in analysis["workload"].items(): + lines.append(f"{name.replace('_', ' ')}: {_format(value)}.") + lines += ["", "## Gate distance", ""] + for name, point in analysis["gate_distance"].items(): + label = ( + "Lowest false alarms at recall at least 0.9" + if name == "false_alarms_at_recall_90" + else "Highest recall under 10 false alarms per 100" + ) + lines.append( + f"{label}: " + + ( + "the curve never reaches this gate bound." + if point is None + else f"recall {_format(point['estimates']['recall'])}; " + f"false alarms {_format(point['estimates']['false_alarms_per_100_pushes'])}." + ) + ) + lines += ["", "## Acceptance comparison", "", "```json", json.dumps(analysis["comparison"], indent=2), "```", ""] + return "\n".join(lines) + + +def pilot_gain( + candidate: Candidate, + decision: accept.Decision, + *, + pilot_manifest: Path, + repos: Mapping[str, str], + training_runs: Sequence[Path], + out_root: Path, + lm: Any, + budget: Mapping[str, float], + baseline_runs: Sequence[Path] = (), + pilot_metrics: Mapping[str, Path], + tools_for: Any = None, + load_changes: Any = None, +) -> dict: + """Run an accepted program once and write analysis.json, instrument.json and NOTES.md. + + pilot_metrics maps raw and filtered to the original metric JSON files. Baseline runs are + single-repeat pilot record directories, each paired with this one accepted run. No command + line scorers or real model clients are constructed here. Budget exhaustion or more than five + percent instrument failures leaves an invalid measurement without performance numbers. + """ + import dspy + + from benchmarks.unseen import score + from benchmarks.unseen.replay import load_slice + + from .program import repository_tools + + if not decision.accepted or decision.comparison.get("candidate_id") != candidate.program_id: + raise ValueError("pilot gain requires the accepted decision for this candidate") + training_runs = [Path(p) for p in training_runs] + if contamination.clean_runs(training_runs) != training_runs: + raise ValueError("contaminated training runs cannot report pilot gain") + before = _before(pilot_metrics) + program = _candidate_program(candidate, tools_for) + changes = list((load_changes or load_slice)(Path(pilot_manifest), "1")) + if not changes or len({c["id"] for c in changes}) != len(changes): + raise ValueError("pilot changes must be nonempty and have unique IDs") + directory = _reserve(Path(out_root), candidate.program_id) + with dspy.context(lm=lm, adapter=dspy.ChatAdapter()): + artifact = runner.run( + Path(pilot_manifest), + repos, + directory / "records", + program=program, + program_id=candidate.program_id, + mask=candidate.mask, + pilot_report=True, + only="all", + budget=budget, + tools_for=tools_for or repository_tools, + load_changes=lambda _path, _slice: changes, + versions={"dspy": dspy.__version__}, + ) + records = _records(directory / "records", changes) + proof = _instrument(records) + _write(directory / "instrument.json", proof) + analysis = dict( + program_id=candidate.program_id, + comparison=decision.comparison, + measurement_directory=directory.name, + valid=proof["valid"] and artifact is not None, + instrument={k: v for k, v in proof.items() if k != "records"}, + ) + if not analysis["valid"]: + analysis["reason"] = "more than 5% instrument failures" if not proof["valid"] else "runner budget exhausted" + else: + records = _clean(records) + # Run the same overall instrument gate and scorer as the pilot before quoting any scores. + score.score(records) + families = _families([dict(change=r, after=_outcome(r, r, 10)) for r in records], ("after",)) + baseline_records = [_clean(_records(Path(path), changes)) for path in baseline_runs] + for baseline in baseline_records: + families = sorted(set(families) | set(_families([dict(change=r, before=_outcome(r, r, 10)) for r in baseline], ("before",)))) + curve, distance = _curve(records) + counts = [search_outcome(r, _findings(r["push"]["admission"]["defects"])).suspicions for r in records] + analysis.update( + before=before, + after=_summaries(records, families), + curve=curve, + gate_distance=distance, + workload=dict( + suspicions_per_100_pushes=_rate(counts, [r["repo"] for r in records], scale=100), + flagged_pushes_per_100_pushes=_rate([int(n > 0) for n in counts], [r["repo"] for r in records], binary=True, scale=100), + ), + ) + if baseline_records: + analysis["baseline"] = [] + for path, baseline in zip(baseline_runs, baseline_records, strict=True): + instrument = _instrument(baseline) + entry = dict(directory=str(path), valid=instrument["valid"], instrument=instrument) + if entry["valid"]: + entry["summary"] = _summaries(baseline, families) + analysis["baseline"].append(entry) + if all(row["valid"] for row in analysis["baseline"]): + analysis["paired"] = _baseline_comparison(baseline_records, records, families) + # Return the same JSON types that a reader of analysis.json receives. + analysis = json.loads(json.dumps(analysis)) + _write(directory / "analysis.json", analysis) + (directory / "NOTES.md").write_text(_notes(analysis), encoding="utf-8") + return analysis diff --git a/src/openultrasast/hunt/rounds.py b/src/openultrasast/hunt/rounds.py new file mode 100644 index 0000000..ef0adaa --- /dev/null +++ b/src/openultrasast/hunt/rounds.py @@ -0,0 +1,299 @@ +"""Measure proposals through the checkout runner before applying the acceptance rule.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import Any + +from .. import stats +from . import accept, runner +from .ablation import _findings, _write +from .metric import search_outcome +from .propose import Candidate +from .table import FeatureTable + + +def _fold_changes(pool_manifest: Path, table: FeatureTable, load_changes: Any) -> list[dict]: + """Resolve folds from table rows, rejecting conflicting change or repository identities.""" + from benchmarks.unseen.replay import load_slice + + runner.refuse_pilot(pool_manifest, pilot_report=False) + folds, repo_folds = {}, {} + for row in table.rows: + key, repo, fold = row["change_id"], row["repo"], row.get("fold") + if type(fold) is not int or fold < 0: + raise ValueError("each table row needs a non-negative fold") + if key in folds and folds[key] != (repo, fold): + raise ValueError("rows of a change must share repository and fold") + if repo in repo_folds and repo_folds[repo] != fold: + raise ValueError("a repository cannot cross folds") + folds[key], repo_folds[repo] = (repo, fold), fold + loaded = (load_changes or load_slice)(pool_manifest, "1") + if len({c["id"] for c in loaded}) != len(loaded): + raise ValueError("change IDs must be unique") + selected = [] + for change in loaded: + if change["id"] not in folds: + raise ValueError("pool changes must exist in the feature table") + repo, fold = folds[change["id"]] + if change["repo"] != repo: + raise ValueError("pool and table disagree on repository") + selected.append({**change, "fold": fold}) + return selected + + +def score_folds( + program: Any, + *, + program_id: str, + mask: Sequence[str], + pool_manifest: Path, + repos: Mapping[str, str], + table: FeatureTable, + folds: Sequence[int], + out_dir: Path, + lm: Any, + budget: Mapping[str, float], + repeats: int = 1, + load_changes: Any = None, + tools_for: Any = None, +) -> Path: + """Run selected folds in repeat directories numbered from zero, with runner resumability. + + The table supplies fold membership; the runner computes inputs at each change's head and + binds repository tools there. The saved study fixes the sample before any records are written. + Budget applies to each runner pass, as it does for ablations. + """ + import dspy + + from .program import repository_tools + + if type(repeats) is not int or repeats < 1: + raise ValueError("repeats must be a positive integer") + if not folds or any(type(f) is not int or f < 0 for f in folds) or len(set(folds)) != len(folds): + raise ValueError("folds must be distinct non-negative integers") + pool_manifest, out = Path(pool_manifest), Path(out_dir) + changes = [c for c in _fold_changes(pool_manifest, table, load_changes) if c["fold"] in folds] + if not changes: + raise ValueError("selected folds contain no changes") + study = dict( + changes=changes, + repeats=repeats, + program_id=program_id, + mask=sorted(set(mask)), + folds=sorted(folds), + table_digest=table.digest(), + pool_digest=hashlib.sha256(pool_manifest.read_bytes()).hexdigest(), + ) + path = out / "study.json" + if path.exists() and json.loads(path.read_text()) != study: + raise ValueError("study differs from the saved fold run") + if not path.exists() and out.exists() and any(out.iterdir()): + raise ValueError("cannot adopt records without a saved fold study") + _write(path, study) + for repeat in range(repeats): + with dspy.context(lm=lm, adapter=dspy.ChatAdapter()): + runner.run( + pool_manifest, + repos, + out / str(repeat), + program=program, + program_id=program_id, + mask=mask, + only="all", + pilot_report=False, + budget=budget, + load_changes=lambda _path, _slice: changes, + tools_for=tools_for or repository_tools, + versions={"dspy": dspy.__version__}, + ) + return out + + +def _outcome(record: Mapping, change: Mapping, tolerance: int) -> dict: + """Keep failed, unparsed and absent records as misses, exactly as ablation analysis does.""" + 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 = [] + return dict( + defects=defects, + hit=search_outcome(change, _findings(defects), tolerance=tolerance).hit, + instrument_failures=int(failed), + parse_failures=int(parsed), + missing_records=int(not record), + ) + + +def _estimate(rows: list[dict], *, seed: int, resamples: int, stat: Any = None) -> stats.Estimate: + """Compare candidate minus current while resampling whole repositories.""" + import pandas as pd + + return stats.paired_bootstrap( + pd.DataFrame(rows, columns=["repo", "a", "b"]), "repo", "a", "b", seed=seed, resamples=resamples, stat=stat + ) + + +def _families(units: list[dict], arms: Sequence[str]) -> list[str]: + """Include labelled families and families reported only on ordinary pushes.""" + return sorted( + {u["change"]["family"] for u in units if u["change"].get("family")} + | {d["family"] for u in units for arm in arms for d in u[arm]["defects"] if d.get("family")} + ) + + +def _paired(units: list[dict], family: str | None, *, seed: int, resamples: int) -> tuple[stats.Estimate, stats.Estimate]: + """Compare hits on labelled positives and family suspicions over every ordinary push.""" + 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"] + recall = _estimate( + [dict(repo=u["change"]["repo"], a=int(u["a"]["hit"]), b=int(u["b"]["hit"])) for u in positives], seed=seed, resamples=resamples + ) + suspicions = _estimate( + [ + { + "repo": u["change"]["repo"], + **{arm: sum(family is None or d.get("family") == family for d in u[arm]["defects"]) for arm in ("a", "b")}, + } + for u in ordinary + ], + seed=seed, + resamples=resamples, + ) + return recall, suspicions + + +def fold_results( + current_dir: Path, candidate_dir: Path, changes: Sequence[Mapping], *, scope: str, tolerance: int, seed: int, resamples: int = 2000 +) -> dict[str, accept.FoldResult]: + """Return acceptance inputs on (change, repeat) pairs, including missing records as misses. + + Pass the changes for this scope. Saved studies fix the expected repeats; for standalone record + fixtures, numeric child directories supply repeats. The union keeps an entirely missing arm. + """ + if scope not in ("compile", "holdout"): + raise ValueError("scope must be compile or holdout") + if tolerance < 0 or resamples < 2: + raise ValueError("tolerance must be non-negative and resamples at least two") + changes = list(changes) + if len({c["id"] for c in changes}) != len(changes): + raise ValueError("change IDs must be unique") + directories = {"a": Path(current_dir), "b": Path(candidate_dir)} + studies, repeats = {}, set() + for arm, directory in directories.items(): + path = directory / "study.json" + if path.exists(): + study = studies[arm] = json.loads(path.read_text()) + saved = {c["id"]: c for c in study["changes"]} + if any(c["id"] not in saved or {"fold": saved[c["id"]]["fold"], **c} != saved[c["id"]] for c in changes): + raise ValueError("analysis changes differ from the saved study") + repeats.update(range(study["repeats"])) + elif directory.exists(): + repeats.update(int(p.name) for p in directory.iterdir() if p.is_dir() and p.name.isdecimal()) + if len(studies) == 2 and studies["a"]["repeats"] != studies["b"]["repeats"]: + raise ValueError("paired studies must have the same repeats") + units = [] + for change in changes: + for repeat in sorted(repeats or {0}): + unit = dict(change=change, repeat=repeat) + for arm, directory in directories.items(): + path = runner.record_path(directory / str(repeat), change["id"]) + record = json.loads(path.read_text()) if path.exists() else {} + if record and arm in studies: + study = studies[arm] + saved_change = next(c for c in study["changes"] if c["id"] == change["id"]) + if ( + record.get("id") != change["id"] + or record.get("program_id") != study["program_id"] + or set(record.get("mask", [])) != set(study["mask"]) + or record.get("input_digest") != runner.input_digest(saved_change, study["program_id"], study["mask"]) + ): + raise ValueError("record does not match the saved change, program or mask") + unit[arm] = _outcome(record, change, tolerance) + units.append(unit) + return { + family: accept.FoldResult(scope, family, *_paired(units, None if family == "pooled" else family, seed=seed, resamples=resamples)) + for family in ["pooled", *_families(units, ("a", "b"))] + } + + +def _candidate_program(candidate: Candidate, tools_for: Any = None) -> Any: + """Restore either a one-call program or its tool loop without losing the proposed state.""" + from .program import HuntProgram, program_id, repository_tools + + uses_tools = "hunt.react" in candidate.state + tools = (tools_for or repository_tools)(Path.cwd()) if uses_tools else () + program = HuntProgram(mask=candidate.mask, tools=tools) + if set(program.dump_state()) != set(candidate.state): + raise ValueError("unsupported candidate program structure") + program.load_state(candidate.state) + if program_id(program, mask=candidate.mask) != candidate.program_id: + raise ValueError("candidate program ID does not match its saved state") + return program + + +def round_trip( + current: Any, + candidate: Candidate, + *, + current_id: str, + pool_manifest: Path, + repos: Mapping[str, str], + table: FeatureTable, + compile_folds: Sequence[int], + holdout_folds: Sequence[int], + out_dir: Path, + lm: Any, + budget: Mapping[str, float], + rule: accept.Rule, + repeats: int = 1, + tolerance: int = 10, + seed: int = 0, + resamples: int = 2000, + load_changes: Any = None, + tools_for: Any = None, +) -> accept.Decision: + """Score both programs on both scopes, decide, and retain compile-only rejection history.""" + from .program import program_id + + if not compile_folds or not holdout_folds or set(compile_folds) & set(holdout_folds): + raise ValueError("compile and holdout folds must be nonempty and disjoint") + if current_id != program_id(current, mask=candidate.base_mask) or tuple(current.mask) != candidate.base_mask: + raise ValueError("current program does not match its ID and base mask") + candidate_program = _candidate_program(candidate, tools_for) + changes = _fold_changes(Path(pool_manifest), table, load_changes) + results = {} + for scope, folds in (("compile", compile_folds), ("holdout", holdout_folds)): + paths = [] + for arm, program, pid, mask in ( + ("current", current, current_id, candidate.base_mask), + ("candidate", candidate_program, candidate.program_id, candidate.mask), + ): + paths.append( + score_folds( + program, + program_id=pid, + mask=mask, + pool_manifest=pool_manifest, + repos=repos, + table=table, + folds=folds, + out_dir=Path(out_dir) / scope / arm, + lm=lm, + budget=budget, + repeats=repeats, + load_changes=load_changes, + tools_for=tools_for, + ) + ) + results[scope] = fold_results( + *paths, [c for c in changes if c["fold"] in folds], scope=scope, tolerance=tolerance, seed=seed, resamples=resamples + ) + decision = accept.decide(current_id, candidate, results["compile"], results["holdout"], rule) + if not decision.accepted: + accept.record_rejection(candidate.out_dir, candidate, decision) + return decision diff --git a/tests/test_hunt_report.py b/tests/test_hunt_report.py new file mode 100644 index 0000000..2c502d8 --- /dev/null +++ b/tests/test_hunt_report.py @@ -0,0 +1,241 @@ +"""Pilot reports use small git fixtures and never run the held-out pilot.""" + +import json +from dataclasses import replace + +import pytest + +pytest.importorskip("dspy") +pytest.importorskip("pandas") +pytest.importorskip("statsmodels") + +from test_hunt_rounds import candidate_at, finding, model # noqa: E402 +from test_hunt_rounds import population as population # noqa: E402 + +from benchmarks.unseen import score # noqa: E402 +from benchmarks.valid import metrics # noqa: E402 +from openultrasast.hunt import accept, report, runner # noqa: E402 + + +@pytest.fixture +def pilot_case(population, tmp_path): + """Create an accepted candidate, a checked training run and recorded before metrics.""" + pool, repos, changes, _ = population + _, _, candidate = candidate_at(tmp_path / "proposal") + decision = accept.Decision(True, [], {"candidate_id": candidate.program_id, "accepted": True, "rule": {"version": "fixture"}}) + training = tmp_path / "training" + training.mkdir() + (training / "contamination.json").write_text('{"contaminated": false}') + # These are fixture numbers in exactly the pilot metrics file's shape. + before = dict( + recall=score.rate([False, True]), + precision=score.rate([False, True, False]), + false_alarms_per_100_pushes={**score.rate([True, True]), "rate": 100, "wilson95": [34.237, 100]}, + ) + paths = {} + for name in ("raw", "filtered"): + path = tmp_path / f"metrics_b_{name}.json" + path.write_text(json.dumps(before)) + paths[name] = path + kwargs = dict( + pilot_manifest=pool, + repos=repos, + training_runs=[training], + out_root=tmp_path / "reports", + lm=model([[finding()], [], [finding()], []]), + budget={"usd": 1, "calls": 100}, + pilot_metrics=paths, + load_changes=lambda m, s: changes, + ) + return candidate, decision, kwargs + + +@pytest.mark.parametrize("accepted,matches", [(False, True), (True, False)]) +def test_refuse_unaccepted_or_unrelated_decision(pilot_case, accepted, matches): + """Only the decision for this accepted program authorizes a pilot attempt.""" + candidate, decision, kwargs = pilot_case + decision = replace(decision, accepted=accepted, comparison={"candidate_id": candidate.program_id if matches else "other"}) + with pytest.raises(ValueError, match="accepted"): + report.pilot_gain(candidate, decision, **kwargs) + assert not kwargs["out_root"].exists() + + +@pytest.mark.parametrize("check", [None, {"contaminated": True}, {"contaminated": "false"}]) +def test_refuse_contaminated_or_unchecked_training(pilot_case, check): + """Missing, malformed and contaminated checks cannot lead to a pilot number.""" + candidate, decision, kwargs = pilot_case + path = kwargs["training_runs"][0] / "contamination.json" + path.unlink() if check is None else path.write_text(json.dumps(check)) + with pytest.raises(ValueError, match="unchecked|contaminat"): + report.pilot_gain(candidate, decision, **kwargs) + assert not kwargs["out_root"].exists() + + +def test_valid_report_before_after_workload_curve_and_ledger(pilot_case, population): + """A readable run writes metrics, per-family tables, instrument proof and a durable reservation.""" + candidate, decision, kwargs = pilot_case + analysis = report.pilot_gain(candidate, decision, **kwargs) + assert analysis["valid"] + directory = kwargs["out_root"] / analysis["measurement_directory"] + assert json.loads((directory / "analysis.json").read_text()) == analysis + instruments = json.loads((directory / "instrument.json").read_text()) + assert instruments["failures"] == 0 and instruments["changes"] == 4 + for row in instruments["records"]: + assert row["changed_bytes"]["app.py"] > 0 + assert row["base_verified"] and row["head_verified"] and row["base"] and row["head"] + assert analysis["before"]["raw"]["metrics"] == json.loads(kwargs["pilot_metrics"]["raw"].read_text()) + records = score.load_records(directory / "records") + assert analysis["after"]["pooled"]["push"] == metrics.summarize(records, scored=score.score(metrics.scoring_records(records))) + assert analysis["after"]["injection"]["search"]["recall"]["rate"] == 1 + assert analysis["after"]["pooled"]["search"]["suspicions_per_push"]["rate"] == 0 + assert analysis["workload"]["suspicions_per_100_pushes"]["rate"] == 50 + assert analysis["workload"]["flagged_pushes_per_100_pushes"]["rate"] == 50 + assert analysis["gate_distance"]["false_alarms_at_recall_90"]["false_alarms_per_100_pushes"] == 0 + assert analysis["gate_distance"]["recall_under_10_false_alarms"]["recall"] == 1 + notes = (directory / "NOTES.md").read_text() + for phrase in ("raw", "filtered", "injection", "pooled", "Workload", "Gate", "Instrument", candidate.program_id): + assert phrase in notes + ledger = [json.loads(line) for line in (kwargs["out_root"] / "hunt-pilot-runs.jsonl").read_text().splitlines()] + assert ledger[0]["program_id"] == candidate.program_id and ledger[0]["date"] + calls = len(kwargs["lm"].history) + with pytest.raises(ValueError, match="already"): + report.pilot_gain(candidate, decision, **kwargs) + assert len(kwargs["lm"].history) == calls + + +def test_invalid_instrument_quotes_no_numbers(pilot_case, population): + """One unreadable change out of four invalidates the run before any metrics are computed.""" + candidate, decision, kwargs = pilot_case + changes = [dict(c) for c in population[2]] + changes[0]["files"] = [{"path": "missing.py", "head_lines": [[1, 2]]}] + kwargs["load_changes"] = lambda m, s: changes + analysis = report.pilot_gain(candidate, decision, **kwargs) + assert not analysis["valid"] and analysis["instrument"]["failures"] == 1 + for name in ("before", "after", "workload", "gate_distance", "paired"): + assert name not in analysis + directory = kwargs["out_root"] / analysis["measurement_directory"] + notes = (directory / "NOTES.md").read_text() + assert "invalid" in notes.lower() and "No performance numbers" in notes + assert "Wilson" not in notes and "0.5" not in notes + assert (directory / "instrument.json").is_file() and (directory / "analysis.json").is_file() + + +def test_crash_consumes_program_id(pilot_case, monkeypatch): + """The reservation is on disk before runner.run starts, including a failed launch.""" + candidate, decision, kwargs = pilot_case + + def crash(*args, **kw): + assert kw["pilot_report"] is True and kw["only"] == "all" + ledger = kwargs["out_root"] / "hunt-pilot-runs.jsonl" + assert json.loads(ledger.read_text())["program_id"] == candidate.program_id + raise RuntimeError("fixture crash") + + monkeypatch.setattr(runner, "run", crash) + with pytest.raises(RuntimeError, match="fixture crash"): + report.pilot_gain(candidate, decision, **kwargs) + with pytest.raises(ValueError, match="already"): + report.pilot_gain(candidate, decision, **kwargs) + + +def test_baseline_repeats_are_paired_with_single_run(pilot_case, population, tmp_path): + """Two baseline repeats yield four positive pairs and two repository clusters.""" + import dspy + + candidate, decision, kwargs = pilot_case + base, _, _ = candidate_at(tmp_path / "baseline-program") + directories = [] + for repeat in range(2): + out = tmp_path / f"baseline-{repeat}" + with dspy.context(lm=model([[]] * 4), adapter=dspy.ChatAdapter()): + runner.run( + population[0], population[1], out, program=base, program_id=candidate.base_program_id, load_changes=kwargs["load_changes"] + ) + directories.append(out) + analysis = report.pilot_gain(candidate, decision, **{**kwargs, "baseline_runs": directories}) + recall = analysis["paired"]["pooled"]["search_recall"] + assert recall["diff"] == 1 and recall["units"] == 4 and recall["groups"] == 2 + assert recall["ci"] == [1, 1] + assert analysis["paired"]["pooled"]["mcnemar"]["c"] == 4 + assert analysis["paired"]["pooled"]["recall_mcnemar"]["c"] == 4 + assert analysis["paired"]["pooled"]["false_alarms_per_100_pushes"]["diff"] == 0 + assert len(analysis["baseline"]) == 2 + + +def test_curve_reports_unreached_bounds_and_counts_unknown_families(pilot_case): + """Low recall and ordinary alarms leave both gate bounds explicitly unreached.""" + candidate, decision, kwargs = pilot_case + kwargs["lm"] = model([[], [finding(0.4, "path"), finding(0.4)], [], [finding(0.4)]]) + analysis = report.pilot_gain(candidate, decision, **kwargs) + assert analysis["gate_distance"]["false_alarms_at_recall_90"] is None + assert analysis["gate_distance"]["recall_under_10_false_alarms"] is None + assert analysis["after"]["path"]["search"]["recall"]["n"] == 0 + assert analysis["after"]["path"]["search"]["suspicions_per_push"]["count"] == 1 + point = analysis["curve"][0] + assert point["false_alarms_per_100_pushes"] == 100 and point["suspicions_per_push"] == 1.5 + notes = (kwargs["out_root"] / analysis["measurement_directory"] / "NOTES.md").read_text() + assert "never reaches" in notes + + +def test_gate_reads_both_bounds_from_different_confidence_cuts(pilot_case): + """A quiet high cut and a noisy complete cut report different gate distances.""" + candidate, decision, kwargs = pilot_case + kwargs["lm"] = model([[finding(0.9)], [finding(0.4)], [finding(0.4)], [finding(0.4)]]) + analysis = report.pilot_gain(candidate, decision, **kwargs) + assert [p["confidence"] for p in analysis["curve"]] == [0.9, 0.4] + assert analysis["gate_distance"]["false_alarms_at_recall_90"]["false_alarms_per_100_pushes"] == 100 + assert analysis["gate_distance"]["recall_under_10_false_alarms"]["recall"] == 0.5 + assert all(p["estimates"]["recall"]["n"] == 2 and p["estimates"]["recall"]["ci"] for p in analysis["curve"]) + + +def test_budget_stop_keeps_unrun_changes_in_instrument_denominator(pilot_case): + """A partial runner input index cannot make an unfinished pilot look complete.""" + candidate, decision, kwargs = pilot_case + analysis = report.pilot_gain(candidate, decision, **{**kwargs, "budget": {"usd": 1, "calls": 1}}) + assert not analysis["valid"] + assert analysis["instrument"]["changes"] == 4 and analysis["instrument"]["failures"] == 3 + assert "after" not in analysis + + +def test_report_restores_tool_program_and_reads_checkout(pilot_case, tmp_path): + """A saved ReAct candidate keeps its tools and reads source through the runner.""" + from dspy.utils import DummyLM + + _, decision, kwargs = pilot_case + _, _, candidate = candidate_at(tmp_path / "tool-proposal", tools=True) + decision = replace(decision, comparison={"candidate_id": candidate.program_id}) + kwargs["lm"] = DummyLM( + [ + answer + for _ in range(4) + for answer in ( + {"next_thought": "Read", "next_tool_name": "read_file", "next_tool_args": {"path": "app.py"}}, + {"next_thought": "Done", "next_tool_name": "finish", "next_tool_args": {}}, + {"reasoning": "The interpreter receives input.", "findings": json.dumps([finding()])}, + ) + ] + ) + analysis = report.pilot_gain(candidate, decision, **kwargs) + assert analysis["valid"] and analysis["after"]["pooled"]["search"]["recall"]["rate"] == 1 + assert all("Failed to execute" not in str(entry.get("messages")) for entry in kwargs["lm"].history) + + +def test_push_metrics_use_pilots_alternate_target_projection(pilot_case, population): + """A catch at an additional verified site counts once in the pilot's push metric.""" + candidate, decision, kwargs = pilot_case + # This test isolates scoring projection; the remote primary site supplies no shown source lines. + kwargs["lm"] = model([[{**finding(), "cited_lines": []}], [], [{**finding(), "cited_lines": []}], []]) + changes = [] + for original in population[2]: + change = dict(original) + if change["kind"] == "vulnerability": + change.update( + path="other.py", + function="other", + function_lines=[40, 50], + files=[{"path": "app.py", "head_lines": [[40, 50]]}], + targets=[dict(path="app.py", function="run", function_lines=[1, 2], family="injection")], + ) + changes.append(change) + analysis = report.pilot_gain(candidate, decision, **{**kwargs, "load_changes": lambda m, s: changes}) + assert analysis["after"]["pooled"]["push"]["recall"]["count"] == 2 + assert analysis["after"]["pooled"]["search"]["recall"]["count"] == 0 diff --git a/tests/test_hunt_rounds.py b/tests/test_hunt_rounds.py new file mode 100644 index 0000000..5382791 --- /dev/null +++ b/tests/test_hunt_rounds.py @@ -0,0 +1,221 @@ +"""Score real fixture checkouts with scripted answers and compare separate folds.""" + +import json +import subprocess +from copy import deepcopy +from dataclasses import replace + +import pytest + +dspy = pytest.importorskip("dspy") +pytest.importorskip("pandas") +pytest.importorskip("statsmodels") +from dspy.utils import DummyLM # noqa: E402 + +from openultrasast.hunt import accept, rounds, runner # noqa: E402 +from openultrasast.hunt import program as hp # noqa: E402 +from openultrasast.hunt.propose import Candidate # noqa: E402 +from openultrasast.hunt.table import FeatureTable # noqa: E402 + + +@pytest.fixture +def population(tmp_path): + """Make two repository groups, each with a positive and an ordinary push.""" + changes, rows, repos = [], [], {} + for fold in range(2): + root = tmp_path / f"repo-{fold}" + root.mkdir() + + def git(*args, root=root): + return subprocess.run( + ["git", "-C", str(root), "-c", "core.hooksPath=/dev/null", *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", ".") + git("commit", "-qm", "base") + base = git("rev-parse", "HEAD") + (root / "app.py").write_text("def run():\n return eval(input())\n") + git("add", ".") + git("commit", "-qm", "head") + head = git("rev-parse", "HEAD") + repo = f"fixture/repo-{fold}" + repos[repo] = str(root) + for kind in ("vulnerability", "ordinary"): + change = dict( + id=f"{fold}-{kind}", + repo=repo, + base=base, + head=head, + kind=kind, + family="injection", + path="app.py", + function="run", + function_lines=[1, 2], + case_id=f"case-{fold}-{kind}", + files=[{"path": "app.py", "head_lines": [[2, 2]]}], + ) + changes.append(change) + rows.append(dict(change_id=change["id"], repo=repo, fold=fold)) + pool = tmp_path / "pool-fixture.toml" + pool.write_text("status = 'fixture'\n") + return pool, repos, changes, FeatureTable(rows) + + +def finding(confidence=0.9, family="injection"): + """A suspicion at the fixture's changed interpreter call.""" + return dict(path="app.py", line=2, family=family, confidence=confidence, rationale="eval", cited_lines=[2]) + + +def model(answers): + """Return a local model whose answers go through ChatAdapter.""" + return DummyLM([{"findings": json.dumps(answer)} for answer in answers]) + + +def candidate_at(out, *, tools=False): + """Save one instruction edit with the same structure as a real proposal.""" + base = hp.HuntProgram(tools=hp.repository_tools(out) if tools else ()) + program = base.deepcopy() + predictor = program.hunt.extract.predict if tools else program.hunt + predictor.signature = predictor.signature.with_instructions("Follow the changed interpreter call.") + candidate = Candidate( + hp.program_id(program), + "instruction", + {"instruction": "Follow the changed interpreter call."}, + {}, + deepcopy(program.dump_state()), + hp.program_id(base), + deepcopy(base.dump_state()), + (), + (), + str(out), + ) + return base, program, candidate + + +def score_args(population, out, lm): + """Supply the fixture manifest and its row folds to the runner wrapper.""" + pool, repos, changes, table = population + return dict( + pool_manifest=pool, + repos=repos, + table=table, + folds=[0], + out_dir=out, + lm=lm, + budget={"usd": 1, "calls": 100}, + load_changes=lambda m, s: changes, + ) + + +def test_fold_selection_repeat_resume_and_tools(population, tmp_path): + """Only selected folds run, and tools read each detached checkout on both repeats.""" + base, _, _ = candidate_at(tmp_path, tools=True) + answers = [] + for _ in range(4): + answers.extend( + [ + {"next_thought": "Read the change", "next_tool_name": "read_file", "next_tool_args": {"path": "app.py"}}, + {"next_thought": "Done", "next_tool_name": "finish", "next_tool_args": {}}, + {"reasoning": "The changed call evaluates input.", "findings": json.dumps([finding()])}, + ] + ) + lm = DummyLM(answers) + kwargs = score_args(population, tmp_path / "folds", lm) + out = rounds.score_folds(base, program_id=hp.program_id(base), mask=(), repeats=2, **kwargs) + assert out == tmp_path / "folds" + for repeat in (0, 1): + entries = json.loads((out / str(repeat) / "inputs.json").read_text()) + assert [e["item_id"] for e in entries] == [c["id"] for c in population[2][:2]] + record = json.loads(runner.record_path(out / str(repeat), entries[0]["item_id"]).read_text()) + assert record["instrument"]["changed_bytes"]["app.py"] > 0 + assert record["instrument"]["base_verified"] and record["instrument"]["head_verified"] + assert record["push"]["admission"]["defects"] + assert any("eval(input())" in str(entry.get("messages")) for entry in lm.history) + assert 2 in base.seen["app.py"] + calls = len(lm.history) + rounds.score_folds(base, program_id=hp.program_id(base), mask=(), repeats=2, **kwargs) + assert len(lm.history) == calls + + +def test_fold_refuses_pilot_and_conflicting_rows(population, tmp_path): + """Fold selection cannot bypass the pilot guard or split repository groups.""" + pool, _, _, table = population + kwargs = score_args(population, tmp_path / "folds", model([])) + (pool.with_name("freeze-fixture.json")).write_text(json.dumps({"freeze_digest": runner.PILOT_FREEZE_DIGEST})) + with pytest.raises(ValueError, match="held out"): + rounds.score_folds(hp.HuntProgram(), program_id="p", mask=(), **kwargs) + pool.with_name("freeze-fixture.json").unlink() + table.rows[1]["fold"] = 1 + with pytest.raises(ValueError, match="repository"): + rounds.score_folds(hp.HuntProgram(), program_id="p", mask=(), **kwargs) + + +def test_fold_results_feed_decide_and_outcomes_flip_it(population, tmp_path): + """Paired repository estimates accept an improvement and reject the reversed outcomes.""" + base, program, candidate = candidate_at(tmp_path / "candidate") + for name, prog, answers in (("current", base, [[]] * 8), ("candidate", program, [[finding()], []] * 4)): + kwargs = score_args(population, tmp_path / name, model(answers)) + rounds.score_folds(prog, program_id=hp.program_id(prog), mask=(), repeats=2, **{**kwargs, "folds": [0, 1]}) + current, proposed = tmp_path / "current", tmp_path / "candidate" + + def results(a, b, scope): + return rounds.fold_results(a, b, population[2], scope=scope, tolerance=10, seed=13, resamples=50) + + compiled = results(current, proposed, "compile") + held = results(current, proposed, "holdout") + assert compiled["pooled"].recall.diff == 1 + assert compiled["pooled"].recall.units == 4 and compiled["pooled"].recall.groups == 2 + assert compiled["injection"].suspicions_per_push.diff == 0 + rule = accept.Rule(0, {}, 10000) + assert accept.decide(candidate.base_program_id, candidate, compiled, held, rule).accepted + assert not accept.decide( + candidate.base_program_id, candidate, results(proposed, current, "compile"), results(proposed, current, "holdout"), rule + ).accepted + # Each failure mode retains a positive unit but removes stale findings. + for repeat, change, field in ((0, population[2][0], "parse_failed"), (1, population[2][0], "status")): + path = runner.record_path(proposed / str(repeat), change["id"]) + record = json.loads(path.read_text()) + record[field] = True if field == "parse_failed" else "instrument_failure" + path.write_text(json.dumps(record)) + runner.record_path(proposed / "0", population[2][2]["id"]).unlink() + remaining = results(current, proposed, "compile")["pooled"].recall + assert remaining.units == 4 and remaining.diff == 0.25 + changed_folds = [{**c, "fold": 99} for c in population[2]] + with pytest.raises(ValueError, match="saved study"): + rounds.fold_results(current, proposed, changed_folds, scope="compile", tolerance=10, seed=13, resamples=50) + + +def test_round_trip_scores_four_passes_and_records_rejection(population, tmp_path): + """Compile and holdout run separately; a rejected tie reaches the proposal history.""" + base, _, candidate = candidate_at(tmp_path / "proposal") + kwargs = score_args(population, tmp_path / "round", model([[]] * 8)) + kwargs.pop("folds") + decision = rounds.round_trip( + base, + candidate, + current_id=candidate.base_program_id, + compile_folds=[0], + holdout_folds=[1], + rule=accept.Rule(0, {}, 10000), + seed=13, + resamples=50, + **kwargs, + ) + assert not decision.accepted and "holdout_tie" in decision.reasons + assert len(list((tmp_path / "round").glob("*/*/0/inputs.json"))) == 4 + rejected = json.loads((tmp_path / "proposal" / "rejected.jsonl").read_text()) + assert rejected["program_id"] == candidate.program_id and "holdout_results" not in rejected + with pytest.raises(ValueError, match="disjoint"): + rounds.round_trip( + base, + replace(candidate), + current_id=candidate.base_program_id, + compile_folds=[0], + holdout_folds=[0], + rule=accept.Rule(0, {}, 10000), + **kwargs, + )