feat(proposal-review): read-only scanner over the comprehension-failure sink (RPT-a)

core/proposal_review/{model,scan}.py: scan reads teaching/proposals/comprehension_failures/*.json into typed PendingProposal records (path, content_hash, failure_family, status, mounted, requires_review, problem_text_sha256, observed_attempts); any unparseable file becomes a MalformedArtifact (never silently dropped). Pure read — never writes/moves/deletes; deterministic (sorted by path); missing sink -> empty. The sink path is computed independently of the emitter (the reporter verifies the contract, doesn't import the writer).

Read-only reporter, NOT an idle_tick (ChatRuntime.idle_tick stays the only one), NOT L10. 4 tests incl. read-only invariant + malformed flagging.

core/proposal_review/__init__.py

@@ -0,0 +1,16 @@
+"""Read-only proposal review reporter (RPT) — surfaces comprehension-failure 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.
+"""
+
+from __future__ import annotations
+
+from core.proposal_review.model import MalformedArtifact, PendingProposal
+from core.proposal_review.scan import DEFAULT_SINK, default_sink, scan
+
+__all__ = ["DEFAULT_SINK", "MalformedArtifact", "PendingProposal", "default_sink", "scan"]

core/proposal_review/model.py

@@ -0,0 +1,39 @@
+"""Typed records for the read-only proposal review reporter (RPT-a).
+
+A ``PendingProposal`` is the parsed view of one ``teaching/proposals/comprehension_failures/
+<hash>.json`` artifact (emitted by the contemplation pass, N5). A ``MalformedArtifact`` is a file
+that could not be parsed into one. Pure value data — the reporter never mutates the sink.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True, slots=True)
+class PendingProposal:
+    """A parsed proposal artifact awaiting human review. ``content_hash`` is the filename stem
+    (the content address). Safety fields (``status`` / ``mounted`` / ``requires_review``) are
+    carried verbatim so the dry-check (RPT-c) can verify them independently of the emitter."""
+
+    path: str
+    content_hash: str
+    failure_family: str
+    status: str
+    mounted: bool
+    requires_review: bool
+    problem_text_sha256: str
+    observed_attempts: tuple[dict[str, Any], ...]
+
+
+@dataclass(frozen=True, slots=True)
+class MalformedArtifact:
+    """A file under the sink that is not a parseable proposal (bad JSON / missing or mistyped
+    fields). Surfaced so a human notices corruption rather than the reporter silently skipping it."""
+
+    path: str
+    reason: str
+
+
+__all__ = ["MalformedArtifact", "PendingProposal"]

core/proposal_review/scan.py

@@ -0,0 +1,85 @@
+"""Read-only scanner over the comprehension-failure proposal sink (RPT-a).
+
+Reads ``teaching/proposals/comprehension_failures/*.json`` into typed ``PendingProposal`` records;
+any file that does not parse into one is reported as a ``MalformedArtifact`` (never silently
+dropped). **Pure read** — opens files, never writes/moves/deletes. Deterministic: results are
+sorted by filename. The sink location is computed here independently of the emitter, so the
+reporter verifies the artifact contract without importing the writer.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from core.proposal_review.model import MalformedArtifact, PendingProposal
+
+#: The proposal sink the contemplation pass (N5) writes to — known independently of the emitter.
+DEFAULT_SINK = (
+    Path(__file__).resolve().parents[2] / "teaching" / "proposals" / "comprehension_failures"
+)
+
+#: Required fields and their JSON types for a well-formed proposal artifact.
+_REQUIRED: tuple[tuple[str, type | tuple[type, ...]], ...] = (
+    ("status", str),
+    ("failure_family", str),
+    ("problem_text_sha256", str),
+    ("mounted", bool),
+    ("requires_review", bool),
+    ("observed_attempts", list),
+)
+
+
+def default_sink() -> Path:
+    return DEFAULT_SINK
+
+
+def _parse(path: Path) -> PendingProposal | MalformedArtifact:
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+        return MalformedArtifact(str(path), f"invalid_json: {exc}")
+    if not isinstance(raw, dict):
+        return MalformedArtifact(str(path), "not_a_json_object")
+    for key, typ in _REQUIRED:
+        if key not in raw:
+            return MalformedArtifact(str(path), f"missing_field: {key}")
+        # bool is a subclass of int; check bools explicitly so 0/1 don't pass as bool.
+        if typ is bool and not isinstance(raw[key], bool):
+            return MalformedArtifact(str(path), f"bad_type: {key}")
+        if typ is not bool and not isinstance(raw[key], typ):
+            return MalformedArtifact(str(path), f"bad_type: {key}")
+    attempts: tuple[dict[str, Any], ...] = tuple(
+        a for a in raw["observed_attempts"] if isinstance(a, dict)
+    )
+    return PendingProposal(
+        path=str(path),
+        content_hash=path.stem,
+        failure_family=raw["failure_family"],
+        status=raw["status"],
+        mounted=raw["mounted"],
+        requires_review=raw["requires_review"],
+        problem_text_sha256=raw["problem_text_sha256"],
+        observed_attempts=attempts,
+    )
+
+
+def scan(root: Path | None = None) -> tuple[list[PendingProposal], list[MalformedArtifact]]:
+    """Scan the sink (default: the comprehension-failure sink). Returns ``(proposals, malformed)``,
+    each sorted by path. A missing sink yields two empty lists (nothing to review yet)."""
+    base = root if root is not None else DEFAULT_SINK
+    if not base.exists():
+        return [], []
+    proposals: list[PendingProposal] = []
+    malformed: list[MalformedArtifact] = []
+    for path in sorted(base.glob("*.json")):
+        parsed = _parse(path)
+        if isinstance(parsed, PendingProposal):
+            proposals.append(parsed)
+        else:
+            malformed.append(parsed)
+    return proposals, malformed
+
+
+__all__ = ["DEFAULT_SINK", "default_sink", "scan"]

tests/test_proposal_review_scan.py

@@ -0,0 +1,56 @@
+"""Tests for the read-only proposal scanner (RPT-a).
+
+Emits real proposals via the contemplation pass into a tmp sink, scans them into typed
+PendingProposal records, flags malformed artifacts, and pins that scanning never mutates the sink.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from core.proposal_review.model import PendingProposal
+from core.proposal_review.scan import scan
+from evals.constraint_oracle.runner import _load_r2_gold
+from generate.contemplation import contemplate
+
+
+def _emit_gold_proposals(root: Path) -> None:
+    for fx in _load_r2_gold():
+        if fx["expect"] == "reader_refuses" and fx["reader_reason"] in (
+            "missing_total_count",
+            "missing_weighted_total",
+        ):
+            contemplate(fx["text"], proposal_root=root, case_id=fx["id"])
+
+
+def test_scan_reads_emitted_proposals(tmp_path: Path) -> None:
+    _emit_gold_proposals(tmp_path)
+    proposals, malformed = scan(tmp_path)
+    assert len(proposals) == 2 and malformed == []
+    assert all(isinstance(p, PendingProposal) for p in proposals)
+    assert {p.failure_family for p in proposals} == {"missing_total_count", "missing_weighted_total"}
+    assert all(p.status == "proposal_only" and not p.mounted and p.requires_review for p in proposals)
+    assert all(p.content_hash == Path(p.path).stem for p in proposals)
+
+
+def test_scan_flags_malformed_artifacts(tmp_path: Path) -> None:
+    (tmp_path / "bad.json").write_text("{ not valid json", encoding="utf-8")
+    (tmp_path / "incomplete.json").write_text(json.dumps({"status": "proposal_only"}), encoding="utf-8")
+    proposals, malformed = scan(tmp_path)
+    assert proposals == [] and len(malformed) == 2
+    reasons = " ".join(m.reason for m in malformed)
+    assert "invalid_json" in reasons and "missing_field" in reasons
+
+
+def test_scan_missing_sink_is_empty(tmp_path: Path) -> None:
+    assert scan(tmp_path / "does_not_exist") == ([], [])
+
+
+def test_scan_is_read_only(tmp_path: Path) -> None:
+    _emit_gold_proposals(tmp_path)
+    before = {p.name: p.read_bytes() for p in tmp_path.glob("*.json")}
+    scan(tmp_path)
+    scan(tmp_path)
+    after = {p.name: p.read_bytes() for p in tmp_path.glob("*.json")}
+    assert before == after and before  # unchanged, and non-empty (the proposals exist)