c0cc8c40c4
tests/test_teaching_queue.py::test_read_only_invariant hardcoded `project_root / "engine_state"` (the real shared dir) in its read-only- invariant snapshot, instead of reading `engine_state._DEFAULT_DIR` (the root-conftest autouse fixture's per-test isolated dir). Same bug pattern as test_workbench_replay.py::test_replay_leaves_no_trace, and this one was caught red-handed: a full `-n auto` fast-lane run failed with "Directory .../engine_state was mutated!" and passed cleanly in serial isolation, the exact order-dependent-flake signature this wave hunts. The other three snapshotted dirs (teaching/proposals, packs, contemplation/runs) have no equivalent per-test isolation idiom in the repo yet and are left as-is — this test's own hitl-queue commands don't write to them, so they were not implicated in the observed failure.
546 lines
17 KiB
Python
546 lines
17 KiB
Python
"""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!"
|