Gitea-Tools/issue_lock_adoption.py

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

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"

# Citable decision labels aligned with the ``assess_own_branch_adoption``
# outcomes, surfaced verbatim in the live ``gitea_lock_issue`` response so
# recovery reports (#473-style) can quote the lock tool output directly
# instead of inferring adoption from separate offline checks (#477).
DECISION_LABELS = {
    ADOPT: "ADOPT",
    BLOCK_COMPETING: "BLOCK_COMPETING",
    NO_MATCH: "NO_MATCH",
}

_SAFE_NEXT_ACTIONS = {
    ADOPT: (
        "Own existing branch adopted for lock recovery; proceed to "
        "gitea_create_pr for this issue and cite this adoption proof."
    ),
    BLOCK_COMPETING: (
        "Competing same-issue branch(es) exist; resolve branch ownership "
        "before locking. No adoption performed (fail closed)."
    ),
    NO_MATCH: (
        "No existing branch carries this issue marker; normal lock path "
        "applied. No adoption performed."
    ),
}


def decision_label(outcome: str) -> str:
    """Map an ``assess_own_branch_adoption`` outcome to its citable label."""
    return DECISION_LABELS.get(outcome, "UNKNOWN")


def safe_next_action(outcome: str) -> str:
    """Return the safe next action string for an adoption *outcome*."""
    return _SAFE_NEXT_ACTIONS.get(
        outcome, "Unknown adoption outcome; treat as fail closed."
    )


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 _matcher_summary(issue_number: int, assessment: dict) -> str:
    """Explain, citably, why the assessed branch did or did not qualify.

    Names the numeric word-boundary rule so reports can show that
    ``issue-42`` was not matched inside ``issue-420`` (#440 / #477 AC3).
    """
    outcome = assessment.get("outcome")
    matched = assessment.get("matched_branch")
    competing = assessment.get("competing_branches") or []
    if outcome == ADOPT and matched:
        return (
            f"branch '{matched}' exactly matches the issue-{int(issue_number)} "
            f"marker (numeric word-boundary; 'issue-{int(issue_number)}' is not "
            f"matched inside 'issue-{int(issue_number)}0')"
        )
    if outcome == BLOCK_COMPETING:
        return (
            f"competing same-issue branch(es) {competing} carry the "
            f"issue-{int(issue_number)} marker but are not the requested "
            f"branch; ownership is ambiguous (fail closed)"
        )
    return (
        f"no existing branch carries the issue-{int(issue_number)} marker "
        f"under the numeric word-boundary rule"
    )


def _competing_branch_check(assessment: dict) -> dict:
    """Structured competing-branch verdict for the proof block."""
    competing = list(assessment.get("competing_branches") or [])
    return {
        "result": "blocked" if competing else "clear",
        "competing_branches": competing,
    }


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.

    #477: additionally surface explicit, citable adoption-proof fields tied to
    the ``assess_own_branch_adoption`` outcome (``adoption_decision``,
    ``adopted``, ``adopted_branch``, ``adopted_branch_head``,
    ``matcher_summary``, ``competing_branch_check``, ``safe_next_action``) so a
    recovery session can quote the live lock response directly. The explicit
    fields are populated for any outcome; ``adopted_branch`` /
    ``adopted_branch_head`` are set only when the outcome is ADOPT so a
    non-adoption proof can never be misread as claiming adoption.
    """
    outcome = assessment.get("outcome")
    adopted = outcome == ADOPT
    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,
        # Explicit citable fields (#477).
        "adoption_decision": decision_label(outcome),
        "adopted": adopted,
        "adopted_branch": branch_name if adopted else None,
        "adopted_branch_head": assessment.get("matched_head_sha") if adopted else None,
        "matcher_summary": _matcher_summary(issue_number, assessment),
        "competing_branch_check": _competing_branch_check(assessment),
        "safe_next_action": safe_next_action(outcome),
    }


def build_non_adoption_lock_proof(*, issue_number: int, branch_name: str) -> dict:
    """Safe, adoption-free proof metadata for a normal (NO_MATCH) lock.

    Requirement #477 AC2: non-adoption lock responses must stay clear and must
    not imply adoption. This returns explicit ``adopted: False`` metadata with
    the ``NO_MATCH`` decision so a normal lock response can carry citable proof
    without ever asserting a branch was adopted.
    """
    return {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "adoption_decision": DECISION_LABELS[NO_MATCH],
        "adopted": False,
        "adopted_branch": None,
        "adopted_branch_head": None,
        "matcher_summary": (
            f"no existing branch carries the issue-{int(issue_number)} marker; "
            f"normal lock path (no adoption)"
        ),
        "competing_branch_check": {"result": "clear", "competing_branches": []},
        "safe_next_action": safe_next_action(NO_MATCH),
    }