"""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-`` 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 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 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. """ marker = f"issue-{issue_number}" requested = (requested_branch or "").strip() matches: list[tuple[str, str | None]] = [] for entry in existing_branches or []: name = _branch_name(entry).strip() if marker in name: 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, }