feat: non-destructive lock recovery for pushed branches (Closes #440)

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]>

docs/llm-workflow-runbooks.md

@@ -308,6 +308,28 @@ metadata. Final-report validation blocks handoffs that hide lock read/write/dele
 under `External-state mutations: none` or mix author PR creation with reviewer
 approval in one run. See also #438 (global lock redesign).
 
+### Non-destructive lock recovery after push (#440)
+
+When work is pushed but the in-memory MCP session lock is lost (server restart,
+crashed process, or stale session pointer):
+
+1. **Do not delete the remote branch.** Branch deletion is not a normal recovery
+   step and can destroy the only copy of unmerged work.
+2. **Re-run `gitea_lock_issue`** with the same `issue_number`, exact
+   `branch_name`, and active `branches/` worktree. Own-branch adoption
+   reacquires the lease when the remote branch is the caller's exact branch and
+   no open PR or competing same-issue branch exists.
+3. **Call `gitea_create_pr`** with the same `head` branch. The server resolves
+   the durable keyed lock file even when the session pointer was cleared at
+   restart.
+4. **Stop fail-closed** when another actor owns a competing branch, an open PR
+   already exists, or a live foreign lease blocks takeover — never adopt their
+   branch.
+
+Branch ownership uses structured evidence
+`(fix|feat|docs|chore)/issue-<n>-<desc>` — not broad substring matches on
+`issue-<n>`.
+
 Remote branches matching the issue number are also treated as active work unless
 the recovery review proves the branch is abandoned or superseded. Never delete
 or clean up a branch when it has an active lease, dirty worktree, open PR, or is

gitea_mcp_server.py

@@ -541,6 +541,7 @@ import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
 import issue_lock_store  # noqa: E402
 import issue_lock_adoption  # noqa: E402
+import issue_branch_ownership  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
@@ -1286,11 +1287,11 @@ def gitea_lock_issue(
         worktree_path: Author scratch-clone path to validate (defaults to
             GITEA_AUTHOR_WORKTREE or the MCP server project root).
     """
-    # 1. Enforce branch name includes issue number
-    expected_pattern = f"issue-{issue_number}"
-    if expected_pattern not in branch_name:
+    # 1. Enforce canonical issue branch ownership (#440)
+    if not issue_branch_ownership.branch_belongs_to_issue(branch_name, issue_number):
         raise ValueError(
-            f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
+            f"Branch name '{branch_name}' must match "
+            f"(fix|feat|docs|chore)/issue-{issue_number}-<desc> (fail closed)"
         )
 
     blocked = _profile_permission_block(

issue_branch_ownership.py

@@ -0,0 +1,28 @@
+"""Structured issue/branch ownership evidence (#440).
+
+Author branches must follow ``(fix|feat|docs|chore)/issue-<n>-<desc>``. Duplicate-work
+and lock-recovery gates use this parser instead of broad substring matching on
+``issue-<n>`` so unrelated branch names cannot false-positive.
+"""
+
+from __future__ import annotations
+
+import re
+
+ISSUE_BRANCH_RE = re.compile(
+    r"^(?P<prefix>fix|feat|docs|chore)/issue-(?P<num>\d+)(?:-|$)"
+)
+
+
+def parse_issue_branch(branch_name: str) -> int | None:
+    """Return the issue number encoded in a canonical author branch, else None."""
+    match = ISSUE_BRANCH_RE.match((branch_name or "").strip())
+    if not match:
+        return None
+    return int(match.group("num"))
+
+
+def branch_belongs_to_issue(branch_name: str, issue_number: int) -> bool:
+    """True when ``branch_name`` is a canonical branch for ``issue_number``."""
+    parsed = parse_issue_branch(branch_name)
+    return parsed is not None and parsed == issue_number

issue_lock_adoption.py

@@ -1,4 +1,4 @@
-"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#442 / #443).
+"""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.
@@ -14,6 +14,8 @@ 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"
@@ -40,13 +42,12 @@ def assess_own_branch_adoption(
     existing_branches,
 ) -> dict:
     """Decide whether an existing matching branch is adoptable."""
-    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:
+        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})

tests/test_issue_branch_ownership.py

