"""The adjudication command: it records a judgement, and never invents one. A proposal a human has not looked at must never be replayable as an adjudication, because replay is exactly what the run path does with a plan -- deterministically and forever. So this command writes a SIBLING record and leaves the proposal untouched: the two files together say who judged what, when, and how long it took, and either can be re-read against the other. Three properties are pinned here rather than described: - **The model leg is OFF by default.** Pre-annotation has been measured LOWERING a good annotator's accuracy, from 98.1 % to 95.8 %, so a leg that cannot be switched off is a leg whose value can never be measured. With it off, no process is spawned at all -- asserted by breaking `subprocess.run`. - **The CLI is named, and the other one is excluded BY NAME.** The model leg shells out to the `claude` CLI. `gemini` is not merely unmentioned; its absence from the module is a test, because "we did not use it" and "nothing stops us using it" look identical in a review. - **Dwell time travels with the verdict** (PM decision B2). A ratified flag with no per-item time is unfalsifiable, and it is the same number that makes adjudication throughput measurable at all. It lives outside `src/`, so it never enters a wheel and no consumer's install surface changes because it exists. """ from __future__ import annotations import json import subprocess import sys from pathlib import Path from typing import Any import pytest from llm_ingestion_okf.segmentation import parse_segmentation_plan sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools")) import okf_adjudicate # noqa: E402 import okf_propose_segments # noqa: E402 DOCUMENT = """# N500 Vegbygging Innledende tekst om vegbygging og dens omfang. ## 3.1 Brannkonsept Krav til seksjonering av bygget. ## 3.2 Roemning To uavhengige roemningsveier. """ ADJUDICATOR = "ktg" AT = "2026-09-02T10:00:00Z" def proposal(tmp_path: Path) -> Path: source = tmp_path / "n500.md" source.write_text(DOCUMENT, encoding="utf-8", newline="") out = tmp_path / "plan.json" assert okf_propose_segments.main([str(source), "--out", str(out), "--proposed-at", AT]) == 0 return out def adjudicate(tmp_path: Path, *extra: str) -> tuple[int, Path]: verdict = tmp_path / "adjudicated.json" code = okf_adjudicate.main( [ "--plan", str(proposal(tmp_path)), "--out", str(verdict), "--adjudicator", ADJUDICATOR, "--adjudicated-at", AT, *extra, ] ) return code, verdict def payload(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def test_the_proposal_survives_untouched(tmp_path: Path) -> None: plan_path = proposal(tmp_path) before = plan_path.read_bytes() okf_adjudicate.main( [ "--plan", str(plan_path), "--out", str(tmp_path / "adjudicated.json"), "--adjudicator", ADJUDICATOR, "--adjudicated-at", AT, ] ) assert plan_path.read_bytes() == before def test_the_verdict_records_adjudicator_timestamp_and_dwell(tmp_path: Path) -> None: code, verdict = adjudicate(tmp_path) assert code == 0 written = payload(verdict) assert written["adjudicated"] is True for entry in written["entries"]: record = entry["adjudication"] assert record["adjudicated_by"] == ADJUDICATOR assert record["adjudicated_at"] == AT assert isinstance(record["adjudication_dwell_s"], int) assert not isinstance(record["adjudication_dwell_s"], bool) def test_the_verdict_parses_as_a_segmentation_plan(tmp_path: Path) -> None: _, verdict = adjudicate(tmp_path) parsed = parse_segmentation_plan(payload(verdict)) assert parsed.adjudicated is True assert all(entry.adjudication is not None for entry in parsed.entries) def test_replaying_the_same_verdict_produces_identical_bytes(tmp_path: Path) -> None: """K4a's mechanism: an adjudication is data, so a re-run is a copy.""" _, first = adjudicate(tmp_path) kept = first.read_bytes() second = tmp_path / "again.json" okf_adjudicate.main( [ "--plan", str(tmp_path / "plan.json"), "--out", str(second), "--adjudicator", ADJUDICATOR, "--adjudicated-at", AT, ] ) assert second.read_bytes() == kept def test_with_the_model_leg_off_no_process_is_spawned( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Asserted by BREAKING the spawn, not by reading the code. A test that merely inspects the default would pass just as happily if the default were ignored. """ def refuse(*args: object, **kwargs: object) -> None: raise AssertionError("the model leg spawned a process while switched off") monkeypatch.setattr(subprocess, "run", refuse) code, _ = adjudicate(tmp_path) assert code == 0 def test_the_model_leg_is_off_unless_asked_for(tmp_path: Path) -> None: assert ( okf_adjudicate.parse_args( ["--plan", "p", "--out", "o", "--adjudicator", "a", "--adjudicated-at", AT] ).model is None ) def test_the_resolved_argv_starts_with_the_claude_binary() -> None: argv = okf_adjudicate.model_argv("claude-opus-5", "spoersmaal") assert argv[0] == okf_adjudicate.CLAUDE_CLI assert Path(argv[0]).name == "claude" assert "--model" in argv assert argv[argv.index("--model") + 1] == "claude-opus-5" def test_the_other_cli_is_excluded_by_name_not_merely_unused() -> None: """ "We did not use it" and "nothing stops us using it" look identical in a review. This is the difference, as a measurement.""" module = Path(okf_adjudicate.__file__).read_text(encoding="utf-8") assert "gemini" not in module.lower() def test_the_gemini_check_can_actually_fire() -> None: """The negative control for the check above: prove it can find the word.""" assert "gemini" in "a line naming gemini".lower() def test_a_missing_plan_exits_two_and_says_so( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: code = okf_adjudicate.main( [ "--plan", str(tmp_path / "nothing.json"), "--out", str(tmp_path / "out.json"), "--adjudicator", ADJUDICATOR, "--adjudicated-at", AT, ] ) assert code == 2 assert "nothing.json" in capsys.readouterr().err