diff --git a/launchpad/review-agent/run_adjudication.py b/launchpad/review-agent/run_adjudication.py new file mode 100644 index 00000000000..67b0ecc6990 --- /dev/null +++ b/launchpad/review-agent/run_adjudication.py @@ -0,0 +1,364 @@ +"""The adjudication stage's CLI. Implements launchpad-26/buzz#118 STEP 3. + +Reads one #117 **merged document** on stdin, adjudicates every finding with an +**injected judge callable** -- defaulting to a stub that returns ``UNPROVEN`` +with a stated reason -- and prints one document on stdout, in the shape +ADJUDICATION.md defines and ``verdicts.validate`` checks. Demonstrable before a +single adjudication prompt is written (STEP 5). + +Three things this module must get right, each a way to lose data rather than a +missing feature: + +**Pass-through is byte-identical where it is pass-through.** ``pr``, +``merge_base_sha``, ``head_sha`` and the whole ``containment`` block leave +exactly as they arrived -- this module builds the output from a +``copy.deepcopy`` of the input and only ever mutates a finding dict's own six +new keys, never touching those four. The evidence inside a containment finding +is raw per FINDINGS.md's contract, and #119 escapes at render time; a stage +that re-serialises through anything lossy would publish an excerpt that no +longer matches what the author wrote. + +**The adjudicator never re-reads raw PR text.** CONTAINMENT.md forbids +re-reading raw PR text "to check for itself". This module makes no +``fetch.fetch_all`` call and no ``gh`` call for any of the seven surfaces -- +the only input it ever reads is the merged document already on stdin. A judge +injected here *may* read the repository at ``head_sha`` -- the file a finding +is anchored at -- because that is the change under review as **code**, the +artefact the finding claims a defect about; it is never the author's PR +title/body/comments/diff-as-prose, which is the surface CONTAINMENT.md +contains and this module must not touch a second time. This module's own +stub judge does neither: it reads only the finding dict it is given. + +**Anchor ``pr`` is normal, not an error.** A finding with ``file`` and +``line`` both null is structurally valid per FINDINGS.md, and this module +adjudicates it without raising. ``_location_description`` below is the one +place this module describes *where* a finding is anchored, and it branches on +``anchor`` first, before ever touching ``file``/``line`` -- never the reverse. + +**The input is validated before a single finding is adjudicated.** +``adjudicate()`` runs #117's own ``findings.validate`` against the input +document first and raises ``InputValidationError`` -- adjudicating nothing -- +when it fails; ``main()`` turns that into a non-zero exit with no document +printed at all. This is what keeps STEP 2's severity guarantee reachable: a +finding whose ``severity`` arrives out-of-ladder (an ``"Info"``, say -- #117's +own field name, before this stage ever renames it to ``reported_severity``) +fails ``findings.validate`` on that ground alone and never reaches this +module's per-finding logic, where "there is no legal value to preserve it as" +would otherwise be a real question with no good answer. + +Two ways to obtain a verdict, and no others: ``--judge stub`` (the default -- +``stub_judge`` below) and ``--replay `` (``make_replay_judge``, reading +STEP 9's future recorded judge outputs). This is also what keeps "choosing the +model" out of scope here, per #117's own framing and #118's issue: this module +never names one, and neither flag lets a caller supply one. + +Severity re-rating, the escalate-only guard, downgrade recording and dedupe +are STEP 6/7's job, layered on top of this module later. This step leaves +every finding's ``severity`` exactly equal to its ``reported_severity`` -- +the honest behaviour for a judge (the stub) that never rates anything -- and +``duplicate_of`` always null. + +``adjudication.notes`` is **deferred to STEP 6/7 too, and left empty here**, +which until now was the one hardcoded-empty field with no deferral stated +anywhere. The judge protocol below carries no ``notes`` key, so a judge that +returns one has it dropped. Recording it explicitly because the silence was +the defect: ADJUDICATION.md declares the field and ``verdicts.py`` carries +it, so a reader had every reason to assume the channel worked. + +**This deferral is in tension with ``adjudicator.md`` (#265), which +normatively tells a judge to "record it in ``adjudication.notes``".** While +that instruction ships against a protocol that discards the key, a judge's +only remaining outlet is ``verdict_evidence`` -- the field with no +structural guard. Whoever resolves this should either plumb ``notes`` +through the protocol here (symmetric with how a future ``severity_reason`` +would be) or amend ``adjudicator.md`` to say the channel is deferred. Not +decided in this step; named so it cannot be merged past unnoticed. +""" + +from __future__ import annotations + +import argparse +import copy +import json +import sys +from pathlib import Path +from typing import Callable + +import findings +import verdicts + +#: The judge protocol: ``judge(finding, input_document) -> dict`` with at +#: least ``{"verdict": ..., "verdict_evidence": ...}``. Anything else -- +#: a raised exception, a missing/illegal ``verdict``, empty +#: ``verdict_evidence`` -- is treated as unusable output and fails closed to +#: UNPROVEN, per ADJUDICATION.md's own default. +Judge = Callable[[dict, dict], dict] + + +class InputValidationError(ValueError): + """Raised by ``adjudicate()`` when the input document fails #117's own + ``findings.validate`` -- carries every violation, never just the first, + the same "report everything" discipline ``findings.validate`` and + ``verdicts.validate`` both already follow. + """ + + def __init__(self, violations: list[str]): + self.violations = violations + super().__init__("input document fails findings.validate: " + "; ".join(violations)) + + +def _location_description(finding: dict) -> str: + """Describe where a finding is anchored, branching on ``anchor`` FIRST -- + never assuming ``file``/``line`` exist. Anchor ``"pr"`` is a normal, valid + shape (file and line both null; see FINDINGS.md and ADJUDICATION.md), not + an error case, so it gets its own branch rather than falling through to a + file/line format string that would render ``"None:None"``. + """ + anchor = finding.get("anchor") + if anchor == "pr": + return "the whole pull request (no file or line anchor)" + if anchor == "file": + return f"{finding.get('file')}" + if anchor == "line": + return f"{finding.get('file')}:{finding.get('line')}" + return "a finding with an unrecognised anchor" + + +def stub_judge(finding: dict, document: dict) -> dict: + """The default judge (``--judge stub``). Establishes nothing about any + finding -- it exists to prove the harness end to end before a single + adjudication prompt is written (STEP 5). Every verdict it returns is + ``UNPROVEN`` with a stated reason, per ADJUDICATION.md's own default, + never ``CONFIRMED`` or ``REFUTED``. + """ + return { + "verdict": "UNPROVEN", + "verdict_evidence": ( + "stub judge: no adjudication was performed; " + f"{_location_description(finding)} was not examined." + ), + } + + +def make_replay_judge(replay_dir: Path) -> Judge: + """Build a judge that replays recorded judge outputs from ``replay_dir`` + (STEP 9's future recordings) instead of calling a live model. + + STEP 9 has not been built yet and ``replay_dir`` will not exist when this + runs in practice today -- this is a real, reachable code path per STEP 3's + own scope, not one exercised end to end here. The format it reads: every + ``*.json`` file directly under ``replay_dir`` is a JSON object mapping + ``finding_id`` -> ``{"verdict": ..., "verdict_evidence": ...}``. Every + file found is loaded and merged into one lookup; a ``finding_id`` with no + matching entry anywhere fails closed to ``UNPROVEN`` with a reason naming + the missing recording -- "no recording for this finding" is "cannot reach + the finding", the same failure family ``_run_judge_safely`` already covers, + not a crash. + """ + recordings: dict[str, dict] = {} + if replay_dir.is_dir(): + for path in sorted(replay_dir.glob("*.json")): + with path.open("r", encoding="utf-8") as fh: + data = json.load(fh) + if isinstance(data, dict): + recordings.update(data) + + def _replay(finding: dict, document: dict) -> dict: + finding_id = finding.get("finding_id") + recorded = recordings.get(finding_id) + if recorded is None: + return { + "verdict": "UNPROVEN", + "verdict_evidence": ( + f"replay: no recorded judge output for finding_id {finding_id!r} " + f"under {replay_dir}" + ), + } + return recorded + + return _replay + + +def _run_judge_safely(judge: Judge, finding: dict, input_document: dict) -> dict: + """Call ``judge`` and fail closed to ``UNPROVEN`` on anything unusable -- + a raised exception, a non-dict return, an illegal/missing ``verdict``, or + ``verdict_evidence`` that is not a string with at least one + non-whitespace character. ADJUDICATION.md's own words: "An adjudicator + that cannot reach the location, cannot parse the finding, times out, or + returns unusable output yields UNPROVEN with a reason." + + "Blank", not "empty", and the distinction is the whole point: a + truthiness test lets ``" "`` through, and a whitespace reason is + indistinguishable from no reason -- which is the case ADJUDICATION.md + says the requirement exists to exclude. The rule is + ``verdicts.is_nonempty_str``, imported rather than re-implemented, so + this producer guard and the contract check in ``verdicts.validate`` + cannot drift apart: they did exactly that, each admitting whitespace + because the other did. + """ + try: + result = judge(finding, input_document) + except Exception as exc: # noqa: BLE001 -- a judge's own crash is exactly + # the "cannot parse / times out" case above, and must fail closed + # rather than propagate and abort the whole run over one finding. + return { + "verdict": "UNPROVEN", + "verdict_evidence": ( + f"adjudicator raised {type(exc).__name__}: {exc}; failing closed " + "to UNPROVEN per ADJUDICATION.md's default." + ), + } + + verdict = result.get("verdict") if isinstance(result, dict) else None + evidence = result.get("verdict_evidence") if isinstance(result, dict) else None + if verdict not in verdicts.VERDICTS or not verdicts.is_nonempty_str(evidence): + return { + "verdict": "UNPROVEN", + "verdict_evidence": ( + "adjudicator returned unusable output (missing or illegal verdict, " + "or verdict_evidence that was blank, whitespace-only or not a " + "string); failing closed to UNPROVEN per ADJUDICATION.md's default." + ), + } + return {"verdict": verdict, "verdict_evidence": evidence} + + +def adjudicate(input_document: dict, judge: Judge) -> dict: + """Adjudicate every finding in ``input_document`` with ``judge`` and + return the adjudicated output document. Never mutates ``input_document``. + + Raises ``InputValidationError`` -- adjudicating nothing, calling ``judge`` + zero times -- when ``input_document`` fails #117's own + ``findings.validate``. This is the boundary STEP 1/STEP 2 call load-bearing: + a finding whose ``severity`` already arrived illegal is refused here, + wholesale, rather than reaching a per-finding fallback with no good answer. + + Pass-through fields (``pr``, ``merge_base_sha``, ``head_sha``, + ``containment``) are never touched: the output starts as a + ``copy.deepcopy`` of the input, and only a finding dict's own six new keys + are ever written. Severity re-rating, the escalate-only guard, downgrade + recording and dedupe are later steps' job -- every finding's ``severity`` + here is left exactly equal to its ``reported_severity``, and + ``duplicate_of`` is always null. + """ + violations = findings.validate(input_document) + if violations: + raise InputValidationError(violations) + + output_document = copy.deepcopy(input_document) + nonce = output_document.get("nonce") + + verdict_counts = {"CONFIRMED": 0, "REFUTED": 0, "UNPROVEN": 0} + findings_in = 0 + + for report in output_document.get("reports", []): + for finding in report.get("findings", []): + findings_in += 1 + result = _run_judge_safely(judge, finding, input_document) + reported_severity = finding["severity"] + finding["verdict"] = result["verdict"] + finding["verdict_evidence"] = result["verdict_evidence"] + finding["reported_severity"] = reported_severity + # No re-rating in this stage: `severity` (#117's own field, already + # present on `finding`) is left exactly as reported. STEP 6 adds + # the guard that lets a judge's re-rating land here safely. + finding["severity_reason"] = None + finding["duplicate_of"] = None + verdict_counts[result["verdict"]] += 1 + + # Nothing is dropped or invented at this stage, so the two counts are the + # same number by construction -- kept as two separate values (rather than + # one variable used twice) because that is the shape STEP 7's dedupe and a + # future drop/invent defect would change independently. + findings_out = findings_in + total_refutation = findings_in > 0 and verdict_counts["REFUTED"] == findings_in + + output_document["adjudication"] = verdicts.Adjudication( + schema_version=1, + verdict_counts=verdict_counts, + findings_in=findings_in, + findings_out=findings_out, + duplicate_groups=[], + downgrades=[], + total_refutation=total_refutation, + # Deferred to STEP 6/7, not an oversight -- see this module's docstring, + # including the unresolved tension with adjudicator.md (#265). The judge + # protocol carries no `notes` key, so nothing can populate this yet. + notes=[], + completion_marker=f"BUZZ-ADJUDICATION-COMPLETE:{nonce}", + ).as_dict() + + return output_document + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="run_adjudication.py", + description=( + "Adjudicate every finding in a #117 merged document (read on stdin) " + "and print the adjudicated document on stdout. See ADJUDICATION.md." + ), + ) + parser.add_argument( + "--judge", + choices=["stub"], + default="stub", + help="the built-in judge to use when --replay is not given (default: %(default)s)", + ) + parser.add_argument( + "--replay", + type=Path, + default=None, + metavar="DIR", + help=( + "replay recorded judge outputs from DIR (STEP 9) instead of calling " + "--judge. Takes precedence over --judge when both are given." + ), + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_arg_parser() + args = parser.parse_args(argv) + + raw = sys.stdin.read() + try: + input_document = json.loads(raw) + except json.JSONDecodeError as exc: + print(f"run_adjudication: malformed JSON on stdin: {exc}", file=sys.stderr) + return 1 + + # Valid JSON does not imply a JSON *object*: `[]`, `"x"`, `42` all parse. + # findings.validate assumes a dict (document.get(...), key not in document) + # and is not guaranteed to raise cleanly on other JSON types -- reachable + # directly from this CLI's untrusted stdin, so it is refused here, before + # that assumption is ever exercised, the same way malformed JSON is. + if not isinstance(input_document, dict): + print( + "run_adjudication: input must be a JSON object, got " + f"{type(input_document).__name__}", + file=sys.stderr, + ) + return 1 + + judge: Judge = make_replay_judge(args.replay) if args.replay is not None else stub_judge + + try: + output_document = adjudicate(input_document, judge) + except InputValidationError as exc: + for violation in exc.violations: + print(f"run_adjudication: {violation}", file=sys.stderr) + return 1 + + print(json.dumps(output_document)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/launchpad/review-agent/test_run_adjudication.py b/launchpad/review-agent/test_run_adjudication.py new file mode 100644 index 00000000000..c91ceb4636c --- /dev/null +++ b/launchpad/review-agent/test_run_adjudication.py @@ -0,0 +1,543 @@ +#!/usr/bin/env python3 +"""Controls for run_adjudication.py -- issue #118 STEP 3's CLI. + +Exercises every behaviour named in STEP 3's own done-when list in +launchpad/plans/2026-08-13-issue-118-adjudication.md: byte-identical +pass-through of ``pr``/``merge_base_sha``/``head_sha``/``containment``, +anchor ``pr`` adjudicated without raising, all three containment kinds +passed through with no verdict field added, malformed JSON and an +already-illegal input ``severity`` both refused before a single finding is +adjudicated (the injected judge's own call count proves the refusal happens +first), and no ``gh`` subprocess or HTTP client invoked during a stub run. + +Deliberately NOT exercised here (later steps' territory, per the plan): +the nonce three-way disagreement diagnosis and the ``stages`` manifest +(STEP 4), the escalate-only guard and downgrade recording for a judge that +actually re-rates severity (STEP 6), and dedupe (STEP 7). Every fixture +below either omits a re-rating entirely or only ever asserts that this +stage's own severity pass-through (``severity == reported_severity``, +always) holds. + +This file is scoped to `run_adjudication.py` alone and is deliberately not +wired into `run_controls.py`'s CONTROLS list -- that is STEP 10's control +suite over the full adjudication surface, not this module in isolation. + +Run: python3 -m unittest test_run_adjudication (from launchpad/review-agent/) + or: python3 test_run_adjudication.py +""" + +from __future__ import annotations + +import contextlib +import io +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import contain +import findings +import run_adjudication +import verdicts + +HERE = Path(__file__).parent +SCRIPT = HERE / "run_adjudication.py" + +NONCE = "deadbeefcafef00d" + + +def make_states(omit: str | None = None) -> dict: + states = {ep: "ok" for ep in contain.ENTRY_POINTS} + if omit is not None: + del states[omit] + return states + + +def make_raw_finding(**overrides) -> dict: + """A well-formed #117 finding dict -- BEFORE adjudication. None of + ADJUDICATION.md's six added fields are present, matching what #117 + actually emits. + """ + base = dict( + dimension="secrets-and-access", + severity="High", + anchor="line", + file="crates/buzz-relay/src/lib.rs", + line=42, + defect="hardcoded credential", + failure="credential leaks to logs", + entry_point=None, + evidence=None, + ) + base.update(overrides) + if "finding_id" not in overrides: + base["finding_id"] = findings.finding_id( + base["dimension"], + base["anchor"], + base["file"], + base["line"], + base["entry_point"], + base["defect"], + base["evidence"], + ) + return base + + +def make_report(dimension="secrets-and-access", nonce=NONCE, findings_list=None, **overrides) -> dict: + findings_list = findings_list if findings_list is not None else [] + report = dict( + schema_version=1, + dimension=dimension, + pr=42, + merge_base_sha="a" * 40, + head_sha="b" * 40, + status="complete", + outcome="findings" if findings_list else "clean", + error=None, + findings=findings_list, + findings_count=len(findings_list), + ) + report.update(overrides) + report["completion_marker"] = f"BUZZ-DIMENSION-COMPLETE:{dimension}:{nonce}" + return report + + +def make_document(reports=None, nonce=NONCE, states=None, containment_findings=None) -> dict: + reports = reports if reports is not None else [make_report(findings_list=[make_raw_finding()])] + return dict( + pr=42, + merge_base_sha="a" * 40, + head_sha="b" * 40, + reports=reports, + containment=dict( + findings=containment_findings if containment_findings is not None else [], + states=states if states is not None else make_states(), + ), + nonce=nonce, + ) + + +def make_containment_finding(kind: str, entry_point="pr_body", evidence="BUZZ-UNTRUSTED forged") -> dict: + return {"kind": kind, "entry_point": entry_point, "evidence": evidence, "severity": "Blocker"} + + +class CountingJudge: + """A judge that records how many times it was called, so a test can + assert it was never invoked -- the mechanism STEP 3's done-when uses to + prove the input-validation refusal happens BEFORE any judge runs, not + merely that the output happens to look refused. + """ + + def __init__(self, verdict="REFUTED", evidence="counting judge: forced verdict"): + self.calls: list[dict] = [] + self._verdict = verdict + self._evidence = evidence + + def __call__(self, finding: dict, document: dict) -> dict: + self.calls.append(finding) + return {"verdict": self._verdict, "verdict_evidence": self._evidence} + + @property + def call_count(self) -> int: + return len(self.calls) + + +class ByteIdenticalPassThroughTests(unittest.TestCase): + def test_pr_merge_base_head_and_containment_survive_untouched(self): + containment_findings = [make_containment_finding("delimiter_forge")] + input_doc = make_document(containment_findings=containment_findings) + output_doc = run_adjudication.adjudicate(input_doc, run_adjudication.stub_judge) + + for key in ("pr", "merge_base_sha", "head_sha", "containment"): + self.assertEqual( + json.dumps(output_doc[key], sort_keys=True), + json.dumps(input_doc[key], sort_keys=True), + f"{key} was not byte-identical", + ) + + def test_output_validates_against_both_contracts(self): + input_doc = make_document() + output_doc = run_adjudication.adjudicate(input_doc, run_adjudication.stub_judge) + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + self.assertEqual(findings.validate(output_doc), []) + + +class AnchorPrTests(unittest.TestCase): + def test_pr_anchored_finding_adjudicates_without_raising(self): + finding = make_raw_finding(anchor="pr", file=None, line=None) + input_doc = make_document(reports=[make_report(findings_list=[finding])]) + + output_doc = run_adjudication.adjudicate(input_doc, run_adjudication.stub_judge) + + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertTrue(adjudicated["verdict_evidence"]) + self.assertIn(adjudicated["verdict"], verdicts.VERDICTS) + + +class ContainmentPassThroughTests(unittest.TestCase): + def test_all_three_containment_kinds_emitted_unchanged(self): + kinds = ["delimiter_forge", "delimiter_lookalike", "injection_attempt"] + containment_findings = [make_containment_finding(k) for k in kinds] + input_doc = make_document(containment_findings=containment_findings) + + output_doc = run_adjudication.adjudicate(input_doc, run_adjudication.stub_judge) + + self.assertEqual(output_doc["containment"]["findings"], containment_findings) + for cf in output_doc["containment"]["findings"]: + self.assertEqual(cf["severity"], "Blocker") + self.assertNotIn("verdict", cf) + self.assertNotIn("verdict_evidence", cf) + + +class MalformedJsonTests(unittest.TestCase): + def _run_main_with_stdin(self, stdin_text: str) -> tuple[int, str, str]: + stdout, stderr = io.StringIO(), io.StringIO() + with mock.patch.object(sys, "stdin", io.StringIO(stdin_text)), \ + contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): + exit_code = run_adjudication.main([]) + return exit_code, stdout.getvalue(), stderr.getvalue() + + def test_malformed_json_exits_nonzero_and_prints_no_document(self): + exit_code, stdout, stderr = self._run_main_with_stdin("{not valid json") + self.assertNotEqual(exit_code, 0) + self.assertEqual(stdout, "") + self.assertTrue(stderr) + + def test_valid_json_non_object_exits_nonzero_and_prints_no_document(self): + # `[]`, a bare string, and a bare number are all VALID JSON but not + # objects -- json.loads succeeds on each, so this is not caught by + # the JSONDecodeError branch above. Refused cleanly before reaching + # findings.validate, which assumes a dict. + for stdin_text in ("[]", '"just a string"', "42"): + with self.subTest(stdin_text=stdin_text): + exit_code, stdout, stderr = self._run_main_with_stdin(stdin_text) + self.assertNotEqual(exit_code, 0) + self.assertEqual(stdout, "") + self.assertTrue(stderr) + + +class IllegalInputSeverityTests(unittest.TestCase): + """A fixture whose one finding arrives with an out-of-ladder `severity` -- + #117's own field name; `reported_severity` does not exist until this + stage produces it. This must be refused wholesale, before any judge runs. + """ + + def _illegal_document(self) -> dict: + finding = make_raw_finding(severity="Info") + return make_document(reports=[make_report(findings_list=[finding])]) + + def test_adjudicate_raises_before_calling_the_judge(self): + judge = CountingJudge(verdict="REFUTED") + input_doc = self._illegal_document() + + with self.assertRaises(run_adjudication.InputValidationError): + run_adjudication.adjudicate(input_doc, judge) + + self.assertEqual(judge.call_count, 0, judge.calls) + + def test_main_exits_nonzero_and_prints_no_document(self): + input_doc = self._illegal_document() + stdout, stderr = io.StringIO(), io.StringIO() + with mock.patch.object(sys, "stdin", io.StringIO(json.dumps(input_doc))), \ + contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): + exit_code = run_adjudication.main([]) + self.assertNotEqual(exit_code, 0) + self.assertEqual(stdout.getvalue(), "") + self.assertTrue(stderr.getvalue()) + + +class NoNetworkOrSubprocessTests(unittest.TestCase): + """A stub run must invoke neither a `gh` subprocess nor any HTTP client -- + this module never fetches PR surfaces itself (CONTAINMENT.md's + "never re-read raw PR text"). Patched to raise on any call, so a + regression that reaches for either fails this test rather than passing + silently because nothing was actually asserted about call counts. + """ + + def test_stub_run_makes_no_subprocess_or_http_call(self): + input_doc = make_document() + + def _boom(*args, **kwargs): + raise AssertionError("run_adjudication must not invoke subprocess/HTTP during a stub run") + + with mock.patch("subprocess.run", side_effect=_boom), \ + mock.patch("subprocess.Popen", side_effect=_boom), \ + mock.patch("urllib.request.urlopen", side_effect=_boom), \ + mock.patch("http.client.HTTPConnection.request", side_effect=_boom), \ + mock.patch("http.client.HTTPSConnection.request", side_effect=_boom): + output_doc = run_adjudication.adjudicate(input_doc, run_adjudication.stub_judge) + + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + + +class ReplayJudgeTests(unittest.TestCase): + """--replay's own code path, proven real (STEP 9's actual recordings do + not exist yet -- this is a hand-built stand-in, not an end-to-end + exercise of STEP 9's eventual format). + """ + + def test_replay_uses_the_recorded_verdict_when_present(self): + finding = make_raw_finding() + with tempfile.TemporaryDirectory() as tmp: + recording_path = Path(tmp) / "recorded.json" + recording_path.write_text( + json.dumps( + { + finding["finding_id"]: { + "verdict": "CONFIRMED", + "verdict_evidence": "replay: read the file at head_sha, credential present.", + } + } + ) + ) + judge = run_adjudication.make_replay_judge(Path(tmp)) + result = judge(finding, {}) + + self.assertEqual(result["verdict"], "CONFIRMED") + self.assertTrue(result["verdict_evidence"]) + + def test_replay_fails_closed_to_unproven_when_no_recording_matches(self): + finding = make_raw_finding() + with tempfile.TemporaryDirectory() as tmp: + judge = run_adjudication.make_replay_judge(Path(tmp)) + result = judge(finding, {}) + + self.assertEqual(result["verdict"], "UNPROVEN") + self.assertIn(finding["finding_id"], result["verdict_evidence"]) + + def test_replay_dir_that_does_not_exist_fails_closed_rather_than_raising(self): + judge = run_adjudication.make_replay_judge(Path("/nonexistent/replay/dir")) + finding = make_raw_finding() + result = judge(finding, {}) + self.assertEqual(result["verdict"], "UNPROVEN") + + +class JudgeFailsClosedTests(unittest.TestCase): + def test_judge_exception_fails_closed_to_unproven(self): + def _raising_judge(finding, document): + raise RuntimeError("boom") + + input_doc = make_document() + output_doc = run_adjudication.adjudicate(input_doc, _raising_judge) + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertEqual(adjudicated["verdict"], "UNPROVEN") + self.assertTrue(adjudicated["verdict_evidence"]) + + def test_judge_returning_illegal_verdict_fails_closed_to_unproven(self): + def _bad_judge(finding, document): + return {"verdict": "APPROVED", "verdict_evidence": "looks fine"} + + input_doc = make_document() + output_doc = run_adjudication.adjudicate(input_doc, _bad_judge) + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertEqual(adjudicated["verdict"], "UNPROVEN") + + def test_judge_returning_empty_evidence_fails_closed_to_unproven(self): + def _empty_evidence_judge(finding, document): + return {"verdict": "CONFIRMED", "verdict_evidence": ""} + + input_doc = make_document() + output_doc = run_adjudication.adjudicate(input_doc, _empty_evidence_judge) + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertEqual(adjudicated["verdict"], "UNPROVEN") + + def test_judge_returning_whitespace_evidence_fails_closed_to_unproven(self): + """A truthiness test is not the "unusable output" check this function + promises: ``not " "`` is False. ADJUDICATION.md's reason for + requiring evidence is that "an UNPROVEN with no reason is + indistinguishable from a stage that skipped the finding", and + whitespace IS no reason -- so a CONFIRMED Blocker could be published + with a blank justification and still validate clean. + """ + for blank in (" ", "\n", "\t", " \n ", "\xa0"): + with self.subTest(evidence=blank): + def _blank_evidence_judge(finding, document, _b=blank): + return {"verdict": "CONFIRMED", "verdict_evidence": _b} + + input_doc = make_document() + output_doc = run_adjudication.adjudicate(input_doc, _blank_evidence_judge) + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertEqual(adjudicated["verdict"], "UNPROVEN") + self.assertTrue(adjudicated["verdict_evidence"].strip()) + + def test_judge_returning_non_string_evidence_fails_closed_to_unproven(self): + """The sibling half: the guard had no type check, so any truthy value + passed. ``verdict_evidence: 42`` is not something a reader can act on. + """ + for wrong_type in (42, True, 0.5, ["x"], {"a": 1}): + with self.subTest(evidence=wrong_type): + def _wrong_type_judge(finding, document, _w=wrong_type): + return {"verdict": "CONFIRMED", "verdict_evidence": _w} + + input_doc = make_document() + output_doc = run_adjudication.adjudicate(input_doc, _wrong_type_judge) + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertEqual(adjudicated["verdict"], "UNPROVEN") + self.assertIsInstance(adjudicated["verdict_evidence"], str) + + def test_blank_evidence_output_still_satisfies_the_verdict_contract(self): + """The half that makes this load-bearing: before the fix, the blank + evidence reached the published document AND `verdicts.validate` + returned zero violations, because the contract check used the same + truthiness idiom. Both ends must now agree. + """ + def _blank_evidence_judge(finding, document): + return {"verdict": "CONFIRMED", "verdict_evidence": " \n "} + + input_doc = make_document() + output_doc = run_adjudication.adjudicate(input_doc, _blank_evidence_judge) + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + self.assertEqual(output_doc["reports"][0]["findings"][0]["verdict"], "UNPROVEN") + + +class ReplayBlankEvidenceTests(unittest.TestCase): + """`--replay` forwards a recording's contents unfiltered, so the blank + shape was reachable from an ordinary command line -- not only from an + injected judge. This drives the guard through the shipped flag. + """ + + @staticmethod + def _replay_run(finding, evidence): + """Drive `adjudicate` through a real replay recording. + + The recording format is a mapping ``finding_id -> {verdict, ...}``, + NOT a flat record -- get that wrong and the lookup misses, the judge + fails closed with "no recorded judge output", and a test asserting + UNPROVEN passes for entirely the wrong reason. + """ + with tempfile.TemporaryDirectory() as tmp: + recording = { + finding["finding_id"]: {"verdict": "CONFIRMED", "verdict_evidence": evidence} + } + (Path(tmp) / "rec.json").write_text(json.dumps(recording), encoding="utf-8") + judge = run_adjudication.make_replay_judge(Path(tmp)) + input_doc = make_document(reports=[make_report(findings_list=[finding])]) + output_doc = run_adjudication.adjudicate(input_doc, judge) + return input_doc, output_doc + + def test_a_matching_recording_is_actually_used(self): + """The control this class needs to be worth anything: prove the lookup + HITS, so a later UNPROVEN is the blank-evidence guard firing and not a + recording that was never found. + """ + finding = make_raw_finding() + input_doc, output_doc = self._replay_run(finding, "the credential is present at that line") + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertEqual(adjudicated["verdict"], "CONFIRMED") + self.assertNotIn("no recorded judge output", adjudicated["verdict_evidence"]) + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + + def test_replay_recording_with_whitespace_evidence_fails_closed(self): + finding = make_raw_finding() + input_doc, output_doc = self._replay_run(finding, " \n ") + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertEqual(adjudicated["verdict"], "UNPROVEN") + self.assertTrue(adjudicated["verdict_evidence"].strip()) + # Specifically the guard, not a missed lookup. + self.assertIn("unusable output", adjudicated["verdict_evidence"]) + self.assertNotIn("no recorded judge output", adjudicated["verdict_evidence"]) + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + + +class NotesDeferralTests(unittest.TestCase): + """`adjudication.notes` is hardcoded empty at this step and the judge + protocol does not honour a `notes` key. That is deliberate, but it was + undocumented -- and `adjudicator.md` (#265) normatively tells a judge to + record new observations there. These tests pin the CURRENT behaviour so + the deferral is asserted rather than assumed, and so whichever way #118 + resolves it, a test changes with the decision. + """ + + def test_a_judge_supplied_notes_key_is_not_carried(self): + def _noting_judge(finding, document): + return { + "verdict": "CONFIRMED", + "verdict_evidence": "the credential is present at that line", + "notes": ["a genuinely new defect noticed while adjudicating"], + } + + input_doc = make_document() + output_doc = run_adjudication.adjudicate(input_doc, _noting_judge) + # Documented deferral, not an accident: see this module's docstring. + self.assertEqual(output_doc["adjudication"]["notes"], []) + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + + def test_the_deferral_is_stated_in_the_module_docstring(self): + """The finding was that nothing said so. If the sentence goes, this + test goes red rather than the gap reopening silently. + """ + self.assertIn("notes", run_adjudication.__doc__) + self.assertRegex(run_adjudication.__doc__, r"notes.*(defer|STEP 6/7|left empty)") + + +class NoRerateInThisStepTests(unittest.TestCase): + """This step performs no re-rating at all (STEP 6's job): every finding's + `severity` equals its `reported_severity`, even when the injected judge + returns a verdict -- the judge protocol here carries no severity field. + """ + + def test_severity_always_equals_reported_severity(self): + finding = make_raw_finding(severity="Blocker") + input_doc = make_document(reports=[make_report(findings_list=[finding])]) + + output_doc = run_adjudication.adjudicate(input_doc, run_adjudication.stub_judge) + + adjudicated = output_doc["reports"][0]["findings"][0] + self.assertEqual(adjudicated["reported_severity"], "Blocker") + self.assertEqual(adjudicated["severity"], "Blocker") + self.assertIsNone(adjudicated["severity_reason"]) + self.assertIsNone(adjudicated["duplicate_of"]) + + +class SubprocessInvocationTests(unittest.TestCase): + """The literal CLI form STEP 3's done-when names: + `python3 run_adjudication.py < fixture.json`. + """ + + def test_real_process_stub_run_exits_zero_and_prints_a_valid_document(self): + input_doc = make_document() + proc = subprocess.run( + [sys.executable, str(SCRIPT)], + input=json.dumps(input_doc), + capture_output=True, + text=True, + cwd=str(HERE), + timeout=30, + ) + self.assertEqual(proc.returncode, 0, proc.stderr) + output_doc = json.loads(proc.stdout) + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + self.assertEqual(findings.validate(output_doc), []) + + def test_real_process_malformed_json_exits_nonzero_no_stdout(self): + proc = subprocess.run( + [sys.executable, str(SCRIPT)], + input="{not json", + capture_output=True, + text=True, + cwd=str(HERE), + timeout=30, + ) + self.assertNotEqual(proc.returncode, 0) + self.assertEqual(proc.stdout, "") + + def test_real_process_illegal_severity_exits_nonzero_no_stdout(self): + finding = make_raw_finding(severity="Info") + input_doc = make_document(reports=[make_report(findings_list=[finding])]) + proc = subprocess.run( + [sys.executable, str(SCRIPT)], + input=json.dumps(input_doc), + capture_output=True, + text=True, + cwd=str(HERE), + timeout=30, + ) + self.assertNotEqual(proc.returncode, 0) + self.assertEqual(proc.stdout, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/launchpad/review-agent/test_verdicts.py b/launchpad/review-agent/test_verdicts.py new file mode 100644 index 00000000000..9c4a09f7a8e --- /dev/null +++ b/launchpad/review-agent/test_verdicts.py @@ -0,0 +1,554 @@ +#!/usr/bin/env python3 +"""Controls for verdicts.py -- issue #118 STEP 2's verdict contract in code. + +Exercises every behaviour named in STEP 2's own done-when list in +launchpad/plans/2026-08-13-issue-118-adjudication.md, plus the shared-object +guarantee (`verdicts.SEVERITY_ORDER is review.SEVERITY_ORDER`) and the +re-run of #117's own `findings.validate` against the output document. + +This file is scoped to `verdicts.py` alone and is deliberately not wired into +`run_controls.py`'s CONTROLS list -- that is STEP 10's control suite, over the +full adjudication surface (`run_adjudication.py` included), not this module in +isolation. `python3 -m unittest test_verdicts` (or `python3 test_verdicts.py`) +run directly is a real, path-invoked suite in its own right. + +Run: python3 -m unittest test_verdicts (from launchpad/review-agent/) + or: python3 test_verdicts.py +""" + +from __future__ import annotations + +import copy +import unittest + +import contain +import findings +import review +import verdicts + +NONCE = "deadbeefcafef00d" + + +def make_states(omit: str | None = None): + states = {ep: "ok" for ep in contain.ENTRY_POINTS} + if omit is not None: + del states[omit] + return states + + +def make_raw_finding(**overrides): + """A well-formed #117 finding dict -- BEFORE adjudication. None of the six + ADJUDICATION.md fields are present, matching what #117 actually emits. + """ + base = dict( + dimension="secrets-and-access", + severity="High", + anchor="line", + file="crates/buzz-relay/src/lib.rs", + line=42, + defect="hardcoded credential", + failure="credential leaks to logs", + entry_point=None, + evidence=None, + ) + base.update(overrides) + if "finding_id" not in overrides: + base["finding_id"] = findings.finding_id( + base["dimension"], + base["anchor"], + base["file"], + base["line"], + base["entry_point"], + base["defect"], + base["evidence"], + ) + return base + + +def make_finding(**overrides): + """A well-formed, ALREADY-ADJUDICATED finding dict: #117's ten fields plus + the six ADJUDICATION.md adds. ``finding_id`` is recomputed from the seven + hash inputs, which never include ``severity`` or any of the six added + fields -- so a re-rating never invalidates the id. + """ + base = dict( + dimension="secrets-and-access", + severity="High", + anchor="line", + file="crates/buzz-relay/src/lib.rs", + line=42, + defect="hardcoded credential", + failure="credential leaks to logs", + entry_point=None, + evidence=None, + verdict="CONFIRMED", + verdict_evidence="Read the file at head_sha; the credential is present at that line.", + reported_severity="High", + severity_reason=None, + duplicate_of=None, + ) + base.update(overrides) + if "finding_id" not in overrides: + base["finding_id"] = findings.finding_id( + base["dimension"], + base["anchor"], + base["file"], + base["line"], + base["entry_point"], + base["defect"], + base["evidence"], + ) + return base + + +def make_report(dimension="secrets-and-access", nonce=NONCE, findings_list=None, **overrides): + findings_list = findings_list if findings_list is not None else [] + report = dict( + schema_version=1, + dimension=dimension, + pr=42, + merge_base_sha="a" * 40, + head_sha="b" * 40, + status="complete", + outcome="findings" if findings_list else "clean", + error=None, + findings=findings_list, + findings_count=len(findings_list), + ) + report.update(overrides) + report["completion_marker"] = f"BUZZ-DIMENSION-COMPLETE:{dimension}:{nonce}" + return report + + +def make_document(reports=None, nonce=NONCE, states=None, containment_findings=None, adjudication=None): + reports = reports if reports is not None else [make_report(findings_list=[make_raw_finding()])] + doc = dict( + pr=42, + merge_base_sha="a" * 40, + head_sha="b" * 40, + reports=reports, + containment=dict( + findings=containment_findings if containment_findings is not None else [], + states=states if states is not None else make_states(), + ), + nonce=nonce, + ) + if adjudication is not None: + doc["adjudication"] = adjudication + return doc + + +def make_adjudication(findings_list, nonce=NONCE, **overrides): + """A well-formed `adjudication` block for exactly ``findings_list``.""" + counts = {"CONFIRMED": 0, "REFUTED": 0, "UNPROVEN": 0} + for f in findings_list: + counts[f["verdict"]] = counts.get(f["verdict"], 0) + 1 + block = dict( + schema_version=1, + verdict_counts=counts, + findings_in=len(findings_list), + findings_out=len(findings_list), + duplicate_groups=[], + downgrades=[], + total_refutation=bool(findings_list) and all(f["verdict"] == "REFUTED" for f in findings_list), + notes=[], + ) + block.update(overrides) + block["completion_marker"] = f"BUZZ-ADJUDICATION-COMPLETE:{nonce}" + return block + + +def make_well_formed_pair(nonce=NONCE): + """A matching (input_document, output_document) pair: one finding, no + re-rating, no dedupe, nothing refuted -- the baseline every mutation test + below starts from and breaks in exactly one way. + """ + raw = make_raw_finding() + verdicted = make_finding() + assert raw["finding_id"] == verdicted["finding_id"] + + input_doc = make_document( + reports=[make_report(findings_list=[raw], nonce=nonce)], + nonce=nonce, + ) + output_doc = make_document( + reports=[make_report(findings_list=[verdicted], nonce=nonce)], + nonce=nonce, + adjudication=make_adjudication([verdicted], nonce=nonce), + ) + return input_doc, output_doc + + +class SharedSeverityOrderTests(unittest.TestCase): + def test_severity_order_is_the_same_object_as_review(self): + self.assertIs(verdicts.SEVERITY_ORDER, review.SEVERITY_ORDER) + + +class WellFormedPairTests(unittest.TestCase): + def test_well_formed_pair_validates_clean(self): + input_doc, output_doc = make_well_formed_pair() + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + + def test_findings_validate_still_accepts_the_output(self): + # STEP 2's own requirement: verdicts.validate re-runs findings.validate + # over the output, but the output document must also pass it stood alone. + _, output_doc = make_well_formed_pair() + self.assertEqual(findings.validate(output_doc), []) + + def test_four_independent_violations_surface_at_once(self): + input_doc, output_doc = make_well_formed_pair() + # 1: verdict not one of the three legal values. + output_doc["reports"][0]["findings"][0]["verdict"] = "APPROVED" + # 2: verdict_evidence empty. + output_doc["reports"][0]["findings"][0]["verdict_evidence"] = "" + # 3: reported_severity out of the ladder. + output_doc["reports"][0]["findings"][0]["reported_severity"] = "Info" + # 4: a forbidden key present in the document. + output_doc["adjudication"]["approved"] = None + violations = verdicts.validate(input_doc, output_doc) + self.assertGreaterEqual(len(violations), 4, violations) + + +class FindingIdSetTests(unittest.TestCase): + """The two cases a mere findings_out == findings_in COUNT comparison + cannot tell apart from a well-formed document, because both leave that + equality true. + """ + + def test_dropped_finding_id_is_named(self): + raw_a = make_raw_finding(defect="defect A") + raw_b = make_raw_finding(defect="defect B") + verdicted_a = make_finding(defect="defect A", finding_id=raw_a["finding_id"]) + + input_doc = make_document( + reports=[make_report(findings_list=[raw_a, raw_b])], + ) + # Output drops raw_b's finding_id entirely, and its own adjudication + # block is internally self-consistent (findings_out == findings_in == 1) + # -- a count-only check would see nothing wrong here. + output_doc = make_document( + reports=[make_report(findings_list=[verdicted_a])], + adjudication=make_adjudication([verdicted_a]), + ) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue( + any(raw_b["finding_id"] in v and "missing from output" in v for v in violations), + violations, + ) + + def test_invented_finding_id_is_named(self): + raw_a = make_raw_finding(defect="defect A") + verdicted_a = make_finding(defect="defect A", finding_id=raw_a["finding_id"]) + invented = make_finding(defect="defect NEVER ON INPUT", finding_id="0" * 16) + + input_doc = make_document( + reports=[make_report(findings_list=[raw_a])], + ) + # Output carries an extra finding absent from input, and its own + # adjudication block is internally self-consistent (findings_out == + # findings_in == 2) -- a count-only check would see nothing wrong here + # either. + output_doc = make_document( + reports=[make_report(findings_list=[verdicted_a, invented])], + adjudication=make_adjudication([verdicted_a, invented]), + ) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue( + any("0000000000000000" in v and "invented" in v for v in violations), + violations, + ) + + +class SeverityLadderTests(unittest.TestCase): + def test_severity_out_of_ladder_is_rejected(self): + input_doc, output_doc = make_well_formed_pair() + output_doc["reports"][0]["findings"][0]["severity"] = "Info" + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("severity" in v and "Info" in v for v in violations), violations) + + def test_reported_severity_out_of_ladder_is_rejected(self): + # The case a guard watching only re-ratings never sees: the finding + # ARRIVED broken and the judge never touched severity at all. + input_doc, output_doc = make_well_formed_pair() + output_doc["reports"][0]["findings"][0]["reported_severity"] = "Info" + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue( + any("reported_severity" in v and "Info" in v for v in violations), violations + ) + + +class DowngradeTests(unittest.TestCase): + def test_a_real_fall_with_no_downgrade_entry_is_rejected(self): + raw = make_raw_finding(severity="Blocker") + verdicted = make_finding(severity="Low", reported_severity="Blocker", severity_reason="re-rated down") + input_doc = make_document(reports=[make_report(findings_list=[raw])]) + output_doc = make_document( + reports=[make_report(findings_list=[verdicted])], + adjudication=make_adjudication([verdicted]), # downgrades left empty + ) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue( + any("severity fell" in v and "not" in v and "recorded" in v for v in violations), + violations, + ) + + def test_a_downgrade_entry_with_no_real_fall_is_rejected(self): + input_doc, output_doc = make_well_formed_pair() # severity == reported_severity, no fall + fid = output_doc["reports"][0]["findings"][0]["finding_id"] + output_doc["adjudication"]["downgrades"] = [ + {"finding_id": fid, "from": "High", "to": "High", "reason": "not actually a fall"} + ] + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue( + any("did not fall" in v or "did not actually fall" in v for v in violations) + or any("severity did not fall" in v for v in violations), + violations, + ) + + +class TotalRefutationTests(unittest.TestCase): + def test_all_refuted_with_flag_false_is_rejected(self): + raw = make_raw_finding() + verdicted = make_finding(verdict="REFUTED", verdict_evidence="checked and found absent") + input_doc = make_document(reports=[make_report(findings_list=[raw])]) + output_doc = make_document( + reports=[make_report(findings_list=[verdicted])], + adjudication=make_adjudication([verdicted], total_refutation=False), + ) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("total_refutation" in v for v in violations), violations) + + def test_mixed_verdicts_with_flag_true_is_rejected(self): + raw_a = make_raw_finding(defect="A") + raw_b = make_raw_finding(defect="B") + confirmed = make_finding(defect="A", finding_id=raw_a["finding_id"], verdict="CONFIRMED") + refuted = make_finding( + defect="B", + finding_id=raw_b["finding_id"], + verdict="REFUTED", + verdict_evidence="checked and found absent", + ) + input_doc = make_document(reports=[make_report(findings_list=[raw_a, raw_b])]) + output_doc = make_document( + reports=[make_report(findings_list=[confirmed, refuted])], + adjudication=make_adjudication([confirmed, refuted], total_refutation=True), + ) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("total_refutation" in v for v in violations), violations) + + +class DedupeTests(unittest.TestCase): + def test_group_naming_a_finding_whose_duplicate_of_is_null_is_rejected(self): + raw_a = make_raw_finding(defect="A") + raw_b = make_raw_finding(defect="B") + survivor = make_finding(defect="A", finding_id=raw_a["finding_id"]) + # duplicate_of left null (default), but a group claims it as a duplicate. + not_pointing_back = make_finding(defect="B", finding_id=raw_b["finding_id"], duplicate_of=None) + input_doc = make_document(reports=[make_report(findings_list=[raw_a, raw_b])]) + adjudication = make_adjudication( + [survivor, not_pointing_back], + duplicate_groups=[{"survivor": survivor["finding_id"], "duplicates": [not_pointing_back["finding_id"]]}], + ) + output_doc = make_document( + reports=[make_report(findings_list=[survivor, not_pointing_back])], + adjudication=adjudication, + ) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("does not point back" in v for v in violations), violations) + + def test_finding_pointing_at_a_survivor_listed_in_no_group_is_rejected(self): + raw_a = make_raw_finding(defect="A") + raw_b = make_raw_finding(defect="B") + survivor = make_finding(defect="A", finding_id=raw_a["finding_id"]) + orphaned_duplicate = make_finding( + defect="B", finding_id=raw_b["finding_id"], duplicate_of=survivor["finding_id"] + ) + input_doc = make_document(reports=[make_report(findings_list=[raw_a, raw_b])]) + # No duplicate_groups entry at all names this pairing. + output_doc = make_document( + reports=[make_report(findings_list=[survivor, orphaned_duplicate])], + adjudication=make_adjudication([survivor, orphaned_duplicate]), + ) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue( + any("no" in v and "duplicate_groups" in v and "lists this finding" in v for v in violations), + violations, + ) + + +class VerdictEvidenceTests(unittest.TestCase): + def test_refuted_with_empty_verdict_evidence_is_rejected(self): + input_doc, output_doc = make_well_formed_pair() + output_doc["reports"][0]["findings"][0]["verdict"] = "REFUTED" + output_doc["reports"][0]["findings"][0]["verdict_evidence"] = "" + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("verdict_evidence" in v for v in violations), violations) + + def test_whitespace_only_verdict_evidence_is_rejected(self): + """ADJUDICATION.md requires evidence "non-empty" because "an UNPROVEN + with no reason is indistinguishable from a stage that skipped the + finding". Whitespace IS no reason, so a truthiness test is the wrong + check -- ``not " "`` is False and the value sails through. + """ + # "\xa0" is a non-breaking space: whitespace to str.strip(), but not + # something an eye catches in a diff. Kept deliberately, written as + # an escape so it is visible in the source rather than invisible. + for blank in (" ", "\n", "\t", " \n ", "\xa0"): + with self.subTest(evidence=blank): + input_doc, output_doc = make_well_formed_pair() + output_doc["reports"][0]["findings"][0]["verdict_evidence"] = blank + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("verdict_evidence" in v for v in violations), violations) + + def test_non_string_verdict_evidence_is_rejected(self): + """The sibling half: the guard had no type check at all, so every + truthy value passed. ``verdict_evidence: 42`` is not defensible under + any reading of the contract, and ``verdict`` one field above already + does ``not isinstance(verdict, str)``. + """ + for wrong_type in (42, True, 0.5, ["x"], {"a": 1}): + with self.subTest(evidence=wrong_type): + input_doc, output_doc = make_well_formed_pair() + output_doc["reports"][0]["findings"][0]["verdict_evidence"] = wrong_type + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("verdict_evidence" in v for v in violations), violations) + + +class SeverityReasonTests(unittest.TestCase): + """``severity_reason`` is required when severity differs from + reported_severity, and it was guarded by the same truthiness test as + ``verdict_evidence`` -- so a re-rating could be justified by whitespace. + """ + + def _falling_pair(self, reason): + raw = make_raw_finding() + fell = make_finding( + finding_id=raw["finding_id"], severity="Medium", severity_reason=reason + ) + input_doc = make_document(reports=[make_report(findings_list=[raw])]) + output_doc = make_document( + reports=[make_report(findings_list=[fell])], + adjudication=make_adjudication( + [fell], + downgrades=[ + { + "finding_id": fell["finding_id"], + "from": "High", + "to": "Medium", + "reason": reason, + } + ], + ), + ) + return input_doc, output_doc + + def test_whitespace_only_severity_reason_is_rejected(self): + for blank in (" ", "\n", "\t"): + with self.subTest(reason=blank): + input_doc, output_doc = self._falling_pair(blank) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("severity_reason" in v for v in violations), violations) + + def test_non_string_severity_reason_is_rejected(self): + for wrong_type in (42, True, ["because"]): + with self.subTest(reason=wrong_type): + input_doc, output_doc = self._falling_pair(wrong_type) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("severity_reason" in v for v in violations), violations) + + def test_a_real_reason_still_validates_clean(self): + input_doc, output_doc = self._falling_pair("the judge re-rated it after reading the guard") + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + + +class CountFieldTypeTests(unittest.TestCase): + """``bool`` subclasses ``int`` in Python, so ``isinstance(True, int)`` is + True and every count field accepted a boolean. ``total_refutation`` two + checks away uses a strict ``isinstance(..., bool)``, which is the file + showing the strictness was intended. + """ + + def test_boolean_findings_in_and_out_are_rejected(self): + input_doc, output_doc = make_well_formed_pair() + output_doc["adjudication"]["findings_in"] = True + output_doc["adjudication"]["findings_out"] = True + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue( + any("findings_in" in v and "integer" in v for v in violations), violations + ) + + def test_boolean_report_findings_count_is_rejected(self): + """The one the equality check cannot catch on its own: with a single + finding, ``True == 1 == len(findings)``, so #117's own count comparison + stays silent and only a type check sees it. + """ + input_doc, output_doc = make_well_formed_pair() + output_doc["reports"][0]["findings_count"] = True + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("findings_count" in v for v in violations), violations) + + def test_honest_integer_counts_still_validate_clean(self): + input_doc, output_doc = make_well_formed_pair() + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + + +class DuplicateGroupSurvivorTests(unittest.TestCase): + """A group's ``survivor`` was only ever validated through a member pointing + back at it, so a group with an empty ``duplicates`` list never had its + survivor checked at all. Validator-only today -- the producer gates on + ``len(candidate_ids) < 2`` -- but this is the check STEP 10's malformed-field + controls will look for. + """ + + def _group_pair(self, group): + input_doc, output_doc = make_well_formed_pair() + output_doc["adjudication"]["duplicate_groups"] = [group] + return input_doc, output_doc + + def test_empty_duplicates_with_a_survivor_that_is_no_finding_is_rejected(self): + input_doc, output_doc = self._group_pair({"survivor": "no-such-id", "duplicates": []}) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("survivor" in v for v in violations), violations) + + def test_empty_duplicates_with_a_missing_survivor_key_is_rejected(self): + input_doc, output_doc = self._group_pair({"duplicates": []}) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("survivor" in v for v in violations), violations) + + def test_empty_duplicates_with_a_non_string_survivor_is_rejected(self): + input_doc, output_doc = self._group_pair({"survivor": 12345, "duplicates": []}) + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("survivor" in v for v in violations), violations) + + def test_a_real_survivor_with_empty_duplicates_still_validates_clean(self): + _, baseline = make_well_formed_pair() + real_id = baseline["reports"][0]["findings"][0]["finding_id"] + input_doc, output_doc = self._group_pair({"survivor": real_id, "duplicates": []}) + self.assertEqual(verdicts.validate(input_doc, output_doc), []) + + +class FindingsValidateReRunTests(unittest.TestCase): + def test_output_breaking_findings_own_contract_is_caught(self): + # A document that leaves this stage still satisfies the contract it + # arrived under -- here the top-level nonce is emptied, which #117's + # own findings.validate rejects on its own terms. + input_doc, output_doc = make_well_formed_pair() + output_doc["nonce"] = "" + violations = verdicts.validate(input_doc, output_doc) + self.assertTrue(any("nonce" in v for v in violations), violations) + # And findings.validate, run directly, agrees. + self.assertTrue(any("nonce" in v for v in findings.validate(output_doc))) + + +class DeepCopyIsolationTests(unittest.TestCase): + def test_validate_does_not_mutate_either_document(self): + input_doc, output_doc = make_well_formed_pair() + input_before = copy.deepcopy(input_doc) + output_before = copy.deepcopy(output_doc) + verdicts.validate(input_doc, output_doc) + self.assertEqual(input_doc, input_before) + self.assertEqual(output_doc, output_before) + + +if __name__ == "__main__": + unittest.main() diff --git a/launchpad/review-agent/verdicts.py b/launchpad/review-agent/verdicts.py new file mode 100644 index 00000000000..ee9c60b1bb5 --- /dev/null +++ b/launchpad/review-agent/verdicts.py @@ -0,0 +1,552 @@ +"""The adjudication stage's verdict contract, in code. + +Implements launchpad-26/buzz#118 (STEP 2). The normative contract is +ADJUDICATION.md in this directory; this module is that document made +executable. Where the two disagree, the document wins and this file is the +bug. + +Pure dataclasses and validation over an already-produced pair of documents: no +subprocess, no network, no model call. STEP 3's `run_adjudication.py` is the +process that reads a #117 merged document, adjudicates every finding with an +injected judge, and writes the adjudicated document back out; this module only +describes the shape of the six added finding fields and the nine +`adjudication` block keys, and checks the movement between the two documents +(`validate`). + +Severity is imported from `review.py`, never redeclared here -- ADJUDICATION.md +is explicit that a second copy of the four-value ladder drifts, the same +reason FINDINGS.md gives and #117's `findings.py` already follows. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from review import SEVERITY_ORDER + +#: The three legal verdicts. ADJUDICATION.md § The verdict: "exactly one... +#: from CONFIRMED | REFUTED | UNPROVEN". +VERDICTS = frozenset({"CONFIRMED", "REFUTED", "UNPROVEN"}) + +#: The machine-checkable half of "escalate, never approve" (prohibition 1): +#: no key anywhere in the document may carry an approval, a merge +#: recommendation, or a pass. +_FORBIDDEN_KEYS = frozenset({"approved", "mergeable", "merge_recommendation"}) + + +def is_nonempty_str(value: object) -> bool: + """A string carrying at least one non-whitespace character. + + ADJUDICATION.md requires several fields "non-empty", and § The verdict + gives the reason: "An ``UNPROVEN`` with no reason is indistinguishable + from a stage that skipped the finding." Whitespace IS no reason by that + standard, so a bare truthiness test is the wrong check -- ``not " "`` + is False and the value sails through. The type half matters for the same + reason: truthiness also admits ``42``, ``True`` and ``["x"]``, none of + which a reader can act on. + """ + return isinstance(value, str) and bool(value.strip()) + + +def is_int(value: object) -> bool: + """An integer that is not a boolean. + + ``bool`` subclasses ``int`` in Python, so ``isinstance(True, int)`` is + True and a boolean satisfies a naive count check. That matters most where + the count is 1: ``True == 1 == len(findings)``, so the equality + comparison downstream stays silent and only a type check sees it. + ``total_refutation`` is guarded with a strict ``isinstance(..., bool)`` + below, which is this module already showing the strictness it intends. + """ + return isinstance(value, int) and not isinstance(value, bool) + + +@dataclass +class Verdict: + """The six fields ADJUDICATION.md adds on top of FINDINGS.md's ten. + + A plain data carrier, like `findings.Finding` -- it does not validate + itself. Structural checking of a finding already in dict form is + `validate`'s job. + """ + + verdict: str + verdict_evidence: str + reported_severity: str + severity: str + severity_reason: str | None + duplicate_of: str | None + + def as_dict(self) -> dict: + return { + "verdict": self.verdict, + "verdict_evidence": self.verdict_evidence, + "reported_severity": self.reported_severity, + "severity": self.severity, + "severity_reason": self.severity_reason, + "duplicate_of": self.duplicate_of, + } + + @classmethod + def from_dict(cls, d: dict) -> Verdict: + return cls( + verdict=d["verdict"], + verdict_evidence=d["verdict_evidence"], + reported_severity=d["reported_severity"], + severity=d["severity"], + severity_reason=d.get("severity_reason"), + duplicate_of=d.get("duplicate_of"), + ) + + +@dataclass +class Adjudication: + """The top-level `adjudication` block. Nine keys, per ADJUDICATION.md § The + `adjudication` block. + + Field order matches the contract's table, and `completion_marker` -- + last both here and in `as_dict()` -- stays last on purpose: it is the + marker a truncated emit must lose before it loses anything else, the same + reason `findings.Report` keeps its own `completion_marker` last. + """ + + schema_version: int + verdict_counts: dict + findings_in: int + findings_out: int + duplicate_groups: list = field(default_factory=list) + downgrades: list = field(default_factory=list) + total_refutation: bool = False + notes: list = field(default_factory=list) + completion_marker: str = "" + + def as_dict(self) -> dict: + return { + "schema_version": self.schema_version, + "verdict_counts": self.verdict_counts, + "findings_in": self.findings_in, + "findings_out": self.findings_out, + "duplicate_groups": self.duplicate_groups, + "downgrades": self.downgrades, + "total_refutation": self.total_refutation, + "notes": self.notes, + "completion_marker": self.completion_marker, + } + + @classmethod + def from_dict(cls, d: dict) -> Adjudication: + return cls( + schema_version=d["schema_version"], + verdict_counts=d["verdict_counts"], + findings_in=d["findings_in"], + findings_out=d["findings_out"], + duplicate_groups=d.get("duplicate_groups", []), + downgrades=d.get("downgrades", []), + total_refutation=d.get("total_refutation", False), + notes=d.get("notes", []), + completion_marker=d["completion_marker"], + ) + + +def _iter_findings(document: dict): + """Yield ``(label, index, finding)`` for every finding in every report. + + ``label`` matches FINDINGS.md's own labelling convention (``report + {dimension!r}``, falling back to a positional label when a report has no + ``dimension``) so a violation string from this module reads consistently + next to one from `findings.validate`. + + Every container assumed here (``reports``, a report, its ``findings``) is + type-checked before being treated as its expected shape, same as + `findings.validate` -- a malformed document is exactly the input this + module exists to describe with a violation string, never the input it + crashes on. + """ + reports = document.get("reports") + if not isinstance(reports, list): + return + for report_index, report in enumerate(reports): + if not isinstance(report, dict): + continue + dimension = report.get("dimension") + label = f"report {dimension!r}" if dimension is not None else f"document.reports[{report_index}]" + findings_raw = report.get("findings") + if not isinstance(findings_raw, list): + continue + for finding_index, finding in enumerate(findings_raw): + yield label, finding_index, finding + + +def _finding_ids(document: dict) -> set[str]: + """The set of ``finding_id`` values across every report's findings. + + Extracted from the document directly, never from a count -- a count + cannot tell a dropped id from an invented one, or notice a swap that + leaves the count unchanged. This is what `validate`'s finding_id + SET-equality check needs, and the only reason `validate` takes both + documents rather than one. + """ + ids: set[str] = set() + for _, _, finding in _iter_findings(document): + if isinstance(finding, dict): + fid = finding.get("finding_id") + if isinstance(fid, str): + ids.add(fid) + return ids + + +def _severity_fell(reported_severity: object, severity: object) -> bool: + """Whether ``severity`` is a real fall below ``reported_severity``. + + Both must already be legal ladder values -- an out-of-ladder value is + reported as its own violation elsewhere, not treated as a fall here. + """ + if not isinstance(reported_severity, str) or not isinstance(severity, str): + return False + if reported_severity not in SEVERITY_ORDER or severity not in SEVERITY_ORDER: + return False + return SEVERITY_ORDER[severity] > SEVERITY_ORDER[reported_severity] + + +def _find_forbidden_keys(value: object, path: str) -> list[str]: + """Walk the whole document for a key named `approved`, `mergeable` or + `merge_recommendation`, anywhere -- the machine-checkable half of + prohibition 1. Walked, not grepped: a grep over the serialised text would + also flag those words inside `verdict_evidence` prose, which prohibition 1 + explicitly does not bind. + """ + violations: list[str] = [] + if isinstance(value, dict): + for key, sub in value.items(): + if key in _FORBIDDEN_KEYS: + violations.append(f"{path}.{key}: forbidden key present -- no field may carry an approval") + violations.extend(_find_forbidden_keys(sub, f"{path}.{key}")) + elif isinstance(value, list): + for index, item in enumerate(value): + violations.extend(_find_forbidden_keys(item, f"{path}[{index}]")) + return violations + + +def validate(input_document: dict, output_document: dict) -> list[str]: + """Every violation of ADJUDICATION.md that STEP 2's own done-when requires + this function to catch -- never raises, never stops early. + + NOT yet a check of every one of the nine ``adjudication`` block keys. + ``schema_version``, ``verdict_counts`` and ``notes`` are unchecked here on + purpose: ADJUDICATION.md's own STEP 1 text assigns "one control per key" + to STEP 10's separate ``check_adjudication.py`` suite, and STEP 2's + done-when never names these three. A document with a fabricated + ``verdict_counts`` or a wrong-typed ``schema_version`` passes this + function with zero violations today; STEP 10 is where that gap closes. + + Takes **both** documents, not one. An earlier revision declared + ``validate(document) -> list[str]`` and separately required it to report + the finding_id symmetric difference between input and output -- but + ``findings_in`` is a count, and a count cannot tell a dropped id from an + invented one, or notice a swap that leaves the count unchanged. The check + the contract promises needs the actual input id SET, and the only place + that set exists is the input document itself. + + Also re-runs #117's own ``findings.validate`` against the output document, + so a document that leaves this stage still satisfies the contract it + arrived under -- a stage that quietly breaks its input's own rules is the + same defect as one that drops a finding. ``findings`` is imported here, + locally, rather than at module scope, the same reason `findings.py` itself + imports `contain` locally: this module has no other reason to depend on it. + """ + import findings as findings_module + + violations: list[str] = list(findings_module.validate(output_document)) + + input_ids = _finding_ids(input_document) + output_ids = _finding_ids(output_document) + if input_ids != output_ids: + dropped = sorted(input_ids - output_ids) + invented = sorted(output_ids - input_ids) + if dropped: + violations.append(f"finding_id set: present on input, missing from output: {dropped}") + if invented: + violations.append(f"finding_id set: present on output, absent from input (invented): {invented}") + + findings_by_id: dict[str, dict] = {} + for _, _, finding in _iter_findings(output_document): + if isinstance(finding, dict): + fid = finding.get("finding_id") + if isinstance(fid, str): + findings_by_id[fid] = finding + + # Per-finding verdict fields: verdict, verdict_evidence, reported_severity, + # severity, severity_reason. duplicate_of is checked separately below, + # against the full id set built above. + for label, index, finding in _iter_findings(output_document): + if not isinstance(finding, dict): + continue + finding_label = f"{label} finding[{index}]" + fid = finding.get("finding_id") + if isinstance(fid, str): + finding_label = f"{finding_label} (finding_id={fid!r})" + + verdict = finding.get("verdict") + if not isinstance(verdict, str) or verdict not in VERDICTS: + violations.append( + f"{finding_label}: verdict must be one of {sorted(VERDICTS)}, got {verdict!r}" + ) + + if not is_nonempty_str(finding.get("verdict_evidence")): + violations.append( + f"{finding_label}: verdict_evidence must be a non-empty string, got " + f"{finding.get('verdict_evidence')!r}" + ) + + reported_severity = finding.get("reported_severity") + severity = finding.get("severity") + if not isinstance(reported_severity, str) or reported_severity not in SEVERITY_ORDER: + violations.append( + f"{finding_label}: reported_severity {reported_severity!r} is not a key of " + "review.SEVERITY_ORDER" + ) + if not isinstance(severity, str) or severity not in SEVERITY_ORDER: + violations.append( + f"{finding_label}: severity {severity!r} is not a key of review.SEVERITY_ORDER" + ) + + if severity != reported_severity and not is_nonempty_str(finding.get("severity_reason")): + violations.append( + f"{finding_label}: severity_reason must be a non-empty string when severity " + f"({severity!r}) differs from reported_severity ({reported_severity!r}), got " + f"{finding.get('severity_reason')!r}" + ) + + dup = finding.get("duplicate_of") + if dup is not None: + if not isinstance(dup, str): + violations.append( + f"{finding_label}: duplicate_of must be a string or null, got " + f"{type(dup).__name__}" + ) + elif dup == fid: + violations.append(f"{finding_label}: duplicate_of names itself") + elif dup not in output_ids: + violations.append( + f"{finding_label}: duplicate_of {dup!r} is not a finding_id present in the document" + ) + + adjudication_raw = output_document.get("adjudication") + if adjudication_raw is None: + violations.append("document: missing 'adjudication' key") + adjudication: dict = {} + elif not isinstance(adjudication_raw, dict): + violations.append( + f"document.adjudication: expected an object, got {type(adjudication_raw).__name__}" + ) + adjudication = {} + else: + adjudication = adjudication_raw + + # findings_out == findings_in == the sum of every report's findings_count. + findings_in = adjudication.get("findings_in") + findings_out = adjudication.get("findings_out") + reports = output_document.get("reports") + report_findings_count_sum = 0 + if isinstance(reports, list): + for report_index, report in enumerate(reports): + if not isinstance(report, dict): + continue + report_count = report.get("findings_count") + if is_int(report_count): + report_findings_count_sum += report_count + else: + # Named rather than silently skipped: a skipped count makes the + # sum wrong, and the equality violation below would then blame + # findings_in for a defect that is actually here. + violations.append( + f"document.reports[{report_index}]: findings_count must be an integer, " + f"got {report_count!r}" + ) + if not (is_int(findings_in) and is_int(findings_out)): + violations.append( + "document.adjudication: findings_in and findings_out must both be integers, got " + f"{findings_in!r} and {findings_out!r}" + ) + elif not (findings_out == findings_in == report_findings_count_sum): + violations.append( + "document.adjudication: findings_in " + f"({findings_in!r}), findings_out ({findings_out!r}), and the sum of every " + f"report's findings_count ({report_findings_count_sum!r}) must all be equal" + ) + + # Downgrades, both directions: every recorded entry is a real fall, and + # every real fall is recorded. + downgrades = adjudication.get("downgrades") + if not isinstance(downgrades, list): + violations.append( + f"document.adjudication.downgrades: expected an array, got {type(downgrades).__name__}" + ) + downgrades = [] + downgrade_ids: set[str] = set() + for index, entry in enumerate(downgrades): + if not isinstance(entry, dict): + violations.append( + f"document.adjudication.downgrades[{index}]: expected an object, got " + f"{type(entry).__name__}" + ) + continue + entry_fid = entry.get("finding_id") + if not isinstance(entry_fid, str): + violations.append( + f"document.adjudication.downgrades[{index}]: finding_id must be a string, " + f"got {type(entry_fid).__name__}" + ) + continue + downgrade_ids.add(entry_fid) + finding = findings_by_id.get(entry_fid) + if finding is None: + violations.append( + f"document.adjudication.downgrades[{index}]: finding_id {entry_fid!r} is not " + "present in the document" + ) + continue + reported = finding.get("reported_severity") + actual = finding.get("severity") + if entry.get("from") != reported or entry.get("to") != actual: + violations.append( + f"document.adjudication.downgrades[{index}]: from/to " + f"({entry.get('from')!r}/{entry.get('to')!r}) do not match finding " + f"{entry_fid!r}'s reported_severity/severity ({reported!r}/{actual!r})" + ) + if not _severity_fell(reported, actual): + violations.append( + f"document.adjudication.downgrades[{index}]: finding {entry_fid!r} is recorded " + f"as a downgrade but its severity did not fall ({reported!r} -> {actual!r})" + ) + for fid, finding in findings_by_id.items(): + reported = finding.get("reported_severity") + actual = finding.get("severity") + if _severity_fell(reported, actual) and fid not in downgrade_ids: + violations.append( + f"finding_id {fid!r}: severity fell from {reported!r} to {actual!r} but is not " + "recorded in document.adjudication.downgrades" + ) + + # total_refutation, both directions: it is true iff findings_in > 0 and + # every verdict is REFUTED. + all_verdicts = [ + finding.get("verdict") for _, _, finding in _iter_findings(output_document) if isinstance(finding, dict) + ] + declared_total_refutation = adjudication.get("total_refutation") + actual_total_refutation = bool(all_verdicts) and all(v == "REFUTED" for v in all_verdicts) + if not isinstance(declared_total_refutation, bool): + violations.append( + "document.adjudication.total_refutation: must be a boolean, got " + f"{type(declared_total_refutation).__name__}" + ) + elif declared_total_refutation != actual_total_refutation: + violations.append( + f"document.adjudication.total_refutation: declared {declared_total_refutation!r} but " + f"the findings' verdicts imply {actual_total_refutation!r}" + ) + + # duplicate_groups, both directions: a group's members point back at it via + # duplicate_of, and every finding with duplicate_of set is named by the + # group it points at. + duplicate_groups = adjudication.get("duplicate_groups") + if not isinstance(duplicate_groups, list): + violations.append( + f"document.adjudication.duplicate_groups: expected an array, got " + f"{type(duplicate_groups).__name__}" + ) + duplicate_groups = [] + grouped_as: dict[str, object] = {} + for index, group in enumerate(duplicate_groups): + if not isinstance(group, dict): + violations.append( + f"document.adjudication.duplicate_groups[{index}]: expected an object, got " + f"{type(group).__name__}" + ) + continue + survivor = group.get("survivor") + duplicates = group.get("duplicates") + # Checked here, not only inside the loop below. The loop validates the + # survivor indirectly -- via a member pointing back at it -- so a group + # with an empty `duplicates` list never had its survivor checked at all. + if not isinstance(survivor, str): + violations.append( + f"document.adjudication.duplicate_groups[{index}]: survivor must be a string, " + f"got {type(survivor).__name__}" + ) + elif survivor not in findings_by_id: + violations.append( + f"document.adjudication.duplicate_groups[{index}]: survivor {survivor!r} is not " + "a finding_id present in the document" + ) + if not isinstance(duplicates, list): + violations.append( + f"document.adjudication.duplicate_groups[{index}]: duplicates must be an array, " + f"got {type(duplicates).__name__}" + ) + continue + for dup_id in duplicates: + if not isinstance(dup_id, str): + violations.append( + f"document.adjudication.duplicate_groups[{index}]: duplicates entries " + f"must be strings, got {type(dup_id).__name__}" + ) + continue + grouped_as[dup_id] = survivor + finding = findings_by_id.get(dup_id) + if finding is None: + violations.append( + f"document.adjudication.duplicate_groups[{index}]: duplicate {dup_id!r} is " + "not a finding_id present in the document" + ) + continue + if finding.get("duplicate_of") != survivor: + violations.append( + f"document.adjudication.duplicate_groups[{index}]: finding {dup_id!r} does " + f"not point back at survivor {survivor!r} via its own duplicate_of " + f"({finding.get('duplicate_of')!r})" + ) + for fid, finding in findings_by_id.items(): + dup_of = finding.get("duplicate_of") + if dup_of is not None and grouped_as.get(fid) != dup_of: + violations.append( + f"finding_id {fid!r}: duplicate_of names {dup_of!r} but no " + "document.adjudication.duplicate_groups entry lists this finding as one of its " + "duplicates" + ) + + # The top-level nonce: present, unchanged from the input, and (via the + # completion_marker check below) equal to this stage's own marker. + # `findings_module.validate` above already checked it against every + # report's own marker. + nonce = output_document.get("nonce") + if nonce != input_document.get("nonce"): + violations.append( + f"document.nonce: output nonce ({nonce!r}) differs from the input document's " + f"nonce ({input_document.get('nonce')!r})" + ) + + # The adjudication block's own completion_marker: present, LAST key of the + # block, and matching the top-level nonce. + if "completion_marker" not in adjudication: + violations.append("document.adjudication: missing 'completion_marker'") + else: + keys = list(adjudication.keys()) + if keys[-1] != "completion_marker": + violations.append( + "document.adjudication: completion_marker must be the last key, found last key " + f"{keys[-1]!r}" + ) + marker = adjudication["completion_marker"] + expected_marker = f"BUZZ-ADJUDICATION-COMPLETE:{nonce}" + if marker != expected_marker: + violations.append( + f"document.adjudication.completion_marker: expected {expected_marker!r}, got " + f"{marker!r}" + ) + + # Prohibition 1's machine-checkable half: no key anywhere named `approved`, + # `mergeable` or `merge_recommendation`. + violations.extend(_find_forbidden_keys(output_document, "document")) + + return violations