sysadminandClaude Opus 4.8
43874bf092
Adds structured issue-branch ownership parsing, wires it into lock adoption and gitea_lock_issue validation, documents the restart recovery workflow, and adds regression tests for adoption, durable create_pr resolution, and open-PR blocking without remote branch deletion. Built on keyed persistent lock store and own-branch adoption (#443 / #442). Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#440 / #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 issue_branch_ownership
|
|
|
|
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."""
|
|
requested = (requested_branch or "").strip()
|
|
|
|
matches: list[tuple[str, str | None]] = []
|
|
for entry in existing_branches or []:
|
|
name = _branch_name(entry).strip()
|
|
if issue_branch_ownership.branch_belongs_to_issue(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]
|
|
|
|
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."""
|
|
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,
|
|
} |