Gitea-Tools/reviewer_proof_backed_handoff.py

"""Proof-backed reviewer handoff claim verifier (#395)."""

from __future__ import annotations

import re
from typing import Any

_PAGINATION_FINALITY = re.compile(
    r"has_more\s*:\s*false|is_final_page\s*:\s*true|"
    r"inventory_complete\s*:\s*true|pages_fetched|total_count\s*:",
    re.I,
)
_PAGE_SIZE_ONLY = re.compile(
    r"(?:less|fewer) than (?:the )?(?:default )?page[- ]?size|"
    r"under (?:the )?50[- ]?(?:item )?limit|"
    r"returned \d+ (?:open )?prs?.*less than 50",
    re.I,
)
_INVENTORY_COMPLETE_CLAIM = re.compile(
    r"\b(?:inventory (?:is )?complete|inventory exhaustive|"
    r"complete (?:pr )?inventory)\b",
    re.I,
)
_SKIP_CLAIM = re.compile(
    r"\b(?:earlier prs? skipped|skipped (?:earlier )?pr|"
    r"skip(?:ped)? pr #?\d+|non-mergeable|prior request.changes)\b",
    re.I,
)
_CONFLICT_PROOF = re.compile(
    r"merge simulation|git merge --no-commit|conflicting files|"
    r"conflict proof|merge_exit|non-mergeable",
    re.I,
)
_BASELINE_CLAIM = re.compile(
    r"\b(?:baseline (?:validation|worktree|comparison)|"
    r"same as master|pre-existing (?:on )?master|"
    r"failure signatures match)\b",
    re.I,
)
_BASELINE_PATH = re.compile(r"baseline worktree path\s*:\s*(\S+)", re.I)
_BASELINE_SHA = re.compile(r"baseline (?:target )?sha\s*:\s*([0-9a-f]{7,40})", re.I)
_BASELINE_DIRTY_BEFORE = re.compile(
    r"baseline.*dirty before|dirty before.*baseline", re.I
)
_BASELINE_DIRTY_AFTER = re.compile(
    r"baseline.*dirty after|dirty after.*baseline", re.I
)
_BASELINE_COMMAND = re.compile(
    r"baseline.*(?:validation )?command|pytest.*baseline", re.I
)
_BASELINE_RESULT = re.compile(
    r"baseline.*(?:validation )?result|baseline_exit|baseline failures", re.I
)
_MASTER_INTEGRATION = re.compile(
    r"\b(?:merged master into|merge(?:d)? (?:remote[- ]tracking )?branch.*master|"
    r"master integration|integrated master|rebase.*master)\b",
    re.I,
)
_CLEANUP_CLAIM = re.compile(
    r"\b(?:worktree(?:s)? (?:were )?cleaned|cleanup (?:result|mutations)|"
    r"removed (?:session[- ]owned )?worktree|git worktree remove)\b",
    re.I,
)
_WORKTREE_LIST = re.compile(r"git worktree list|worktree list proof", re.I)
_PROOF_SOURCE = re.compile(
    r"proof source\s*:\s*(command|mcp metadata|prior blocker|not checked)",
    re.I,
)
_HANDOFF_SECTION = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)


def _handoff_fields(report_text: str) -> dict[str, str]:
    text = report_text or ""
    match = _HANDOFF_SECTION.search(text)
    if not match:
        return {}
    fields: dict[str, str] = {}
    for line in text[match.end() :].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 _command_text(entry: Any) -> str:
    if isinstance(entry, dict):
        return str(entry.get("command") or entry.get("tool") or "").strip()
    return str(entry or "").strip()


def _action_log_has(action_log: list | None, pattern: re.Pattern[str]) -> bool:
    for entry in action_log or []:
        blob = " ".join(
            filter(
                None,
                [
                    _command_text(entry),
                    str(entry.get("result") or "") if isinstance(entry, dict) else "",
                    str(entry.get("reason") or "") if isinstance(entry, dict) else "",
                ],
            )
        )
        if pattern.search(blob):
            return True
    return False


