Gitea-Tools/issue_lock_adoption.py

"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#442).

When an issue's own already-pushed branch exists, lock reacquisition must be
allowed (adoption) instead of being treated as #400 duplicate competing work.
This module isolates the pure decision so it can be unit-tested apart from the
MCP server's live Gitea calls.

Adoption is granted only for the issue's *exact* requested branch. Any other
branch that merely contains the same ``issue-<n>`` marker is competing work and
stays fail-closed. Open-PR, competing-live-lock, capability, and worktree
safety checks are enforced by the caller before this decision is consulted;
this module additionally records whether they passed for proof purposes.
"""

from __future__ import annotations

import re

ADOPT = "adopt_existing_branch"
BLOCK_COMPETING = "block_competing_branch"
NO_MATCH = "no_matching_branch"


def _branch_name(entry) -> str:
    if isinstance(entry, dict):
        return str(entry.get("name") or "")
    return str(entry or "")


def _branch_sha(entry) -> str | None:
    if isinstance(entry, dict):
        sha = entry.get("commit_sha")
        if sha:
            return str(sha)
    return None


def _branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
    """Return True when *branch_name* references issue *issue_number* exactly.

    Uses a numeric word-boundary so ``issue-42`` does not match inside
    ``issue-420`` (AC6 / #440).
    """
    name = (branch_name or "").strip()
    if not name:
        return False
    pattern = rf"(?:^|/)issue-{int(issue_number)}(?![0-9])"
    return re.search(pattern, name) is not None


def assess_own_branch_adoption(
    *,
    issue_number: int,
    requested_branch: str,
    existing_branches,
) -> dict:
    """Decide whether an existing matching branch is adoptable.

    Args:
        issue_number: The tracking issue number being locked.
        requested_branch: The exact branch the caller wants to lock.
        existing_branches: Iterable of remote branch entries — either names or
            dicts with ``name`` and optional ``commit_sha``.

    Returns:
        dict with:
          * ``outcome`` — one of ADOPT / BLOCK_COMPETING / NO_MATCH
          * ``adopt`` (bool), ``block`` (bool)
          * ``reason`` (str)
          * ``matched_branch`` (str | None), ``matched_head_sha`` (str | None)
          * ``competing_branches`` (list[str])

    ADOPT: the issue's exact branch exists and no other same-issue branch does.
    BLOCK_COMPETING: at least one same-issue branch is not the requested branch.
    NO_MATCH: no branch carries the issue marker — normal lock path applies.
    """
    requested = (requested_branch or "").strip()

    matches: list[tuple[str, str | None]] = []
    for entry in existing_branches or []:
        name = _branch_name(entry).strip()
        if _branch_carries_issue_marker(name, issue_number):
            matches.append((name, _branch_sha(entry)))

    competing = sorted({name for name, _ in matches if name != requested})
    exact = [(name, sha) for name, sha in matches if name == requested]

    # Fail closed whenever any non-requested same-issue branch exists, even if
    # the requested branch is also present: ownership is then ambiguous.
    if competing:
        return {
            "outcome": BLOCK_COMPETING,
            "adopt": False,
            "block": True,
            "reason": (
                f"issue #{issue_number} already has matching branch(es) "
                f"{competing} that are not the requested branch "
                f"'{requested}' (fail closed)"
            ),
            "matched_branch": None,
            "matched_head_sha": None,
            "competing_branches": competing,
        }

    if exact:
        name, sha = exact[0]
        return {
            "outcome": ADOPT,
            "adopt": True,
            "block": False,
            "reason": (
                f"existing branch '{name}' is the exact requested branch for "
                f"issue #{issue_number}; adopting it for lock recovery"
            ),
            "matched_branch": name,
            "matched_head_sha": sha,
            "competing_branches": [],
        }

    return {
        "outcome": NO_MATCH,
        "adopt": False,
        "block": False,
        "reason": f"no existing branch matches issue #{issue_number}",
        "matched_branch": None,
        "matched_head_sha": None,
        "competing_branches": [],
    }


def build_adoption_proof(
    *,
    issue_number: int,
    branch_name: str,
    assessment: dict,
    open_pr_checked: bool,
    competing_lock_checked: bool,
    lock_file_path: str,
    lock_file_status: str,
) -> dict:
    """Assemble the proof block returned by ``gitea_lock_issue`` on adoption.

    Requirement #4: adoption results must carry issue number, branch name,
    branch head commit, adoption reason, no-existing-PR proof, no-competing-
    live-lock proof, and lock file path/status.
    """
    return {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "branch_head_commit": assessment.get("matched_head_sha"),
        "adoption_reason": assessment.get("reason"),
        "no_existing_pr_proof": bool(open_pr_checked),
        "no_competing_live_lock_proof": bool(competing_lock_checked),
        "lock_file_path": lock_file_path,
        "lock_file_status": lock_file_status,
    }