Merge branch 'feat/proposal-queue-cli' into main

core/cli.py

@@ -3695,6 +3695,10 @@ def build_parser() -> argparse.ArgumentParser:
 
     _register_formation(subparsers)
 
+    from core.cli_proposal_queue import register as _register_proposal_queue
+
+    _register_proposal_queue(subparsers)
+
     contemplation = subparsers.add_parser(
         "contemplation",
         help="run ADR-0080 read-only contemplation over explicit evidence files",

core/cli_proposal_queue.py

@@ -0,0 +1,145 @@
+"""core/cli_proposal_queue.py — ``core proposal-queue`` CLI (Tier S4).
+
+Operator-facing surface over :mod:`core.proposal_review.queue`: list and
+review pending proposals from the contemplation/idle sinks
+(``comprehension_failures``, ``derived_close_facts``). Read + review-state
+transitions ONLY — no command here can ratify, mutate the teaching corpus,
+mount anything, or flip a flag. Ratification stays ``core teaching
+proposals`` / ``core teaching review`` over the separate ADR-0057 sink.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from dataclasses import asdict
+
+from core.proposal_review.queue import (
+    SINKS,
+    QueueEntry,
+    record_review,
+    reviewed_keys,
+    scan_all,
+    queue_status,
+)
+
+_SINK_NAMES = tuple(spec.name for spec in SINKS)
+
+
+def _print_entry(entry: QueueEntry, *, reviewed: bool) -> None:
+    mark = "reviewed" if reviewed else ("pending" if entry.requires_review else "no-review-needed")
+    print(f"[{entry.sink}] {entry.content_hash}  {mark:<17} status={entry.status} family={entry.family}")
+
+
+def cmd_proposal_queue_status(args: argparse.Namespace) -> int:
+    sinks = (args.sink,) if args.sink else None
+    status = queue_status(sinks)
+    if args.json:
+        print(json.dumps(status, indent=2, sort_keys=True))
+        return 0
+    for name, counts in status.items():
+        print(
+            f"{name}: total={counts['total']} pending_review={counts['pending_review']} "
+            f"reviewed={counts['reviewed']} malformed={counts['malformed']}"
+        )
+    return 0
+
+
+def cmd_proposal_queue_list(args: argparse.Namespace) -> int:
+    sinks = (args.sink,) if args.sink else None
+    entries, malformed = scan_all(sinks)
+    reviewed = reviewed_keys()
+    if args.pending_only:
+        entries = [
+            e for e in entries
+            if e.requires_review and (e.sink, e.content_hash) not in reviewed
+        ]
+    if args.json:
+        payload = {
+            "entries": [
+                {**asdict(e), "reviewed": (e.sink, e.content_hash) in reviewed}
+                for e in entries
+            ],
+            "malformed": [asdict(m) for m in malformed],
+        }
+        print(json.dumps(payload, indent=2, sort_keys=True))
+        return 0
+    if not entries:
+        print("(no pending proposals)")
+    for entry in entries:
+        _print_entry(entry, reviewed=(entry.sink, entry.content_hash) in reviewed)
+    for m in malformed:
+        print(f"[{m.sink}] MALFORMED {m.path}: {m.reason}")
+    return 1 if malformed else 0
+
+
+def cmd_proposal_queue_show(args: argparse.Namespace) -> int:
+    entries, _malformed = scan_all((args.sink,))
+    match = next((e for e in entries if e.content_hash == args.content_hash), None)
+    if match is None:
+        print(f"error: no entry {args.content_hash!r} in sink {args.sink!r}")
+        return 2
+    with open(match.path, encoding="utf-8") as fh:
+        raw = json.load(fh)
+    print(json.dumps(raw, indent=2, sort_keys=True))
+    return 0
+
+
+def cmd_proposal_queue_review(args: argparse.Namespace) -> int:
+    try:
+        record = record_review(args.sink, args.content_hash, note=args.note or "")
+    except KeyError as exc:
+        print(f"error: {exc}")
+        return 2
+    print(f"recorded review: {args.sink}/{args.content_hash} at {record.reviewed_at}")
+    return 0
+
+
+def register(subparsers: argparse._SubParsersAction) -> None:
+    """Attach the ``core proposal-queue`` subcommand tree to a top-level parser."""
+    queue = subparsers.add_parser(
+        "proposal-queue",
+        help="list and review pending contemplation/idle proposals (HITL, read-only)",
+        description=(
+            "Read-only listing plus human-review-state tracking over the "
+            "comprehension_failures and derived_close_facts proposal sinks. "
+            "Never ratifies, mutates the teaching corpus, mounts anything, or "
+            "flips a flag — that stays `core teaching proposals`/`review`."
+        ),
+    )
+    sub = queue.add_subparsers(
+        dest="proposal_queue_command",
+        metavar="proposal-queue-command",
+        required=True,
+    )
+
+    status = sub.add_parser("status", help="per-sink pending/reviewed/malformed counts")
+    status.add_argument("--sink", choices=_SINK_NAMES, default=None)
+    status.add_argument("--json", action="store_true")
+    status.set_defaults(func=cmd_proposal_queue_status)
+
+    list_cmd = sub.add_parser("list", help="list proposal-queue entries")
+    list_cmd.add_argument("--sink", choices=_SINK_NAMES, default=None)
+    list_cmd.add_argument(
+        "--pending-only", action="store_true",
+        help="only entries that require review and have no review record yet",
+    )
+    list_cmd.add_argument("--json", action="store_true")
+    list_cmd.set_defaults(func=cmd_proposal_queue_list)
+
+    show = sub.add_parser("show", help="print one entry's full artifact JSON")
+    show.add_argument("sink", choices=_SINK_NAMES)
+    show.add_argument("content_hash")
+    show.set_defaults(func=cmd_proposal_queue_show)
+
+    review = sub.add_parser(
+        "review",
+        help="record that a human reviewed one entry (append-only; no ratification)",
+    )
+    review.add_argument("sink", choices=_SINK_NAMES)
+    review.add_argument("content_hash")
+    review.add_argument("--note", default=None, help="optional free-text note")
+    review.set_defaults(func=cmd_proposal_queue_review)
+
+
+__all__ = ["register"]

core/cli_test.py

@@ -75,6 +75,7 @@ TEST_SUITES: dict[str, tuple[str, ...]] = {
         "tests/test_math_frame_ratification.py",
         "tests/test_math_composition_ratification.py",
         "tests/test_teaching_coverage_cli.py",
+        "tests/test_proposal_queue.py",
     ),
     "packs": (
         "tests/test_pack_draft_serve_boundary.py",

core/proposal_review/__init__.py

@@ -1,16 +1,33 @@
-"""Read-only proposal review reporter (RPT) — surfaces comprehension-failure proposals for review.
+"""Proposal review reporter (RPT) — surfaces contemplation/idle proposals for review.
 
 Observes ``teaching/proposals/comprehension_failures/*.json`` (emitted by the contemplation pass,
-N5), validates them, and reports pending review obligations. It is **read-only**: it does not
-advance the teaching loop, ratify, mount, modify readers, or affect serving. It is **not** an
-``idle_tick`` (``ChatRuntime.idle_tick`` remains the only one) and **not** L10 — it is the review
-surface that keeps proposal artifacts from becoming inert files. A future PR may call this reporter
-from ``idle_tick`` as a read-only sub-pass.
+N5), validates them, and reports pending review obligations. It does not advance the teaching
+loop, ratify, mount, modify readers, or affect serving. It is **not** an ``idle_tick``
+(``ChatRuntime.idle_tick`` remains the only one) and **not** L10 — it is the review surface that
+keeps proposal artifacts from becoming inert files. A future PR may call this reporter from
+``idle_tick`` as a read-only sub-pass.
+
+:mod:`core.proposal_review.queue` (Tier S4) extends this to a second sink
+(``derived_close_facts``) and adds human review-STATE tracking — an append-only sidecar log
+(``teaching/proposals/review_log.jsonl``) recording that a human looked at an artifact. That sidecar
+is the one write anywhere in this package: it never touches a sink artifact's own ``status`` /
+``mounted`` / ``requires_review`` fields, never ratifies, and never mounts anything — ratification
+stays ``teaching/proposals.py``'s separate ADR-0057 corridor (``core teaching proposals`` / ``core
+teaching review``, a different sink entirely).
 """
 
 from __future__ import annotations
 
 from core.proposal_review.model import MalformedArtifact, PendingProposal
+from core.proposal_review.queue import (
+    MalformedEntry,
+    QueueEntry,
+    ReviewRecord,
+    queue_status,
+    record_review,
+    reviewed_keys,
+    scan_all,
+)
 from core.proposal_review.report import (
     ProposalReviewReport,
     build_report,
@@ -24,15 +41,22 @@ from core.proposal_review.summary import ProposalReviewIdleSummary, idle_summary
 __all__ = [
     "DEFAULT_SINK",
     "MalformedArtifact",
+    "MalformedEntry",
     "PendingProposal",
     "ProposalReviewIdleSummary",
     "ProposalReviewReport",
+    "QueueEntry",
+    "ReviewRecord",
     "SafetyVerdict",
     "build_report",
     "default_sink",
     "dry_check",
     "idle_summary",
+    "queue_status",
+    "record_review",
     "report_json",
     "report_text",
+    "reviewed_keys",
     "scan",
+    "scan_all",
 ]

core/proposal_review/queue.py

@@ -0,0 +1,301 @@
+"""core/proposal_review/queue.py — generic multi-sink proposal queue (Tier S4).
+
+**Read + review-state transitions only.** NEVER ratification, NEVER corpus
+mutation, NEVER flag flips — those stay `teaching/proposals.py`'s own
+`accept_proposal`/`reject_proposal` path over a DIFFERENT sink
+(`teaching/proposals/proposals.jsonl`, already CLI-exposed via
+``core teaching proposals`` / ``core teaching review``, ADR-0057).
+
+This module covers the other two proposal sinks the generalization-arc
+brief calls "contemplation/idle sinks" — populated by background passes,
+not by the ratification corridor, and until now readable only through a
+Python API with no CLI surface at all:
+
+- ``comprehension_failures`` (the N5 contemplation pass) — already has a
+  hardened typed reader and an independent safety dry-check:
+  :mod:`core.proposal_review` (``scan`` / ``dry_check``). Reused here, not
+  duplicated.
+- ``derived_close_facts`` (the idle_tick PR-2 bridge,
+  :mod:`generate.determine.derived_close_proposals`) — emitter only, no
+  reader existed before this module. Its own docstring says it is
+  "reviewable by the same HITL tooling" as ``comprehension_failures`` —
+  this is that tooling. Read GENERICALLY here (the shared minimal
+  ``status``/``mounted``/``requires_review`` contract every proposal-only
+  artifact carries), not via a new typed dataclass: the sink is currently
+  empty and default-off (``review_derived_close_proposals``), so committing
+  to a schema-specific reader now would be speculative in a way the
+  populated ``comprehension_failures`` sink is not.
+
+"Reviewing" here means recording that a HUMAN looked at an artifact — an
+append-only sidecar log (``teaching/proposals/review_log.jsonl``), never a
+write to the artifact itself. The artifact's own ``status`` / ``mounted`` /
+``requires_review`` fields stay exactly what the emitter wrote; nothing in
+this module can change what a proposal ratifies to.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from core.proposal_review.scan import DEFAULT_SINK as _COMPREHENSION_SINK
+from core.proposal_review.scan import scan as _scan_comprehension
+from generate.determine.derived_close_proposals import DEFAULT_SINK as _DERIVED_CLOSE_SINK
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+
+#: The append-only human-review sidecar. Lives beside the sinks it reviews,
+#: not inside either of them — a review record is metadata ABOUT an
+#: artifact, never a mutation of it.
+REVIEW_LOG_PATH = _REPO_ROOT / "teaching" / "proposals" / "review_log.jsonl"
+
+#: Fields every proposal-only artifact in EITHER sink carries, regardless of
+#: family-specific schema (verified against both emitters:
+#: `core/comprehension_attempt/proposal.py`,
+#: `generate/determine/derived_close_proposals.py`).
+_SHARED_REQUIRED: tuple[str, ...] = ("status", "requires_review", "mounted")
+
+
+@dataclass(frozen=True, slots=True)
+class QueueEntry:
+    """One proposal artifact, sink-agnostic. ``family`` is a short label for
+    display only (``failure_family`` for comprehension-failures, ``source``
+    for derived-close-facts) — it carries no safety meaning here."""
+
+    sink: str
+    content_hash: str
+    family: str
+    status: str
+    requires_review: bool
+    mounted: bool
+    path: str
+
+
+@dataclass(frozen=True, slots=True)
+class MalformedEntry:
+    sink: str
+    path: str
+    reason: str
+
+
+@dataclass(frozen=True, slots=True)
+class SinkSpec:
+    name: str
+    path: Path
+    label: str
+
+
+#: The two contemplation/idle sinks this queue reads. Order is display order.
+SINKS: tuple[SinkSpec, ...] = (
+    SinkSpec(
+        "comprehension_failures",
+        _COMPREHENSION_SINK,
+        "comprehension-failure proposals (N5 contemplation pass)",
+    ),
+    SinkSpec(
+        "derived_close_facts",
+        _DERIVED_CLOSE_SINK,
+        "derived CLOSE-fact proposals (idle_tick PR-2 bridge)",
+    ),
+)
+
+_SINKS_BY_NAME: dict[str, SinkSpec] = {spec.name: spec for spec in SINKS}
+
+
+def _scan_derived_close(root: Path) -> tuple[list[QueueEntry], list[MalformedEntry]]:
+    """Generic reader for ``derived_close_facts`` — no typed dataclass (module
+    docstring): validates only the shared minimal contract, tolerating the
+    family's own extra fields (``predicate``/``subject``/``object``/...)."""
+    if not root.exists():
+        return [], []
+    entries: list[QueueEntry] = []
+    malformed: list[MalformedEntry] = []
+    for path in sorted(root.glob("*.json")):
+        try:
+            raw: Any = json.loads(path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+            malformed.append(MalformedEntry("derived_close_facts", str(path), f"invalid_json: {exc}"))
+            continue
+        if not isinstance(raw, dict):
+            malformed.append(MalformedEntry("derived_close_facts", str(path), "not_a_json_object"))
+            continue
+        missing = [key for key in _SHARED_REQUIRED if key not in raw]
+        if missing:
+            malformed.append(
+                MalformedEntry("derived_close_facts", str(path), f"missing_fields: {missing}")
+            )
+            continue
+        predicate = str(raw.get("predicate", ""))
+        subject = str(raw.get("subject", ""))
+        obj = str(raw.get("object", ""))
+        family = f"{predicate}:{subject}:{obj}" if predicate else str(raw.get("source", "unknown"))
+        entries.append(
+            QueueEntry(
+                sink="derived_close_facts",
+                content_hash=path.stem,
+                family=family,
+                status=str(raw["status"]),
+                requires_review=bool(raw["requires_review"]),
+                mounted=bool(raw["mounted"]),
+                path=str(path),
+            )
+        )
+    return entries, malformed
+
+
+def scan_all(
+    sink_names: tuple[str, ...] | None = None,
+    *,
+    roots: dict[str, Path] | None = None,
+) -> tuple[list[QueueEntry], list[MalformedEntry]]:
+    """Scan the requested sinks (default: all). Pure read, sorted by
+    ``(sink, content_hash)`` for a deterministic listing order. ``roots``
+    overrides a sink's directory by name — test isolation, never used by the
+    CLI (which always reads the real sinks)."""
+    names = sink_names or tuple(spec.name for spec in SINKS)
+    entries: list[QueueEntry] = []
+    malformed: list[MalformedEntry] = []
+    for name in names:
+        spec = _SINKS_BY_NAME[name]
+        root = (roots or {}).get(name, spec.path)
+        if name == "comprehension_failures":
+            proposals, bad = _scan_comprehension(root)
+            entries.extend(
+                QueueEntry(
+                    sink=name,
+                    content_hash=p.content_hash,
+                    family=p.failure_family,
+                    status=p.status,
+                    requires_review=p.requires_review,
+                    mounted=p.mounted,
+                    path=p.path,
+                )
+                for p in proposals
+            )
+            malformed.extend(MalformedEntry(name, m.path, m.reason) for m in bad)
+        elif name == "derived_close_facts":
+            good, close_bad = _scan_derived_close(root)
+            entries.extend(good)
+            malformed.extend(close_bad)
+        else:  # pragma: no cover - closed SINKS tuple above
+            raise ValueError(f"unknown proposal-queue sink: {name!r}")
+    entries.sort(key=lambda e: (e.sink, e.content_hash))
+    malformed.sort(key=lambda m: (m.sink, m.path))
+    return entries, malformed
+
+
+@dataclass(frozen=True, slots=True)
+class ReviewRecord:
+    sink: str
+    content_hash: str
+    note: str
+    reviewed_at: str
+
+
+def _load_review_log(path: Path | None = None) -> list[ReviewRecord]:
+    log_path = path or REVIEW_LOG_PATH
+    if not log_path.exists():
+        return []
+    records: list[ReviewRecord] = []
+    for line in log_path.read_text(encoding="utf-8").splitlines():
+        if not line.strip():
+            continue
+        row = json.loads(line)
+        records.append(
+            ReviewRecord(
+                sink=row["sink"],
+                content_hash=row["content_hash"],
+                note=row.get("note", ""),
+                reviewed_at=row["reviewed_at"],
+            )
+        )
+    return records
+
+
+def reviewed_keys(path: Path | None = None) -> frozenset[tuple[str, str]]:
+    """``(sink, content_hash)`` pairs that have at least one review record."""
+    return frozenset((r.sink, r.content_hash) for r in _load_review_log(path))
+
+
+def record_review(
+    sink: str,
+    content_hash: str,
+    *,
+    note: str = "",
+    path: Path | None = None,
+    roots: dict[str, Path] | None = None,
+) -> ReviewRecord:
+    """Append one human-review record. Does NOT touch the artifact itself,
+    does not ratify, does not mount, does not flip any flag — purely an
+    append to the sidecar log. Raises ``KeyError`` if ``(sink, content_hash)``
+    does not resolve in a fresh scan, so a typo cannot silently log a review
+    of nothing."""
+    entries, _malformed = scan_all((sink,), roots=roots)
+    if not any(e.content_hash == content_hash for e in entries):
+        raise KeyError(f"no pending entry {content_hash!r} in sink {sink!r}")
+    log_path = path or REVIEW_LOG_PATH
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    record = ReviewRecord(
+        sink=sink,
+        content_hash=content_hash,
+        note=note,
+        reviewed_at=datetime.now(timezone.utc).isoformat(),
+    )
+    with log_path.open("a", encoding="utf-8") as fh:
+        fh.write(json.dumps(asdict(record), sort_keys=True) + "\n")
+    return record
+
+
+def queue_status(
+    sink_names: tuple[str, ...] | None = None,
+    *,
+    roots: dict[str, Path] | None = None,
+    log_path: Path | None = None,
+) -> dict[str, Any]:
+    """Per-sink counts: total, pending_review (requires_review AND not yet
+    human-reviewed), reviewed (has a review record), no_review_needed
+    (requires_review is False — not currently emitted by either sink, kept
+    distinct rather than folded into "reviewed" for correctness if that ever
+    changes), malformed. Not a safety verdict — for ``comprehension_failures``,
+    use :func:`core.proposal_review.dry_check` for that; this is a triage
+    summary only."""
+    entries, malformed = scan_all(sink_names, roots=roots)
+    reviewed = reviewed_keys(log_path)
+    by_sink: dict[str, dict[str, int]] = {}
+    for spec in SINKS:
+        if sink_names is not None and spec.name not in sink_names:
+            continue
+        sink_entries = [e for e in entries if e.sink == spec.name]
+        no_review_needed = sum(1 for e in sink_entries if not e.requires_review)
+        has_review = sum(
+            1 for e in sink_entries if (e.sink, e.content_hash) in reviewed
+        )
+        pending = sum(
+            1 for e in sink_entries
+            if e.requires_review and (e.sink, e.content_hash) not in reviewed
+        )
+        by_sink[spec.name] = {
+            "total": len(sink_entries),
+            "pending_review": pending,
+            "reviewed": has_review,
+            "no_review_needed": no_review_needed,
+            "malformed": sum(1 for m in malformed if m.sink == spec.name),
+        }
+    return by_sink
+
+
+__all__ = [
+    "REVIEW_LOG_PATH",
+    "MalformedEntry",
+    "QueueEntry",
+    "ReviewRecord",
+    "SINKS",
+    "SinkSpec",
+    "queue_status",
+    "record_review",
+    "reviewed_keys",
+    "scan_all",
+]

docs/research/hitl-proposal-queue-cli-2026-07-24.md

@@ -0,0 +1,98 @@
+# HITL Proposal-Queue CLI — `core proposal-queue`
+
+**Tier S4** (generalization-arc-2026-07-24 §6). A `core` CLI surface for
+listing and reviewing pending proposals from the contemplation/idle sinks —
+`teaching/proposals/comprehension_failures/` (the N5 contemplation pass) and
+`teaching/proposals/derived_close_facts/` (the idle_tick PR-2 bridge). Read +
+review-state transitions only: no command here ratifies, mutates the
+teaching corpus, mounts anything, or flips a flag. That corridor
+(`teaching/proposals.py`'s `accept_proposal`/`reject_proposal`, a DIFFERENT
+sink — `teaching/proposals/proposals.jsonl`, ADR-0057) already has its own
+CLI (`core teaching proposals` / `core teaching review`) and is untouched.
+
+## What shipped
+
+- `core/proposal_review/queue.py` — a sink-agnostic scanner
+  (`scan_all`), an append-only human-review sidecar log
+  (`teaching/proposals/review_log.jsonl`, via `record_review` /
+  `reviewed_keys`), and a triage summary (`queue_status`).
+- `core/cli_proposal_queue.py` — `core proposal-queue {status,list,show,review}`,
+  registered via the same `register(subparsers)` pattern
+  `core/cli_ingest.py` and `formation/cli.py` use.
+- `tests/test_proposal_queue.py` (13 tests, registered in the `teaching`
+  suite), fully isolated from the real sinks via `roots=`/`path=`
+  overrides — never touches the real `teaching/proposals/review_log.jsonl`.
+
+```
+core proposal-queue status [--sink NAME] [--json]
+core proposal-queue list [--sink NAME] [--pending-only] [--json]
+core proposal-queue show <sink> <content-hash>
+core proposal-queue review <sink> <content-hash> [--note TEXT]
+```
+
+`review` only appends a `{sink, content_hash, note, reviewed_at}` record to
+the sidecar log after confirming the hash resolves in a fresh scan (a typo
+cannot silently "review" nothing); it never opens the artifact file for
+writing. `test_record_review_does_not_touch_the_artifact` pins that
+byte-for-byte.
+
+## A real bug found while building the second sink's reader
+
+`generate/determine/derived_close_proposals.py::DEFAULT_SINK` computed
+`Path(__file__).resolve().parents[3] / "teaching" / "proposals" /
+"derived_close_facts"`. From `<repo>/generate/determine/derived_close_proposals.py`,
+`parents[2]` is the repo root (0=`determine/`, 1=`generate/`, 2=`<repo>`) —
+`parents[3]` is one directory **above** the repository. Confirmed live: a
+`core proposal-queue list` run before the fix listed an entry at
+`/Users/kaizenpro/Projects/teaching/proposals/derived_close_facts/…` — a
+directory sitting outside `core/` entirely, containing one stray artifact
+dated 2026-06-16, untracked by git, invisible to anything that searches
+inside the repository. Fixed to `parents[2]`
+(`test_derived_close_default_sink_resolves_inside_the_repo` pins it).
+
+**Why nothing caught this before:** the sink is gated by a default-off flag
+(`review_derived_close_proposals`) and no existing test asserted
+`DEFAULT_SINK` resolves inside the repo — its own tests, where they exist,
+pass an explicit `sink=` override. The bug was silent because the feature
+has essentially never run in its default configuration. The stray file
+outside the repo was left as-is (it is not tracked by git and this session
+has no standing to delete files outside the repository without being
+asked); a future real run of `emit_derived_close_proposals` will now write
+to the correct, in-repo location.
+
+## Design choices, made explicit
+
+- **No typed dataclass for `derived_close_facts`.** Unlike
+  `comprehension_failures` (which has a hardened `PendingProposal` reader
+  and an independent safety `dry_check`, both reused unchanged here), this
+  queue reads `derived_close_facts` generically — only the shared minimal
+  contract (`status`/`requires_review`/`mounted`) is validated. Committing
+  to a schema-specific reader for a sink that is currently empty and
+  default-off would be speculative; the generic reader is honest about
+  that and will not silently misclassify whatever the emitter's schema
+  settles into.
+- **No new safety dry-check for `derived_close_facts`.** The existing
+  `core/proposal_review/safety.py::dry_check` is scoped to
+  `comprehension_failures`'s specific content-address formula. Extending
+  it to the second sink's different dedupe-key formula would be a real
+  design decision about what "inert" means for that family — out of
+  scope for a read/review CLI, and not requested.
+- **Review vocabulary stays minimal: a note, not a state machine.** No spec
+  exists for review-state vocabulary (accept/flag/defer/etc.); inventing
+  one would risk becoming ratification-adjacent policy. `record_review`
+  answers exactly one question — "has a human looked at this?" — with an
+  optional free-text note, which is the smallest thing that satisfies
+  "review-state transitions" without overstepping into ratification.
+
+## Non-goals
+
+- Does not touch `teaching/proposals.py`, `accept_proposal`, or
+  `core teaching proposals`/`review` — that corridor is unchanged.
+- Does not run `emit_derived_close_proposals` or otherwise populate either
+  sink; both are read as they currently stand (one real
+  `comprehension_failures` entry, zero `derived_close_facts` entries once
+  read from the corrected path).
+- Does not wire `core proposal-queue` into `idle_tick` or any automated
+  loop — operator-invoked only, per the brief's HITL framing.
+
+Relates to [[project-generalization-arc]].

generate/determine/derived_close_proposals.py

@@ -26,8 +26,16 @@ from generate.realize import RealizedRecord, recall_realized
 
 #: Dedicated sink for derived CLOSE proposals (distinct from comprehension-failures
 #: to keep families clean while still reviewable by the same HITL tooling).
+#: ``parents[2]`` reaches the repo root from
+#: ``<repo>/generate/determine/derived_close_proposals.py`` (0=determine/,
+#: 1=generate/, 2=<repo>) — found and fixed while building the S4 HITL
+#: proposal-queue CLI (`core/proposal_review/queue.py`): the prior
+#: ``parents[3]`` resolved one directory ABOVE the repo, silently writing/
+#: reading outside git entirely. Dormant because the feature default-off
+#: flag (``review_derived_close_proposals``) means this sink has never been
+#: populated in normal use.
 DEFAULT_SINK = (
-    Path(__file__).resolve().parents[3]
+    Path(__file__).resolve().parents[2]
     / "teaching"
     / "proposals"
     / "derived_close_facts"

tests/test_proposal_queue.py

@@ -0,0 +1,224 @@
+"""tests/test_proposal_queue.py — HITL proposal-queue (Tier S4).
+
+Fully isolated: every test passes its own ``roots``/``log_path`` (or
+``tmp_path`` fixtures) rather than touching the real
+``teaching/proposals/`` sinks or ``teaching/proposals/review_log.jsonl`` —
+that file is real operator data, never a test fixture.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+import pytest
+
+from core.proposal_review.queue import (
+    QueueEntry,
+    queue_status,
+    record_review,
+    reviewed_keys,
+    scan_all,
+)
+from generate.determine.derived_close_proposals import DEFAULT_SINK as DERIVED_CLOSE_SINK
+
+
+def _write_comprehension_artifact(root: Path, content_hash: str, **overrides: object) -> None:
+    root.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "status": "proposal_only",
+        "failure_family": "missing_total_count",
+        "problem_text_sha256": "a" * 64,
+        "mounted": False,
+        "requires_review": True,
+        "observed_attempts": [],
+        **overrides,
+    }
+    (root / f"{content_hash}.json").write_text(json.dumps(payload), encoding="utf-8")
+
+
+def _write_derived_close_artifact(root: Path, content_hash: str, **overrides: object) -> None:
+    root.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "source": "derived_close_fact",
+        "predicate": "member",
+        "subject": "socrates",
+        "object": "mortal",
+        "status": "proposal_only",
+        "requires_review": True,
+        "mounted": False,
+        **overrides,
+    }
+    (root / f"{content_hash}.json").write_text(json.dumps(payload), encoding="utf-8")
+
+
+def test_derived_close_default_sink_resolves_inside_the_repo() -> None:
+    """Pins the off-by-one fix: parents[2], not parents[3] — the sink must
+    resolve under <repo>/teaching/proposals/derived_close_facts, not one
+    directory above the repository."""
+    repo_root = Path(__file__).resolve().parents[1]
+    assert DERIVED_CLOSE_SINK == repo_root / "teaching" / "proposals" / "derived_close_facts"
+    assert DERIVED_CLOSE_SINK.is_relative_to(repo_root)
+
+
+def test_scan_all_reads_both_sinks_via_roots_override(tmp_path: Path) -> None:
+    comp_root = tmp_path / "comprehension_failures"
+    close_root = tmp_path / "derived_close_facts"
+    _write_comprehension_artifact(comp_root, "a" * 64)
+    _write_derived_close_artifact(close_root, "b" * 64)
+
+    entries, malformed = scan_all(roots={"comprehension_failures": comp_root, "derived_close_facts": close_root})
+
+    assert malformed == []
+    assert {e.sink for e in entries} == {"comprehension_failures", "derived_close_facts"}
+    comp_entry = next(e for e in entries if e.sink == "comprehension_failures")
+    assert comp_entry.family == "missing_total_count"
+    close_entry = next(e for e in entries if e.sink == "derived_close_facts")
+    assert close_entry.family == "member:socrates:mortal"
+
+
+def test_scan_all_sink_filter(tmp_path: Path) -> None:
+    comp_root = tmp_path / "comprehension_failures"
+    close_root = tmp_path / "derived_close_facts"
+    _write_comprehension_artifact(comp_root, "a" * 64)
+    _write_derived_close_artifact(close_root, "b" * 64)
+
+    entries, _ = scan_all(
+        ("comprehension_failures",),
+        roots={"comprehension_failures": comp_root, "derived_close_facts": close_root},
+    )
+    assert [e.sink for e in entries] == ["comprehension_failures"]
+
+
+def test_scan_all_missing_sink_directory_is_empty_not_an_error(tmp_path: Path) -> None:
+    entries, malformed = scan_all(
+        ("derived_close_facts",), roots={"derived_close_facts": tmp_path / "does_not_exist"}
+    )
+    assert entries == []
+    assert malformed == []
+
+
+def test_derived_close_malformed_missing_required_field(tmp_path: Path) -> None:
+    root = tmp_path / "derived_close_facts"
+    root.mkdir(parents=True)
+    (root / ("c" * 64 + ".json")).write_text(
+        json.dumps({"status": "proposal_only", "mounted": False}), encoding="utf-8"
+    )
+    _entries, malformed = scan_all(("derived_close_facts",), roots={"derived_close_facts": root})
+    assert len(malformed) == 1
+    assert "missing_fields" in malformed[0].reason
+
+
+def test_derived_close_malformed_invalid_json(tmp_path: Path) -> None:
+    root = tmp_path / "derived_close_facts"
+    root.mkdir(parents=True)
+    (root / "d.json").write_text("not json", encoding="utf-8")
+    _entries, malformed = scan_all(("derived_close_facts",), roots={"derived_close_facts": root})
+    assert len(malformed) == 1
+    assert "invalid_json" in malformed[0].reason
+
+
+def test_record_review_unknown_hash_raises(tmp_path: Path) -> None:
+    comp_root = tmp_path / "comprehension_failures"
+    _write_comprehension_artifact(comp_root, "a" * 64)
+    with pytest.raises(KeyError):
+        record_review(
+            "comprehension_failures",
+            "nonexistent",
+            path=tmp_path / "review_log.jsonl",
+            roots={"comprehension_failures": comp_root},
+        )
+
+
+def test_record_review_does_not_touch_the_artifact(tmp_path: Path) -> None:
+    comp_root = tmp_path / "comprehension_failures"
+    content_hash = "a" * 64
+    _write_comprehension_artifact(comp_root, content_hash)
+    artifact_path = comp_root / f"{content_hash}.json"
+    before = artifact_path.read_text(encoding="utf-8")
+
+    record_review(
+        "comprehension_failures",
+        content_hash,
+        note="looks fine",
+        path=tmp_path / "review_log.jsonl",
+        roots={"comprehension_failures": comp_root},
+    )
+
+    after = artifact_path.read_text(encoding="utf-8")
+    assert before == after  # the review NEVER mutates the artifact
+
+
+def test_record_review_then_reviewed_keys(tmp_path: Path) -> None:
+    comp_root = tmp_path / "comprehension_failures"
+    content_hash = "a" * 64
+    _write_comprehension_artifact(comp_root, content_hash)
+    log_path = tmp_path / "review_log.jsonl"
+
+    assert reviewed_keys(log_path) == frozenset()
+    record_review(
+        "comprehension_failures", content_hash, note="ok",
+        path=log_path, roots={"comprehension_failures": comp_root},
+    )
+    assert reviewed_keys(log_path) == {("comprehension_failures", content_hash)}
+
+
+def test_reviewed_keys_missing_log_is_empty(tmp_path: Path) -> None:
+    assert reviewed_keys(tmp_path / "no_such_log.jsonl") == frozenset()
+
+
+def test_queue_status_counts(tmp_path: Path) -> None:
+    comp_root = tmp_path / "comprehension_failures"
+    close_root = tmp_path / "derived_close_facts"
+    _write_comprehension_artifact(comp_root, "a" * 64)
+    _write_comprehension_artifact(comp_root, "b" * 64)
+    _write_derived_close_artifact(close_root, "c" * 64)
+    log_path = tmp_path / "review_log.jsonl"
+
+    record_review(
+        "comprehension_failures", "a" * 64, note="reviewed",
+        path=log_path, roots={"comprehension_failures": comp_root},
+    )
+
+    status = queue_status(
+        roots={"comprehension_failures": comp_root, "derived_close_facts": close_root},
+        log_path=log_path,
+    )
+    assert status["comprehension_failures"] == {
+        "total": 2, "pending_review": 1, "reviewed": 1,
+        "no_review_needed": 0, "malformed": 0,
+    }
+    assert status["derived_close_facts"] == {
+        "total": 1, "pending_review": 1, "reviewed": 0,
+        "no_review_needed": 0, "malformed": 0,
+    }
+
+
+def test_queue_entry_is_the_expected_shape() -> None:
+    entry = QueueEntry(
+        sink="comprehension_failures", content_hash="x", family="f",
+        status="proposal_only", requires_review=True, mounted=False, path="/tmp/x.json",
+    )
+    assert entry.sink == "comprehension_failures"
+
+
+def test_cli_review_command_reports_error_on_unknown_hash(tmp_path: Path, monkeypatch, capsys) -> None:
+    from core.proposal_review import queue as queue_module
+    import core.cli_proposal_queue as cli_module
+
+    comp_root = tmp_path / "comprehension_failures"
+    _write_comprehension_artifact(comp_root, "a" * 64)
+    monkeypatch.setattr(
+        queue_module,
+        "SINKS",
+        (queue_module.SinkSpec("comprehension_failures", comp_root, "test"),),
+    )
+    monkeypatch.setattr(queue_module, "_SINKS_BY_NAME", {"comprehension_failures": queue_module.SINKS[0]})
+    monkeypatch.setattr(queue_module, "REVIEW_LOG_PATH", tmp_path / "review_log.jsonl")
+
+    args = argparse.Namespace(sink="comprehension_failures", content_hash="does-not-exist", note=None)
+    rc = cli_module.cmd_proposal_queue_review(args)
+    captured = capsys.readouterr()
+    assert rc == 2
+    assert "error" in captured.out