Gitea-Tools/reviewer_already_landed_handoff.py

"""Already-landed controller handoff consistency verifier (#299)."""

from __future__ import annotations

import re
from typing import Any

from review_proofs import git_ref_mutating_commands

ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED"

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

_STALE_HANDOFF_FIELDS = (
    "pinned reviewed head",
    "scratch worktree used",
)

_LEGACY_MUTATIONS_NONE_RE = re.compile(
    r"^\s*[-*]?\s*mutations\s*:\s*none\s*$",
    re.I | re.M,
)
_LEGACY_WORKSPACE_NONE_RE = re.compile(
    r"^\s*[-*]?\s*workspace\s+mutations\s*:\s*none\s*$",
    re.I | re.M,
)

_FORBIDDEN_REVIEW_STATES_RE = re.compile(
    r"review decision\s*:\s*approved?\b|"
    r"merge result\s*:\s*merged\b|"
    r"\bready[_ ]to[_ ]merge\b",
    re.I,
)

_NARRATIVE_ELIGIBILITY_RE = re.compile(
    r"eligibility class\s*:\s*([^\n]+)",
    re.I,
)
_NARRATIVE_REVIEWED_SHA_RE = re.compile(
    r"reviewed head sha\s*:\s*([^\n]+)",
    re.I,
)
_NARRATIVE_WORKTREE_RE = re.compile(
    r"review worktree used\s*:\s*([^\n]+)",
    re.I,
)


def _handoff_field_map(report_text: str) -> dict[str, str]:
    text = report_text or ""
    match = _HANDOFF_SECTION_RE.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 _narrative_before_handoff(report_text: str) -> str:
    text = report_text or ""
    match = _HANDOFF_SECTION_RE.search(text)
    if not match:
        return text
    return text[: match.start()]


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


def _gate_active(text: str, fields: dict[str, str], session: dict) -> bool:
    if session.get("gate_fired") or session.get("already_landed_gate_fired"):
        return True
    eligibility = (fields.get("eligibility class") or "").upper()
    if ALREADY_LANDED_STATE in eligibility:
        return True
    return ALREADY_LANDED_STATE.lower() in text.lower()


def assess_already_landed_handoff_report(
    report_text: str,
    *,
    handoff_session: dict | None = None,
    command_log: list | None = None,
) -> dict[str, Any]:
    """Reject stale handoff fields and narrative/handoff drift after the gate (#299)."""
    text = report_text or ""
    session = dict(handoff_session or {})
    fields = _handoff_field_map(text)
    reasons: list[str] = []

    if not _gate_active(text, fields, session):
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "gate_active": False,
            "safe_next_action": "proceed",
        }

    for stale_field in _STALE_HANDOFF_FIELDS:
        if stale_field in fields:
            reasons.append(
                f"already-landed handoff must not include stale field "
                f"'{stale_field.title()}' (#299)"
            )

    if _LEGACY_WORKSPACE_NONE_RE.search(text):
        reasons.append(
            "already-landed handoff must not include legacy "
            "'Workspace mutations: None' (#299)"
        )

    ref_commands = git_ref_mutating_commands(command_log or session.get("command_log"))
    mutations_observed = bool(
        ref_commands
        or session.get("mutations_observed")
        or session.get("mcp_mutations")
        or session.get("review_mutations")
    )
    if mutations_observed and _LEGACY_MUTATIONS_NONE_RE.search(text):
        reasons.append(
            "already-landed handoff must not claim 'Mutations: None' when "
            "git ref, MCP, review, merge, or cleanup mutations occurred (#299)"
        )

    git_ref_value = fields.get("git ref mutations", "").strip().lower()
    if ref_commands and (not git_ref_value or git_ref_value == "none"):
        reasons.append(
            "git fetch/ref updates must be reported under Git ref mutations (#299)"
        )

    reviewed_sha = fields.get("reviewed head sha", "")
    if reviewed_sha and _is_truthy(reviewed_sha):
        reasons.append(
            "already-landed handoff must use 'Reviewed head SHA: none' (#299)"
        )

    worktree_used = fields.get("review worktree used", "")
    if worktree_used and _is_truthy(worktree_used):
        reasons.append(
            "already-landed handoff must set Review worktree used: false (#299)"
        )

    candidate_sha = fields.get("candidate head sha", "")
    if not candidate_sha or candidate_sha.lower() in {"none", "n/a", "unknown"}:
        reasons.append(
            "already-landed handoff must include Candidate head SHA (#299)"
        )

    if _FORBIDDEN_REVIEW_STATES_RE.search(text):
        reasons.append(
            "already-landed handoff must not claim approved/merged/ready-to-merge (#299)"
        )

    narrative = _narrative_before_handoff(text)
    handoff_eligibility = (fields.get("eligibility class") or "").strip()
    narrative_eligibility = (
        _NARRATIVE_ELIGIBILITY_RE.search(narrative).group(1).strip()
        if _NARRATIVE_ELIGIBILITY_RE.search(narrative)
        else ""
    )
    if handoff_eligibility and narrative_eligibility:
        if handoff_eligibility.upper() != narrative_eligibility.upper():
            reasons.append(
                "narrative Eligibility class disagrees with controller handoff (#299)"
            )

    narrative_reviewed = (
        _NARRATIVE_REVIEWED_SHA_RE.search(narrative).group(1).strip()
        if _NARRATIVE_REVIEWED_SHA_RE.search(narrative)
        else ""
    )
    if narrative_reviewed and reviewed_sha:
        if narrative_reviewed.lower() != reviewed_sha.lower():
            reasons.append(
                "narrative Reviewed head SHA disagrees with controller handoff (#299)"
            )

    narrative_worktree = (
        _NARRATIVE_WORKTREE_RE.search(narrative).group(1).strip()
        if _NARRATIVE_WORKTREE_RE.search(narrative)
        else ""
    )
    if narrative_worktree and worktree_used:
        if narrative_worktree.lower() != worktree_used.lower():
            reasons.append(
                "narrative Review worktree used disagrees with controller handoff (#299)"
            )

    git_ref_narrative = "git ref mutations" in narrative.lower()
    if git_ref_narrative and git_ref_value:
        if "none" in git_ref_value and ref_commands:
            reasons.append(
                "narrative and handoff disagree on git ref mutation reporting (#299)"
            )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": list(dict.fromkeys(reasons)),
        "gate_active": True,
        "safe_next_action": (
            "emit canonical ALREADY_LANDED_RECONCILE_REQUIRED handoff without "
            "stale review/merge fields; align narrative and controller handoff"
            if reasons
            else "proceed"
        ),
    }