feat(ADR-0161.1): core teaching queue list|show — read-only queue projection (#296)

* 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: <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

@@ -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 <candidate-jsonl-path>\n  core teaching proposals --state pending\n  core teaching review <proposal_id> --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 <proposal_id>\n  core teaching propose <candidate-jsonl-path>\n  core teaching proposals --state pending\n  core teaching review <proposal_id> --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",

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 <proposal_id>` — prints the full derived
+- `core teaching hitl-queue show <proposal_id>` — 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).
 

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,

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

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!"