diff --git a/review_proofs.py b/review_proofs.py index fdebea9..b6ab8c5 100644 --- a/review_proofs.py +++ b/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 diff --git a/tests/test_reconciliation_inventory_proof.py b/tests/test_reconciliation_inventory_proof.py new file mode 100644 index 0000000..60ed9d3 --- /dev/null +++ b/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()