@@ -0,0 +1,36 @@
+"""Structured issue branch ownership (#440)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_branch_ownership as ibo  # noqa: E402
+
+
+class TestIssueBranchOwnership(unittest.TestCase):
+    def test_canonical_branch_parses_issue_number(self):
+        self.assertEqual(
+            ibo.parse_issue_branch("feat/issue-420-server-code-parity"),
+            420,
+        )
+
+    def test_prefix_variants_supported(self):
+        for prefix in ("fix", "feat", "docs", "chore"):
+            branch = f"{prefix}/issue-440-lock-recovery"
+            self.assertTrue(ibo.branch_belongs_to_issue(branch, 440))
+
+    def test_substring_false_positive_rejected(self):
+        self.assertFalse(
+            ibo.branch_belongs_to_issue("feat/my-issue-420-backport", 420)
+        )
+        self.assertFalse(ibo.branch_belongs_to_issue("release/issue-420-hotfix", 420))
+
+    def test_different_issue_number_rejected(self):
+        self.assertFalse(
+            ibo.branch_belongs_to_issue("feat/issue-421-server-code-parity", 420)
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_lock_adoption.py

@@ -43,6 +43,15 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
         )
         self.assertEqual(result["outcome"], NO_MATCH)
 
+    def test_substring_only_branch_name_is_ignored(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/my-issue-420-backport"}],
+        )
+        self.assertEqual(result["outcome"], NO_MATCH)
+        self.assertFalse(result["block"])
+
 
 class TestBuildAdoptionProof(unittest.TestCase):
     def test_proof_has_required_fields(self):

tests/test_issue_lock_recovery.py

@@ -0,0 +1,157 @@
+"""End-to-end lock recovery scenarios for issue #440."""
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_adoption  # noqa: E402
+import issue_lock_store as ils  # noqa: E402
+import mcp_server  # noqa: E402
+from mcp_server import gitea_create_pr, gitea_lock_issue  # noqa: E402
+
+PRGS_REPO = mcp_server.REMOTES["prgs"]["repo"]
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+BRANCH = "feat/issue-420-server-code-parity"
+ISSUE_WRITE_ENV = {
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+}
+CREATE_PR_ENV = {
+    "GITEA_PROFILE_NAME": "author-test",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.create,gitea.branch.push",
+    "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.approve,gitea.pr.merge,gitea.pr.review",
+}
+
+
+def _clean_master_git_state_for_lock():
+    return {
+        "current_branch": "master",
+        "porcelain_status": "",
+        "base_equivalent": True,
+        "inspected_git_root": "/tmp/repo",
+        "base_branch": "master",
+    }
+
+
+def _lock_record(**overrides):
+    record = {
+        "issue_number": 420,
+        "branch_name": BRANCH,
+        "remote": "prgs",
+        "org": "Scaled-Tech-Consulting",
+        "repo": PRGS_REPO,
+        "worktree_path": os.path.realpath(os.getcwd()),
+        "work_lease": {
+            "operation_type": "author_issue_work",
+            "expires_at": "2999-01-01T00:00:00Z",
+        },
+    }
+    record.update(overrides)
+    return record
+
+
+class TestIssueLockRecovery(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmpdir.cleanup)
+        self.lock_dir = self._tmpdir.name
+
+    def test_substring_branch_does_not_trigger_adoption(self):
+        result = issue_lock_adoption.assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=BRANCH,
+            existing_branches=[{"name": "feat/my-issue-420-backport"}],
+        )
+        self.assertEqual(result["outcome"], issue_lock_adoption.NO_MATCH)
+
+    def test_competing_actor_branch_blocks_adoption(self):
+        result = issue_lock_adoption.assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=BRANCH,
+            existing_branches=[{"name": "feat/issue-420-other-work"}],
+        )
+        self.assertTrue(result["block"])
+
+    @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_recovery_after_restart_adopts_own_branch(self, _auth, mock_api, _git):
+        mock_api.side_effect = [
+            [],
+            [{"name": BRANCH, "commit": {"id": "934688a"}}],
+        ]
+        env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
+        with patch.dict(os.environ, env, clear=True):
+            with patch("os.getpid", return_value=9999):
+                res = gitea_lock_issue(
+                    issue_number=420,
+                    branch_name=BRANCH,
+                    remote="prgs",
+                )
+        self.assertTrue(res["success"])
+        self.assertIn("adoption", res)
+        found = ils.find_lock_for_branch(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo=PRGS_REPO,
+            branch_name=BRANCH,
+            lock_dir=self.lock_dir,
+        )
+        self.assertEqual(found["branch_name"], BRANCH)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_resolves_durable_lock_after_session_loss(self, _auth, _role, mock_api):
+        mock_api.return_value = {"number": 421, "html_url": "https://example/pr/421"}
+        worktree = os.path.realpath(os.getcwd())
+        path = ils.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo=PRGS_REPO,
+            issue_number=420,
+            lock_dir=self.lock_dir,
+        )
+        ils.save_lock_file(path, _lock_record(worktree_path=worktree))
+
+        env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
+        with patch.dict(os.environ, env, clear=True):
+            with patch("os.getpid", return_value=8888):
+                self.assertIsNone(ils.read_session_issue_lock(lock_dir=self.lock_dir))
+                res = gitea_create_pr(
+                    title=f"feat: server parity Closes #420",
+                    head=BRANCH,
+                    remote="prgs",
+                    worktree_path=worktree,
+                )
+        self.assertEqual(res["number"], 421)
+
+    @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_open_pr_blocks_recovery_lock(self, _auth, mock_api, _git):
+        mock_api.side_effect = [
+            [{"number": 99, "head": {"ref": BRANCH}, "title": "WIP", "body": ""}],
+            [],
+        ]
+        env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
+        with patch.dict(os.environ, env, clear=True):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_lock_issue(issue_number=420, branch_name=BRANCH, remote="prgs")
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_server.py

@@ -3142,7 +3142,7 @@ class TestIssueLocking(unittest.TestCase):
     def test_lock_issue_mismatch_branch_fails(self):
         with self.assertRaises(ValueError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
-        self.assertIn("must contain locked issue pattern", str(ctx.exception))
+        self.assertIn("must match", str(ctx.exception))
 
     @patch(
         "mcp_server.issue_lock_worktree.read_worktree_git_state",