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

@@ -291,9 +291,29 @@ recovery path.** That global slot is deprecated and can clobber unrelated live
 leases (#438). After an MCP restart, call `gitea_lock_issue` again — own-branch
 adoption rebinds the session when the issue's exact branch already exists (#442).
 `gitea_create_pr` resolves the durable keyed lock by session pointer or by
-matching `head` branch without unsafe manual seeding (#440). Branch ownership
+matching `head` branch without unsafe manual seeding.
-uses structured ``(fix|feat|docs|chore)/issue-<n>-`` parsing — not broad
+
-substring matches like ``issue-420`` inside ``issue-4200``.
+### 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

gitea_mcp_server.py

@@ -537,10 +537,10 @@ import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import agent_temp_artifacts
-import issue_branch_ownership  # noqa: E402
 import issue_lock_worktree  # 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
@@ -1172,11 +1172,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
+    # 1. Enforce canonical issue branch ownership (#440)
-    expected_pattern = f"issue-{issue_number}"
+    if not issue_branch_ownership.branch_belongs_to_issue(branch_name, issue_number):
-    if expected_pattern not in branch_name:
         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(
@@ -1227,7 +1227,7 @@ def gitea_lock_issue(
         pr_title = pr.get("title", "")
         pr_body = pr.get("body", "")
 
-        if issue_branch_ownership.branch_tracks_issue(pr_head, issue_number):
+        if issue_branch_ownership.branch_belongs_to_issue(pr_head, issue_number):
             raise ValueError(
                 f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
             )

issue_branch_ownership.py

@@ -1,31 +1,28 @@
 """Structured issue/branch ownership evidence (#440).
 
 Author branches must follow ``(fix|feat|docs|chore)/issue-<n>-<desc>``. Duplicate-work
-and recovery gates use this parser instead of broad substring checks like
+and lock-recovery gates use this parser instead of broad substring matching on
-``issue-420`` inside unrelated names (for example ``issue-4200``).
+``issue-<n>`` so unrelated branch names cannot false-positive.
 """
 
 from __future__ import annotations
 
 import re
 
-IMPLEMENTATION_BRANCH_RE = re.compile(
+ISSUE_BRANCH_RE = re.compile(
-    r"^(?:fix|feat|docs|chore)/issue-(\d+)(?:-.+)?$"
+    r"^(?P<prefix>fix|feat|docs|chore)/issue-(?P<num>\d+)(?:-|$)"
 )
 
 
-def parse_tracked_issue_number(branch_name: str) -> int | None:
+def parse_issue_branch(branch_name: str) -> int | None:
-    """Return the issue number encoded in a canonical author branch name."""
+    """Return the issue number encoded in a canonical author branch, else None."""
-    text = (branch_name or "").strip()
+    match = ISSUE_BRANCH_RE.match((branch_name or "").strip())
-    if not text:
-        return None
-    match = IMPLEMENTATION_BRANCH_RE.match(text)
     if not match:
         return None
-    return int(match.group(1))
+    return int(match.group("num"))
 
 
-def branch_tracks_issue(branch_name: str, issue_number: int) -> bool:
+def branch_belongs_to_issue(branch_name: str, issue_number: int) -> bool:
-    """True when ``branch_name`` structurally belongs to ``issue_number``."""
+    """True when ``branch_name`` is a canonical branch for ``issue_number``."""
-    parsed = parse_tracked_issue_number(branch_name)
+    parsed = parse_issue_branch(branch_name)
-    return parsed == issue_number if parsed is not None else False
+    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.
@@ -47,7 +47,7 @@ def assess_own_branch_adoption(
     matches: list[tuple[str, str | None]] = []
     for entry in existing_branches or []:
         name = _branch_name(entry).strip()
-        if issue_branch_ownership.branch_tracks_issue(name, issue_number):
+        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})

skills/llm-project-workflow/workflows/work-issue.md

@@ -309,15 +309,6 @@ Do not implement unclaimed work.
 
 If the claim/lock gates are broken, produce a recovery handoff.
 
-### Lost lock recovery after push (non-destructive)
-
-If the MCP server restarts after the issue branch was pushed but before PR creation:
-
-* **Do not** delete the remote branch as the normal recovery path.
-* Call `gitea_lock_issue` again with the same issue number and exact branch name. Own-branch adoption rebinds the session when open-PR, competing-lock, and worktree safety checks pass.
-* Call `gitea_create_pr` afterward. Durable keyed locks resolve from the session pointer or by matching the PR `head` branch without manual lock seeding.
-* If adoption is blocked by an open PR, a competing live lease, or a different same-issue branch, stop and produce a recovery handoff.
-
 Create a tooling issue only if this run is explicitly authorized to switch to issue-creation mode and exact `create_issue` capability is proven.
 
 Report:

tests/test_issue_branch_ownership.py

@@ -1,34 +1,35 @@
-"""Unit tests for structured issue/branch ownership parsing (#440)."""
+"""Structured issue branch ownership (#440)."""
 import sys
 import unittest
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
-from issue_branch_ownership import (  # noqa: E402
+import issue_branch_ownership as ibo  # noqa: E402
-    branch_tracks_issue,
-    parse_tracked_issue_number,
-)
 
 
 class TestIssueBranchOwnership(unittest.TestCase):
-    def test_parses_canonical_author_branch(self):
+    def test_canonical_branch_parses_issue_number(self):
         self.assertEqual(
-            parse_tracked_issue_number("feat/issue-420-server-code-parity"),
+            ibo.parse_issue_branch("feat/issue-420-server-code-parity"),
             420,
         )
 
-    def test_issue_4200_does_not_track_issue_420(self):
+    def test_prefix_variants_supported(self):
-        self.assertEqual(parse_tracked_issue_number("feat/issue-4200-unrelated"), 4200)
+        for prefix in ("fix", "feat", "docs", "chore"):
-        self.assertFalse(branch_tracks_issue("feat/issue-4200-unrelated", 420))
+            branch = f"{prefix}/issue-440-lock-recovery"
+            self.assertTrue(ibo.branch_belongs_to_issue(branch, 440))
 
-    def test_branch_tracks_issue_exact_match(self):
+    def test_substring_false_positive_rejected(self):
-        self.assertTrue(
+        self.assertFalse(
-            branch_tracks_issue("fix/issue-440-branch-recovery", 440)
+            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_unrelated_branch_does_not_track(self):
+    def test_different_issue_number_rejected(self):
-        self.assertFalse(branch_tracks_issue("feat/issue-999-other", 440))
+        self.assertFalse(
+            ibo.branch_belongs_to_issue("feat/issue-421-server-code-parity", 420)
+        )
 
 
 if __name__ == "__main__":

tests/test_issue_lock_adoption.py

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

tests/test_issue_lock_recovery.py

@@ -1,10 +1,9 @@
-"""Integration tests for non-destructive lock/PR recovery (#440)."""
+"""End-to-end lock recovery scenarios for issue #440."""
 import os
 import sys
 import tempfile
 import unittest
 from pathlib import Path
-from unittest import mock
 from unittest.mock import patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
@@ -13,10 +12,31 @@ 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
-from tests.test_mcp_server import CREATE_PR_ENV, ISSUE_WRITE_ENV  # noqa: E402
 
-FAKE_AUTH = "token fake"
+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):
@@ -25,8 +45,8 @@ def _lock_record(**overrides):
         "branch_name": BRANCH,
         "remote": "prgs",
         "org": "Scaled-Tech-Consulting",
-        "repo": mcp_server.REMOTES["prgs"]["repo"],
+        "repo": PRGS_REPO,
-        "worktree_path": "/tmp/wt-420",
+        "worktree_path": os.path.realpath(os.getcwd()),
         "work_lease": {
             "operation_type": "author_issue_work",
             "expires_at": "2999-01-01T00:00:00Z",
@@ -36,116 +56,101 @@ def _lock_record(**overrides):
     return record
 
 
-def _clean_git_state():
-    return {
-        "current_branch": BRANCH,
-        "porcelain_status": "",
-        "base_equivalent": True,
-        "inspected_git_root": os.getcwd(),
-        "base_branch": "master",
-    }
-
-
 class TestIssueLockRecovery(unittest.TestCase):
     def setUp(self):
         self._tmpdir = tempfile.TemporaryDirectory()
-        self._lock_dir = self._tmpdir.name
+        self.addCleanup(self._tmpdir.cleanup)
-        self._env = {
+        self.lock_dir = self._tmpdir.name
-            **ISSUE_WRITE_ENV,
-            "GITEA_ISSUE_LOCK_DIR": self._lock_dir,
-        }
-        self._create_pr_env = {
-            **CREATE_PR_ENV,
-            "GITEA_ISSUE_LOCK_DIR": self._lock_dir,
-        }
 
-    def tearDown(self):
+    def test_substring_branch_does_not_trigger_adoption(self):
-        self._tmpdir.cleanup()
+        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_adoption_after_restart_without_session_pointer(self):
+    def test_competing_actor_branch_blocks_adoption(self):
-        with patch.dict(os.environ, self._env, clear=True):
+        result = issue_lock_adoption.assess_own_branch_adoption(
-            with mock.patch("os.getpid", return_value=9999):
+            issue_number=420,
-                self.assertIsNone(ils.read_session_issue_lock())
+            requested_branch=BRANCH,
+            existing_branches=[{"name": "feat/issue-420-other-work"}],
+        )
+        self.assertTrue(result["block"])
 
-            with mock.patch(
+    @patch(
-                "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
-                return_value=_clean_git_state(),
+        return_value=_clean_master_git_state_for_lock(),
-            ), mock.patch("mcp_server.api_get_all") as mock_api, mock.patch(
+    )
-                "mcp_server.get_auth_header", return_value=FAKE_AUTH
+    @patch("mcp_server.api_get_all")
-            ):
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-                mock_api.side_effect = [
+    def test_lock_recovery_after_restart_adopts_own_branch(self, _auth, mock_api, _git):
-                    [],
+        mock_api.side_effect = [
-                    [{"name": BRANCH, "commit": {"id": "934688a"}}],
+            [],
-                ]
+            [{"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",
-                    worktree_path=os.path.realpath(os.getcwd()),
                 )
         self.assertTrue(res["success"])
         self.assertIn("adoption", res)
-        self.assertEqual(res["adoption"]["branch_head_commit"], "934688a")
+        found = ils.find_lock_for_branch(
-
+            remote="prgs",
-    @mock.patch(
+            org="Scaled-Tech-Consulting",
-        "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+            repo=PRGS_REPO,
-        return_value=(True, []),
+            branch_name=BRANCH,
-    )
+            lock_dir=self.lock_dir,
-    @mock.patch("mcp_server.api_request")
-    @mock.patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_resolves_keyed_lock_after_restart(self, _auth, mock_api, _role):
-        mock_api.return_value = {"number": 501, "html_url": "https://example/pr/501"}
-        worktree = os.path.realpath(os.getcwd())
-        with patch.dict(os.environ, self._create_pr_env, clear=True):
-            path = ils.lock_file_path(
-                remote="prgs",
-                org=mcp_server.REMOTES["prgs"]["org"],
-                repo=mcp_server.REMOTES["prgs"]["repo"],
-                issue_number=420,
-                lock_dir=self._lock_dir,
-            )
-            ils.save_lock_file(path, _lock_record(worktree_path=worktree))
-
-            with mock.patch("os.getpid", return_value=4242):
-                self.assertIsNone(ils.read_session_issue_lock())
-
-            res = gitea_create_pr(
-                title="feat: recovery Closes #420",
-                head=BRANCH,
-                base="master",
-                remote="prgs",
-                worktree_path=worktree,
-            )
-        self.assertEqual(res["number"], 501)
-
-    def test_open_pr_blocks_adoption(self):
-        with patch.dict(os.environ, self._env, clear=True):
-            with mock.patch(
-                "mcp_server.issue_lock_worktree.read_worktree_git_state",
-                return_value=_clean_git_state(),
-            ), mock.patch("mcp_server.api_get_all") as mock_api, mock.patch(
-                "mcp_server.get_auth_header", return_value=FAKE_AUTH
-            ):
-                mock_api.side_effect = [
-                    [{"number": 99, "head": {"ref": BRANCH}, "title": "", "body": ""}],
-                    [{"name": BRANCH, "commit": {"id": "934688a"}}],
-                ]
-                with self.assertRaises(ValueError) as ctx:
-                    gitea_lock_issue(
-                        issue_number=420,
-                        branch_name=BRANCH,
-                        remote="prgs",
-                        worktree_path=os.path.realpath(os.getcwd()),
-                    )
-        self.assertIn("already tied to an open PR", str(ctx.exception))
-
-    def test_structured_ownership_ignores_issue_4200_collision(self):
-        result = issue_lock_adoption.assess_own_branch_adoption(
-            issue_number=420,
-            requested_branch=BRANCH,
-            existing_branches=[{"name": "feat/issue-4200-unrelated"}],
         )
-        self.assertEqual(result["outcome"], issue_lock_adoption.NO_MATCH)
+        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__":

tests/test_mcp_server.py

@@ -3115,7 +3115,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",