Gitea-Tools/reviewer_handoff_consistency.py

"""Reviewer handoff consistency validation (#501).

Detects contradictions between narrative claims and mutation ledger fields in
reviewer/final-review controller handoffs. Pure validation only — does not post
comments or mutate Gitea state.
"""

from __future__ import annotations

import re

_HANDOFF_FIELD_RE = re.compile(
    r"^\s*[-*]\s*([^:]+):\s*(.*)$",
    re.MULTILINE | re.IGNORECASE,
)

_MERGE_NARRATIVE_RE = re.compile(
    r"\b(?:merged\s+pr\s*#|pr\s+#\d+\s+(?:was\s+)?merged|"
    r"merge\s+result\s*:\s*merged|successfully\s+merged|"
    r"terminal\s+mutation\s+budget.{0,80}\bmerge)\b",
    re.IGNORECASE | re.DOTALL,
)
_REVIEW_SUBMIT_BLOCKED_RE = re.compile(
    r"\b(?:review\s+submission\s+blocked|submit_pr_review\s+(?:failed|blocked)|"
    r"gitea_submit_pr_review\s+(?:failed|blocked|not\s+(?:attempted|performed))|"
    r"review\s+not\s+submitted|could\s+not\s+submit\s+review)\b",
    re.IGNORECASE,
)
_FINAL_DECISION_MARKED_RE = re.compile(
    r"\b(?:gitea_mark_final_review_decision|final\s+review\s+decision\s+marked|"
    r"server-side\s+final\s+decision\s+marked|marked\s+final\s+review\s+decision)\b",
    re.IGNORECASE,
)
_TERMINAL_BUDGET_CONSUMED_RE = re.compile(
    r"\b(?:terminal\s+mutation\s+budget\s+(?:already\s+)?consumed|"
    r"single[- ]terminal\s+mutation\s+(?:already\s+)?consumed|"
    r"mutation\s+budget\s+exhausted)\b",
    re.IGNORECASE,
)
_LEASE_NARRATIVE_RE = re.compile(
    r"\b(?:reviewer\s+lease\s+acquired|acquired\s+reviewer\s+pr\s+lease|"
    r"gitea_acquire_reviewer_pr_lease)\b",
    re.IGNORECASE,
)
_REJECTED_MUTATION_RE = re.compile(
    r"\b(?:mutation\s+rejected|tool\s+failed\s+closed|blocked\s+before\s+mutation|"
    r"no\s+server-side\s+state\s+changed)\b",
    re.IGNORECASE,
)
_MERGE_IN_LEDGER_RE = re.compile(r"\bmerge\b", re.IGNORECASE)
_REVIEW_SUBMITTED_IN_LEDGER_RE = re.compile(
    r"\b(?:submitted|approve|request_changes|review\s+submitted)\b",
    re.IGNORECASE,
)
_NONE_VALUE_RE = re.compile(r"^\s*(?:none|not\s+attempted|n/a)\s*$", re.IGNORECASE)

_BLOCKED_REVIEW_PROOF_FIELDS = (
    "tool called",
    "mutation attempted",
    "mutation rejected",
    "no server-side state changed",
)


def render_blocked_review_handoff_template() -> str:
    """Return a corrected handoff template for blocked review submissions."""
    return """## Blocked Review Submission Handoff

- Tool called: gitea_submit_pr_review
- Mutation attempted: yes
- Mutation rejected: yes
- No server-side state changed: confirmed
- Proof/source: <paste exact tool error or gate reason>
- Prior terminal mutation that consumed budget: <name exact prior mutation or none>
- Review decision (local intent): <approve | request_changes | comment only>
- Server-side final decision marked: <yes with tool proof | no>
- Gitea review submitted: no
- Gitea review blocked because: <exact gate/error>
- Review mutations: none (submission blocked)
- MCP/Gitea mutations: <list only mutations that actually occurred>
- Safe next action: rewrite handoff with consistent ledger before posting
"""


def _handoff_fields(text: str | None) -> dict[str, str]:
    fields: dict[str, str] = {}
    for match in _HANDOFF_FIELD_RE.finditer(text or ""):
        key = match.group(1).strip().lower()
        value = match.group(2).strip()
        fields[key] = value
    return fields


def _is_none_value(value: str | None) -> bool:
    return bool(_NONE_VALUE_RE.match(value or ""))


def _ledger_mentions_merge(*values: str | None) -> bool:
    blob = " ".join(v for v in values if v)
    return bool(blob) and bool(_MERGE_IN_LEDGER_RE.search(blob))


def _ledger_mentions_review_submission(*values: str | None) -> bool:
    blob = " ".join(v for v in values if v)
    return bool(blob) and bool(_REVIEW_SUBMITTED_IN_LEDGER_RE.search(blob))


def assess_reviewer_handoff_consistency(report_text: str | None) -> dict:
    """Validate reviewer handoff narrative against mutation ledger fields."""
    text = report_text or ""
    fields = _handoff_fields(text)
    reasons: list[str] = []

    review_mutations = fields.get("review mutations", "")
    merge_mutations = fields.get("merge mutations", "")
    mcp_mutations = fields.get("mcp/gitea mutations", "")
    review_decision = fields.get("review decision", "")

    if _MERGE_NARRATIVE_RE.search(text) and not _ledger_mentions_merge(
        merge_mutations, mcp_mutations, review_mutations
    ):
        reasons.append(
            "narrative claims a merge occurred but mutation ledger omits merge"
        )

    if _TERMINAL_BUDGET_CONSUMED_RE.search(text):
        if _ledger_mentions_merge(merge_mutations, mcp_mutations):
            pass
        elif _LEASE_NARRATIVE_RE.search(text) and not _ledger_mentions_merge(
            merge_mutations, mcp_mutations
        ):
            reasons.append(
                "terminal mutation budget attributed to merge but ledger only "
                "shows lease or non-merge activity"
            )
        elif not _ledger_mentions_merge(merge_mutations, mcp_mutations):
            reasons.append(
                "terminal mutation budget consumed claim must identify the "
                "exact prior mutation in the ledger"
            )

    if _REVIEW_SUBMIT_BLOCKED_RE.search(text) and _FINAL_DECISION_MARKED_RE.search(text):
        reasons.append(
            "handoff claims review submission blocked but also claims "
            "server-side final decision was marked"
        )

    lease_claimed = _LEASE_NARRATIVE_RE.search(text) or (
        not _is_none_value(mcp_mutations)
        and "lease" in mcp_mutations.lower()
    )
    if lease_claimed and _is_none_value(review_decision):
        reasons.append(
            "reviewer lease acquired but Review decision field is missing or none"
        )

    if _REVIEW_SUBMIT_BLOCKED_RE.search(text) or _REJECTED_MUTATION_RE.search(text):
        lower = text.lower()
        missing_proof = [
            field
            for field in _BLOCKED_REVIEW_PROOF_FIELDS
            if field not in lower
        ]
        if missing_proof:
            reasons.append(
                "blocked/rejected mutation claim missing proof fields: "
                + ", ".join(missing_proof)
            )

    if (
        _FINAL_DECISION_MARKED_RE.search(text)
        and _is_none_value(review_mutations)
        and not _ledger_mentions_review_submission(mcp_mutations)
        and not _REVIEW_SUBMIT_BLOCKED_RE.search(text)
    ):
        reasons.append(
            "final review decision marked but Review mutations ledger is empty"
        )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "fields": fields,
    }