feat: add own-branch lock adoption recovery to gitea_lock_issue (Closes #442)

Closes #442.

gitea_lock_issue treated ANY remote branch containing issue-<n> as
duplicate competing work (#400) and failed closed — including the issue's
own already-pushed branch. If the in-memory lock was lost (e.g. MCP server
restart) after the branch was pushed but before the PR, there was no
non-destructive way to reacquire the lock, deadlocking PR creation. This
surfaced during recovery of #420 (feat/issue-420-server-code-parity).

Adds a safe own-branch adoption path.

Changes:
- issue_lock_adoption.py — assess_own_branch_adoption() distinguishes the
  issue's exact requested branch (adopt) from a different same-issue branch
  (competing, still fail-closed); ambiguous (own + other) fails closed.
  build_adoption_proof() emits the required proof block.
- gitea_mcp_server.py — gitea_lock_issue consults the adoption assessment in
  place of the blanket branch-match block; on adoption the result carries an
  `adoption` proof (issue, branch, head commit, reason, no-existing-PR proof,
  no-competing-live-lock proof, lock file path/status). Open-PR and active-
  lease checks already run first, so adoption implies neither exists. Adds
  _branch_entry_commit_sha helper.
- Tests: tests/test_issue_lock_adoption.py (8 cases) + 3 MCP-level cases in
  test_mcp_server.py (adopt own branch, open PR blocks adoption, create_pr
  proceeds after adopted lock). Existing competing-branch block preserved
  (still raises "already has matching branch").

Safety:
- Different same-issue branch, competing branch, existing open PR, ambiguous
  ownership all remain fail-closed. #400 duplicate-work protection intact.
- #420's feature branch was only read (git ls-remote); not modified/deleted.

Validation:
  venv/bin/python -m pytest tests/ -q  ->  1611 passed, 6 skipped

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

gitea_mcp_server.py

@@ -503,6 +503,7 @@ import issue_lock_worktree  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
+import issue_lock_adoption  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
@@ -641,6 +642,19 @@ def _branch_entry_name(branch: dict | str) -> str:
     return str(branch.get("name") or branch.get("ref") or "")
 
 
+def _branch_entry_commit_sha(branch: dict | str) -> str | None:
+    """Best-effort head SHA for a Gitea branch entry (None when absent)."""
+    if not isinstance(branch, dict):
+        return None
+    commit = branch.get("commit")
+    if isinstance(commit, dict):
+        sha = commit.get("id") or commit.get("sha")
+        if sha:
+            return str(sha)
+    sha = branch.get("commit_sha")
+    return str(sha) if sha else None
+
+
 def _reveal_endpoints() -> bool:
     """Admin/debug opt-in (#120): include endpoint URLs and token source
     names in tool output. Off by default so normal LLM-facing responses
@@ -1146,13 +1160,28 @@ def gitea_lock_issue(
         branches = api_get_all(branch_url, auth)
     except Exception as e:
         raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
-    for branch in branches:
-        name = _branch_entry_name(branch)
-        if expected_pattern in name:
-            raise ValueError(
-                f"Issue #{issue_number} already has matching branch '{name}' "
-                "(fail closed)"
-            )
+    # #442: distinguish the issue's own already-pushed branch (adoptable for
+    # lock recovery) from a different same-issue branch (competing work, still
+    # fail-closed per #400). Open-PR and active-lease checks already ran above,
+    # so reaching here means no open PR and no competing live lock.
+    existing_branch_entries = [
+        {
+            "name": _branch_entry_name(branch),
+            "commit_sha": _branch_entry_commit_sha(branch),
+        }
+        for branch in branches
+    ]
+    adoption = issue_lock_adoption.assess_own_branch_adoption(
+        issue_number=issue_number,
+        requested_branch=branch_name,
+        existing_branches=existing_branch_entries,
+    )
+    if adoption["block"]:
+        competing = ", ".join(adoption["competing_branches"])
+        raise ValueError(
+            f"Issue #{issue_number} already has matching branch '{competing}' "
+            "that is not the requested branch (fail closed)"
+        )
 
     work_lease = _build_author_issue_work_lease(
         issue_number=issue_number,
@@ -1190,6 +1219,21 @@ def gitea_lock_issue(
         "worktree_path": resolved_worktree,
         "work_lease": work_lease,
     }
+    if adoption["adopt"]:
+        # #442: recovery lock over the issue's own already-pushed branch.
+        result["adoption"] = issue_lock_adoption.build_adoption_proof(
+            issue_number=issue_number,
+            branch_name=branch_name,
+            assessment=adoption,
+            open_pr_checked=True,
+            competing_lock_checked=True,
+            lock_file_path=ISSUE_LOCK_FILE,
+            lock_file_status="written",
+        )
+        result["message"] = (
+            f"Adopted existing branch '{branch_name}' and locked issue "
+            f"#{issue_number} for recovery (fail-closed check complete)."
+        )
     if agent_artifacts:
         result["warnings"] = [
             "Agent temp artifacts at repo root (delete before implementation): "

issue_lock_adoption.py

@@ -0,0 +1,142 @@
+"""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
+
+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,
+    }

tests/test_issue_lock_adoption.py

@@ -0,0 +1,124 @@
+"""Unit tests for own-branch lock adoption decision (#442)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from issue_lock_adoption import (  # noqa: E402
+    ADOPT,
+    BLOCK_COMPETING,
+    NO_MATCH,
+    assess_own_branch_adoption,
+    build_adoption_proof,
+)
+
+REQ = "feat/issue-420-server-code-parity"
+
+
+class TestAssessOwnBranchAdoption(unittest.TestCase):
+    def test_exact_own_branch_is_adopted(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
+        )
+        self.assertEqual(result["outcome"], ADOPT)
+        self.assertTrue(result["adopt"])
+        self.assertFalse(result["block"])
+        self.assertEqual(result["matched_branch"], REQ)
+        self.assertEqual(result["matched_head_sha"], "934688a")
+
+    def test_exact_own_branch_adopted_when_sha_missing(self):
+        result = assess_own_branch_adoption(
+            issue_number=420, requested_branch=REQ, existing_branches=[REQ]
+        )
+        self.assertEqual(result["outcome"], ADOPT)
+        self.assertIsNone(result["matched_head_sha"])
+
+    def test_different_branch_same_issue_blocks(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-420-other-work"}],
+        )
+        self.assertEqual(result["outcome"], BLOCK_COMPETING)
+        self.assertTrue(result["block"])
+        self.assertFalse(result["adopt"])
+        self.assertIn("feat/issue-420-other-work", result["competing_branches"])
+        self.assertIn("fail closed", result["reason"])
+
+    def test_own_branch_plus_competing_branch_blocks(self):
+        # Ambiguous ownership: fail closed even though the exact branch exists.
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": REQ}, {"name": "feat/issue-420-rogue"}],
+        )
+        self.assertEqual(result["outcome"], BLOCK_COMPETING)
+        self.assertTrue(result["block"])
+        self.assertEqual(result["competing_branches"], ["feat/issue-420-rogue"])
+
+    def test_no_matching_branch_is_normal_path(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-999-unrelated"}],
+        )
+        self.assertEqual(result["outcome"], NO_MATCH)
+        self.assertFalse(result["block"])
+        self.assertFalse(result["adopt"])
+
+    def test_empty_branch_list_is_normal_path(self):
+        result = assess_own_branch_adoption(
+            issue_number=420, requested_branch=REQ, existing_branches=[]
+        )
+        self.assertEqual(result["outcome"], NO_MATCH)
+
+    def test_substring_issue_number_not_falsely_matched_as_exact(self):
+        # A different issue's branch must not be adopted as ours.
+        result = assess_own_branch_adoption(
+            issue_number=42,
+            requested_branch="feat/issue-42-thing",
+            existing_branches=[{"name": "feat/issue-420-server-code-parity"}],
+        )
+        # 'issue-42' is a substring of 'issue-420', so marker matches but the
+        # name differs -> competing block, never adoption of the wrong branch.
+        self.assertEqual(result["outcome"], BLOCK_COMPETING)
+        self.assertNotEqual(result["matched_branch"], "feat/issue-420-server-code-parity")
+
+
+class TestBuildAdoptionProof(unittest.TestCase):
+    def test_proof_has_all_required_fields(self):
+        assessment = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
+        )
+        proof = build_adoption_proof(
+            issue_number=420,
+            branch_name=REQ,
+            assessment=assessment,
+            open_pr_checked=True,
+            competing_lock_checked=True,
+            lock_file_path="/tmp/gitea_issue_lock.json",
+            lock_file_status="written",
+        )
+        for key in (
+            "issue_number",
+            "branch_name",
+            "branch_head_commit",
+            "adoption_reason",
+            "no_existing_pr_proof",
+            "no_competing_live_lock_proof",
+            "lock_file_path",
+            "lock_file_status",
+        ):
+            self.assertIn(key, proof)
+        self.assertEqual(proof["branch_head_commit"], "934688a")
+        self.assertTrue(proof["no_existing_pr_proof"])
+        self.assertTrue(proof["no_competing_live_lock_proof"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_server.py

@@ -3127,6 +3127,105 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already has matching branch", str(ctx.exception))
 
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_adopts_own_existing_branch(self, _auth, mock_api, _git_state):
+        # #442: the issue's own already-pushed branch is adoptable for recovery.
+        sha = "934688a71dd19d9b46369e73fb3ac356ee0cc211"
+
+        def _api(url, *a, **k):
+            if "/pulls" in url:
+                return []
+            if "/branches" in url:
+                return [{"name": "feat/issue-196-mutations", "commit": {"id": sha}}]
+            return []
+
+        mock_api.side_effect = _api
+        res = gitea_lock_issue(
+            issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs"
+        )
+        self.assertTrue(res["success"])
+        self.assertIn("adoption", res)
+        proof = res["adoption"]
+        self.assertEqual(proof["issue_number"], 196)
+        self.assertEqual(proof["branch_name"], "feat/issue-196-mutations")
+        self.assertEqual(proof["branch_head_commit"], sha)
+        self.assertTrue(proof["no_existing_pr_proof"])
+        self.assertTrue(proof["no_competing_live_lock_proof"])
+        self.assertEqual(proof["lock_file_path"], ISSUE_LOCK_FILE)
+        self.assertIn("adopt", proof["adoption_reason"].lower())
+        self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
+
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_open_pr_blocks_adoption(self, _auth, mock_api, _git_state):
+        # Even with the exact own branch present, an open PR blocks adoption.
+        def _api(url, *a, **k):
+            if "/pulls" in url:
+                return [{
+                    "number": 200,
+                    "head": {"ref": "feat/issue-196-mutations"},
+                    "title": "WIP",
+                    "body": "",
+                }]
+            if "/branches" in url:
+                return [{"name": "feat/issue-196-mutations", "commit": {"id": "abc"}}]
+            return []
+
+        mock_api.side_effect = _api
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(
+                issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs"
+            )
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+        self.assertFalse(os.path.exists(ISSUE_LOCK_FILE))
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_proceeds_after_adopted_lock(
+        self, _auth, mock_api, mock_req, _git_state, _role
+    ):
+        # #442 recovery goal: after adopting the own branch, create_pr proceeds.
+        sha = "934688a71dd19d9b46369e73fb3ac356ee0cc211"
+
+        def _api(url, *a, **k):
+            if "/pulls" in url:
+                return []
+            if "/branches" in url:
+                return [{"name": "feat/issue-196-mutations", "commit": {"id": sha}}]
+            return []
+
+        mock_api.side_effect = _api
+        lock_res = gitea_lock_issue(
+            issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs"
+        )
+        self.assertIn("adoption", lock_res)
+
+        mock_req.return_value = {"number": 7, "html_url": "https://gitea.prgs.cc/pulls/7"}
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            pr = gitea_create_pr(
+                title="feat: parity gate Closes #196",
+                head="feat/issue-196-mutations",
+                base="master",
+                remote="prgs",
+            )
+        self.assertEqual(pr["number"], 7)
+
     def test_lock_issue_blocks_active_same_operation_lease(self):
         with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
             json.dump({