diff --git a/.kiro/specs/valid-detection-benchmark/tasks.md b/.kiro/specs/valid-detection-benchmark/tasks.md index 8b91600..acd3fbe 100644 --- a/.kiro/specs/valid-detection-benchmark/tasks.md +++ b/.kiro/specs/valid-detection-benchmark/tasks.md @@ -51,14 +51,16 @@ Order is dependency order. Each task is small enough to review on its own. New c target set size from `required_n` for the full ~40-positive benchmark. ## Group 6 - hunter arm (the LLM tier required on pre-push) -- [ ] 6.1 `benchmarks/valid/hunter_arm.py`: for each change in the frozen manifest, check out `head` in - the local clone, build `FileTarget`s scoped to the changed files, run `hunter.run_hunter_pool` - (`src/openultrasast/hunter.py`, wiring in `cli.py` `_attach_tool_hunter`), and emit a record `score.py` - can read (an alerts/advisory stream keyed by change id). Runs off-cluster; the model key never enters - a replay container. Honors the instrument rule (bytes read per target, fail loudly on an empty read). -- [ ] 6.2 Combine arm (a) and arm (b) records per change id and rescore: this is the number the tool is +- [x] 6.1 `benchmarks/valid/hunter_arm.py`: for each change in the frozen manifest, check out `head` in + the local clone (detached worktree), build hotspots for the changed files, run the shipped deep-mode + arm `tool_hunter.run_tool_hunter` with `resolve_hunter_client()` (the hunter that beat overlay 12/14; + wiring in `cli.py` `_attach_tool_hunter`), and emit a record `score.py` can read: findings go under + `push.admission.defects` with `family` and `locations = [[path, line]]` (the alerts stream), keyed by + change id. Runs off-cluster; the model key never enters a replay container. Honors the instrument rule + (bytes read per hotspot, fail loudly on an empty read). +- [x] 6.2 Combine arm (a) and arm (b) records per change id and rescore: this is the number the tool is held to (R6). Report (a), (a)+(b), and the difference as the pre-push product gap. -- [ ] 6.3 Offline test with a stub hunter: a flagged change joins the (a) record; a missing (b) record +- [x] 6.3 Offline test with a stub hunter: a flagged change joins the (a) record; a missing (b) record leaves (a) unchanged; duplicate ids are rejected. ## Out of this spec diff --git a/benchmarks/valid/hunter_arm.py b/benchmarks/valid/hunter_arm.py new file mode 100644 index 0000000..78efc70 --- /dev/null +++ b/benchmarks/valid/hunter_arm.py @@ -0,0 +1,299 @@ +"""Run the shipped tool hunter on local frozen changes, then join replay records.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +import tempfile +from copy import deepcopy +from pathlib import Path + +from openultrasast import tool_hunter +from openultrasast.complexity.map import Hotspot +from openultrasast.config import load_config +from openultrasast.hunter_tools import clamp_repo_path +from openultrasast.model.endpoint import resolve_models +from openultrasast.model.taxonomy import FAMILY_IDS, load_families + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from benchmarks.unseen.extract import SHA, InstrumentFailure +from benchmarks.unseen.replay import canonical, load_slice +from benchmarks.unseen.score import load_records +from benchmarks.valid.catalog import local_git + +# These names describe findings, never benchmark labels. The taxonomy owns the +# family mapping; an ambiguous family alone does not establish a particular CWE. +_WEAKNESSES = ( + (r"\bsql[ _-]injection\b", "CWE-89"), + (r"\b(?:os[ _-])?command[ _-]injection\b", "CWE-78"), + (r"\b(?:code[ _-]injection|dynamic execution|remote code execution)\b", "CWE-94"), + (r"\b(?:ssti|server[ -]side template injection)\b", "CWE-1336"), + (r"\bcypher[ _-]injection\b", "CWE-943"), + (r"\b(?:path[ _-]traversal|directory[ _-]traversal)\b", "CWE-22"), + (r"\b(?:unsafe[ _-]deserialization|insecure[ _-]deserialization)\b", "CWE-502"), + (r"\b(?:xss|cross[ -]site scripting)\b", "CWE-79"), + (r"\b(?:ssrf|server[ -]side request forgery)\b", "CWE-918"), + (r"\bopen[ _-]redirect\b", "CWE-601"), + (r"\bprototype[ _-]pollution\b", "CWE-1321"), + (r"\b(?:missing[ _-]authentication|missing[ _-]auth[ _-]guard)\b", "CWE-306"), + (r"\b(?:missing[ _-]authorization|broken[ _-]access[ _-]control)\b", "CWE-862"), + (r"\bhardcoded[ _-](?:secret|credential|password)s?\b", "CWE-798"), +) + + +def finding_defect(finding): + """Translate the hunter's own CWE, family tag or weakness title to a defect.""" + taxonomy = load_families() + raw_family = next((tag.split(":", 1)[1].strip() for tag in finding.tags if tag.startswith("family:")), "") + evidence = " ".join([*finding.tags, finding.title]) + explicit = re.search(r"\bCWE-\d+\b", evidence, re.I) + cwe = explicit.group().upper() if explicit else next((cwe for pattern, cwe in _WEAKNESSES if re.search(pattern, evidence, re.I)), None) + mapped = taxonomy.family_of_cwe(cwe or "") + mechanism = taxonomy.family_of_mechanism(raw_family) + family = raw_family if raw_family in FAMILY_IDS else mechanism.id if mechanism else mapped.id if mapped else "unknown" + if mapped and mapped.id != family: + cwe = None + return { + "defect_id": finding.finding_id, + "family": family, + "cwe": cwe, + "function": finding.function_name, + "locations": [[finding.path, finding.line if finding.line is not None else 0]], + "source": "hunter", + } + + +def _write_json(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(".json.tmp") + temporary.write_text(json.dumps(value, sort_keys=True, indent=2) + "\n") + temporary.replace(path) + + +def _record_path(out, identity): + return out / (hashlib.sha256(identity.encode()).hexdigest() + ".json") + + +def _inputs(out): + path = out / "inputs.json" + entries = json.loads(path.read_text()) if path.exists() else [] + for key in (lambda entry: entry["item_id"], lambda entry: entry["change"]["id"]): + if len({key(entry) for entry in entries}) != len(entries): + raise ValueError(f"duplicate ids in {path}") + return entries + + +def _hunt_change(change, repos, *, client, model, max_steps): + instrument = { + "hook_exit": 0, + "hook_bytes_read": 0, + "changed_bytes": {}, + "head_verified": False, + "base_verified": False, + "failure": None, + "suspect_fast": False, + } + record = { + "id": change["id"], + "done": True, + "status": "hunter_covered", + "instrument": instrument, + "push": {"admission": {"defects": [], "dispositions": []}, "quick_tier": []}, + "skipped": [], + } + try: + if not all(SHA.fullmatch(change[side]) for side in ("base", "head")): + raise InstrumentFailure("invalid_revision") + git = local_git(Path(repos[change["repo"]]).expanduser().resolve()) + git.run("cat-file", "-e", change["base"] + "^{commit}") + instrument["base_verified"] = True + with tempfile.TemporaryDirectory(prefix="ousast-hunter-") as temporary: + root = Path(temporary) / "head" + added = False + try: + git.run("-c", "core.hooksPath=/dev/null", "worktree", "add", "--detach", str(root), change["head"]) + added = True + if local_git(root).run("rev-parse", "HEAD").strip() != change["head"]: + raise InstrumentFailure("head_mismatch") + instrument["head_verified"] = True + hotspots = [] + for relative in dict.fromkeys(file["path"] for file in change.get("files", [])): + path = clamp_repo_path(root, relative) + try: + path.stat() + except FileNotFoundError: + reason = {"path": relative, "reason": "missing_at_head"} + record["skipped"].append(reason) + print(f"{change['id']} {relative}: skipped (missing_at_head)", flush=True) + continue + try: + data = path.read_bytes() + except OSError: + print(f"{change['id']} {relative}: 0 bytes read", flush=True) + raise InstrumentFailure(f"unreadable_hotspot:{relative}") from None + size = len(data) + print(f"{change['id']} {relative}: {size} bytes read", flush=True) + instrument["changed_bytes"][relative] = size + instrument["hook_bytes_read"] += size + if not size: + raise InstrumentFailure(f"empty_hotspot:{relative}") + hotspots.append( + Hotspot( + path=relative, + function_name=None, + score=1.0, + band="low", + signals={"bytes_read": size}, + rationale="Changed file at the frozen head", + test_hint=None, + inventory_finding_ids=(), + ) + ) + if not hotspots: + raise InstrumentFailure("no_readable_hotspots") + found = tool_hunter.run_tool_hunter(root, hotspots, client=client, model=model, max_steps=max_steps) + record["push"]["admission"]["defects"] = [finding_defect(finding) for finding in found] + finally: + if added: + git.run("worktree", "remove", "--force", str(root)) + except Exception as error: # noqa: BLE001 + # Provider exception messages can contain credentials. Store only the + # error type, except for our own bounded instrument reasons. + reason = str(error) if isinstance(error, InstrumentFailure) else type(error).__name__ + instrument.update(hook_exit=2, failure=reason) + record["status"] = "instrument_failure" + record["push"]["admission"]["defects"] = [] + print(f"{change['id']}: instrument_failure ({reason})", file=sys.stderr, flush=True) + return record + + +def run(manifest, repos_path, out, *, slice_id="1", max_steps=tool_hunter.DEFAULT_MAX_STEPS, only="all", client_mode=None): + if max_steps < 1: + raise ValueError("max_steps must be positive") + changes = load_slice(manifest, slice_id) + if len({change["id"] for change in changes}) != len(changes): + raise ValueError("duplicate change ids") + changes = [change for change in changes if only == "all" or change["kind"] == only] + entries = _inputs(out) + indexed = {entry["change"]["id"]: entry for entry in entries} + pending = [] + for change in changes: + path = _record_path(out, change["id"]) + previous = json.loads(path.read_text()) if path.exists() else {} + entry = indexed.get(change["id"]) + if entry is None: + entry = { + "item_id": change["id"], + "change": change, + "input_digest": previous.get("input_digest") or hashlib.sha256(canonical(change)).hexdigest(), + } + entries.append(entry) + elif entry["change"] != change or entry["item_id"] != change["id"]: + raise ValueError("existing hunter inputs do not match frozen change") + if previous.get("done") is True: + print(f"{change['id']}: skipped (done)", flush=True) + continue + pending.append(entry) + if not pending: + _write_json(out / "inputs.json", entries) + return 0 + config_path = Path("openultrasast.toml") + config = load_config(config_path if config_path.exists() else None, dotenv=client_mode != "scripted") + model = resolve_models(config)[0] + client = tool_hunter.scripted_hunter_client() if client_mode == "scripted" else tool_hunter.resolve_hunter_client() + if client is None: + print("No hunter client resolved. Configure a hunter provider or use --client scripted for offline tests.", file=sys.stderr) + return 2 + repos = json.loads(repos_path.read_text()) + _write_json(out / "inputs.json", entries) + failed = False + for entry in pending: + # The built-in script is a finite sequence, so each change needs a fresh + # sequence. Live clients remain shared and are resolved only once. + if client_mode == "scripted": + client = tool_hunter.scripted_hunter_client() + record = _hunt_change(entry["change"], repos, client=client, model=model, max_steps=max_steps) + record["input_digest"] = entry["input_digest"] + _write_json(_record_path(out, entry["item_id"]), record) + failed |= record["status"] == "instrument_failure" + return 2 if failed else 0 + + +def _joined(out): + _inputs(out) + rows = load_records(out) + if len({row["id"] for row in rows}) != len(rows): + raise ValueError("duplicate joined change ids") + return {row["id"]: row for row in rows} + + +def _extend_defects(existing, extra): + seen = set() + merged = [] + for defect in [*existing, *extra]: + locations = [] + for path, line in defect.get("locations", []): + key = (path, line, defect.get("family")) + if key not in seen: + locations.append([path, line]) + seen.add(key) + if locations: + merged.append({**defect, "locations": locations}) + elif not defect.get("locations"): + merged.append(defect) + return merged + + +def combine(replay_out, hunter_out, out): + """Preserve replay coverage while adding independently detected hunter defects.""" + a = _joined(replay_out) + b = _joined(hunter_out) if (hunter_out / "inputs.json").exists() else {} + combined = [] + for identity, left in a.items(): + right = b.get(identity, {}) + has_a, has_b = "push" in left, "push" in right + merged = deepcopy(left if has_a or not has_b else right) + if has_a and has_b: + admission = merged["push"].setdefault("admission", {}) + admission["defects"] = _extend_defects( + admission.get("defects", []), + right["push"].get("admission", {}).get("defects", []), + ) + merged["arms"] = { + "a": left.get("status", "missing") if has_a else "missing", + "b": right.get("status", "missing") if has_b else "missing", + } + combined.append(merged) + _write_json(out, combined) + return combined + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + hunt = commands.add_parser("run") + hunt.add_argument("manifest", type=Path) + hunt.add_argument("--slice", default="1") + hunt.add_argument("--repos", type=Path, required=True) + hunt.add_argument("--out", type=Path, required=True) + hunt.add_argument("--max-steps", type=int, default=tool_hunter.DEFAULT_MAX_STEPS) + hunt.add_argument("--only", choices=("vulnerability", "ordinary", "all"), default="all") + hunt.add_argument("--client", choices=("scripted",)) + join = commands.add_parser("combine") + join.add_argument("replay_out", type=Path) + join.add_argument("hunter_out", type=Path) + join.add_argument("--out", type=Path, required=True) + args = parser.parse_args(argv) + if args.command == "combine": + combine(args.replay_out, args.hunter_out, args.out) + return 0 + return run(args.manifest, args.repos, args.out, slice_id=args.slice, max_steps=args.max_steps, only=args.only, client_mode=args.client) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_hunter_arm.py b/tests/test_hunter_arm.py new file mode 100644 index 0000000..956690a --- /dev/null +++ b/tests/test_hunter_arm.py @@ -0,0 +1,306 @@ +"""Offline coverage of the frozen benchmark hunter arm and replay join.""" + +import hashlib +import importlib +import json +import socket +import sys +from dataclasses import replace +from pathlib import Path + +import pytest + +from openultrasast import tool_hunter +from openultrasast.complexity.map import Hotspot +from openultrasast.config import load_config +from openultrasast.model.endpoint import resolve_models + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +hunter_arm = importlib.import_module("benchmarks.valid.hunter_arm") +catalog = importlib.import_module("benchmarks.valid.catalog") +score = importlib.import_module("benchmarks.unseen.score") + + +@pytest.fixture(autouse=True) +def offline(monkeypatch): + def forbidden(*args, **kwargs): + pytest.fail("network access attempted") + + monkeypatch.setattr(socket.socket, "connect", forbidden) + monkeypatch.setattr(socket, "create_connection", forbidden) + for name in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GIT_OBJECT_DIRECTORY", "GIT_ALTERNATE_OBJECT_DIRECTORIES"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(tool_hunter, "resolve_hunter_client", lambda: tool_hunter.scripted_hunter_client()) + + +@pytest.fixture +def fixture(tmp_path): + root = tmp_path / "repo" + root.mkdir() + git = catalog.local_git(root) + git.run("init", "-q") + git.run("config", "user.name", "Fixture") + git.run("config", "user.email", "fixture@example.invalid") + heads = [] + for text in ('def search(q):\n return db.execute("select " + q)\n', "# Later checkout\n"): + (root / "app.py").write_text(text) + git.run("add", "app.py") + git.run("-c", "core.hooksPath=/dev/null", "commit", "-qm", "fixture") + heads.append(git.run("rev-parse", "HEAD").strip()) + change = dict( + id="fixture-positive", + repo="fixture/app", + repository_url="https://example.invalid/app", + kind="vulnerability", + base=heads[1], + head=heads[0], + family="injection", + cwe="CWE-89", + files=[{"path": "app.py", "head_lines": [[2, 2]]}], + function="search", + function_lines=[1, 2], + ) + manifest = tmp_path / "frozen.json" + manifest.write_text( + json.dumps({"changes": [change], "sha256": hashlib.sha256(json.dumps([change], sort_keys=True).encode()).hexdigest()}) + ) + repos = tmp_path / "repos.json" + repos.write_text(json.dumps({"fixture/app": str(root)})) + return git, change, manifest, repos, tmp_path / "out" + + +def flagged_client(): + return tool_hunter.ScriptedChatClient( + [ + tool_hunter.ChatResponse(tool_calls=(tool_hunter.ToolCall("read", "read_file", {"path": "app.py"}),)), + tool_hunter.ChatResponse( + content=json.dumps( + [ + { + "path": "app.py", + "line": 2, + "function_name": "search", + "title": "SQL injection (CWE-89)", + "rationale": "Input reaches a SQL interpreter", + "family": "sql-injection", + } + ] + ) + ), + ] + ) + + +def record_path(out, item_id): + return out / (hashlib.sha256(item_id.encode()).hexdigest() + ".json") + + +def test_positive_is_scoreable_and_checkout_is_removed(fixture, monkeypatch, capsys): + git, change, manifest, repos, out = fixture + client = flagged_client() + monkeypatch.setattr(tool_hunter, "resolve_hunter_client", lambda: client) + assert hunter_arm.main(["run", str(manifest), "--repos", str(repos), "--out", str(out), "--max-steps", "2"]) == 0 + (row,) = score.load_records(out) + (defect,) = row["push"]["admission"]["defects"] + assert score.catches(row) == ["alerts"] + assert defect["family"] == "injection" and defect["cwe"] == "CWE-89" + assert defect["locations"] == [["app.py", 2]] and defect["function"] == "search" + assert defect["source"] == "hunter" + assert row["status"] == "hunter_covered" and row["done"] + assert row["instrument"]["hook_bytes_read"] == len(git.blob(change["head"], "app.py").encode()) + assert score.coverage(row) != "instrument_failure" + assert row["input_digest"] == json.loads((out / "inputs.json").read_text())[0]["input_digest"] + assert client.calls[0]["model"] == resolve_models(load_config(dotenv=False))[0] + assert "fixture-positive app.py:" in capsys.readouterr().out + assert git.run("worktree", "list", "--porcelain").count("worktree ") == 1 + assert (git.path / "app.py").read_text() == "# Later checkout\n" + + +@pytest.mark.parametrize("failure", ["empty", "unreadable", "hunter", "missing", "bad_head", "bad_base"]) +def test_failed_input_never_emits_findings(fixture, monkeypatch, capsys, failure): + git, change, manifest, repos, out = fixture + if failure in {"empty", "missing"}: + change["files"] = [{"path": "empty.py" if failure == "empty" else "absent.py"}] + if failure == "empty": + (git.path / "empty.py").write_bytes(b"") + git.run("add", "empty.py") + git.run("-c", "core.hooksPath=/dev/null", "commit", "-qm", "empty") + change["head"] = git.run("rev-parse", "HEAD").strip() + if failure in {"bad_head", "bad_base"}: + change[failure.removeprefix("bad_")] = "f" * 40 + monkeypatch.setattr(hunter_arm, "load_slice", lambda *args: [change]) + if failure == "unreadable": + original = Path.read_bytes + + def unreadable(path): + if path.name == "app.py": + raise PermissionError("fixture") + return original(path) + + monkeypatch.setattr(Path, "read_bytes", unreadable) + + def hunter(*args, **kwargs): + if failure == "hunter": + raise RuntimeError("private model key must not be logged") + pytest.fail("hunter must not run on unread input") + + monkeypatch.setattr(tool_hunter, "run_tool_hunter", hunter) + assert hunter_arm.main(["run", str(manifest), "--repos", str(repos), "--out", str(out)]) != 0 + (row,) = score.load_records(out) + assert row["status"] == "instrument_failure" + assert list(score.findings(row)) == [] and row["instrument"]["failure"] + assert "instrument_failure" in capsys.readouterr().err + assert "private model key" not in record_path(out, change["id"]).read_text() + assert git.run("worktree", "list", "--porcelain").count("worktree ") == 1 + + +def test_missing_file_skipped_and_hotspots_are_unlabelled(fixture, monkeypatch): + _, change, manifest, repos, out = fixture + change["files"].append({"path": "removed.py", "head_lines": []}) + monkeypatch.setattr(hunter_arm, "load_slice", lambda *args: [change]) + + def hunter(root, hotspots, **kwargs): + (hotspot,) = hotspots + assert isinstance(hotspot, Hotspot) and hotspot.path == "app.py" + assert hotspot.function_name is None and "injection" not in hotspot.rationale + assert kwargs["max_steps"] == tool_hunter.DEFAULT_MAX_STEPS + return [] + + monkeypatch.setattr(tool_hunter, "run_tool_hunter", hunter) + assert hunter_arm.main(["run", str(manifest), "--repos", str(repos), "--out", str(out)]) == 0 + (row,) = score.load_records(out) + assert row["skipped"] == [{"path": "removed.py", "reason": "missing_at_head"}] + + +@pytest.mark.parametrize("with_inputs", [True, False]) +def test_resume_skips_done_without_client_or_checkout(fixture, monkeypatch, with_inputs): + _, change, manifest, repos, out = fixture + out.mkdir() + existing = {"id": change["id"], "done": True, "input_digest": "existing-digest", "status": "hunter_covered"} + record_path(out, change["id"]).write_text(json.dumps(existing)) + if with_inputs: + (out / "inputs.json").write_text(json.dumps([{"item_id": change["id"], "change": change, "input_digest": "existing-digest"}])) + + def forbidden(*args, **kwargs): + pytest.fail("completed change was executed again") + + monkeypatch.setattr(tool_hunter, "resolve_hunter_client", forbidden) + monkeypatch.setattr(hunter_arm, "local_git", forbidden) + assert hunter_arm.main(["run", str(manifest), "--repos", str(repos), "--out", str(out)]) == 0 + assert json.loads(record_path(out, change["id"]).read_text()) == existing + assert score.load_records(out)[0]["input_digest"] == "existing-digest" + + +def test_no_client_fails_clearly(fixture, monkeypatch, capsys): + _, _, manifest, repos, out = fixture + monkeypatch.setattr(tool_hunter, "resolve_hunter_client", lambda: None) + assert hunter_arm.main(["run", str(manifest), "--repos", str(repos), "--out", str(out)]) != 0 + assert "no hunter client" in capsys.readouterr().err.lower() + assert not list(out.glob("*.json")) + + +def test_classification_does_not_borrow_change_label(fixture): + _, change, _, _, _ = fixture + (finding,) = tool_hunter._findings_from_content( + Path("/tmp"), + [], + json.dumps( + [ + { + "path": "app.py", + "line": 2, + "title": "Path traversal", + "family": "path", + } + ] + ), + ) + defect = hunter_arm.finding_defect(finding) + assert defect["family"] == "path" and defect["cwe"] == "CWE-22" + row = {**change, "push": {"admission": {"defects": [defect]}}} + assert not score.catches(row) + unknown = hunter_arm.finding_defect(replace(finding, tags=[], title="Unclassified concern", rationale="")) + assert unknown["family"] == "unknown" and unknown["cwe"] is None + + +def write_arm(out, entries, records): + out.mkdir() + (out / "inputs.json").write_text(json.dumps(entries)) + for item_id, record in records.items(): + record_path(out, item_id).write_text(json.dumps(record)) + + +def test_combine_preserves_a_and_scores_b(fixture, tmp_path): + _, change, _, _, _ = fixture + changes = [{**change, "id": name} for name in ("both", "a-only", "b-only")] + entries_a = [{"item_id": "replay-" + c["id"], "change": c, "input_digest": "a"} for c in changes] + entries_b = [{"item_id": c["id"], "change": c, "input_digest": "b"} for c in changes] + defect = { + "defect_id": "hunter", + "family": "injection", + "cwe": "CWE-89", + "function": "search", + "locations": [["app.py", 2]], + "source": "hunter", + } + a = { + "status": "quick_only", + "input_digest": "a", + "done": True, + "instrument": {"sentinel": "keep"}, + "push": {"admission": {"defects": [], "dispositions": [{"sentinel": "keep"}]}, "quick_tier": [{"sentinel": "keep"}]}, + } + b = { + "status": "hunter_covered", + "input_digest": "b", + "done": True, + "push": {"admission": {"defects": [defect], "dispositions": []}, "quick_tier": []}, + } + replay_out, hunter_out, combined = tmp_path / "a", tmp_path / "b", tmp_path / "combined.json" + write_arm(replay_out, entries_a, {"replay-both": a, "replay-a-only": a}) + write_arm(hunter_out, entries_b, {"both": b, "b-only": b}) + assert hunter_arm.main(["combine", str(replay_out), str(hunter_out), "--out", str(combined)]) == 0 + both, a_only, b_only = score.load_records(combined) + assert score.catches(both) == ["alerts"] and score.catches(b_only) == ["alerts"] + assert both["instrument"] == a["instrument"] and both["status"] == a["status"] + assert both["arms"] == {"a": "quick_only", "b": "hunter_covered"} + assert both["push"]["quick_tier"] == a["push"]["quick_tier"] + assert both["push"]["admission"]["dispositions"] == [{"sentinel": "keep"}] + assert a_only == {**changes[1], **a, "arms": {"a": "quick_only", "b": "missing"}} + assert b_only["arms"] == {"a": "missing", "b": "hunter_covered"} + a["push"]["admission"]["defects"] = [{**defect, "defect_id": "a-defect", "source": "engine"}] + record_path(replay_out, "replay-both").write_text(json.dumps(a)) + hunter_arm.main(["combine", str(replay_out), str(hunter_out), "--out", str(combined)]) + defects = score.load_records(combined)[0]["push"]["admission"]["defects"] + assert len(defects) == 1 and defects[0]["source"] == "engine" + + +@pytest.mark.parametrize("arm", ["a", "b"]) +def test_combine_rejects_duplicate_ids(fixture, tmp_path, arm): + _, change, _, _, _ = fixture + entry = {"item_id": change["id"], "change": change, "input_digest": "digest"} + a, b = tmp_path / "a", tmp_path / "b" + write_arm(a, [entry, entry] if arm == "a" else [entry], {}) + write_arm(b, [entry, entry] if arm == "b" else [entry], {}) + with pytest.raises(ValueError, match="duplicate"): + hunter_arm.combine(a, b, tmp_path / "combined.json") + + +@pytest.mark.parametrize("directory_exists", [True, False]) +def test_combine_without_hunter_index_keeps_replay(fixture, tmp_path, directory_exists): + _, change, _, _, _ = fixture + replay_out, hunter_out = tmp_path / "a", tmp_path / "b" + entry = {"item_id": "replay-item", "change": change, "input_digest": "replay-digest"} + a = { + "status": "quick_only", + "input_digest": "replay-digest", + "done": True, + "instrument": {"sentinel": "keep"}, + "push": {"admission": {"defects": []}}, + } + write_arm(replay_out, [entry], {"replay-item": a}) + if directory_exists: + hunter_out.mkdir() + combined = tmp_path / "combined.json" + hunter_arm.combine(replay_out, hunter_out, combined) + assert score.load_records(combined) == [{**change, **a, "arms": {"a": "quick_only", "b": "missing"}}]