Gitea-Tools/post_merge_cleanup_proof.py

"""Post-merge cleanup proof verifier for reviewer final reports (#402)."""

from __future__ import annotations

import re
from typing import Any

CLEANUP_SKIPPED = "CLEANUP_SKIPPED"
CLEANUP_PERFORMED = "CLEANUP_PERFORMED"

_CLEANUP_SECTION_HINT = re.compile(
    r"(?:cleanup (?:status|result|mutations)|post-merge cleanup|"
    r"gitea_delete_branch|remote branch.*deleted|worktree remove)",
    re.IGNORECASE,
)
_CLEANUP_SKIPPED_RE = re.compile(r"\bCLEANUP_SKIPPED\b", re.IGNORECASE)
_CLEANUP_BLOCKER_RE = re.compile(
    r"(?:cleanup blocker|cleanup skip(?:ped)? reason)\s*:\s*(.+)",
    re.IGNORECASE,
)
_REMOTE_DELETE_CLAIM_RE = re.compile(
    r"(?:gitea_delete_branch|remote (?:head )?branch (?:was )?deleted|"
    r"deleted remote branch|delete_branch)",
    re.IGNORECASE,
)
_WORKTREE_REMOVE_CLAIM_RE = re.compile(
    r"(?:git worktree remove|worktree (?:was )?removed|removed (?:local )?worktree|"
    r"worktree cleanup performed)",
    re.IGNORECASE,
)
_DELETE_CAPABILITY_RE = re.compile(
    r"(?:delete[- ]branch capability resolved|gitea\.branch\.delete)\s*:\s*"
    r".*(?:gitea\.branch\.delete|delete_branch).*(?:resolved|allowed|proven)|"
    r"gitea\.branch\.delete\s+(?:resolved|allowed|proven)",
    re.IGNORECASE,
)
_DELETE_TASK_RE = re.compile(
    r"(?:delete_branch|cleanup_branch|reconcile_merged_cleanups)",
    re.IGNORECASE,
)
_MERGE_RESULT_RE = re.compile(
    r"merge result\s*:\s*(?:merged|success|performed)",
    re.IGNORECASE,
)
_MERGE_COMMIT_SHA_RE = re.compile(
    r"(?:merge commit sha|merged commit sha|merge commit)\s*:\s*([0-9a-f]{7,40})",
    re.IGNORECASE,
)
_PR_HEAD_BRANCH_RE = re.compile(
    r"(?:merged pr head branch|pr head branch|deleted branch)\s*:\s*(\S+)",
    re.IGNORECASE,
)
_BRANCH_NOT_PROTECTED_RE = re.compile(
    r"branch (?:is )?not protected|branch protection\s*:\s*(?:none|false|no)",
    re.IGNORECASE,
)
_OPEN_PR_INVENTORY_RE = re.compile(
    r"(?:no other open pr(?:\s+references)?(?:\s+\S+)?|open pr inventory proof|"
    r"open pr references).*(?:none|zero|0|clear|inventory complete)|"
    r"no other open pr references branch",
    re.IGNORECASE,
)
_ACTIVE_CLAIM_LEASE_RE = re.compile(
    r"(?:no active (?:heartbeat|claim|lease)|"
    r"(?:active )?(?:heartbeat|claim|lease)(?:/(?:claim|lease))*\s*:\s*none)",
    re.IGNORECASE,
)
_SESSION_OWNED_WORKTREE_RE = re.compile(
    r"(?:removed worktree path|cleanup worktree path|session-owned worktree)\s*:\s*(\S+)",
    re.IGNORECASE,
)
_BRANCHES_PATH_RE = re.compile(r"\bbranches/", re.IGNORECASE)
_CLEAN_TRACKED_RE = re.compile(
    r"(?:pre-removal tracked state|tracked state before removal)\s*:\s*clean",
    re.IGNORECASE,
)
_CLEAN_UNTRACKED_RE = re.compile(
    r"(?:pre-removal untracked state|untracked state before removal)\s*:\s*clean",
    re.IGNORECASE,
)
_WORKTREE_LIST_AFTER_RE = re.compile(
    r"(?:git worktree list after|post-removal worktree list|worktree list after)",
    re.IGNORECASE,
)
_WRONG_BRANCH_RE = re.compile(
    r"deleted branch (?:does not match|!=|differs from) (?:merged )?pr head",
    re.IGNORECASE,
)


def _claims_remote_delete(text: str) -> bool:
    return bool(_REMOTE_DELETE_CLAIM_RE.search(text))


def _claims_worktree_remove(text: str) -> bool:
    return bool(_WORKTREE_REMOVE_CLAIM_RE.search(text))


