Gitea-Tools/reviewer_inventory_worktree.py

"""Reviewer inventory completeness and worktree state verifier (#293)."""

from __future__ import annotations

import re
from typing import Any

_PAGINATION_FINALITY_EVIDENCE = re.compile(
    r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|"
    r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|"
    r"no next page|final[- ]page|pr_inventory_trust_gate\.status|"
    r"total_count\s*:|pagination.*(?:final|complete)",
    re.I,
)

_INCOMPLETE_PAGINATION_PROOF = re.compile(
    r"first page only|partial page|page 1 only|truncated|"
    r"returned \d+ open prs?(?:\s*$|\s*[,;])",
    re.I,
)

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

_ELIGIBILITY_CLAIM_RE = re.compile(
    r"\b(?:oldest eligible pr|next eligible pr|next pr to review|"
    r"inventory (?:is )?complete|inventory exhaustive|exhaustive inventory)\b",
    re.I,
)

_EXACT_PAGE_LIMIT_RE = re.compile(
    r"(?:returned|listed|fetched)\s+(\d+)\s+open prs?|"
    r"open pr count\s*:\s*(\d+)|"
    r"page[- ]?size\s*(?:=|:)\s*(\d+)",
    re.I,
)

_LEGACY_SCRATCH_FALSE_RE = re.compile(
    r"scratch worktree used\s*:\s*false",
    re.I,
)

_BRANCHES_WORKTREE_RE = re.compile(r"\bbranches/", re.I)

_CONTRADICTORY_HEAD_RE = re.compile(
    r"detached head\s*/\s*branch\s+master|"
    r"branch\s+master\s*/\s*detached head|"
    r"detached head.*branch\s+(?:master|main|dev)\b.*detached|"
    r"checkout\s*:\s*detached head\s*/\s*branch",
    re.I,
)

_MAIN_CHECKOUT_BRANCH_RE = re.compile(
    r"main checkout branch\s*:\s*(\S+)",
    re.I,
)

_REVIEW_HEAD_STATE_RE = re.compile(
    r"review worktree head state\s*:\s*(\S+)",
    re.I,
)

_HANDOFF_SECTION_RE = re.compile(
    r"^##\s*Controller Handoff\s*$",
    re.I | re.M,
)


def _handoff_field_map(report_text: str) -> dict[str, str]:
    """Parse ``- Field: value`` lines from the Controller Handoff section."""
    text = report_text or ""
    match = _HANDOFF_SECTION_RE.search(text)
    if not match:
        return {}
    section = text[match.end() :]
    fields: dict[str, str] = {}
    for line in section.splitlines():
        stripped = line.strip().lstrip("-*").strip()
        if ":" not in stripped:
            continue
        key, value = stripped.split(":", 1)
        fields[key.strip().lower()] = value.strip()
    return fields


def _truthy_field(value: str) -> bool:
    lowered = (value or "").strip().lower()
    if not lowered:
        return False
    if lowered in {"false", "no", "none", "not applicable", "n/a", "—", "-"}:
        return False
    return True


def _field_proves_pagination(value: str) -> bool:
    proof = (value or "").strip()
    if not proof or proof.lower() in {"none", "n/a", "not applicable", "unknown"}:
        return False
    if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(proof):
        return False
    if _INCOMPLETE_PAGINATION_PROOF.search(proof):
        return False
    return bool(_PAGINATION_FINALITY_EVIDENCE.search(proof))


def _pagination_proven(text: str, session: dict, *, pagination_field: str = "") -> bool:
    if session.get("pagination_complete") or session.get("inventory_complete"):
        return True
    session_proof = session.get("inventory_pagination_proof") or ""
    if _field_proves_pagination(str(session_proof)):
        return True
    if _field_proves_pagination(pagination_field):
        return True
    if _PAGINATION_FINALITY_EVIDENCE.search(text):
        return True
    return False


def _exact_page_limit_unproven(
    text: str, session: dict, *, pagination_proven: bool = False
) -> bool:
    """True when a full page was returned without final-page proof."""
    if pagination_proven:
        return False
    requested = session.get("requested_page_size")
    returned = session.get("returned_page_size")
    if isinstance(requested, int) and isinstance(returned, int):
        if returned >= requested > 0:
            return True
    for match in _EXACT_PAGE_LIMIT_RE.finditer(text):
        count = next((g for g in match.groups() if g), None)
        if not count:
            continue
        try:
            n = int(count)
        except ValueError:
            continue
        if n in {10, 20, 50}:
            return True
    return False


