Merge pull request #634 from AssetOverflow/feat/proposal-review-b

feat(proposal-review): RPT-b — deterministic review report [merge after A]

core/proposal_review/__init__.py

@@ -11,6 +11,22 @@ 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.report import (
+    ProposalReviewReport,
+    build_report,
+    report_json,
+    report_text,
+)
 from core.proposal_review.scan import DEFAULT_SINK, default_sink, scan
 
-__all__ = ["DEFAULT_SINK", "MalformedArtifact", "PendingProposal", "default_sink", "scan"]
+__all__ = [
+    "DEFAULT_SINK",
+    "MalformedArtifact",
+    "PendingProposal",
+    "ProposalReviewReport",
+    "build_report",
+    "default_sink",
+    "report_json",
+    "report_text",
+    "scan",
+]

core/proposal_review/report.py

@@ -0,0 +1,80 @@
+"""Deterministic review report over the scanned proposals (RPT-b).
+
+A pure projection of the scan results into a summary: total pending, counts by failure_family,
+counts by status, the malformed count, and the review-needed list. Deterministic given the sink
+contents (no clock): counts are sorted, the review-needed list is sorted by content hash.
+
+Time-based "oldest/newest" is intentionally **omitted**: the proposal artifacts are
+content-addressed and carry no timestamp (the emitter is clock-free for idempotence), so an honest
+temporal ordering is not available from the data — only from non-deterministic filesystem mtime,
+which would make this report non-deterministic. A human can sort the sink by mtime if needed.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Any
+
+from core.proposal_review.model import MalformedArtifact, PendingProposal
+
+
+@dataclass(frozen=True, slots=True)
+class ProposalReviewReport:
+    """A deterministic snapshot of the review obligations in the proposal sink."""
+
+    total: int
+    by_family: dict[str, int]
+    by_status: dict[str, int]
+    malformed: int
+    review_needed: tuple[str, ...]
+
+    def to_json_dict(self) -> dict[str, Any]:
+        return {
+            "total": self.total,
+            "by_family": self.by_family,
+            "by_status": self.by_status,
+            "malformed": self.malformed,
+            "review_needed": list(self.review_needed),
+        }
+
+
+def build_report(
+    proposals: list[PendingProposal], malformed: list[MalformedArtifact]
+) -> ProposalReviewReport:
+    by_family: dict[str, int] = {}
+    by_status: dict[str, int] = {}
+    review_needed: list[str] = []
+    for p in proposals:
+        by_family[p.failure_family] = by_family.get(p.failure_family, 0) + 1
+        by_status[p.status] = by_status.get(p.status, 0) + 1
+        if p.requires_review:
+            review_needed.append(p.content_hash)
+    return ProposalReviewReport(
+        total=len(proposals),
+        by_family=dict(sorted(by_family.items())),
+        by_status=dict(sorted(by_status.items())),
+        malformed=len(malformed),
+        review_needed=tuple(sorted(review_needed)),
+    )
+
+
+def report_json(report: ProposalReviewReport) -> str:
+    """Deterministic JSON (sorted keys)."""
+    return json.dumps(report.to_json_dict(), indent=2, sort_keys=True)
+
+
+def report_text(report: ProposalReviewReport) -> str:
+    """Human-readable summary."""
+    lines = [
+        f"comprehension-failure proposals: {report.total} pending · {report.malformed} malformed",
+        "  by family:",
+        *(f"    {fam}: {n}" for fam, n in report.by_family.items()),
+        "  by status:",
+        *(f"    {status}: {n}" for status, n in report.by_status.items()),
+        f"  review-needed: {len(report.review_needed)}",
+    ]
+    return "\n".join(lines)
+
+
+__all__ = ["ProposalReviewReport", "build_report", "report_json", "report_text"]

tests/test_proposal_review_report.py

@@ -0,0 +1,51 @@
+"""Tests for the deterministic proposal review report (RPT-b)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from core.proposal_review.report import build_report, report_json, report_text
+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 test_report_counts(tmp_path: Path) -> None:
+    _emit(tmp_path)
+    report = build_report(*scan(tmp_path))
+    assert report.total == 2
+    assert report.by_family == {"missing_total_count": 1, "missing_weighted_total": 1}
+    assert report.by_status == {"proposal_only": 2}
+    assert report.malformed == 0
+    assert len(report.review_needed) == 2
+
+
+def test_report_json_is_deterministic(tmp_path: Path) -> None:
+    _emit(tmp_path)
+    assert report_json(build_report(*scan(tmp_path))) == report_json(build_report(*scan(tmp_path)))
+
+
+def test_report_counts_malformed(tmp_path: Path) -> None:
+    (tmp_path / "bad.json").write_text("{ not json", encoding="utf-8")
+    report = build_report(*scan(tmp_path))
+    assert report.total == 0 and report.malformed == 1
+
+
+def test_report_text_lists_families(tmp_path: Path) -> None:
+    _emit(tmp_path)
+    text = report_text(build_report(*scan(tmp_path)))
+    assert "missing_total_count" in text and "2 pending" in text
+
+
+def test_empty_sink_report(tmp_path: Path) -> None:
+    report = build_report(*scan(tmp_path))
+    assert report.total == 0 and report.by_family == {} and report.review_needed == ()