def _branch_safety_fields_present(text: str) -> list[str]:
    missing: list[str] = []
    if not _DELETE_CAPABILITY_RE.search(text):
        missing.append("delete-branch capability resolved (gitea.branch.delete)")
    if not _DELETE_TASK_RE.search(text):
        missing.append("delete-branch task named (delete_branch or cleanup)")
    if not _MERGE_RESULT_RE.search(text):
        missing.append("merge result: merged")
    if not _MERGE_COMMIT_SHA_RE.search(text):
        missing.append("merge commit SHA")
    if not _PR_HEAD_BRANCH_RE.search(text):
        missing.append("merged PR head branch / deleted branch name")
    if not _BRANCH_NOT_PROTECTED_RE.search(text):
        missing.append("branch not protected proof")
    if not _OPEN_PR_INVENTORY_RE.search(text):
        missing.append("open PR inventory proof (no other PR references branch)")
    if not _ACTIVE_CLAIM_LEASE_RE.search(text):
        missing.append("no active heartbeat/claim/lease proof")
    return missing


def _worktree_cleanup_fields_present(text: str) -> list[str]:
    missing: list[str] = []
    match = _SESSION_OWNED_WORKTREE_RE.search(text)
    path = match.group(1).strip() if match else ""
    if not path:
        missing.append("session-owned worktree path")
    elif not _BRANCHES_PATH_RE.search(path.replace("\\", "/")):
        missing.append("worktree path under branches/")
    if not _CLEAN_TRACKED_RE.search(text):
        missing.append("pre-removal tracked state: clean")
    if not _CLEAN_UNTRACKED_RE.search(text):
        missing.append("pre-removal untracked state: clean")
    if not _WORKTREE_LIST_AFTER_RE.search(text):
        missing.append("git worktree list after removal")
    return missing


def assess_post_merge_cleanup_proof(
    report_text: str,
    *,
    cleanup_session: dict | None = None,
) -> dict[str, Any]:
    """Validate post-merge cleanup claims carry safety-gate proof (#402)."""
    text = report_text or ""
    session = dict(cleanup_session or {})
    reasons: list[str] = []

    if _CLEANUP_SKIPPED_RE.search(text) or session.get("outcome") == CLEANUP_SKIPPED:
        blocker = (session.get("blocker") or "").strip()
        if not blocker:
            match = _CLEANUP_BLOCKER_RE.search(text)
            blocker = match.group(1).strip() if match else ""
        if blocker.upper() == CLEANUP_SKIPPED:
            blocker = ""
        if not blocker:
            reasons.append(
                "CLEANUP_SKIPPED requires exact cleanup blocker reason (#402)"
            )
        return {
            "block": bool(reasons),
            "proven": not reasons,
            "outcome": CLEANUP_SKIPPED,
            "remote_delete_claimed": False,
            "worktree_remove_claimed": False,
            "reasons": reasons,
            "safe_next_action": (
                "report CLEANUP_SKIPPED with exact blocker; do not claim performed cleanup"
                if reasons
                else "proceed"
            ),
        }

    if not _CLEANUP_SECTION_HINT.search(text) and not session.get("cleanup_claimed"):
        return {
            "block": False,
            "proven": True,
            "outcome": None,
            "remote_delete_claimed": False,
            "worktree_remove_claimed": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }

    remote_delete = bool(
        session.get("remote_delete_claimed") or _claims_remote_delete(text)
    )
    worktree_remove = bool(
        session.get("worktree_remove_claimed") or _claims_worktree_remove(text)
    )

    if _WRONG_BRANCH_RE.search(text):
        reasons.append(
            "cleanup report claims deleted branch that is not the merged PR head branch"
        )

    if remote_delete:
        reasons.extend(
            f"remote branch deletion missing {field}"
            for field in _branch_safety_fields_present(text)
        )

    if worktree_remove:
        reasons.extend(
            f"worktree removal missing {field}"
            for field in _worktree_cleanup_fields_present(text)
        )

    if (remote_delete or worktree_remove) and not (remote_delete or worktree_remove):
        pass

    if not remote_delete and not worktree_remove:
        cleanup_mutations = re.search(
            r"cleanup mutations\s*:\s*(?!none\b)\S",
            text,
            re.IGNORECASE,
        )
        if cleanup_mutations:
            reasons.append(
                "cleanup mutations reported without post-merge cleanup proof checklist"
            )

    outcome = CLEANUP_PERFORMED if (remote_delete or worktree_remove) and not reasons else None
    if remote_delete or worktree_remove:
        outcome = CLEANUP_PERFORMED if not reasons else "CLEANUP_CLAIMED_UNPROVEN"

    block = bool(reasons)
    return {
        "block": block,
        "proven": not block,
        "outcome": outcome,
        "remote_delete_claimed": remote_delete,
        "worktree_remove_claimed": worktree_remove,
        "reasons": reasons,
        "safe_next_action": (
            "report CLEANUP_SKIPPED with exact blocker or include the full cleanup "
            "checklist before claiming remote delete or worktree removal"
            if reasons
            else "proceed"
        ),
    }