def assess_inventory_worktree_report(
    report_text: str,
    *,
    inventory_session: dict | None = None,
) -> dict[str, Any]:
    """Validate PR inventory pagination proof and worktree field consistency (#293)."""
    text = report_text or ""
    session = dict(inventory_session or {})
    reasons: list[str] = []

    fields = _handoff_field_map(text)
    pagination_proof_field = fields.get("inventory pagination proof", "")

    pagination_proven = _pagination_proven(
        text, session, pagination_field=pagination_proof_field
    )

    if _ELIGIBILITY_CLAIM_RE.search(text):
        if not pagination_proven:
            reasons.append(
                "oldest/next eligible PR or inventory-complete claim requires "
                "final-page/no-next-page/pagination_complete proof"
            )

    if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text):
        if not pagination_proven:
            reasons.append(
                "inventory pagination assumed from default page size without "
                "final-page/no-next-page/total-count/traversal proof"
            )

    if pagination_proof_field:
        if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(pagination_proof_field):
            reasons.append(
                "Inventory pagination proof field relies on page-size assumption"
            )
        elif _INCOMPLETE_PAGINATION_PROOF.search(pagination_proof_field):
            reasons.append(
                "Inventory pagination proof field does not prove final page"
            )
        elif not _field_proves_pagination(pagination_proof_field):
            if _ELIGIBILITY_CLAIM_RE.search(text) or _EXACT_PAGE_LIMIT_RE.search(text):
                reasons.append(
                    "Inventory pagination proof field missing final-page metadata"
                )

    if _exact_page_limit_unproven(text, session, pagination_proven=pagination_proven):
        reasons.append(
            "exactly-full first page returned without final-page or "
            "no-next-page proof"
        )

    review_worktree_used = fields.get("review worktree used", "")
    review_worktree_path = fields.get("review worktree path", "")
    scratch_used = fields.get("scratch worktree used", "")
    inside_branches = fields.get("review worktree inside branches:", "") or fields.get(
        "review worktree inside branches", ""
    )

    branches_path_in_text = bool(
        _BRANCHES_WORKTREE_RE.search(review_worktree_path)
        or _BRANCHES_WORKTREE_RE.search(text)
    )

    if branches_path_in_text:
        if review_worktree_used and not _truthy_field(review_worktree_used):
            reasons.append(
                "reports using a branches/ review worktree must set "
                "Review worktree used: true"
            )
        if scratch_used and re.search(r"\bfalse\b", scratch_used, re.I):
            reasons.append(
                "Scratch worktree used: false rejected when a branches/ "
                "review worktree was created or used"
            )
        if _LEGACY_SCRATCH_FALSE_RE.search(text) and not _truthy_field(
            review_worktree_used
        ):
            reasons.append(
                "legacy Scratch worktree used: false contradicts branches/ "
                "review worktree usage"
            )

    if review_worktree_path and _truthy_field(review_worktree_used):
        if "branches/" not in review_worktree_path.replace("\\", "/").lower():
            reasons.append(
                "Review worktree path must be under branches/ when "
                "Review worktree used is true"
            )
        if inside_branches and re.search(r"\bfalse\b", inside_branches, re.I):
            reasons.append(
                "Review worktree inside branches must be true when path is "
                "under branches/"
            )

    main_branch = (
        fields.get("main checkout branch", "")
        or _MAIN_CHECKOUT_BRANCH_RE.search(text).group(1)
        if _MAIN_CHECKOUT_BRANCH_RE.search(text)
        else ""
    )
    head_state = (
        fields.get("review worktree head state", "")
        or (
            _REVIEW_HEAD_STATE_RE.search(text).group(1)
            if _REVIEW_HEAD_STATE_RE.search(text)
            else ""
        )
    )
    if main_branch and head_state:
        main_norm = main_branch.strip().lower()
        head_norm = head_state.strip().lower()
        if head_norm in {"branch", "on branch"} and main_norm in {
            "master",
            "main",
            "dev",
        }:
            if "detached" in text.lower() and "review worktree" in text.lower():
                reasons.append(
                    "review worktree HEAD state must not reuse main checkout "
                    "branch wording"
                )

    if _CONTRADICTORY_HEAD_RE.search(text):
        reasons.append(
            "contradictory checkout wording such as 'Detached HEAD / Branch master'"
        )

    if review_worktree_used and _truthy_field(review_worktree_used):
        if not review_worktree_path or review_worktree_path.lower() in {
            "none",
            "n/a",
            "not applicable",
        }:
            reasons.append(
                "Review worktree used: true requires Review worktree path"
            )
        if not head_state or head_state.lower() in {
            "none",
            "n/a",
            "not applicable",
            "unknown",
        }:
            reasons.append(
                "Review worktree used: true requires Review worktree HEAD state"
            )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": list(dict.fromkeys(reasons)),
        "pagination_proven": pagination_proven,
        "branches_worktree_used": branches_path_in_text,
        "safe_next_action": (
            "prove inventory pagination with final-page metadata; align "
            "Review worktree used/path/HEAD state with branches/ usage"
            if reasons
            else "proceed"
        ),
    }