Gitea-Tools/branch_cleanup_guard.py

"""Guards for merged-PR branch cleanup and raw git delete bypasses (#514)."""

from __future__ import annotations

import re
from typing import Any

PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})

# Evidence / preservation branches must never be removed by cleanup tools
# (e.g. chore/issue-681-preserve-review-session-wip).
_PRESERVATION_MARKERS = ("preserve", "preservation", "evidence")


def is_preservation_or_evidence_branch(branch: str | None) -> bool:
    """Return True when *branch* is a preservation/evidence ref that must stay."""
    if not branch:
        return False
    name = str(branch).lower()
    return any(marker in name for marker in _PRESERVATION_MARKERS)


_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),
)


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 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_cleanup_merged_pr_branch or another approved cleanup "
            "helper with explicit branch.delete capability"
            if reasons
            else "proceed"
        ),
    }


def assess_merged_pr_branch_cleanup(
    *,
    pr_number: int,
    head_branch: str,
    merged: bool,
    remote_branch_exists: bool,
    open_pr_heads: set[str],
    head_on_target: bool | None,
    delete_capability_allowed: bool,
    confirmation: str | None,
    protected_branches: frozenset[str] | None = None,
) -> dict[str, Any]:
    """Assess whether a merged PR source branch can be deleted via MCP."""
    protected = protected_branches or PROTECTED_BRANCHES
    expected_confirmation = f"CLEANUP MERGED PR {pr_number} BRANCH {head_branch}"
    reasons: list[str] = []
    if not merged:
        reasons.append("PR is not merged")
    if not remote_branch_exists:
        reasons.append("remote branch already absent")
    if not head_branch:
        reasons.append("PR head branch is missing")
    if head_branch in protected:
        reasons.append(f"branch '{head_branch}' is protected")
    if is_preservation_or_evidence_branch(head_branch):
        reasons.append(
            f"branch '{head_branch}' is a preservation/evidence branch and "
            "cannot be deleted through merged-PR cleanup"
        )
    if head_branch in open_pr_heads:
        reasons.append("an open PR still references this head branch")
    if head_on_target is False:
        reasons.append("PR head is not an ancestor of the target branch")
    if head_on_target is None:
        reasons.append("PR head ancestry could not be proven")
    if not delete_capability_allowed:
        reasons.append("gitea.branch.delete capability is not allowed")
    if confirmation != expected_confirmation:
        reasons.append(
            "confirmation must equal "
            f"'{expected_confirmation}' for branch cleanup"
        )

    safe = not reasons
    return {
        "pr_number": pr_number,
        "head_branch": head_branch,
        "expected_confirmation": expected_confirmation,
        "remote_branch_exists": remote_branch_exists,
        "safe_to_delete": safe,
        "block_reasons": reasons,
        "recommended_action": "delete_remote_branch" if safe else "keep_remote_branch",
    }


# ---------------------------------------------------------------------------
# #687 remediation: post-delete readback + active ownership protection
# ---------------------------------------------------------------------------

READBACK_NOT_FOUND = "not_found"
READBACK_EXISTS = "exists"
READBACK_AUTHENTICATION = "authentication_error"
READBACK_AUTHORIZATION = "authorization_error"
READBACK_TRANSPORT = "transport_error"
READBACK_UNEXPECTED = "unexpected_response"
READBACK_AMBIGUOUS_404 = "ambiguous_not_found"

# Scope of a 404: only branch-scoped absence may set verified_absent.
NOT_FOUND_SCOPE_BRANCH = "branch"
NOT_FOUND_SCOPE_REPOSITORY = "repository"
NOT_FOUND_SCOPE_HOST = "host"
NOT_FOUND_SCOPE_UNKNOWN = "unknown"

ERROR_CLASS_AUTHENTICATION = "authentication"
ERROR_CLASS_AUTHORIZATION = "authorization"
ERROR_CLASS_TRANSPORT = "transport"
ERROR_CLASS_UNEXPECTED = "unexpected"

