Gitea-Tools/reviewer_blocker_skip.py

"""Prior-blocker skip proof verifier for reviewer queue reports (#318)."""

from __future__ import annotations

import re
from typing import Any

_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE)
_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
_BLOCKING_DECISION_RE = re.compile(
    r"(?:blocking review decision|review decision|blocking decision)\s*:\s*request[_ ]changes|"
    r"request[_ ]changes",
    re.IGNORECASE,
)
_BLOCKING_HEAD_RE = re.compile(
    r"(?:blocking review head(?:\s+sha)?|blocker head(?:\s+sha)?|review head at blocker)",
    re.IGNORECASE,
)
_HEAD_CHANGED_RE = re.compile(
    r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|"
    r"head changed (?:after|since) (?:the )?blocker|head unchanged since blocker)",
    re.IGNORECASE,
)
_BLOCKER_REASON_RE = re.compile(
    r"(?:reason (?:it )?remains blocked|remains blocked because|blocking category)",
    re.IGNORECASE,
)
_LIVE_BLOCKER_PROOF_RE = re.compile(
    r"(?:blocker revalidated live|live proof|gitea_get_pr_review_feedback|"
    r"gitea_view_pr|review feedback fetched|current review state)",
    re.IGNORECASE,
)
_BLOCKER_UNVERIFIED_RE = re.compile(r"\bBLOCKER_STATUS_UNVERIFIED\b")


def _pr_section(text: str, pr_number: int) -> str:
    lines = (text or "").splitlines()
    chunks: list[str] = []
    capture = False
    token = f"#{pr_number}"
    for line in lines:
        lower = line.lower()
        if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""):
            capture = True
            chunks.append(line)
            continue
        if capture:
            if _PR_NUMBER_RE.search(line) and token not in line.lower():
                break
            if line.strip() == "" and len(chunks) > 3:
                break
            chunks.append(line)
    if chunks:
        return "\n".join(chunks)
    return text or ""


def _has_head_sha(text: str, head_sha: str | None) -> bool:
    if not head_sha:
        return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text))
    head = head_sha.strip().lower()
    if head in text.lower():
        return True
    if len(head) >= 7 and head[:7] in text.lower():
        return True
    return bool(_FULL_SHA.search(text))


def assess_prior_blocker_skip_proof(
    report_text: str,
    *,
    skipped_prs: list[dict] | None = None,
) -> dict[str, Any]:
    """Validate live blocker proof for skipped earlier open PRs (#318)."""
    text = report_text or ""
    reasons: list[str] = []
    assessments: list[dict[str, Any]] = []

    for entry in skipped_prs or []:
        pr_number = entry.get("pr_number")
        if pr_number is None:
            reasons.append("skipped PR entry missing pr_number")
            continue
        pr_number = int(pr_number)
        section = _pr_section(text, pr_number)
        verified = entry.get("blocker_verified")
        head_changed = entry.get("head_changed_since_blocker")
        blocking_head = (entry.get("blocking_review_head_sha") or "").strip() or None
        current_head = (entry.get("head_sha") or "").strip() or None
        skip_reason = (entry.get("skip_reason") or entry.get("blocking_decision") or "").lower()

        item_reasons: list[str] = []

        if head_changed is True:
            item_reasons.append(
                f"PR #{pr_number} head changed after blocker; it cannot be skipped "
                "based on stale REQUEST_CHANGES"
            )

        if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower():
            item_reasons.append(f"skipped PR #{pr_number} not documented in final report")

        if "request_changes" in skip_reason or entry.get("blocking_decision") == "request_changes":
            if verified is False:
                if not _BLOCKER_UNVERIFIED_RE.search(section):
                    item_reasons.append(
                        f"PR #{pr_number} blocker proof unavailable; report must classify "
                        "BLOCKER_STATUS_UNVERIFIED"
                    )
            else:
                if not _BLOCKING_DECISION_RE.search(section):
                    item_reasons.append(
                        f"PR #{pr_number} skip missing blocking review decision proof"
                    )
                if not _has_head_sha(section, current_head):
                    item_reasons.append(
                        f"PR #{pr_number} skip missing current head SHA"
                    )
                if blocking_head and not _BLOCKING_HEAD_RE.search(section):
                    if blocking_head.lower() not in section.lower():
                        item_reasons.append(
                            f"PR #{pr_number} skip missing blocking review head SHA"
                        )
                if head_changed is False and not _HEAD_CHANGED_RE.search(section):
                    item_reasons.append(
                        f"PR #{pr_number} skip missing head-changed-since-blocker proof"
                    )
                if not _BLOCKER_REASON_RE.search(section):
                    item_reasons.append(
                        f"PR #{pr_number} skip missing reason-it-remains-blocked"
                    )
                if not _LIVE_BLOCKER_PROOF_RE.search(section):
                    item_reasons.append(
                        f"PR #{pr_number} skip missing live blocker proof for this session"
                    )

        proven = not item_reasons
        assessments.append({
            "pr_number": pr_number,
            "proven": proven,
            "classification": (
                "BLOCKER_STATUS_UNVERIFIED"
                if verified is False
                else "BLOCKED_SKIPPED"
            ),
            "reasons": item_reasons,
        })
        reasons.extend(item_reasons)

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "downgraded": False,
        "assessments": assessments,
        "reasons": reasons,
        "safe_next_action": (
            "fetch live review feedback and document blocker proof for each skipped PR, "
            "or classify BLOCKER_STATUS_UNVERIFIED"
            if reasons
            else "proceed"
        ),
    }