Gitea-Tools/mcp_native_cleanup_proof.py

"""MCP-native post-merge cleanup proof verifier (#517).

Post-merge cleanup of leases, comments, branches, and worktrees must be
proven through explicit MCP tools (or approved helpers), never raw scripts or
ad hoc git/API commands. Merger/reviewer sessions must hand cleanup to a
reconciler profile with the right capability proof.
"""

from __future__ import annotations

import re
from typing import Any

AUTHORIZED_CLEANUP_TOOLS = frozenset({
    "gitea_cleanup_post_merge_moot_lease",
    "gitea_reconcile_merged_cleanups",
    "gitea_delete_branch",
    "gitea_cleanup_merged_pr_branch",
    "gitea_audit_worktree_cleanup",
    "gitea_capture_branches_worktree_snapshot",
    "gitea_assess_worktree_cleanup_integrity",
    "gitea_authorize_reconciliation_cleanup_phase",
})

_RAW_BRANCH_DELETE_PATTERNS = (
    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I),
    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I),
    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s:[^\s`]+", re.I),
)

_RAW_COMMENT_DELETE_PATTERNS = (
    re.compile(
        r"(?:curl|wget|httpie)\b[^\n\r]*\b(?:DELETE|delete)\b[^\n\r]*"
        r"(?:/comments/|issues/\d+/comments)",
        re.I,
    ),
    re.compile(
        r"\bDELETE\b[^\n\r]*/repos/[^\n\r]*/issues/\d+/comments/\d+",
        re.I,
    ),
    re.compile(
        r"\b(?:delete_issue_comment|remove_issue_comment|purge_comments?)\b",
        re.I,
    ),
    re.compile(
        r"\b(?:raw|ad hoc|adhoc)\b[^\n\r]{0,40}\bcomment\b[^\n\r]{0,40}\bdelete",
        re.I,
    ),
)

_CLEANUP_MUTATIONS_RE = re.compile(
    r"^\s*[-*]?\s*cleanup mutations\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_MERGE_MUTATIONS_RE = re.compile(
    r"^\s*[-*]?\s*merge mutations\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_AUTHORIZED_TOOL_RE = re.compile(
    r"\b(" + "|".join(re.escape(t) for t in sorted(AUTHORIZED_CLEANUP_TOOLS)) + r")\b",
    re.IGNORECASE,
)
_RECONCILER_CAPABILITY_RE = re.compile(
    r"(?:reconciler|gitea\.branch\.delete|delete_branch|reconcile_merged_cleanups)",
    re.IGNORECASE,
)
_MERGER_CLEANUP_ROLE_RE = re.compile(
    r"(?:merger|reviewer).{0,80}(?:deleted|removed|cleaned).{0,80}"
    r"(?:branch|worktree|comment|lease)",
    re.IGNORECASE,
)
_HANDOFF_TO_RECONCILER_RE = re.compile(
    r"(?:hand(?:ed)? off|defer(?:red)?|next actor).{0,60}reconciler",
    re.IGNORECASE,
)


def raw_branch_delete_commands(text: str | None) -> list[str]:
    """Return raw git branch-delete commands cited in *text*."""
    if not text:
        return []
    commands: list[str] = []
    for pattern in _RAW_BRANCH_DELETE_PATTERNS:
        commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text))
    return list(dict.fromkeys(commands))


def raw_comment_delete_commands(text: str | None) -> list[str]:
    """Return raw comment-deletion commands/scripts cited in *text*."""
    if not text:
        return []
    commands: list[str] = []
    for pattern in _RAW_COMMENT_DELETE_PATTERNS:
        commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text))
    return list(dict.fromkeys(commands))


def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]:
    """Fail closed when a report uses raw git branch deletion as cleanup proof."""
    commands = raw_branch_delete_commands(text)
    reasons = [
        (
            "raw git branch deletion bypasses MCP branch.delete cleanup gates: "
            f"{command}"
        )
        for command in commands
    ]
    return {
        "proven": not reasons,
        "block": bool(reasons),
        "commands": commands,
        "reasons": reasons,
        "safe_next_action": (
            "use gitea_delete_branch, gitea_reconcile_merged_cleanups, or another "
            "approved MCP cleanup helper with explicit branch.delete capability"
            if reasons
            else "proceed"
        ),
    }


def assess_raw_comment_delete_report(text: str | None) -> dict[str, Any]:
    """Fail closed when a report uses raw comment deletion as cleanup proof."""
    commands = raw_comment_delete_commands(text)
    reasons = [
        (
            "raw comment deletion bypasses MCP lease cleanup gates: "
            f"{command}"
        )
        for command in commands
    ]
    return {
        "proven": not reasons,
        "block": bool(reasons),
        "commands": commands,
        "reasons": reasons,
        "safe_next_action": (
            "use gitea_cleanup_post_merge_moot_lease (append-only release comment) "
            "or hand cleanup to a reconciler session; never delete lease comments"
            if reasons
            else "proceed"
        ),
    }


def assess_merge_cleanup_mutation_separation(text: str | None) -> dict[str, Any]:
    """Require merge mutations and cleanup mutations in separate ledger fields."""
    text = text or ""
    merge_match = _MERGE_MUTATIONS_RE.search(text)
    cleanup_match = _CLEANUP_MUTATIONS_RE.search(text)
    reasons: list[str] = []

    if cleanup_match and not merge_match:
        combined = re.search(
            r"merge.{0,40}cleanup mutations|cleanup.{0,40}merge mutations",
            text,
            re.IGNORECASE,
        )
        if combined:
            reasons.append(
                "merge and cleanup mutations must use separate 'Merge mutations' "
                "and 'Cleanup mutations' ledger fields"
            )

    if merge_match and cleanup_match:
        merge_val = (merge_match.group(1) or "").strip().lower()
        cleanup_val = (cleanup_match.group(1) or "").strip().lower()
        if merge_val == cleanup_val and merge_val not in {"", "none"}:
            reasons.append(
                "merge mutations and cleanup mutations must not duplicate the "
                "same ledger entry"
            )

    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "safe_next_action": (
            "split merge mutations (gitea_merge_pr) from cleanup mutations "
            "(reconciler MCP tools) in the controller handoff"
            if reasons
            else "proceed"
        ),
    }


def assess_authorized_reconciler_cleanup_path(text: str | None) -> dict[str, Any]:
    """Validate cleanup claims cite authorized MCP tools and reconciler capability."""
    text = text or ""
    cleanup_match = _CLEANUP_MUTATIONS_RE.search(text)
    if not cleanup_match:
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }

    cleanup_body = cleanup_match.group(1) or ""
    if cleanup_body.strip().lower() in {"", "none", "n/a"}:
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }

    reasons: list[str] = []
    if not _AUTHORIZED_TOOL_RE.search(cleanup_body):
        reasons.append(
            "cleanup mutations must name an authorized MCP cleanup tool "
            f"({', '.join(sorted(AUTHORIZED_CLEANUP_TOOLS))})"
        )
    if not _RECONCILER_CAPABILITY_RE.search(text):
        reasons.append(
            "post-merge cleanup requires reconciler capability proof "
            "(reconciler profile, gitea.branch.delete, or reconcile_merged_cleanups)"
        )

    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "safe_next_action": (
            "hand cleanup to a prgs-reconciler session and cite the exact MCP tool "
            "plus delete_branch/reconcile_merged_cleanups capability proof"
            if reasons
            else "proceed"
        ),
    }


def assess_merger_cleanup_handoff_guidance(text: str | None) -> dict[str, Any]:
    """Merger sessions that performed cleanup must hand off to reconciler."""
    text = text or ""
    if not _MERGER_CLEANUP_ROLE_RE.search(text):
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }
    if _HANDOFF_TO_RECONCILER_RE.search(text):
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "merger/reviewer session performed cleanup but did not hand off to "
            "reconciler for MCP-native cleanup"
        ],
        "safe_next_action": (
            "merger sessions must end with cleanup handed to prgs-reconciler; "
            "do not perform ad hoc branch/comment/worktree cleanup as merger"
        ),
    }


def assess_mcp_native_cleanup_proof(report_text: str | None) -> dict[str, Any]:
    """Composite #517 verifier for MCP-native post-merge cleanup proof."""
    text = report_text or ""
    checks = (
        assess_raw_branch_delete_report(text),
        assess_raw_comment_delete_report(text),
        assess_merge_cleanup_mutation_separation(text),
        assess_authorized_reconciler_cleanup_path(text),
        assess_merger_cleanup_handoff_guidance(text),
    )
    reasons: list[str] = []
    safe_next = "proceed"
    for result in checks:
        reasons.extend(result.get("reasons") or [])
        if result.get("block") and result.get("safe_next_action"):
            safe_next = result["safe_next_action"]

    block = bool(reasons)
    return {
        "proven": not block,
        "block": block,
        "reasons": reasons,
        "safe_next_action": safe_next,
        "raw_branch_commands": raw_branch_delete_commands(text),
        "raw_comment_commands": raw_comment_delete_commands(text),
    }