OWNERSHIP_CATEGORY_AUTHOR_SESSION = "author_session"
OWNERSHIP_CATEGORY_AUTHOR_LEASE = "author_lease"
OWNERSHIP_CATEGORY_REVIEWER_LEASE = "reviewer_lease"
OWNERSHIP_CATEGORY_MERGER_LEASE = "merger_lease"
OWNERSHIP_CATEGORY_CONTROLLER_LEASE = "controller_lease"
OWNERSHIP_CATEGORY_RECONCILER_LEASE = "reconciler_lease"
OWNERSHIP_CATEGORY_WORKTREE_BINDING = "worktree_binding"
OWNERSHIP_CATEGORY_INVENTORY_ERROR = "ownership_inventory_error"

_ROLE_TO_OWNERSHIP_CATEGORY = {
    "author": OWNERSHIP_CATEGORY_AUTHOR_LEASE,
    "reviewer": OWNERSHIP_CATEGORY_REVIEWER_LEASE,
    "merger": OWNERSHIP_CATEGORY_MERGER_LEASE,
    "controller": OWNERSHIP_CATEGORY_CONTROLLER_LEASE,
    "reconciler": OWNERSHIP_CATEGORY_RECONCILER_LEASE,
}

_ACTIVE_OWNERSHIP_STATUSES = frozenset(
    {"active", "live", "claimed", "in_progress", "working", "pushing", "pushed"}
)
_TERMINAL_OWNERSHIP_STATUSES = frozenset(
    {"released", "abandoned", "done", "blocked", "terminal", "closed"}
)
_EXPIRED_STATUSES = frozenset({"expired"})
_STALE_STATUSES = frozenset(
    {
        "stale",
        "stale_dead_process",
        "stale_missing_worktree",
        # #790 Slice A heartbeat-lifecycle bands. Listed here so they are
        # *classified* rather than falling through to the unknown-status branch;
        # they still block unless the ownership record proves
        # ``reclaim_allowed is True``, so the O2 fail-closed rule is unchanged.
        "stale_missed_heartbeat",
        "stale_absolute_cap",
    }
)


def _norm_str(value: Any) -> str:
    return str(value or "").strip()


def normalize_host(host: str | None) -> str:
    """Normalize a host identity for ownership matching (no credentials)."""
    text = _norm_str(host).lower()
    for prefix in ("https://", "http://"):
        if text.startswith(prefix):
            text = text[len(prefix) :]
    # Drop path/query if a full URL slipped through.
    text = text.split("/", 1)[0]
    text = text.split("?", 1)[0]
    return text.rstrip(".")


def ownership_category_for_role(role: str | None) -> str:
    """Map a role kind to a non-secret ownership category label."""
    key = _norm_str(role).lower()
    return _ROLE_TO_OWNERSHIP_CATEGORY.get(key, f"{key or 'unknown'}_lease")


