Merge pull request 'Add pagination metadata to gitea_list_prs' (#342) from feat/issue-340-list-prs-pagination-metadata into master

review_proofs.py

@@ -968,6 +968,90 @@ def assess_mutation_ledger_report(
     }
 
 
+def assess_list_prs_pagination_proof(
+    list_prs_response: dict | list | None,
+    *,
+    inventory_complete_claimed: bool = False,
+) -> dict:
+    """#340: verify ``gitea_list_prs`` pagination metadata proves inventory scope.
+
+    Accepts the wrapped ``{"prs": [...], "pagination": {...}}`` response from
+    ``gitea_list_prs``. Legacy bare lists fail closed when completeness is
+    claimed. Single-page fetches may not claim full inventory unless
+    ``inventory_complete`` or ``is_final_page`` is true.
+    """
+    reasons: list[str] = []
+    if list_prs_response is None:
+        reasons.append("gitea_list_prs response missing; fail closed")
+        return {"proven": False, "block": True, "reasons": reasons, "pagination": None}
+
+    if isinstance(list_prs_response, list):
+        pagination = None
+        if inventory_complete_claimed:
+            reasons.append(
+                "bare PR list lacks pagination metadata; cannot prove inventory "
+                "completeness"
+            )
+        return {
+            "proven": not reasons,
+            "block": bool(reasons),
+            "reasons": reasons,
+            "pagination": pagination,
+        }
+
+    if not isinstance(list_prs_response, dict):
+        reasons.append("gitea_list_prs response is not a dict or list; fail closed")
+        return {"proven": False, "block": True, "reasons": reasons, "pagination": None}
+
+    prs = list_prs_response.get("prs")
+    pagination = list_prs_response.get("pagination")
+    if not isinstance(prs, list):
+        reasons.append("gitea_list_prs response missing 'prs' list")
+    if not isinstance(pagination, dict):
+        reasons.append("gitea_list_prs response missing 'pagination' metadata")
+        return {
+            "proven": False,
+            "block": True,
+            "reasons": reasons,
+            "pagination": pagination if isinstance(pagination, dict) else None,
+        }
+
+    required_keys = ("page", "per_page", "returned_count", "has_more", "is_final_page")
+    for key in required_keys:
+        if key not in pagination:
+            reasons.append(f"pagination metadata missing '{key}'")
+
+    if inventory_complete_claimed:
+        complete = pagination.get("inventory_complete") is True
+        final_page = pagination.get("is_final_page") is True and pagination.get("has_more") is False
+        if not (complete or final_page):
+            reasons.append(
+                "inventory completeness claimed but pagination metadata does not "
+                "prove final page (inventory_complete or is_final_page with "
+                "has_more=false required)"
+            )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "pagination": pagination,
+    }
+
+
+def _prs_from_list_response(list_prs_response: list | dict | None) -> list | None:
+    """Normalize ``gitea_list_prs`` output to a PR list."""
+    if list_prs_response is None:
+        return None
+    if isinstance(list_prs_response, list):
+        return list_prs_response
+    if isinstance(list_prs_response, dict):
+        prs = list_prs_response.get("prs")
+        return prs if isinstance(prs, list) else None
+    return None
+
+
 def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
                          justification=None):
     """Assess reviewer/author role separation for blind queue workflows.
@@ -1862,14 +1946,19 @@ def pr_inventory_trust_gate(
             "queue_target_lock": lock_status,
         }
 
-    if list_prs_response is None or not isinstance(list_prs_response, list):
+    prs_list = _prs_from_list_response(list_prs_response)
+    pagination_meta = (
+        list_prs_response.get("pagination")
+        if isinstance(list_prs_response, dict) else {}
+    ) or {}
+    if prs_list is None:
         return {
             "status": "inventory_error",
-            "reasons": ["PR list response is invalid (not a list or None)"],
+            "reasons": ["PR list response is invalid (missing prs list or None)"],
             "corroborated": False,
         }
 
-    if len(list_prs_response) > 0:
+    if len(prs_list) > 0:
         return {
             "status": "trusted_nonempty",
             "reasons": [],
@@ -1900,6 +1989,13 @@ def pr_inventory_trust_gate(
     corroborated = False
     if has_finality_metadata:
         corroborated = True
+    elif pagination_meta.get("inventory_complete") is True:
+        corroborated = True
+    elif (
+        pagination_meta.get("is_final_page") is True
+        and pagination_meta.get("has_more") is False
+    ):
+        corroborated = True
     elif corroboration_open_pr_counter == 0:
         corroborated = True
     else: