Merge pull request #635 from AssetOverflow/feat/proposal-review-c

feat(proposal-review): RPT-c — safety dry-check + read-only CLI [merge after B]

core/proposal_review/__init__.py

@@ -17,6 +17,7 @@ from core.proposal_review.report import (
     report_json,
     report_text,
 )
+from core.proposal_review.safety import SafetyVerdict, dry_check
 from core.proposal_review.scan import DEFAULT_SINK, default_sink, scan
 
 __all__ = [
@@ -24,8 +25,10 @@ __all__ = [
     "MalformedArtifact",
     "PendingProposal",
     "ProposalReviewReport",
+    "SafetyVerdict",
     "build_report",
     "default_sink",
+    "dry_check",
     "report_json",
     "report_text",
     "scan",

core/proposal_review/__main__.py

@@ -0,0 +1,60 @@
+"""CLI for the read-only proposal review reporter (RPT-c).
+
+    python -m core.proposal_review                    # text report + safety dry-check
+    python -m core.proposal_review comprehension-failures
+    python -m core.proposal_review --json             # machine-readable
+    python -m core.proposal_review --root <path>      # override the sink
+
+Read-only: scans, reports, and dry-checks the comprehension-failure proposal sink. **Mutates
+nothing.** Exit 0 iff the safety dry-check passes (every artifact inert + serving unconsumed).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from core.proposal_review.report import build_report, report_text
+from core.proposal_review.safety import dry_check
+from core.proposal_review.scan import scan
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        prog="python -m core.proposal_review",
+        description="Read-only comprehension-failure proposal review (observes; never mounts/ratifies).",
+    )
+    parser.add_argument(
+        "target", nargs="?", default="comprehension-failures", choices=["comprehension-failures"]
+    )
+    parser.add_argument("--json", action="store_true", help="emit JSON instead of text")
+    parser.add_argument("--root", default=None, help="override the sink path")
+    args = parser.parse_args(argv)
+
+    root = Path(args.root) if args.root else None
+    proposals, malformed = scan(root)
+    report = build_report(proposals, malformed)
+    verdict = dry_check(proposals, malformed, root=root)
+
+    if args.json:
+        print(
+            json.dumps(
+                {
+                    "report": report.to_json_dict(),
+                    "safety": {"ok": verdict.ok, "violations": list(verdict.violations)},
+                },
+                indent=2,
+                sort_keys=True,
+            )
+        )
+    else:
+        print(report_text(report))
+        print(f"  safety: {'OK' if verdict.ok else f'VIOLATIONS ({len(verdict.violations)})'}")
+        for v in verdict.violations:
+            print(f"    ! {v}")
+    return 0 if verdict.ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

core/proposal_review/safety.py

@@ -0,0 +1,102 @@
+"""Safety dry-check over the proposal sink (RPT-c) — the load-bearing part of the reporter.
+
+The reporter's value is not just visibility; it is **independent safety verification**. The
+dry-check confirms — without trusting the emitter — that every artifact in the sink is inert:
+
+```text
+status == "proposal_only"
+mounted == false
+requires_review == true
+content-address consistent: filename == sha256(failure_family : problem_text_sha256)
+path under the sink
+no malformed artifact (an unverifiable file in a safety-critical sink is a violation)
+serving never imports/reads the sink
+```
+
+Any failure is a violation; the CLI exits non-zero. **Pure read** — verifies, never repairs.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from dataclasses import dataclass
+from pathlib import Path
+
+from core.proposal_review.model import MalformedArtifact, PendingProposal
+from core.proposal_review.scan import DEFAULT_SINK
+
+#: Serving-path roots that must never read the proposal sink (CLAUDE.md forbidden/serving sites).
+_SERVING_TARGETS = (
+    ("generate", "stream.py"),
+    ("field", "propagate.py"),
+    ("vault", "store.py"),
+    ("generate", "derivation"),
+    ("core", "reliability_gate"),
+)
+_SINK_MARKER = "comprehension_failures"
+
+
+@dataclass(frozen=True, slots=True)
+class SafetyVerdict:
+    """The outcome of the dry-check: ``ok`` iff there are no violations."""
+
+    ok: bool
+    violations: tuple[str, ...]
+
+
+def _repo_root() -> Path:
+    return Path(__file__).resolve().parents[2]
+
+
+def _serving_references_sink(repo_root: Path) -> list[str]:
+    violations: list[str] = []
+    for parts in _SERVING_TARGETS:
+        target = repo_root.joinpath(*parts)
+        if not target.exists():
+            continue
+        files = [target] if target.is_file() else sorted(target.rglob("*.py"))
+        for f in files:
+            try:
+                if _SINK_MARKER in f.read_text(encoding="utf-8"):
+                    violations.append(f"serving reads the sink: {f.relative_to(repo_root)}")
+            except (UnicodeDecodeError, OSError):  # pragma: no cover - defensive
+                continue
+    return violations
+
+
+def dry_check(
+    proposals: list[PendingProposal],
+    malformed: list[MalformedArtifact],
+    *,
+    root: Path | None = None,
+    repo_root: Path | None = None,
+) -> SafetyVerdict:
+    """Verify every artifact is inert and the sink is serving-unconsumed. Returns a SafetyVerdict."""
+    base = (root if root is not None else DEFAULT_SINK).resolve()
+    violations: list[str] = []
+
+    for p in proposals:
+        tag = p.content_hash
+        if p.status != "proposal_only":
+            violations.append(f"{tag}: status={p.status!r} (must be 'proposal_only')")
+        if p.mounted:
+            violations.append(f"{tag}: mounted=True (must be False)")
+        if not p.requires_review:
+            violations.append(f"{tag}: requires_review=False (must be True)")
+        expected = hashlib.sha256(
+            f"{p.failure_family}:{p.problem_text_sha256}".encode("utf-8")
+        ).hexdigest()
+        if p.content_hash != expected:
+            violations.append(f"{tag}: content-address mismatch (expected {expected})")
+        if not str(Path(p.path).resolve()).startswith(str(base)):
+            violations.append(f"{tag}: path outside the sink")
+
+    for m in malformed:
+        violations.append(f"malformed (unverifiable): {m.path} — {m.reason}")
+
+    violations.extend(_serving_references_sink(repo_root if repo_root is not None else _repo_root()))
+
+    return SafetyVerdict(not violations, tuple(violations))
+
+
+__all__ = ["SafetyVerdict", "dry_check"]

tests/test_proposal_review_safety.py

@@ -0,0 +1,90 @@
+"""Tests for the proposal-review safety dry-check + CLI (RPT-c).
+
+Each safety assertion is proven meaningful-fail: planting exactly one violating artifact makes
+the dry-check fail on exactly that check. Also confirms serving never reads the sink (real repo).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+from core.proposal_review.safety import dry_check
+from core.proposal_review.scan import scan
+from evals.constraint_oracle.runner import _load_r2_gold
+from generate.contemplation import contemplate
+
+
+def _emit(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 _write(root: Path, *, family: str = "missing_total_count", text_sha: str = "a" * 64,
+           name: str | None = None, **overrides: object) -> str:
+    doc = {
+        "status": "proposal_only", "failure_family": family, "problem_text_sha256": text_sha,
+        "observed_attempts": [], "suggested_next_fixture": None, "requires_review": True,
+        "mounted": False,
+    }
+    doc.update(overrides)
+    stem = name or hashlib.sha256(f"{family}:{text_sha}".encode("utf-8")).hexdigest()
+    (root / f"{stem}.json").write_text(json.dumps(doc), encoding="utf-8")
+    return stem
+
+
+def test_dry_check_passes_for_real_proposals(tmp_path: Path) -> None:
+    _emit(tmp_path)
+    verdict = dry_check(*scan(tmp_path), root=tmp_path)
+    assert verdict.ok and verdict.violations == ()
+
+
+def test_dry_check_flags_mounted(tmp_path: Path) -> None:
+    _write(tmp_path, mounted=True)
+    verdict = dry_check(*scan(tmp_path), root=tmp_path)
+    assert not verdict.ok and any("mounted" in v for v in verdict.violations)
+
+
+def test_dry_check_flags_non_proposal_status(tmp_path: Path) -> None:
+    _write(tmp_path, status="ratified")
+    verdict = dry_check(*scan(tmp_path), root=tmp_path)
+    assert not verdict.ok and any("status" in v for v in verdict.violations)
+
+
+def test_dry_check_flags_requires_review_false(tmp_path: Path) -> None:
+    _write(tmp_path, requires_review=False)
+    verdict = dry_check(*scan(tmp_path), root=tmp_path)
+    assert not verdict.ok and any("requires_review" in v for v in verdict.violations)
+
+
+def test_dry_check_flags_content_address_mismatch(tmp_path: Path) -> None:
+    _write(tmp_path, name="deadbeef")  # filename != sha256(family:text_sha)
+    verdict = dry_check(*scan(tmp_path), root=tmp_path)
+    assert not verdict.ok and any("content-address" in v for v in verdict.violations)
+
+
+def test_dry_check_flags_malformed(tmp_path: Path) -> None:
+    (tmp_path / "bad.json").write_text("{ not json", encoding="utf-8")
+    verdict = dry_check(*scan(tmp_path), root=tmp_path)
+    assert not verdict.ok and any("malformed" in v for v in verdict.violations)
+
+
+def test_serving_never_reads_the_sink_real_repo() -> None:
+    # Against the real tree: no serving module references the sink, so an empty sink is OK.
+    verdict = dry_check([], [], root=None)
+    assert verdict.ok and not any("serving reads" in v for v in verdict.violations)
+
+
+def test_cli_exit_codes(tmp_path: Path) -> None:
+    from core.proposal_review.__main__ import main
+
+    _emit(tmp_path)
+    assert main(["--root", str(tmp_path)]) == 0
+    assert main(["--root", str(tmp_path), "--json"]) == 0
+    _write(tmp_path, mounted=True)
+    assert main(["--root", str(tmp_path)]) == 1