def classify_branch_readback_http_status(
    status_code: int | None,
    *,
    not_found_scope: str | None = None,
) -> dict[str, Any]:
    """Classify a GET-branch HTTP status into a secret-free readback result.

    R1: A bare/generic/repository/wrong-host 404 never yields
    ``verified_absent=True``. Only an authoritative *branch-scoped* not-found
    (``not_found_scope='branch'``) may verify deletion.
    """
    if status_code == 404:
        scope = _norm_str(not_found_scope).lower() or NOT_FOUND_SCOPE_UNKNOWN
        if scope == NOT_FOUND_SCOPE_BRANCH:
            return {
                "status": READBACK_NOT_FOUND,
                "error_class": None,
                "verified_absent": True,
                "branch_present": False,
                "not_found_scope": NOT_FOUND_SCOPE_BRANCH,
                "reasons": [],
            }
        # repository / host / unknown / generic 404 — not verified absence
        reason = {
            NOT_FOUND_SCOPE_REPOSITORY: (
                "post-delete readback 404 is repository-scoped, not branch absence"
            ),
            NOT_FOUND_SCOPE_HOST: (
                "post-delete readback 404 is host-scoped, not branch absence"
            ),
        }.get(
            scope,
            "post-delete readback 404 is ambiguous (not branch-scoped); "
            "cannot verify absence",
        )
        return {
            "status": READBACK_AMBIGUOUS_404,
            "error_class": ERROR_CLASS_UNEXPECTED,
            "verified_absent": False,
            "branch_present": None,
            "not_found_scope": scope,
            "reasons": [reason],
        }
    if status_code in (401, 407):
        return {
            "status": READBACK_AUTHENTICATION,
            "error_class": ERROR_CLASS_AUTHENTICATION,
            "verified_absent": False,
            "branch_present": None,
            "reasons": ["post-delete branch readback authentication failed"],
        }
    if status_code == 403:
        return {
            "status": READBACK_AUTHORIZATION,
            "error_class": ERROR_CLASS_AUTHORIZATION,
            "verified_absent": False,
            "branch_present": None,
            "reasons": ["post-delete branch readback authorization failed"],
        }
    if status_code is not None and 200 <= int(status_code) < 300:
        return {
            "status": READBACK_EXISTS,
            "error_class": None,
            "verified_absent": False,
            "branch_present": True,
            "reasons": ["post-delete readback found branch still present"],
        }
    if status_code is not None and int(status_code) >= 500:
        return {
            "status": READBACK_TRANSPORT,
            "error_class": ERROR_CLASS_TRANSPORT,
            "verified_absent": False,
            "branch_present": None,
            "reasons": ["post-delete branch readback transport/upstream failure"],
        }
    return {
        "status": READBACK_UNEXPECTED,
        "error_class": ERROR_CLASS_UNEXPECTED,
        "verified_absent": False,
        "branch_present": None,
        "reasons": ["post-delete branch readback returned an unexpected response"],
        "http_status": status_code,
    }


def _extract_http_status(exc: BaseException) -> int | None:
    """Extract an HTTP status code from an exception chain (secret-free)."""
    seen: set[int] = set()
    current: BaseException | None = exc
    while current is not None and id(current) not in seen:
        seen.add(id(current))
        for attr in ("code", "status", "status_code"):
            value = getattr(current, attr, None)
            if isinstance(value, int) and 100 <= value <= 599:
                return value
        text = str(current) if current is not None else ""
        match = re.match(r"HTTP\s+(\d{3})\b", text)
        if match:
            return int(match.group(1))
        current = current.__cause__ or current.__context__
    return None


def classify_branch_readback_exception(
    exc: BaseException,
    *,
    not_found_scope: str | None = None,
) -> dict[str, Any]:
    """Classify a GET-branch exception without leaking credentials or bodies.

    R1: substring ``404`` / ``not found`` alone never becomes verified_absent.
    Callers must pass ``not_found_scope='branch'`` only after authoritative
    proof that the repository/host is still reachable and the 404 is branch-level.
    """
    status_code = _extract_http_status(exc)
    if status_code is None:
        lower = str(exc).lower() if exc is not None else ""
        if any(
            token in lower
            for token in (
                "timed out",
                "timeout",
                "connection",
                "network",
                "temporarily unavailable",
                "name or service not known",
                "nodename nor servname",
            )
        ):
            return {
                "status": READBACK_TRANSPORT,
                "error_class": ERROR_CLASS_TRANSPORT,
                "verified_absent": False,
                "branch_present": None,
                "reasons": [
                    "post-delete branch readback transport/upstream failure"
                ],
            }
        if "unauthorized" in lower:
            status_code = 401
        elif "forbidden" in lower:
            status_code = 403
        elif "404" in lower or "not found" in lower:
            # Ambiguous: do NOT treat as branch absence without scope proof.
            status_code = 404
            if not_found_scope is None:
                not_found_scope = NOT_FOUND_SCOPE_UNKNOWN

    if status_code is not None:
        return classify_branch_readback_http_status(
            status_code, not_found_scope=not_found_scope
        )

    return {
        "status": READBACK_UNEXPECTED,
        "error_class": ERROR_CLASS_UNEXPECTED,
        "verified_absent": False,
        "branch_present": None,
        "reasons": ["post-delete branch readback returned an unexpected response"],
    }