def assess_proof_backed_handoff_report(
    report_text: str,
    *,
    action_log: list | None = None,
    inventory_session: dict | None = None,
    baseline_session: dict | None = None,
    skip_session: list[dict] | None = None,
) -> dict[str, Any]:
    """Require explicit command/tool evidence for proof-sensitive review claims (#395)."""
    text = report_text or ""
    fields = _handoff_fields(text)
    reasons: list[str] = []
    inventory = dict(inventory_session or {})
    baseline = dict(baseline_session or {})
    skips = list(skip_session or [])

    pagination_field = fields.get("inventory pagination proof", "")
    if _INVENTORY_COMPLETE_CLAIM.search(text) or _PAGE_SIZE_ONLY.search(text):
        has_meta = bool(
            inventory.get("inventory_complete")
            or inventory.get("pagination_complete")
            or _PAGINATION_FINALITY.search(pagination_field)
            or _PAGINATION_FINALITY.search(text)
            or _action_log_has(action_log, re.compile(r"gitea_list_prs", re.I))
        )
        if _PAGE_SIZE_ONLY.search(text) and not has_meta:
            reasons.append(
                "inventory completeness claimed from page-size assumption only; "
                "require has_more=false, is_final_page=true, or inventory_complete=true"
            )
        elif _INVENTORY_COMPLETE_CLAIM.search(text) and not has_meta:
            reasons.append(
                "inventory complete claim lacks explicit pagination metadata proof"
            )

    if _SKIP_CLAIM.search(text) or fields.get("earlier prs skipped", "").lower() not in {
        "",
        "none",
        "n/a",
    }:
        skip_proof = bool(
            skips
            or _CONFLICT_PROOF.search(text)
            or _action_log_has(
                action_log, re.compile(r"git merge --no-commit|gitea_get_pr_review_feedback", re.I)
            )
        )
        if not skip_proof:
            reasons.append(
                "earlier PR skip claim lacks command/tool conflict or blocker proof"
            )

    if _BASELINE_CLAIM.search(text) or fields.get("baseline worktree used", "").lower() == "true":
        baseline_ok = bool(
            baseline.get("complete")
            or (
                _BASELINE_PATH.search(text)
                and _BASELINE_SHA.search(text)
                and (_BASELINE_DIRTY_BEFORE.search(text) or baseline.get("clean_before"))
                and (_BASELINE_COMMAND.search(text) or baseline.get("command"))
                and (_BASELINE_RESULT.search(text) or baseline.get("result"))
                and (_BASELINE_DIRTY_AFTER.search(text) or baseline.get("clean_after"))
            )
        )
        if not baseline_ok:
            reasons.append(
                "baseline validation claim missing worktree path, target SHA, "
                "dirty-before/after status, command, or result proof"
            )

    if _MASTER_INTEGRATION.search(text):
        if not _action_log_has(
            action_log,
            re.compile(r"git merge.*master|git rebase.*master", re.I),
        ) and not re.search(r"merge_exit\s*=\s*\d+|merge simulation", text, re.I):
            reasons.append(
                "master integration claim lacks exact merge/rebase command and result"
            )

    if _CLEANUP_CLAIM.search(text) or fields.get("cleanup mutations", "").lower() not in {
        "",
        "none",
        "n/a",
    }:
        if not (_WORKTREE_LIST.search(text) or _action_log_has(action_log, _WORKTREE_LIST)):
            reasons.append(
                "cleanup claim lacks final git worktree list or equivalent proof"
            )

    if re.search(r"\blive proof\b", text, re.I) and not _PROOF_SOURCE.search(text):
        if not action_log:
            reasons.append(
                "live proof wording requires proof source classification "
                "(command, MCP metadata, prior blocker, or not checked)"
            )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "safe_next_action": (
            "cite explicit command/tool evidence or structured MCP pagination metadata "
            "for each proof-sensitive claim"
            if not proven
            else "proceed"
        ),
    }