Gitea-Tools/reviewer_reconcile_inventory.py

"""Reconciliation PR inventory pagination proof verifier (#308)."""

from __future__ import annotations

import re
from typing import Any

_COMPLETE_SCAN_CLAIM_RE = re.compile(
    r"\b(?:all already[- ]landed(?:\s+prs?)?(?:\s+were)?\s+found|"
    r"complete queue scan|all open prs? (?:were )?(?:found|checked|inventoried|listed)|"
    r"inventory (?:is )?complete|exhaustive (?:pr )?inventory|"
    r"no additional already[- ]landed|scanned (?:the )?(?:full )?open pr queue)\b",
    re.I,
)

_FINALITY_RE = re.compile(
    r"is_final_page\s*:\s*true|has_more\s*:\s*false|no next page|final[- ]page|"
    r"pagination_complete\s*:\s*true|inventory pagination proof\s*:\s*(?!assumed|none\b).+",
    re.I,
)

_REQUESTED_PAGE_SIZE_RE = re.compile(
    r"requested page size\s*:\s*(\d+)|page[- ]?size\s*(?:=|:)\s*(\d+)",
    re.I,
)
_PAGES_FETCHED_RE = re.compile(
    r"pages? fetched\s*:\s*(\d+)|page count\s*:\s*(\d+)",
    re.I,
)
_RETURNED_PER_PAGE_RE = re.compile(
    r"returned pr count(?: per page)?\s*:|open pr count per page|"
    r"returned \d+ open prs? per page|per[- ]page counts?\s*:",
    re.I,
)
_EXACT_LIMIT_RETURN_RE = re.compile(
    r"returned\s+(\d+)\s+open prs?|listed\s+(\d+)\s+open prs?",
    re.I,
)

_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
    r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
    r"default (?:gitea )?page[- ]?size|assumed complete",
    re.I,
)


def _int_from_groups(match: re.Match[str] | None) -> int | None:
    if not match:
        return None
    for group in match.groups():
        if group:
            return int(group)
    return None


def _pagination_proven(text: str, session: dict) -> bool:
    if session.get("pagination_complete") or session.get("inventory_complete"):
        return True
    if _FINALITY_RE.search(text):
        return True
    pages = session.get("pages_fetched")
    if isinstance(pages, int) and pages >= 1 and session.get("is_final_page"):
        return True
    return False


def _metadata_present(text: str, session: dict) -> list[str]:
    missing: list[str] = []
    has_page_size = bool(
        _REQUESTED_PAGE_SIZE_RE.search(text)
        or session.get("requested_page_size")
    )
    has_pages_fetched = bool(
        _PAGES_FETCHED_RE.search(text) or session.get("pages_fetched")
    )
    has_returned_counts = bool(
        _RETURNED_PER_PAGE_RE.search(text) or session.get("returned_per_page")
    )
    if not has_page_size:
        missing.append("requested page size")
    if not has_pages_fetched:
        missing.append("page count fetched")
    if not has_returned_counts:
        missing.append("returned PR count per page")
    if not _pagination_proven(text, session):
        missing.append("final-page/no-next-page proof")
    return missing


def assess_reconcile_inventory_report(
    report_text: str,
    *,
    inventory_session: dict | None = None,
) -> dict[str, Any]:
    """Validate PR inventory pagination proof in reconciliation reports (#308)."""
    text = report_text or ""
    session = dict(inventory_session or {})
    reasons: list[str] = []

    claims_scan = bool(_COMPLETE_SCAN_CLAIM_RE.search(text))
    lists_open_prs = bool(
        re.search(r"open prs? listed|listed open prs?|pr inventory", text, re.I)
    )

    if not claims_scan and not lists_open_prs and not session.get("inventory_required"):
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "inventory_claimed": False,
            "safe_next_action": "proceed",
        }

    if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text) and not _pagination_proven(text, session):
        reasons.append(
            "reconciliation inventory assumed complete from page-size guess "
            "without final-page proof (#308)"
        )

    if claims_scan and not _pagination_proven(text, session):
        reasons.append(
            "complete queue scan or all already-landed claim requires "
            "pagination finality proof (#308)"
        )

    missing = _metadata_present(text, session)
    if (claims_scan or lists_open_prs) and missing:
        reasons.append(
            "reconciliation inventory missing pagination metadata: "
            + ", ".join(missing)
        )

    requested = session.get("requested_page_size")
    returned = session.get("returned_page_size")
    if isinstance(requested, int) and isinstance(returned, int):
        if returned >= requested > 0 and not _pagination_proven(text, session):
            reasons.append(
                "exactly-full first reconciliation page without final-page proof (#308)"
            )
    else:
        for match in _EXACT_LIMIT_RETURN_RE.finditer(text):
            count = _int_from_groups(match)
            if count in {10, 20, 50} and not _pagination_proven(text, session):
                reasons.append(
                    "exactly-full first reconciliation page without final-page proof (#308)"
                )
                break

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": list(dict.fromkeys(reasons)),
        "inventory_claimed": claims_scan or lists_open_prs,
        "pagination_proven": _pagination_proven(text, session),
        "safe_next_action": (
            "include requested page size, pages fetched, per-page counts, and "
            "final-page/no-next-page proof before claiming complete scan"
            if reasons
            else "proceed"
        ),
    }