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,
)
_GIT_REF_MUTATIONS_RE = re.compile(
    r"^\s*[-*]?\s*git ref mutations\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_WORKTREE_MUTATIONS_RE = re.compile(
    r"^\s*[-*]?\s*worktree mutations\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_MERGE_MUTATIONS_RE = re.compile(
    r"^\s*[-*]?\s*merge mutations\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_NONE_LEDGER_VALUES = frozenset({"", "none", "n/a"})
_MUTATION_LEDGER_PATTERNS = (
    _CLEANUP_MUTATIONS_RE,
    _GIT_REF_MUTATIONS_RE,
    _WORKTREE_MUTATIONS_RE,
)
_RAW_WORKTREE_REMOVE_PATTERNS = (
    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+worktree\s+remove\b", re.I),
)
_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 _ledger_field_body(text: str, pattern: re.Pattern[str]) -> str | None:
    match = pattern.search(text)
    if not match:
        return None
    return (match.group(1) or "").strip()


def _is_none_ledger_value(value: str) -> bool:
    return value.strip().lower() in _NONE_LEDGER_VALUES


def scoped_cleanup_mutation_ledger_text(text: str | None) -> str:
    """Return mutation-ledger bodies scoped to cleanup enforcement (#517)."""
    text = text or ""
    parts: list[str] = []
    for pattern in _MUTATION_LEDGER_PATTERNS:
        body = _ledger_field_body(text, pattern)
        if body is not None and not _is_none_ledger_value(body):
            parts.append(body)
    return "\n".join(parts)


def _git_ref_cleanup_claimed(body: str) -> bool:
    return bool(raw_branch_delete_commands(body))


def _worktree_cleanup_claimed(body: str) -> bool:
    for pattern in _RAW_WORKTREE_REMOVE_PATTERNS:
        if pattern.search(body):
            return True
    return bool(
        re.search(
            r"\b(?:removed|deleted)\b[^\n]{0,40}\bworktree\b",
            body,
            re.IGNORECASE,
        )
    )


def _cleanup_claiming_ledger_fields(text: str) -> list[tuple[str, str]]:
    claiming: list[tuple[str, str]] = []
    cleanup_body = _ledger_field_body(text, _CLEANUP_MUTATIONS_RE)
    if cleanup_body is not None and not _is_none_ledger_value(cleanup_body):
        claiming.append(("Cleanup mutations", cleanup_body))

    git_ref_body = _ledger_field_body(text, _GIT_REF_MUTATIONS_RE)
    if git_ref_body is not None and not _is_none_ledger_value(git_ref_body):
        if _git_ref_cleanup_claimed(git_ref_body):
            claiming.append(("Git ref mutations", git_ref_body))

    worktree_body = _ledger_field_body(text, _WORKTREE_MUTATIONS_RE)
    if worktree_body is not None and not _is_none_ledger_value(worktree_body):
        if _worktree_cleanup_claimed(worktree_body):
            claiming.append(("Worktree mutations", worktree_body))
    return claiming


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 mutation ledgers cite raw git branch deletion."""
    commands = raw_branch_delete_commands(scoped_cleanup_mutation_ledger_text(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 mutation ledgers cite raw comment deletion."""
    commands = raw_comment_delete_commands(scoped_cleanup_mutation_ledger_text(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 ""
    claiming = _cleanup_claiming_ledger_fields(text)
    if not claiming:
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }

    reasons: list[str] = []
    for field_name, body in claiming:
        if not _AUTHORIZED_TOOL_RE.search(body):
            reasons.append(
                f"{field_name} cleanup 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(
            scoped_cleanup_mutation_ledger_text(text)
        ),
        "raw_comment_commands": raw_comment_delete_commands(
            scoped_cleanup_mutation_ledger_text(text)
        ),
    }