"""Tests for ADR-0161 human-in-the-loop review queue read-only commands.""" from __future__ import annotations import json from io import StringIO from pathlib import Path from unittest.mock import patch import pytest import engine_state from core.cli import main from teaching.discovery import DiscoveryCandidate, EvidencePointer from teaching.proposals import ProposalLog, ReplayEvidence, build_proposal from teaching.queue import derive_queue, find_queue_item def make_candidate(candidate_id: str, subject: str, obj: str = "truth") -> DiscoveryCandidate: return DiscoveryCandidate( candidate_id=candidate_id, proposed_chain={ "subject": subject, "intent": "cause", "connective": "reveals", "object": obj, }, trigger="would_have_grounded", source_turn_trace="trace_1", pack_consistent=True, boundary_clean=True, polarity="affirms", claim_domain="factual", evidence=( EvidencePointer( source="corpus", ref="some_chain", polarity="affirms", epistemic_status="coherent", ), ), ) def run_cli(args: list[str]) -> tuple[int, str, str]: with patch("sys.argv", ["core"] + args), patch("sys.stdout", new_callable=StringIO) as out, patch("sys.stderr", new_callable=StringIO) as err: try: code = main() except SystemExit as e: code = e.code if isinstance(e.code, int) else 0 return code, out.getvalue(), err.getvalue() # --------------------------------------------------------------------------- # derive_queue purity and determinism # --------------------------------------------------------------------------- def test_derive_queue_purity(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) runs_dir = tmp_path / "runs" runs_dir.mkdir() c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) log.record_created(p1) res1 = derive_queue(log, contemplation_runs_dir=runs_dir) res2 = derive_queue(log, contemplation_runs_dir=runs_dir) assert res1 == res2 def test_derive_queue_determinism(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) runs_dir = tmp_path / "runs" runs_dir.mkdir() c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) log.record_created(p1) res1 = derive_queue(log, contemplation_runs_dir=runs_dir) res2 = derive_queue(log, contemplation_runs_dir=runs_dir) assert res1 == res2 # --------------------------------------------------------------------------- # State derivation # --------------------------------------------------------------------------- def test_state_derivation(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) runs_dir = tmp_path / "runs" runs_dir.mkdir() c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) log.record_created(p1) items = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(items) == 1 assert items[0].state == "pending" log.record_transition(p1.proposal_id, "accepted", "review accepted") items = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(items) == 1 assert items[0].state == "accepted" log.record_transition(p1.proposal_id, "withdrawn", "review withdrawn") items = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(items) == 1 assert items[0].state == "withdrawn" # --------------------------------------------------------------------------- # Age proposals # --------------------------------------------------------------------------- def test_age_proposals(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) runs_dir = tmp_path / "runs" runs_dir.mkdir() c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) c2 = make_candidate("cand2", "dark") p2 = build_proposal(c2) c3 = make_candidate("cand3", "shade") p3 = build_proposal(c3) log.record_created(p1) log.record_created(p2) log.record_created(p3) items = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(items) == 3 assert items[0].proposal_id == p1.proposal_id assert items[1].proposal_id == p2.proposal_id assert items[2].proposal_id == p3.proposal_id assert items[0].age_proposals == 2 assert items[1].age_proposals == 1 assert items[2].age_proposals == 0 log.record_transition(p1.proposal_id, "accepted", "accepting first") items = derive_queue(log, contemplation_runs_dir=runs_dir) assert items[0].age_proposals == 0 assert items[1].age_proposals == 1 assert items[2].age_proposals == 0 # --------------------------------------------------------------------------- # Contemplation report path # --------------------------------------------------------------------------- def test_contemplation_report_path(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) c2 = make_candidate("cand2", "dark") p2 = build_proposal(c2) c3 = make_candidate("cand3", "shade") p3 = build_proposal(c3) log.record_created(p1) log.record_created(p2) log.record_created(p3) runs_dir = tmp_path / "runs" runs_dir.mkdir() run1_path = runs_dir / "run1.json" run1_path.write_text(json.dumps({ "proposal_id": p1.proposal_id, "scenes": [] })) run2_path = runs_dir / "run2.json" run2_path.write_text(json.dumps({ "scenes": [ { "scene": "S3_engine_authored_proposal", "detail": { "proposal_id": p2.proposal_id } } ] })) items = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(items) == 3 assert items[0].proposal_id == p1.proposal_id assert items[0].contemplation_report_path == str(run1_path.resolve()) assert items[1].proposal_id == p2.proposal_id assert items[1].contemplation_report_path == str(run2_path.resolve()) assert items[2].proposal_id == p3.proposal_id assert items[2].contemplation_report_path is None # --------------------------------------------------------------------------- # Review history # --------------------------------------------------------------------------- def test_review_history(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) runs_dir = tmp_path / "runs" runs_dir.mkdir() c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) log.record_created(p1) log.record_transition(p1.proposal_id, "rejected", "regression check") log.record_transition(p1.proposal_id, "pending", "re-evaluated") log.record_transition(p1.proposal_id, "accepted", "approved") items = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(items) == 1 item = items[0] assert len(item.review_history) == 3 assert item.review_history[0]["to"] == "rejected" assert item.review_history[0]["note"] == "regression check" assert item.review_history[1]["to"] == "pending" assert item.review_history[1]["note"] == "re-evaluated" assert item.review_history[2]["to"] == "accepted" assert item.review_history[2]["note"] == "approved" # --------------------------------------------------------------------------- # find_queue_item # --------------------------------------------------------------------------- def test_find_queue_item(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) runs_dir = tmp_path / "runs" runs_dir.mkdir() c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) c2 = make_candidate("cand2", "dark") p2 = build_proposal(c2) log.record_created(p1) log.record_created(p2) res = find_queue_item(log, p1.proposal_id, runs_dir) assert res is not None assert res.proposal_id == p1.proposal_id prefix = p2.proposal_id[:12] res_prefix = find_queue_item(log, prefix, runs_dir) assert res_prefix is not None assert res_prefix.proposal_id == p2.proposal_id assert find_queue_item(log, "non_existent_id", runs_dir) is None common_prefix = p1.proposal_id[0] if p2.proposal_id.startswith(common_prefix): assert find_queue_item(log, common_prefix, runs_dir) is None # --------------------------------------------------------------------------- # R2: contemplation_report_path caching # --------------------------------------------------------------------------- def test_contemplation_mapping_caching(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) log.record_created(p1) runs_dir = tmp_path / "runs" runs_dir.mkdir() run_file = runs_dir / "run1.json" run_file.write_text(json.dumps({"proposal_id": p1.proposal_id})) glob_called_count = 0 orig_glob = Path.glob def mock_glob(self, pattern): nonlocal glob_called_count if str(self) == str(runs_dir.resolve()) or str(self) == str(runs_dir): glob_called_count += 1 return orig_glob(self, pattern) monkeypatch.setattr(Path, "glob", mock_glob) # First call walks the directory res1 = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(res1) == 1 assert glob_called_count == 1 # Second call hits cache (mtime is unchanged) res2 = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(res2) == 1 assert glob_called_count == 1 # Change the runs directory modification time (simulated via touch/new file) run_file2 = runs_dir / "run2.json" run_file2.write_text(json.dumps({"proposal_id": "dummy"})) # Third call re-evaluates res3 = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(res3) == 1 assert glob_called_count == 2 # --------------------------------------------------------------------------- # R4: empty string source_id normalization # --------------------------------------------------------------------------- def test_source_id_empty_string_normalization(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) runs_dir = tmp_path / "runs" runs_dir.mkdir() # build_proposal returns proposal with source.source_id = "" by default c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) assert p1.source.source_id == "" log.record_created(p1) items = derive_queue(log, contemplation_runs_dir=runs_dir) assert len(items) == 1 assert items[0].source_id is None # --------------------------------------------------------------------------- # CLI list and show commands # --------------------------------------------------------------------------- def test_cli_list_command(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) log.record_created(p1) log.record_replay(p1.proposal_id, ReplayEvidence( baseline={}, candidate={}, regressed_metrics=(), replay_equivalent=True )) runs_dir = tmp_path / "runs" runs_dir.mkdir() code, stdout, stderr = run_cli([ "teaching", "hitl-queue", "list", "--log-path", str(log_path), "--contemplation-runs-dir", str(runs_dir) ]) assert code == 0 assert "proposal_id" in stdout assert "source_kind" in stdout assert "state" in stdout assert "age" in stdout assert "replay" in stdout assert p1.proposal_id[:12] in stdout assert "ok" in stdout code_j, stdout_j, stderr_j = run_cli([ "teaching", "hitl-queue", "list", "--log-path", str(log_path), "--contemplation-runs-dir", str(runs_dir), "--json" ]) assert code_j == 0 parsed = json.loads(stdout_j) assert isinstance(parsed, list) assert len(parsed) == 1 assert parsed[0]["proposal_id"] == p1.proposal_id assert parsed[0]["source_kind"] == "operator" assert parsed[0]["state"] == "pending" assert parsed[0]["age_proposals"] == 0 assert parsed[0]["replay_evidence"]["replay_equivalent"] is True def test_cli_show_command(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) c1 = make_candidate("cand1", "light") p1 = build_proposal(c1) c2 = make_candidate("cand2", "dark") p2 = build_proposal(c2) log.record_created(p1) log.record_created(p2) runs_dir = tmp_path / "runs" runs_dir.mkdir() code, stdout, stderr = run_cli([ "teaching", "hitl-queue", "show", p1.proposal_id, "--log-path", str(log_path), "--contemplation-runs-dir", str(runs_dir) ]) assert code == 0 assert f"Proposal ID: {p1.proposal_id}" in stdout assert "ADR References:" in stdout prefix = p2.proposal_id[:12] code_p, stdout_p, stderr_p = run_cli([ "teaching", "hitl-queue", "show", prefix, "--log-path", str(log_path), "--contemplation-runs-dir", str(runs_dir) ]) assert code_p == 0 assert f"Proposal ID: {p2.proposal_id}" in stdout_p code_m, stdout_m, stderr_m = run_cli([ "teaching", "hitl-queue", "show", "nonexistent_id", "--log-path", str(log_path), "--contemplation-runs-dir", str(runs_dir) ]) assert code_m != 0 assert "error:" in stderr_m assert "matches zero queue items" in stderr_m common_prefix = "" for char1, char2 in zip(p1.proposal_id, p2.proposal_id): if char1 == char2: common_prefix += char1 else: break if common_prefix: code_a, stdout_a, stderr_a = run_cli([ "teaching", "hitl-queue", "show", common_prefix, "--log-path", str(log_path), "--contemplation-runs-dir", str(runs_dir) ]) assert code_a != 0 assert "error:" in stderr_a assert "ambiguous" in stderr_a # --------------------------------------------------------------------------- # Blocker: legacy queue --threshold 3 CLI command validation # --------------------------------------------------------------------------- def test_cli_legacy_queue_command(tmp_path: Path): gaps_dir = tmp_path / "gaps" sink = gaps_dir / "2026" / "2026-05.jsonl" sink.parent.mkdir(parents=True, exist_ok=True) for i in range(3): entry = { "candidate_id": f"cand-{i}", "proposed_chain": { "subject": "knowledge", "intent": "cause", "connective": None, "object": None, }, "trigger": "would_have_grounded", "source_turn_trace": f"trace-{i}", "pack_consistent": True, "boundary_clean": True, "review_state": "unreviewed", } with sink.open("a", encoding="utf-8") as fh: fh.write(json.dumps(entry) + "\n") code, stdout, stderr = run_cli([ "teaching", "queue", "--root", str(gaps_dir), "--threshold", "3", ]) assert code == 0 assert "rank" in stdout assert "queue_id" in stdout assert "count" in stdout assert "clean" in stdout assert "gap:cause:knowledge@3" in stdout # --------------------------------------------------------------------------- # Read-only invariant # --------------------------------------------------------------------------- def snapshot_dir(directory: Path) -> dict[Path, bytes]: snapshot = {} if not directory.exists(): return snapshot for path in directory.glob("**/*"): if path.is_file(): snapshot[path] = path.read_bytes() return snapshot def test_read_only_invariant(): """Directories the hitl-queue commands must not mutate. ``engine_state`` is read via ``engine_state._DEFAULT_DIR`` (not a hardcoded ``project_root / "engine_state"``) because the root-conftest ``_isolate_engine_state_default`` autouse fixture repoints it at a per-test tmp dir; a hardcoded real path would instead snapshot the shared repo dir and false-fail under ``-n auto`` whenever an unrelated concurrent worker wrote to it (observed: AssertionError on this exact directory under a full `-n auto` fast-lane run). """ project_root = Path(__file__).resolve().parent.parent dirs = [ project_root / "teaching" / "proposals", project_root / "packs", Path(engine_state._DEFAULT_DIR), project_root / "contemplation" / "runs", ] before_snapshots = {} for d in dirs: before_snapshots[d] = snapshot_dir(d) run_cli(["teaching", "hitl-queue", "list"]) run_cli(["teaching", "hitl-queue", "show", "nonexistent"]) for d in dirs: after_snapshot = snapshot_dir(d) assert after_snapshot == before_snapshots[d], f"Directory {d} was mutated!"