From ec5d6f5ac761aacc1bd68c713ee0686dd27bd795 Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 26 May 2026 11:42:51 -0700 Subject: [PATCH] =?UTF-8?q?feat(ADR-0161.1):=20core=20teaching=20queue=20l?= =?UTF-8?q?ist|show=20=E2=80=94=20read-only=20queue=20projection=20(#296)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * docs(math): ADR-0163 — path to GSM8K mastery via candidate-graph admissibility (proposed) Audit reframes the math roadmap entirely. State of main: every named math capability axis (G1..G5, S1) passes at 100% with wrong=0 on its controlled lane. binding_graph, math_versor_arithmetic, math_symbolic_equivalence, math_parser, math_candidate_parser, math_solver, math_verifier, math_realizer, math_problem_graph — all landed. The worktrees on disk are stale forks. State of GSM8K (50-case train sample): correct=0, refused=50, wrong=0. Every refusal reason is identical: "candidate_graph: no admissible candidate for statement: ". The reframe: the gap is NOT in operator algebra, NOT in binding graph internals, NOT in symbolic equivalence. The gap is in generate/math_candidate_graph.py — the admissibility surface that turns a natural-language statement into a candidate the downstream pipeline can consume. The capability axes pass at 100% because they test statement shapes the candidate-graph already admits. GSM8K refuses at 100% because its statements span shapes the candidate-graph has never been taught. Six-phase plan to lift GSM8K under the thesis "decodes, not generates": A. Refusal taxonomy (measure before building) B. Exemplar corpora per shape category (≤20 statements each, ≤3 per round) C. Contemplation runner ingests exemplars; emits DerivedRecognizer proposals D. Operator ratifies through ADR-0161 HITL queue (no new surface) E. Re-baseline GSM8K train sample. Round 1 exit: correct ≥ 10, wrong = 0. Round 2: ≥ 25. Round 3: ≥ 35. F. Scale to public/v1 (200 cases, target correct ≥ 100), then holdout (measurement-only — never tune against). Three non-negotiables: - wrong = 0 at every phase. Auto-rejected by replay gate, not by operator vigilance. - No hand-rolled recognizers in generate/. Every recognizer lands via contemplation → proposal → review corridor. - Active corpus mutation only via accept_proposal. Status: proposed. Implementation lands as three PRs starting with Phase A scaffolding. Scope discipline: docs-only. No code, no eval changes, no corpus mutation. * feat(ADR-0161.1): core teaching queue list|show — read-only queue projection * fix(ADR-0161.1): restore gap-queue CLI + rename new commands to hitl-queue + R1..R5 refinements --- core/cli.py | 199 +++++++- docs/decisions/ADR-0161-hitl-async-queue.md | 10 +- teaching/proposals.py | 11 +- teaching/queue.py | 165 ++++++ tests/test_teaching_queue.py | 535 ++++++++++++++++++++ 5 files changed, 910 insertions(+), 10 deletions(-) create mode 100644 teaching/queue.py create mode 100644 tests/test_teaching_queue.py diff --git a/core/cli.py b/core/cli.py index 808f968f..3185b1a9 100644 --- a/core/cli.py +++ b/core/cli.py @@ -23,7 +23,7 @@ _CORE_RS_DIR = _REPO_ROOT / "core-rs" _CORE_RS_MANIFEST = _CORE_RS_DIR / "Cargo.toml" DESCRIPTION = "CORE versor engine command suite." -EPILOG = "Examples:\n core chat\n core pulse \"What is truth?\"\n core pulse --no-glove --json \"Compare knowledge and wisdom\"\n core bench\n core bench --suite all\n core bench --suite all --json --report bench_all.json\n core bench --suite determinism --runs 50\n core bench --suite speedup --json\n core trace \"word beginning truth\"\n core trace --output-language grc --frame-pack grc --json \"logos\"\n core rust status\n core rust build\n core oov covenant\n core pack list\n core pack verify en_minimal_v1\n core teaching audit\n core teaching audit --json\n core teaching gaps --top 10\n core teaching queue --threshold 3\n core teaching propose \n core teaching proposals --state pending\n core teaching review --accept --review-date 2026-05-18\n core teaching supersede cause_light_reveals_truth --subject light --intent cause --connective grounds --object truth --review-date 2026-05-18\n core teaching supersessions\n core teaching supersessions --json\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo register-tour\n core demo anchor-lens-tour\n core demo orthogonality-tour\n core demo pack-measurements\n core demo long-context-comparison\n core demo anti-regression\n core demo learning-loop\n core demo learning-arc\n core demo articulation\n core demo conversation\n core demo conversation --no-stream\n core demo all\n core demo adr-0024-chain\n core eval --list\n core eval cognition\n core eval cognition --json --save\n core eval cognition --split dev --version v1\n core eval cognition --split holdout\n core eval contemplation_quality\n core eval contemplation_quality --json --save\n core workbench api\n core workbench api --port 9000\n core workbench api --host 0.0.0.0 --allow-nonlocal-bind" +EPILOG = "Examples:\n core chat\n core pulse \"What is truth?\"\n core pulse --no-glove --json \"Compare knowledge and wisdom\"\n core bench\n core bench --suite all\n core bench --suite all --json --report bench_all.json\n core bench --suite determinism --runs 50\n core bench --suite speedup --json\n core trace \"word beginning truth\"\n core trace --output-language grc --frame-pack grc --json \"logos\"\n core rust status\n core rust build\n core oov covenant\n core pack list\n core pack verify en_minimal_v1\n core teaching audit\n core teaching audit --json\n core teaching gaps --top 10\n core teaching queue --threshold 3\n core teaching hitl-queue list\n core teaching hitl-queue list --state all --json\n core teaching hitl-queue show \n core teaching propose \n core teaching proposals --state pending\n core teaching review --accept --review-date 2026-05-18\n core teaching supersede cause_light_reveals_truth --subject light --intent cause --connective grounds --object truth --review-date 2026-05-18\n core teaching supersessions\n core teaching supersessions --json\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo register-tour\n core demo anchor-lens-tour\n core demo orthogonality-tour\n core demo pack-measurements\n core demo long-context-comparison\n core demo anti-regression\n core demo learning-loop\n core demo learning-arc\n core demo articulation\n core demo conversation\n core demo conversation --no-stream\n core demo all\n core demo adr-0024-chain\n core eval --list\n core eval cognition\n core eval cognition --json --save\n core eval cognition --split dev --version v1\n core eval cognition --split holdout\n core eval contemplation_quality\n core eval contemplation_quality --json --save\n core workbench api\n core workbench api --port 9000\n core workbench api --host 0.0.0.0 --allow-nonlocal-bind" _TEST_SUITES: dict[str, tuple[str, ...]] = { "fast": ( @@ -1157,6 +1157,150 @@ def cmd_teaching_queue(args: argparse.Namespace) -> int: return 0 +def _contemplation_runs_dir(args_dir: str | None) -> Path: + if args_dir: + return Path(args_dir) + return _REPO_ROOT / "contemplation" / "runs" + + +def cmd_teaching_hitl_queue_list(args: argparse.Namespace) -> int: + """List queue items in the human-in-the-loop review queue.""" + from teaching.proposals import DEFAULT_PROPOSAL_LOG_PATH, ProposalLog + from teaching.queue import derive_queue + + log_path = Path(args.log_path) if args.log_path else DEFAULT_PROPOSAL_LOG_PATH + runs_dir = _contemplation_runs_dir(args.contemplation_runs_dir) + + log = ProposalLog(log_path) + if not log.path.exists(): + return 0 + + items = derive_queue(log, contemplation_runs_dir=runs_dir) + + if args.state and args.state != "all": + items = tuple(item for item in items if item.state == args.state) + + if args.json: + import dataclasses + payload = [dataclasses.asdict(item) for item in items] + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 + + if not items: + return 0 + + header = ("proposal_id", "source_kind", "state", "age", "replay") + rows = [] + for item in items: + if item.replay_evidence is None: + replay_status = "?" + elif item.replay_evidence.get("replay_equivalent") is True: + replay_status = "ok" + elif item.replay_evidence.get("replay_equivalent") is False: + replay_status = "regressed" + else: + replay_status = "?" + + rows.append(( + item.proposal_id[:12], + item.source_kind, + item.state, + str(item.age_proposals), + replay_status, + )) + + col_widths = [len(h) for h in header] + for row in rows: + for idx, val in enumerate(row): + col_widths[idx] = max(col_widths[idx], len(val)) + + header_str = " ".join(f"{h:<{col_widths[idx]}}" for idx, h in enumerate(header)) + print(header_str) + print(" ".join("-" * w for w in col_widths)) + for row in rows: + row_str = " ".join(f"{val:<{col_widths[idx]}}" for idx, val in enumerate(row)) + print(row_str) + + return 0 + + +def cmd_teaching_hitl_queue_show(args: argparse.Namespace) -> int: + """Show details of a specific queue item in the human-in-the-loop review queue.""" + from teaching.proposals import DEFAULT_PROPOSAL_LOG_PATH, ProposalLog + from teaching.queue import derive_queue + + log_path = Path(args.log_path) if args.log_path else DEFAULT_PROPOSAL_LOG_PATH + runs_dir = _contemplation_runs_dir(args.contemplation_runs_dir) + + log = ProposalLog(log_path) + if not log.path.exists(): + _die(f"no proposal log at {log.path}", code=1) + + items = derive_queue(log, contemplation_runs_dir=runs_dir) + + # 1. Search for exact match + exact_matches = [item for item in items if item.proposal_id == args.proposal_id] + if len(exact_matches) == 1: + item = exact_matches[0] + else: + # 2. Search for prefix match + prefix_matches = [item for item in items if item.proposal_id.startswith(args.proposal_id)] + if len(prefix_matches) == 1: + item = prefix_matches[0] + elif len(prefix_matches) == 0: + _die(f"proposal_id prefix {args.proposal_id!r} matches zero queue items", code=1) + else: + _die(f"proposal_id prefix {args.proposal_id!r} is ambiguous (matches multiple items)", code=1) + + if args.json: + import dataclasses + print(json.dumps(dataclasses.asdict(item), ensure_ascii=False, indent=2, sort_keys=True)) + return 0 + + print(f"Proposal ID: {item.proposal_id}") + print(f"Source Kind: {item.source_kind}") + print(f"Source ID : {item.source_id or '—'}") + print(f"State : {item.state}") + print(f"Age : {item.age_proposals}") + + if item.replay_evidence is None: + replay_status = "?" + elif item.replay_evidence.get("replay_equivalent") is True: + replay_status = "ok" + elif item.replay_evidence.get("replay_equivalent") is False: + replay_status = "regressed" + else: + replay_status = "?" + print(f"Replay : {replay_status}") + print(f"Report Path: {item.contemplation_report_path or '—'}") + print() + print("Proposed Chain:") + chain = item.proposed_chain or {} + print(f" subject : {chain.get('subject', '—')}") + print(f" intent : {chain.get('intent', '—')}") + print(f" connective: {chain.get('connective', '—')}") + print(f" object : {chain.get('object', '—')}") + print() + print("Review History:") + if item.review_history: + for ev in item.review_history: + note = ev.get('note', '') + to_state = ev.get('to', '') + review_date = ev.get('review_date', '') + actor = ev.get('actor', '') + print(f" - [{review_date or '—'}] transitioned to {to_state} by {actor or '—'}") + if note: + print(f" Note: {note}") + else: + print(" (no review history)") + print() + print("ADR References:") + print(" - Queue contract: docs/decisions/ADR-0161-hitl-async-queue.md") + print(" - Proposal/review state machine: docs/decisions/ADR-0057-teaching-chain-proposal-review.md") + + return 0 + + def _load_candidate_jsonl(path: str) -> Any: """Read one enriched DiscoveryCandidate JSONL line from *path*.""" from teaching.discovery import DiscoveryCandidate, EvidencePointer, SubQuestion @@ -3582,6 +3726,59 @@ def build_parser() -> argparse.ArgumentParser: ) teaching_queue.set_defaults(func=cmd_teaching_queue) + teaching_hitl_queue = teaching_sub.add_parser( + "hitl-queue", + help="inspect the asynchronous human-in-the-loop review queue (ADR-0161)", + ) + teaching_hitl_queue_sub = teaching_hitl_queue.add_subparsers( + dest="hitl_queue_command", metavar="hitl-queue-command", required=True, + ) + + teaching_hitl_queue_list = teaching_hitl_queue_sub.add_parser( + "list", + help="list queue items", + ) + teaching_hitl_queue_list.add_argument( + "--state", default="pending", + choices=("pending", "accepted", "rejected", "withdrawn", "all"), + help="filter by state (default: pending)", + ) + teaching_hitl_queue_list.add_argument( + "--json", action="store_true", + help="output machine-readable JSON", + ) + teaching_hitl_queue_list.add_argument( + "--log-path", default=None, + help="path to the proposal log file", + ) + teaching_hitl_queue_list.add_argument( + "--contemplation-runs-dir", default=None, + help="path to contemplation runs directory", + ) + teaching_hitl_queue_list.set_defaults(func=cmd_teaching_hitl_queue_list) + + teaching_hitl_queue_show = teaching_hitl_queue_sub.add_parser( + "show", + help="show details of a queue item", + ) + teaching_hitl_queue_show.add_argument( + "proposal_id", + help="proposal ID or prefix", + ) + teaching_hitl_queue_show.add_argument( + "--json", action="store_true", + help="output machine-readable JSON", + ) + teaching_hitl_queue_show.add_argument( + "--log-path", default=None, + help="path to the proposal log file", + ) + teaching_hitl_queue_show.add_argument( + "--contemplation-runs-dir", default=None, + help="path to contemplation runs directory", + ) + teaching_hitl_queue_show.set_defaults(func=cmd_teaching_hitl_queue_show) + teaching_gaps = teaching_sub.add_parser( "gaps", help="rank (subject, intent) cells discovery candidates would have grounded", diff --git a/docs/decisions/ADR-0161-hitl-async-queue.md b/docs/decisions/ADR-0161-hitl-async-queue.md index f21259f6..fb7498ec 100644 --- a/docs/decisions/ADR-0161-hitl-async-queue.md +++ b/docs/decisions/ADR-0161-hitl-async-queue.md @@ -166,10 +166,10 @@ non-zero exit and emits no transition event. The proposal stays Two new CLI commands expose the queue projection: -- `core teaching queue list [--state pending|accepted|rejected|withdrawn|all]` +- `core teaching hitl-queue list [--state pending|accepted|rejected|withdrawn|all]` — prints `proposal_id`, source kind, age (in proposals, not wall-clock — see §4), replay status, and current state. -- `core teaching queue show ` — prints the full derived +- `core teaching hitl-queue show ` — prints the full derived record including `review_history` and the contemplation-report reference if one exists. @@ -352,11 +352,11 @@ this one. Five small PRs, each a self-contained step, none of which mutate existing recorded queue history: -### Step 1 — `core teaching queue` read commands +### Step 1 — `core teaching hitl-queue` read commands - New module `teaching/queue.py` exposing a pure `derive_queue(log)` function that returns the projection in §1. -- New CLI subcommand `core teaching queue list|show` wired in +- New CLI subcommand `core teaching hitl-queue list|show` wired in `core/cli.py`. - Tests: pure derivation over fixture proposals.jsonl; states match ADR-0057's alphabet; replay-equivalence in derivation. @@ -433,7 +433,7 @@ This ADR is ratifiable when: surfaces. 4. The workflow's actor guard fails closed in a CI test that fakes a non-allowlisted `github.actor`. -5. `core teaching queue list` and `... show` succeed against the +5. `core teaching hitl-queue list` and `... show` succeed against the current `teaching/proposals/proposals.jsonl` on `main` without mutating any file (snapshot assertion). diff --git a/teaching/proposals.py b/teaching/proposals.py index cfc27993..c68f53a8 100644 --- a/teaching/proposals.py +++ b/teaching/proposals.py @@ -311,19 +311,22 @@ class ProposalLog: # -- read side ---------------------------------------------------- - def _events(self) -> list[dict[str, Any]]: + def events(self) -> list[dict[str, Any]]: if not self.path.exists(): return [] - events: list[dict[str, Any]] = [] + events_list: list[dict[str, Any]] = [] for line in self.path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue try: - events.append(json.loads(line)) + events_list.append(json.loads(line)) except json.JSONDecodeError: continue - return events + return events_list + + def _events(self) -> list[dict[str, Any]]: + return self.events() def current_state(self) -> dict[str, dict[str, Any]]: """Replay the log → ``{proposal_id: {state, proposal, replay, diff --git a/teaching/queue.py b/teaching/queue.py new file mode 100644 index 00000000..fe4eb58e --- /dev/null +++ b/teaching/queue.py @@ -0,0 +1,165 @@ +"""ADR-0161 Step 1 — Read-only queue projection.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from functools import lru_cache +from pathlib import Path +from typing import Any + +from teaching.proposals import ProposalLog, ReviewState + + +@dataclass(frozen=True, slots=True) +class QueueItem: + proposal_id: str + source_kind: str + source_id: str | None + proposed_chain: dict[str, Any] + replay_evidence: dict[str, Any] | None + state: ReviewState + review_history: tuple[dict[str, Any], ...] + contemplation_report_path: str | None + age_proposals: int + + +@lru_cache(maxsize=1) +def _load_contemplation_mapping(runs_dir: Path, runs_dir_mtime: float) -> dict[str, str]: + """Cache the mapping of proposal_id to JSON file path under runs_dir. + + Keyed on runs_dir and its modification time (mtime) for invalidation. + """ + mapping: dict[str, str] = {} + if runs_dir.exists() and runs_dir.is_dir(): + for path in runs_dir.glob("*.json"): + if not path.is_file(): + continue + try: + with path.open("r", encoding="utf-8") as f: + data = json.load(f) + except Exception: + continue + + if not isinstance(data, dict): + continue + + pids: set[str] = set() + top_pid = data.get("proposal_id") + if isinstance(top_pid, str): + pids.add(top_pid) + + scenes = data.get("scenes") + if isinstance(scenes, list): + for scene in scenes: + if isinstance(scene, dict): + detail = scene.get("detail") + if isinstance(detail, dict): + scene_pid = detail.get("proposal_id") + if isinstance(scene_pid, str): + pids.add(scene_pid) + + for pid in pids: + mapping[pid] = str(path.resolve()) + return mapping + + +def derive_queue( + log: ProposalLog, + contemplation_runs_dir: Path, +) -> tuple[QueueItem, ...]: + """Derive a read-only queue projection from the ProposalLog. + + Order: FIFO by first-pending-event order in the log. + """ + events = log.events() + proposals_data: dict[str, dict[str, Any]] = {} + created_order: list[str] = [] + + for ev in events: + kind = ev.get("event") + if kind == "created": + p = ev.get("proposal") or {} + pid = p.get("proposal_id") + if not pid: + continue + if pid not in proposals_data: + created_order.append(pid) + source_dict = p.get("source") or {} + source_kind = source_dict.get("kind", "") + source_id = source_dict.get("source_id") + # Normalize empty string to None + if source_id == "": + source_id = None + + proposals_data[pid] = { + "proposal_id": pid, + "source_kind": source_kind, + "source_id": source_id, + "proposed_chain": p.get("proposed_chain"), + "replay_evidence": p.get("replay_evidence"), + "state": p.get("review_state", "pending"), + "review_history": [], + } + elif kind == "replay": + pid = ev.get("proposal_id") + if pid in proposals_data: + proposals_data[pid]["replay_evidence"] = ev.get("replay_evidence") + elif kind == "transition": + pid = ev.get("proposal_id") + if pid in proposals_data: + proposals_data[pid]["state"] = ev.get("to") + # Append transition event to review_history + proposals_data[pid]["review_history"].append(dict(ev)) + + # Retrieve cached mapping using runs_dir mtime + try: + mtime = contemplation_runs_dir.stat().st_mtime + except OSError: + mtime = 0.0 + + contemplation_mapping = _load_contemplation_mapping(contemplation_runs_dir, mtime) + + # Build the final tuple of QueueItem + items: list[QueueItem] = [] + for i, pid in enumerate(created_order): + data = proposals_data[pid] + state = data["state"] + + if state == "pending": + # Per ADR-0161 §4, age is subsequent proposals appended regardless of state. + age_proposals = len(created_order) - 1 - i + else: + age_proposals = 0 + + item = QueueItem( + proposal_id=pid, + source_kind=data["source_kind"], + source_id=data["source_id"], + proposed_chain=data["proposed_chain"], + replay_evidence=data["replay_evidence"], + state=state, + review_history=tuple(data["review_history"]), + contemplation_report_path=contemplation_mapping.get(pid), + age_proposals=age_proposals, + ) + items.append(item) + + return tuple(items) + + +def find_queue_item( + log: ProposalLog, + proposal_id: str, + contemplation_runs_dir: Path, +) -> QueueItem | None: + """Find a specific queue item by exact ID or unique prefix.""" + items = derive_queue(log, contemplation_runs_dir) + for item in items: + if item.proposal_id == proposal_id: + return item + + matches = [item for item in items if item.proposal_id.startswith(proposal_id)] + if len(matches) == 1: + return matches[0] + return None diff --git a/tests/test_teaching_queue.py b/tests/test_teaching_queue.py new file mode 100644 index 00000000..acb34315 --- /dev/null +++ b/tests/test_teaching_queue.py @@ -0,0 +1,535 @@ +"""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 + +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(): + project_root = Path(__file__).resolve().parent.parent + dirs = [ + project_root / "teaching" / "proposals", + project_root / "packs", + project_root / "engine_state", + 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!"