def assess_post_delete_readback(readback: dict[str, Any] | None) -> dict[str, Any]:
    """Decide whether DELETE success may be reported after branch readback.

    Success requires authoritative branch-scoped not-found
    (``verified_absent=True``). DELETE HTTP success alone is never enough.
    Generic/repository/host 404 cannot verify absence (R1).
    """
    payload = dict(readback or {})
    status = _norm_str(payload.get("status")) or READBACK_UNEXPECTED
    # Only explicit verified_absent flag counts — never infer from status alone
    # when status is a bare not_found without branch scope proof.
    verified = bool(payload.get("verified_absent")) is True
    if verified and status == READBACK_NOT_FOUND:
        return {
            "ok": True,
            "success": True,
            "verified_absent": True,
            "readback": {
                "status": READBACK_NOT_FOUND,
                "verified_absent": True,
                "branch_present": False,
                "error_class": None,
                "not_found_scope": NOT_FOUND_SCOPE_BRANCH,
            },
            "reasons": [],
        }

    reasons = list(payload.get("reasons") or [])
    if not reasons:
        if status == READBACK_EXISTS:
            reasons = ["post-delete readback found branch still present"]
        elif status == READBACK_AUTHENTICATION:
            reasons = ["post-delete branch readback authentication failed"]
        elif status == READBACK_AUTHORIZATION:
            reasons = ["post-delete branch readback authorization failed"]
        elif status == READBACK_TRANSPORT:
            reasons = ["post-delete branch readback transport/upstream failure"]
        elif status == READBACK_AMBIGUOUS_404:
            reasons = [
                "post-delete readback 404 is not branch-scoped; "
                "cannot verify absence"
            ]
        else:
            reasons = ["post-delete branch readback could not verify deletion"]

    return {
        "ok": False,
        "success": False,
        "verified_absent": False,
        "readback": {
            "status": status,
            "verified_absent": False,
            "branch_present": payload.get("branch_present"),
            "error_class": payload.get("error_class"),
            "not_found_scope": payload.get("not_found_scope"),
        },
        "reasons": reasons,
        "blocker_kind": "post_delete_readback_failed",
    }


def cleanup_result_envelope(
    *,
    success: bool,
    performed: bool,
    delete_acknowledged: bool,
    verified_absent: bool,
    **extra: Any,
) -> dict[str, Any]:
    """R2: consistent top-level cleanup result fields on every return path."""
    out: dict[str, Any] = {
        "success": bool(success),
        "performed": bool(performed),
        "delete_acknowledged": bool(delete_acknowledged),
        "verified_absent": bool(verified_absent),
    }
    out.update(extra)
    return out


def _repo_matches(
    record: dict[str, Any],
    *,
    remote: str,
    org: str,
    repo: str,
    host: str | None = None,
) -> bool:
    """Match ownership record to target remote/org/repo and normalized host."""
    if (
        _norm_str(record.get("remote")).lower() != _norm_str(remote).lower()
        or _norm_str(record.get("org")).lower() != _norm_str(org).lower()
        or _norm_str(record.get("repo")).lower() != _norm_str(repo).lower()
    ):
        return False
    expected_host = normalize_host(host)
    record_host = normalize_host(record.get("host") or record.get("host_name"))
    # When both sides declare a host, they must agree after normalization.
    # A record host that disagrees with the expected host is out of scope.
    # Legacy records without host still match when remote/org/repo agree.
    if expected_host and record_host and expected_host != record_host:
        return False
    return True


def _branch_matches(record: dict[str, Any], branch: str) -> bool:
    return _norm_str(record.get("branch")) == _norm_str(branch)


