feat: require pagination proof for reconciliation inventory claims (Closes #308)

Add assess_reconciliation_inventory_proof to review_proofs:
already-landed reconciliation reports may not claim a complete queue
scan or that all already-landed PRs were found unless the report
carries the pagination proof fields (Requested page size, Pages
fetched, Returned PR count per page, Final page proof) and the
per-page gitea_list_prs pagination metadata proves the final page was
reached — inventory_complete, is_final_page with has_more=false, or a
returned count below the requested page size. Declared page counts and
per-page counts must match the evidence. Reports without a
completeness claim pass without proof. Fails closed on missing
evidence or malformed page metadata. Tests cover first page below
limit, first page exactly at the page limit, multi-page traversal to
the final page, missing pagination metadata, missing report fields,
page-count and per-page-count mismatches, and the no-claim case.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

review_proofs.py

@@ -1052,6 +1052,113 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non
     return None
 
 
+# ── Reconciliation inventory pagination proof (Issue #308) ───────────────────
+#
+# Already-landed reconciliation may not claim a complete queue scan (or
+# that every already-landed PR was found) unless the report carries the
+# pagination proof fields and the per-page gitea_list_prs metadata
+# proves the final page was reached or the page limit was not hit.
+
+_RECONCILE_COMPLETE_CLAIM_RE = re.compile(
+    r"complete queue scan|all already-landed|inventory complete|"
+    r"all open prs (?:were )?(?:inspected|scanned|listed)|found all",
+    re.IGNORECASE,
+)
+
+RECONCILIATION_INVENTORY_PROOF_FIELDS = (
+    "requested page size",
+    "pages fetched",
+    "returned pr count per page",
+    "final page proof",
+)
+
+
+def assess_reconciliation_inventory_proof(report_text, *,
+                                          pagination_evidence=None):
+    """#308: reconciliation completeness claims need pagination proof.
+
+    *pagination_evidence* is the list of per-page ``pagination`` dicts
+    from the session's ``gitea_list_prs`` calls, in fetch order. Reports
+    that do not claim inventory completeness pass without proof; any
+    completeness claim requires the report proof fields and evidence
+    that the final page was reached (``inventory_complete``,
+    ``is_final_page`` with ``has_more=false``, or a returned count below
+    the requested page size).
+
+    Returns {'complete', 'downgraded', 'reasons'}; fails closed.
+    """
+    reasons = []
+    if not _RECONCILE_COMPLETE_CLAIM_RE.search(report_text or ""):
+        return {"complete": True, "downgraded": False, "reasons": []}
+
+    fields = _report_labeled_fields(report_text)
+    for required in RECONCILIATION_INVENTORY_PROOF_FIELDS:
+        if required not in fields:
+            reasons.append(
+                "reconciliation completeness claimed but report missing "
+                f"pagination proof field '{required}'"
+            )
+
+    if not pagination_evidence:
+        reasons.append(
+            "reconciliation completeness claimed without pagination "
+            "evidence from gitea_list_prs"
+        )
+        return {"complete": False, "downgraded": True, "reasons": reasons}
+
+    evidence_valid = True
+    for page in pagination_evidence:
+        if not isinstance(page, dict) or not all(
+            key in page for key in ("per_page", "returned_count")
+        ):
+            reasons.append(
+                "pagination evidence page missing 'per_page'/"
+                "'returned_count' metadata"
+            )
+            evidence_valid = False
+
+    if evidence_valid:
+        last = pagination_evidence[-1]
+        final_proven = (
+            last.get("inventory_complete") is True
+            or (last.get("is_final_page") is True
+                and last.get("has_more") is False)
+            or last["returned_count"] < last["per_page"]
+        )
+        if not final_proven:
+            reasons.append(
+                "pagination evidence does not prove final page (need "
+                "inventory_complete, is_final_page with has_more=false, "
+                "or a returned count below the requested page size)"
+            )
+
+        pages_fetched = fields.get("pages fetched")
+        if pages_fetched:
+            declared = re.findall(r"\d+", pages_fetched)
+            if declared and int(declared[0]) != len(pagination_evidence):
+                reasons.append(
+                    f"report 'Pages fetched' ({declared[0]}) does not match "
+                    f"pagination evidence ({len(pagination_evidence)} pages)"
+                )
+
+        counts_field = fields.get("returned pr count per page")
+        if counts_field:
+            declared_counts = [int(n) for n in re.findall(r"\d+", counts_field)]
+            actual_counts = [p["returned_count"] for p in pagination_evidence]
+            if declared_counts != actual_counts:
+                reasons.append(
+                    "report 'Returned PR count per page' "
+                    f"({declared_counts}) does not match pagination "
+                    f"evidence {actual_counts}"
+                )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
 # ── Linked-issue consistency (Issue #314) ────────────────────────────────────
 #
 # Stale linked-issue text from a previous PR must not leak into a final

tests/test_reconciliation_inventory_proof.py

@@ -0,0 +1,150 @@
+"""Tests for reconciliation PR-inventory pagination proof (#308).
+
+Already-landed reconciliation reports may not claim a complete queue
+scan (or that all already-landed PRs were found) unless the report
+carries pagination proof fields and the per-page evidence proves the
+final page was reached or the page limit was not hit.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_reconciliation_inventory_proof  # noqa: E402
+
+
+def _report(claim=True, fields=True, pages="1", counts="12",
+            final_proof="is_final_page=true, has_more=false"):
+    lines = []
+    if claim:
+        lines.append(
+            "Inventory scope: complete queue scan; all already-landed "
+            "PRs were found."
+        )
+    if fields:
+        lines += [
+            "- Requested page size: 50",
+            f"- Pages fetched: {pages}",
+            f"- Returned PR count per page: {counts}",
+            f"- Final page proof: {final_proof}",
+        ]
+    return "\n".join(lines) + "\n"
+
+
+def _page(page, returned, per_page=50, has_more=False, final=True,
+          inventory_complete=None):
+    d = {
+        "page": page,
+        "per_page": per_page,
+        "returned_count": returned,
+        "has_more": has_more,
+        "is_final_page": final,
+    }
+    if inventory_complete is not None:
+        d["inventory_complete"] = inventory_complete
+    return d
+
+
+class TestCompleteScanProven(unittest.TestCase):
+    def test_first_page_below_limit_passes(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(),
+            pagination_evidence=[_page(1, 12)],
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_multiple_pages_to_final_passes(self):
+        report = _report(
+            pages="3", counts="50, 50, 12",
+            final_proof="has_more=false on page 3",
+        )
+        evidence = [
+            _page(1, 50, has_more=True, final=False),
+            _page(2, 50, has_more=True, final=False),
+            _page(3, 12),
+        ]
+        result = assess_reconciliation_inventory_proof(
+            report, pagination_evidence=evidence
+        )
+        self.assertTrue(result["complete"])
+
+    def test_no_completeness_claim_needs_no_proof(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(claim=False, fields=False),
+            pagination_evidence=None,
+        )
+        self.assertTrue(result["complete"])
+
+
+class TestUnprovenScanFails(unittest.TestCase):
+    def test_first_page_exactly_at_limit_fails(self):
+        # 50 returned with per_page 50 and no final-page signal: the page
+        # limit was hit, so completeness is unproven.
+        evidence = [_page(1, 50, has_more=True, final=False)]
+        result = assess_reconciliation_inventory_proof(
+            _report(counts="50", final_proof="none"),
+            pagination_evidence=evidence,
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("final page" in r.lower() for r in result["reasons"])
+        )
+
+    def test_missing_pagination_evidence_fails(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(), pagination_evidence=None
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("pagination" in r.lower() for r in result["reasons"])
+        )
+
+    def test_missing_report_fields_fails(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(fields=False),
+            pagination_evidence=[_page(1, 12)],
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("requested page size" in r.lower()
+                for r in result["reasons"])
+        )
+
+    def test_page_count_mismatch_fails(self):
+        # Report claims 1 page; evidence shows 2 were fetched.
+        evidence = [
+            _page(1, 50, has_more=True, final=False),
+            _page(2, 3),
+        ]
+        result = assess_reconciliation_inventory_proof(
+            _report(pages="1", counts="50"),
+            pagination_evidence=evidence,
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("pages fetched" in r.lower() for r in result["reasons"])
+        )
+
+    def test_per_page_count_mismatch_fails(self):
+        evidence = [_page(1, 12)]
+        result = assess_reconciliation_inventory_proof(
+            _report(counts="11"),
+            pagination_evidence=evidence,
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("count per page" in r.lower() for r in result["reasons"])
+        )
+
+    def test_evidence_page_missing_metadata_fails(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(),
+            pagination_evidence=[{"page": 1}],
+        )
+        self.assertFalse(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()