From cc3003f425c9ee5910fa3b132d255a2034cb17c5 Mon Sep 17 00:00:00 2001 From: Shay Date: Sun, 7 Jun 2026 21:43:18 -0700 Subject: [PATCH] feat(proposal-review): safety dry-check + read-only CLI (RPT-c) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit core/proposal_review/safety.py: dry_check INDEPENDENTLY verifies (without trusting the emitter) that every artifact is inert — status==proposal_only, mounted==false, requires_review==true, content-address consistent (filename == sha256(family:problem_text_sha256)), path under the sink, no malformed (unverifiable) file — and that no serving-path module reads the sink. Any failure is a violation. __main__.py: python -m core.proposal_review [comprehension-failures] [--json] [--root] — read-only report + dry-check; exit 0 iff safe. Mutates nothing. 8 tests, each safety assertion proven meaningful-fail; serving-unconsumed verified on the real tree. --- core/proposal_review/__init__.py | 3 + core/proposal_review/__main__.py | 60 ++++++++++++++++ core/proposal_review/safety.py | 102 +++++++++++++++++++++++++++ tests/test_proposal_review_safety.py | 90 +++++++++++++++++++++++ 4 files changed, 255 insertions(+) create mode 100644 core/proposal_review/__main__.py create mode 100644 core/proposal_review/safety.py create mode 100644 tests/test_proposal_review_safety.py diff --git a/core/proposal_review/__init__.py b/core/proposal_review/__init__.py index 3cdc6493..384e5daf 100644 --- a/core/proposal_review/__init__.py +++ 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", diff --git a/core/proposal_review/__main__.py b/core/proposal_review/__main__.py new file mode 100644 index 00000000..6b85c0ec --- /dev/null +++ b/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 # 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()) diff --git a/core/proposal_review/safety.py b/core/proposal_review/safety.py new file mode 100644 index 00000000..c539f980 --- /dev/null +++ b/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"] diff --git a/tests/test_proposal_review_safety.py b/tests/test_proposal_review_safety.py new file mode 100644 index 00000000..7de5dd7b --- /dev/null +++ b/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