def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]:
    """Classify one ownership record as blocking or non-blocking.

    Distinguishes active ownership from expired/released/terminal/stale records.
    Expired/stale records block unless reclaim_allowed is *explicitly* True
    (O2: never treat missing/unknown reclaim as auto-allowed).
    """
    status = _norm_str(record.get("status")).lower()
    category = _norm_str(record.get("category")) or "unknown"
    reclaim_allowed = record.get("reclaim_allowed")

    if category == OWNERSHIP_CATEGORY_INVENTORY_ERROR:
        return {
            "blocks": True,
            "status": status or "unknown",
            "category": category,
            "reason": "ownership inventory failed closed",
        }

    if status in _TERMINAL_OWNERSHIP_STATUSES:
        return {
            "blocks": False,
            "status": status,
            "category": category,
            "reason": f"{category} ownership is terminal/released ({status})",
        }
    if status in _ACTIVE_OWNERSHIP_STATUSES:
        return {
            "blocks": True,
            "status": status,
            "category": category,
            "reason": f"active {category} ownership still uses the target branch",
        }
    if status in _EXPIRED_STATUSES or status in _STALE_STATUSES:
        # O2: only explicit reclaim_allowed=True skips the block.
        if reclaim_allowed is True:
            return {
                "blocks": False,
                "status": status,
                "category": category,
                "reason": (
                    f"{category} ownership is {status} and reclaimable; "
                    "does not block deletion"
                ),
            }
        return {
            "blocks": True,
            "status": status,
            "category": category,
            "reason": (
                f"{status} {category} ownership still protects the target "
                "branch (reclaim not proven; fail closed)"
            ),
        }
    # Unknown status → fail closed
    return {
        "blocks": True,
        "status": status or "unknown",
        "category": category,
        "reason": (
            f"unclassified {category} ownership status "
            f"'{status or 'unknown'}'; fail closed"
        ),
    }


def assess_active_branch_ownership(
    *,
    remote: str,
    org: str,
    repo: str,
    branch: str,
    host: str | None = None,
    records: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Assess whether any active ownership still uses *branch* in *repo*.

    Matching requires remote/org/repo/branch and, when provided, normalized
    host identity. Other repositories, hosts, or branches never false-block.
    """
    target_branch = _norm_str(branch)
    expected_host = normalize_host(host)
    considered: list[dict[str, Any]] = []
    blocking: list[dict[str, Any]] = []
    ignored: list[dict[str, Any]] = []

    for raw in records or []:
        if not isinstance(raw, dict):
            continue
        if not _repo_matches(
            raw, remote=remote, org=org, repo=repo, host=expected_host or None
        ):
            ignored.append(
                {
                    "category": _norm_str(raw.get("category")) or "unknown",
                    "reason": "different repository or host scope",
                }
            )
            continue
        if not _branch_matches(raw, target_branch):
            ignored.append(
                {
                    "category": _norm_str(raw.get("category")) or "unknown",
                    "reason": "different branch",
                }
            )
            continue
        activity = assess_ownership_record_activity(raw)
        entry = {
            "category": activity["category"],
            "status": activity["status"],
            "blocks": activity["blocks"],
            "reason": activity["reason"],
        }
        considered.append(entry)
        if activity["blocks"]:
            blocking.append(entry)

    block = bool(blocking)
    categories = sorted({b["category"] for b in blocking})
    reasons = [
        (
            "active ownership protects branch "
            f"'{target_branch}': " + "; ".join(b["reason"] for b in blocking)
        )
    ] if block else []
    return {
        "block": block,
        "safe_to_delete": not block,
        "remote": remote,
        "org": org,
        "repo": repo,
        "host": expected_host or None,
        "branch": target_branch,
        "blocking_categories": categories,
        "blocking": blocking,
        "considered": considered,
        "ignored_out_of_scope": ignored,
        "reasons": reasons,
        "blocker_kind": "active_branch_ownership" if block else None,
        "recommended_action": "keep_remote_branch" if block else "delete_remote_branch",
    }