Merge branch 'master' into fix/issue-842-conflict-fix-lease-lifecycle

gitea_mcp_server.py

@@ -11242,22 +11242,25 @@ def gitea_reconcile_merged_cleanups(
     if dry_run:
         report["dry_run"] = True
         report["executed"] = False
-        # #851: surface planned lifecycle order so dry-run matches execute.
-        report["planned_execution_orders"] = {
-            str(entry.get("pr_number")): entry.get("planned_execution_order") or []
-            for entry in (report.get("entries") or [])
-        }
         return {"success": True, "performed": False, **report}
 
     verify_preflight_purity(
         remote, task="reconcile_merged_cleanups", org=org, repo=repo
     )
     actions: list[dict] = []
-    project_root = _canonical_local_git_root()
+    for entry in report.get("entries") or []:
+        head_branch = entry.get("head_branch") or ""
+        remote_assessment = entry.get("remote_branch") or {}
+        local_assessment = entry.get("local_worktree") or {}
 
-    def _ownership_records_for_branch(
-        head_branch: str, pr_num_int: int | None
-    ) -> list[dict]:
+        if remote_assessment.get("safe_to_delete_remote"):
+            import urllib.parse
+
+            pr_num = entry.get("pr_number")
+            try:
+                pr_num_int = int(pr_num) if pr_num is not None else None
+            except (TypeError, ValueError):
+                pr_num_int = None
             ownership_bundle = _collect_branch_ownership_records(
                 remote=remote,
                 host=h,
@@ -11265,7 +11268,7 @@ def gitea_reconcile_merged_cleanups(
                 repo=r,
                 branch=head_branch,
                 pr_number=pr_num_int,
-            project_root=project_root,
+                project_root=_canonical_local_git_root(),
                 auth=auth,
                 base_api=base,
             )
@@ -11286,18 +11289,6 @@ def gitea_reconcile_merged_cleanups(
                         "role": "inventory",
                     }
                 )
-        return ownership_records
-
-    def _attempt_owned_remote_delete(
-        *,
-        head_branch: str,
-        pr_num_int: int | None,
-        after_worktree_removal: bool = False,
-    ) -> dict:
-        """Fail-closed remote delete with live ownership reassessment (#851)."""
-        import urllib.parse
-
-        ownership_records = _ownership_records_for_branch(head_branch, pr_num_int)
             ownership = branch_cleanup_guard.assess_active_branch_ownership(
                 remote=remote,
                 org=o,
@@ -11307,7 +11298,8 @@ def gitea_reconcile_merged_cleanups(
                 records=ownership_records,
             )
             if ownership.get("block"):
-            return {
+                actions.append(
+                    {
                         "action": "delete_remote_branch",
                         "branch": head_branch,
                         "success": False,
@@ -11316,10 +11308,13 @@ def gitea_reconcile_merged_cleanups(
                         "verified_absent": False,
                         "blocker_kind": "active_branch_ownership",
                         "reasons": ownership.get("reasons") or [],
-                "blocking_categories": ownership.get("blocking_categories") or [],
-                "after_worktree_removal": after_worktree_removal,
-                "ownership_reassessed": after_worktree_removal,
+                        "blocking_categories": ownership.get(
+                            "blocking_categories"
+                        )
+                        or [],
                     }
+                )
+                continue
 
             encoded = urllib.parse.quote(head_branch, safe="")
             url = f"{base}/branches/{encoded}"
@@ -11334,7 +11329,6 @@ def gitea_reconcile_merged_cleanups(
                     "branch": head_branch,
                     "source": "reconcile_merged_cleanups",
                     "ownership_checked": True,
-                "after_worktree_removal": after_worktree_removal,
                 },
             ):
                 api_request("DELETE", url, auth)
@@ -11343,7 +11337,8 @@ def gitea_reconcile_merged_cleanups(
                 readback
             )
             verified = bool(readback_assessment.get("verified_absent"))
-        return {
+            actions.append(
+                {
                     "action": "delete_remote_branch",
                     "branch": head_branch,
                     "success": bool(readback_assessment.get("ok")),
@@ -11352,53 +11347,22 @@ def gitea_reconcile_merged_cleanups(
                     "verified_absent": verified,
                     "readback": readback_assessment.get("readback"),
                     "reasons": readback_assessment.get("reasons") or [],
-            "after_worktree_removal": after_worktree_removal,
-            "ownership_reassessed": after_worktree_removal,
                 }
+            )
 
-    for entry in report.get("entries") or []:
-        head_branch = entry.get("head_branch") or ""
-        remote_assessment = entry.get("remote_branch") or {}
-        local_assessment = entry.get("local_worktree") or {}
-        pr_num = entry.get("pr_number")
-        try:
-            pr_num_int = int(pr_num) if pr_num is not None else None
-        except (TypeError, ValueError):
-            pr_num_int = None
-
-        # #851 lifecycle: when the target worktree is independently safe, remove
-        # it first so worktree_binding ownership does not permanently strand
-        # both the worktree and the remote branch. Never skip worktree removal
-        # merely because remote delete would be blocked by that binding.
-        # Ownership protection for remote delete remains fail-closed below.
-        worktree_removed = False
         if local_assessment.get("safe_to_remove_worktree"):
             result = merged_cleanup_reconcile.remove_local_worktree(
-                project_root,
+                _canonical_local_git_root(),
                 head_branch,
                 worktree_path=local_assessment.get("worktree_path"),
             )
             actions.append({"action": "remove_local_worktree", **result})
-            # Idempotent resume: absent worktree is already gone.
-            msg = (result.get("message") or "").lower()
-            worktree_removed = bool(result.get("success")) or (
-                "not found" in msg
-            )
-
-        if remote_assessment.get("safe_to_delete_remote"):
-            actions.append(
-                _attempt_owned_remote_delete(
-                    head_branch=head_branch,
-                    pr_num_int=pr_num_int,
-                    after_worktree_removal=worktree_removed,
-                )
-            )
 
     for scratch in report.get("reviewer_scratch_entries") or []:
         if not scratch.get("safe_to_remove_worktree"):
             continue
         result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
-            project_root, scratch.get("worktree_path") or ""
+            _canonical_local_git_root(), scratch.get("worktree_path") or ""
         )
         actions.append({"action": "remove_reviewer_scratch_worktree", **result})
 

merged_cleanup_reconcile.py

@@ -566,10 +566,6 @@ def build_pr_cleanup_entry(
         worktree_state=worktree_state,
         active_lock=active_lock,
     )
-    planned = plan_cleanup_execution_order(
-        remote_assessment=remote,
-        local_assessment=local,
-    )
     return {
         "pr_number": pr_number,
         "issue_number": issue_number,
@@ -580,63 +576,9 @@ def build_pr_cleanup_entry(
         "merged": merged,
         "remote_branch": remote,
         "local_worktree": local,
-        # #851: dry-run and execute share the same lifecycle order description.
-        "planned_execution_order": planned,
     }
 
 
-def plan_cleanup_execution_order(
-    *,
-    remote_assessment: dict[str, Any] | None,
-    local_assessment: dict[str, Any] | None,
-) -> list[dict[str, Any]]:
-    """Describe independent worktree-then-reassess-then-remote cleanup order (#851).
-
-    Remote ownership protection remains fail-closed at execute time. A worktree
-    that is independently safe to remove is never skipped merely because remote
-    deletion may be blocked by that same ``worktree_binding``.
-    """
-    remote = remote_assessment or {}
-    local = local_assessment or {}
-    steps: list[dict[str, Any]] = []
-    worktree_safe = bool(local.get("safe_to_remove_worktree"))
-    remote_safe = bool(remote.get("safe_to_delete_remote"))
-
-    if worktree_safe:
-        steps.append(
-            {
-                "action": "remove_local_worktree",
-                "reason": "independently_safe_to_remove",
-                "phase": 1,
-            }
-        )
-    if remote_safe:
-        if worktree_safe:
-            steps.append(
-                {
-                    "action": "reassess_branch_ownership",
-                    "reason": "after_worktree_removal_clear_worktree_binding",
-                    "phase": 2,
-                }
-            )
-            steps.append(
-                {
-                    "action": "delete_remote_branch",
-                    "reason": "only_if_independently_safe_after_reassessment",
-                    "phase": 3,
-                }
-            )
-        else:
-            steps.append(
-                {
-                    "action": "delete_remote_branch",
-                    "reason": "safe_to_delete_and_no_independent_worktree_removal",
-                    "phase": 1,
-                }
-            )
-    return steps
-
-
 def build_reconciliation_report(
     *,
     project_root: str,

pr_work_lease.py

@@ -228,25 +228,74 @@ def find_active_reviewer_lease(
     return None
 
 
+def _conflict_fix_chain_key(lease: dict) -> tuple | None:
+    """Identity of the lease chain a conflict-fix marker belongs to (#842).
+
+    Keyed by PR number, profile, head_before, and branch. Returns None when any
+    required component (pr_number, profile, head_before) is missing or malformed.
+    """
+    raw = lease.get("raw_fields") or {}
+    pr_number = lease.get("pr_number")
+    profile = (lease.get("profile") or "").strip().lower()
+    head_before = lease.get("head_before")
+    branch = (lease.get("branch") or raw.get("branch") or "").strip()
+    if not (pr_number and profile and head_before):
+        return None
+    return (pr_number, profile, head_before, branch)
+
+
+def _conflict_fix_chain_matches(key1: tuple, key2: tuple) -> bool:
+    """True when two conflict-fix chain keys refer to the same lease chain."""
+    pr1, profile1, head1, branch1 = key1
+    pr2, profile2, head2, branch2 = key2
+    if pr1 != pr2 or profile1 != profile2 or head1 != head2:
+        return False
+    if branch1 and branch2 and branch1 != branch2:
+        return False
+    return True
+
+
+def _conflict_fix_chain_terminated_after(entries: list[dict], index: int) -> bool:
+    """True when a later marker terminates the conflict-fix chain of ``entries[index]``.
+
+    Append-only newest-wins: a terminal marker (phase=released/blocked/done)
+    ends only its matching claim chain (#842).
+    """
+    key = _conflict_fix_chain_key(entries[index])
+    if key is None:
+        return False
+    for later in entries[index + 1:]:
+        phase = (later.get("phase") or "").strip().lower()
+        if phase not in _TERMINAL_CONFLICT_FIX_PHASES:
+            continue
+        later_key = _conflict_fix_chain_key(later)
+        if later_key and _conflict_fix_chain_matches(key, later_key):
+            return True
+    return False
+
+
 def find_active_conflict_fix_lease(
     comments: list[dict],
     *,
     pr_number: int,
     now: datetime | None = None,
 ) -> dict[str, Any] | None:
-    """Return the newest unexpired conflict-fix lease for *pr_number*, if any."""
+    """Return the newest unexpired, non-terminated conflict-fix lease for *pr_number*, if any."""
     now = now or datetime.now(timezone.utc)
     candidates = [
         entry for entry in _comment_entries(comments, pr_number=pr_number)
         if entry.get("lease_kind") == "conflict_fix"
     ]
-    for lease in reversed(candidates):
+    for index in range(len(candidates) - 1, -1, -1):
+        lease = candidates[index]
         if _lease_expired(lease, now=now):
             continue
         phase = (lease.get("phase") or "").strip().lower()
         if phase in _TERMINAL_CONFLICT_FIX_PHASES:
             continue
         if phase in _ACTIVE_CONFLICT_FIX_PHASES or phase:
+            if _conflict_fix_chain_terminated_after(candidates, index):
+                continue
             return lease
     return None
 

tests/test_branch_cleanup_guard.py

@@ -1266,378 +1266,6 @@ class TestSecondRemediationIntegration(unittest.TestCase):
         self.assertIn("delete_acknowledged", delete_actions[0])
         self.assertTrue(delete_actions[0].get("verified_absent"))
 
-    def test_issue_851_worktree_removed_when_remote_blocked_only_by_worktree_binding(self):
-        """#851: remote blocked by worktree_binding must not skip safe worktree removal.
-
-        Lifecycle: remove clean owned worktree → reassess ownership → delete
-        remote only if independently safe. Unrelated entries stay untouched.
-        """
-        from mcp_server import gitea_reconcile_merged_cleanups
-
-        target_branch = "fix/issue-844-exclude-epic-containers"
-        foreign_branch = "fix/issue-999-unrelated-active"
-        worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers"
-        ownership_calls = []
-        remove_calls = []
-        delete_api_calls = []
-
-        def fake_collect(**kwargs):
-            ownership_calls.append(dict(kwargs))
-            # Ownership is reassessed *after* independent worktree removal (#851).
-            # Target worktree is already gone → no worktree_binding remains.
-            # Foreign branch keeps an active author lease → remote delete blocked.
-            if kwargs.get("branch") == foreign_branch:
-                # Match session-bound org/repo + host used by the tool resolve path.
-                return {
-                    "records": [
-                        {
-                            "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
-                            "status": "active",
-                            "remote": kwargs.get("remote") or "prgs",
-                            "host": kwargs.get("host") or "gitea.example.com",
-                            "org": kwargs.get("org") or "Scaled-Tech-Consulting",
-                            "repo": kwargs.get("repo") or "Gitea-Tools",
-                            "branch": foreign_branch,
-                            "reclaim_allowed": False,
-                        }
-                    ],
-                    "inventory_error": False,
-                }
-            return {"records": [], "inventory_error": False}
-
-        def fake_remove(project_root, branch, worktree_path=None):
-            remove_calls.append(
-                {"branch": branch, "worktree_path": worktree_path}
-            )
-            return {
-                "success": True,
-                "performed": True,
-                "message": f"removed worktree {worktree_path}",
-                "worktree_path": worktree_path,
-            }
-
-        def fake_probe(h, o, r, auth, br):
-            return guard.classify_branch_readback_http_status(
-                404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
-            )
-
-        def fake_api(method, url, auth, **kwargs):
-            if method == "DELETE":
-                delete_api_calls.append(url)
-            return {}
-
-        report = {
-            "entries": [
-                {
-                    "pr_number": 848,
-                    "head_branch": target_branch,
-                    "remote_branch": {"safe_to_delete_remote": True},
-                    "local_worktree": {
-                        "safe_to_remove_worktree": True,
-                        "worktree_path": worktree_path,
-                    },
-                },
-                {
-                    "pr_number": 999,
-                    "head_branch": foreign_branch,
-                    "remote_branch": {"safe_to_delete_remote": True},
-                    "local_worktree": {
-                        "safe_to_remove_worktree": False,
-                        "worktree_path": None,
-                    },
-                },
-            ],
-            "reviewer_scratch_entries": [],
-        }
-        patch(
-            "mcp_server.get_profile",
-            return_value={
-                "profile_name": "prgs-reconciler",
-                "role": "reconciler",
-                "allowed_operations": [
-                    "gitea.read",
-                    "gitea.branch.delete",
-                    "gitea.pr.close",
-                ],
-                "forbidden_operations": [],
-            },
-        ).start()
-        patch("mcp_server.api_get_all", return_value=[]).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
-            return_value=report,
-        ).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
-            return_value=[],
-        ).start()
-        patch(
-            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
-            return_value=(True, []),
-        ).start()
-        patch("mcp_server.verify_preflight_purity", return_value=None).start()
-        patch(
-            "mcp_server._collect_branch_ownership_records",
-            side_effect=fake_collect,
-        ).start()
-        patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
-            side_effect=fake_remove,
-        ).start()
-        self.mock_api.side_effect = fake_api
-
-        res = gitea_reconcile_merged_cleanups(
-            dry_run=False,
-            execute_confirmed=True,
-            remote="prgs",
-        )
-        self.assertTrue(res.get("performed") or res.get("executed"))
-        actions = res.get("actions") or []
-
-        remove_actions = [
-            a for a in actions if a.get("action") == "remove_local_worktree"
-        ]
-        self.assertEqual(len(remove_actions), 1, actions)
-        self.assertTrue(remove_actions[0].get("success"))
-        self.assertEqual(remove_calls[0]["branch"], target_branch)
-        self.assertEqual(remove_calls[0]["worktree_path"], worktree_path)
-
-        # Target remote delete succeeds after worktree removal + reassessment.
-        target_deletes = [
-            a
-            for a in actions
-            if a.get("action") == "delete_remote_branch"
-            and a.get("branch") == target_branch
-        ]
-        self.assertEqual(len(target_deletes), 1, actions)
-        self.assertTrue(target_deletes[0].get("success"))
-        self.assertTrue(target_deletes[0].get("after_worktree_removal"))
-        self.assertTrue(target_deletes[0].get("ownership_reassessed"))
-        self.assertTrue(target_deletes[0].get("verified_absent"))
-
-        # Foreign branch remains protected (author lease) and is not deleted.
-        foreign_deletes = [
-            a
-            for a in actions
-            if a.get("action") == "delete_remote_branch"
-            and a.get("branch") == foreign_branch
-        ]
-        self.assertEqual(len(foreign_deletes), 1, actions)
-        self.assertFalse(foreign_deletes[0].get("success"))
-        self.assertEqual(
-            foreign_deletes[0].get("blocker_kind"), "active_branch_ownership"
-        )
-        self.assertIn(
-            guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
-            foreign_deletes[0].get("blocking_categories") or [],
-        )
-        # Only the target branch should hit the DELETE API.
-        self.assertEqual(len(delete_api_calls), 1)
-
-        # Ownership collected for target (post-removal) and foreign; worktree
-        # removal happened before target remote delete in the action log.
-        target_idx = next(
-            i
-            for i, a in enumerate(actions)
-            if a.get("action") == "remove_local_worktree"
-        )
-        delete_idx = next(
-            i
-            for i, a in enumerate(actions)
-            if a.get("action") == "delete_remote_branch"
-            and a.get("branch") == target_branch
-            and a.get("success")
-        )
-        self.assertLess(target_idx, delete_idx)
-
-    def test_issue_851_dirty_worktree_not_removed_and_remote_stays_protected(self):
-        """#851: dirty/foreign worktrees remain protected; no unsafe cleanup."""
-        from mcp_server import gitea_reconcile_merged_cleanups
-
-        branch = "fix/issue-851-dirty"
-        remove_calls = []
-
-        def fake_collect(**kwargs):
-            return {
-                "records": [
-                    {
-                        "category": guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
-                        "status": "active",
-                        "remote": kwargs.get("remote") or "prgs",
-                        "host": kwargs.get("host") or "gitea.example.com",
-                        "org": kwargs.get("org") or "Scaled-Tech-Consulting",
-                        "repo": kwargs.get("repo") or "Gitea-Tools",
-                        "branch": branch,
-                        "reclaim_allowed": False,
-                    }
-                ],
-                "inventory_error": False,
-            }
-
-        report = {
-            "entries": [
-                {
-                    "pr_number": 851,
-                    "head_branch": branch,
-                    "remote_branch": {"safe_to_delete_remote": True},
-                    "local_worktree": {
-                        "safe_to_remove_worktree": False,
-                        "worktree_path": "/tmp/dirty-wt",
-                    },
-                }
-            ],
-            "reviewer_scratch_entries": [],
-        }
-        patch(
-            "mcp_server.get_profile",
-            return_value={
-                "profile_name": "prgs-reconciler",
-                "role": "reconciler",
-                "allowed_operations": [
-                    "gitea.read",
-                    "gitea.branch.delete",
-                ],
-                "forbidden_operations": [],
-            },
-        ).start()
-        patch("mcp_server.api_get_all", return_value=[]).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
-            return_value=report,
-        ).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
-            return_value=[],
-        ).start()
-        patch(
-            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
-            return_value=(True, []),
-        ).start()
-        patch("mcp_server.verify_preflight_purity", return_value=None).start()
-        patch(
-            "mcp_server._collect_branch_ownership_records",
-            side_effect=fake_collect,
-        ).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
-            side_effect=lambda *a, **k: remove_calls.append(k) or {
-                "success": True,
-                "performed": True,
-            },
-        ).start()
-        self.mock_api.side_effect = lambda *a, **k: {}
-
-        res = gitea_reconcile_merged_cleanups(
-            dry_run=False,
-            execute_confirmed=True,
-            remote="prgs",
-        )
-        actions = res.get("actions") or []
-        self.assertEqual(remove_calls, [])
-        self.assertFalse(
-            any(a.get("action") == "remove_local_worktree" for a in actions)
-        )
-        deletes = [
-            a for a in actions if a.get("action") == "delete_remote_branch"
-        ]
-        self.assertEqual(len(deletes), 1)
-        self.assertFalse(deletes[0].get("success"))
-        self.assertEqual(deletes[0].get("blocker_kind"), "active_branch_ownership")
-        self.assertIn(
-            guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
-            deletes[0].get("blocking_categories") or [],
-        )
-
-    def test_issue_851_idempotent_resume_when_worktree_already_absent(self):
-        """#851: partial failures remain resumable and idempotent."""
-        from mcp_server import gitea_reconcile_merged_cleanups
-
-        branch = "fix/issue-851-resume"
-        ownership_calls = []
-
-        def fake_collect(**kwargs):
-            ownership_calls.append(kwargs)
-            return {"records": [], "inventory_error": False}
-
-        def fake_remove(project_root, branch, worktree_path=None):
-            return {
-                "success": False,
-                "performed": False,
-                "message": f"worktree not found: {worktree_path}",
-            }
-
-        def fake_probe(h, o, r, auth, br):
-            return guard.classify_branch_readback_http_status(
-                404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
-            )
-
-        report = {
-            "entries": [
-                {
-                    "pr_number": 851,
-                    "head_branch": branch,
-                    "remote_branch": {"safe_to_delete_remote": True},
-                    "local_worktree": {
-                        "safe_to_remove_worktree": True,
-                        "worktree_path": "/tmp/already-gone",
-                    },
-                }
-            ],
-            "reviewer_scratch_entries": [],
-        }
-        patch(
-            "mcp_server.get_profile",
-            return_value={
-                "profile_name": "prgs-reconciler",
-                "role": "reconciler",
-                "allowed_operations": [
-                    "gitea.read",
-                    "gitea.branch.delete",
-                ],
-                "forbidden_operations": [],
-            },
-        ).start()
-        patch("mcp_server.api_get_all", return_value=[]).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
-            return_value=report,
-        ).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
-            return_value=[],
-        ).start()
-        patch(
-            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
-            return_value=(True, []),
-        ).start()
-        patch("mcp_server.verify_preflight_purity", return_value=None).start()
-        patch(
-            "mcp_server._collect_branch_ownership_records",
-            side_effect=fake_collect,
-        ).start()
-        patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
-            side_effect=fake_remove,
-        ).start()
-        self.mock_api.side_effect = lambda *a, **k: {}
-
-        res = gitea_reconcile_merged_cleanups(
-            dry_run=False,
-            execute_confirmed=True,
-            remote="prgs",
-        )
-        actions = res.get("actions") or []
-        removes = [a for a in actions if a.get("action") == "remove_local_worktree"]
-        deletes = [a for a in actions if a.get("action") == "delete_remote_branch"]
-        self.assertEqual(len(removes), 1)
-        self.assertFalse(removes[0].get("success"))
-        self.assertEqual(len(deletes), 1)
-        self.assertTrue(deletes[0].get("success"))
-        self.assertTrue(deletes[0].get("after_worktree_removal"))
-        self.assertTrue(ownership_calls)
-
 
 
 if __name__ == "__main__":

tests/test_merged_cleanup_reconcile.py

@@ -12,59 +12,6 @@ import merged_cleanup_reconcile as mcr  # noqa: E402
 
 
 class TestMergedCleanupAssessment(unittest.TestCase):
-    def test_issue_851_plan_order_worktree_then_reassess_then_remote(self):
-        """#851 dry-run plan: remove worktree, reassess ownership, then remote."""
-        plan = mcr.plan_cleanup_execution_order(
-            remote_assessment={"safe_to_delete_remote": True},
-            local_assessment={"safe_to_remove_worktree": True},
-        )
-        actions = [s["action"] for s in plan]
-        self.assertEqual(
-            actions,
-            [
-                "remove_local_worktree",
-                "reassess_branch_ownership",
-                "delete_remote_branch",
-            ],
-        )
-        self.assertEqual(plan[0]["phase"], 1)
-        self.assertEqual(plan[-1]["phase"], 3)
-        self.assertIn("independently_safe", plan[0]["reason"])
-        self.assertIn("reassessment", plan[-1]["reason"])
-
-    def test_issue_851_plan_remote_only_when_worktree_not_safe(self):
-        plan = mcr.plan_cleanup_execution_order(
-            remote_assessment={"safe_to_delete_remote": True},
-            local_assessment={"safe_to_remove_worktree": False},
-        )
-        self.assertEqual([s["action"] for s in plan], ["delete_remote_branch"])
-        self.assertNotIn("reassess_branch_ownership", [s["action"] for s in plan])
-
-    def test_issue_851_plan_worktree_only_when_remote_not_safe(self):
-        plan = mcr.plan_cleanup_execution_order(
-            remote_assessment={"safe_to_delete_remote": False},
-            local_assessment={"safe_to_remove_worktree": True},
-        )
-        self.assertEqual([s["action"] for s in plan], ["remove_local_worktree"])
-
-    def test_issue_851_entry_includes_planned_execution_order(self):
-        entry = mcr.build_pr_cleanup_entry(
-            pr={
-                "number": 848,
-                "title": "Closes #844",
-                "body": "",
-                "merged_at": "2026-07-23T00:00:00Z",
-                "head": {"ref": "fix/issue-844-x", "sha": "a" * 40},
-            },
-            project_root="/tmp/not-a-real-root",
-            open_pr_heads=set(),
-            remote_branch_exists=True,
-            head_on_master=True,
-            delete_capability_allowed=True,
-        )
-        self.assertIn("planned_execution_order", entry)
-        self.assertIsInstance(entry["planned_execution_order"], list)
-
     def test_extract_linked_issue_from_closes(self):
         issue = mcr.extract_linked_issue(
             "feat: cleanup (Closes #269)",

tests/test_pr_work_lease.py

@@ -19,6 +19,7 @@ from pr_work_lease import (  # noqa: E402
     assess_reviewer_mutation_blocked,
     assess_reviewer_stale_head_final_report,
     format_conflict_fix_lease_body,
+    find_active_conflict_fix_lease,
     parse_conflict_fix_lease_comment,
     parse_reviewer_lease_comment,
 )
@@ -203,5 +204,157 @@ class TestFormatLease(unittest.TestCase):
         self.assertEqual(parsed["pr_number"], 376)
 
 
+class TestConflictFixLeaseLifecycle(unittest.TestCase):
+    def test_claim_followed_by_matching_release(self):
+        claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
+        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
+        release_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "branch: feat/fix-376",
+            "worktree: branches/fix-376",
+            "profile: prgs-author",
+            "phase: released",
+            f"head_before: {HEAD_A}",
+            f"head_after: {HEAD_B}",
+            f"expires_at: {expires}",
+        ])
+        comments = [{"body": claim_body}, {"body": release_body}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNone(lease)
+
+    def test_expired_claim_without_release(self):
+        past_expires = (NOW - timedelta(minutes=10)).isoformat().replace("+00:00", "Z")
+        claim_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "phase: claimed",
+            f"head_before: {HEAD_A}",
+            f"expires_at: {past_expires}",
+            "profile: prgs-author",
+        ])
+        comments = [{"body": claim_body}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNone(lease)
+
+    def test_mismatched_release_different_head(self):
+        claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
+        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
+        release_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "profile: prgs-author",
+            "phase: released",
+            f"head_before: {HEAD_B}",
+            f"expires_at: {expires}",
+        ])
+        comments = [{"body": claim_body}, {"body": release_body}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease["phase"], "claimed")
+
+    def test_mismatched_release_different_branch(self):
+        claim_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "branch: feat/branch-A",
+            "phase: claimed",
+            f"head_before: {HEAD_A}",
+            f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}",
+            "profile: prgs-author",
+        ])
+        release_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "branch: feat/branch-B",
+            "phase: released",
+            f"head_before: {HEAD_A}",
+            f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}",
+            "profile: prgs-author",
+        ])
+        comments = [{"body": claim_body}, {"body": release_body}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease["phase"], "claimed")
+
+    def test_release_followed_by_newer_claim(self):
+        claim_1 = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
+        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
+        release_1 = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "profile: prgs-author",
+            "phase: released",
+            f"head_before: {HEAD_A}",
+            f"head_after: {HEAD_B}",
+            f"expires_at: {expires}",
+        ])
+        claim_2 = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "profile: prgs-author",
+            "phase: claimed",
+            f"head_before: {HEAD_B}",
+            f"expires_at: {expires}",
+        ])
+        comments = [{"body": claim_1}, {"body": release_1}, {"body": claim_2}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease["head_before"], HEAD_B)
+
+    def test_malformed_or_ambiguous_markers(self):
+        malformed_release = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "phase: released",
+            # missing head_before and profile
+        ])
+        claim_body = _conflict_fix_body(phase="claimed")
+        comments = [{"body": claim_body}, {"body": malformed_release}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNotNone(lease)
+
+    def test_pr818_historical_sequence(self):
+        comment_14696 = "\n".join([
+            "<!-- mcp-conflict-fix-lease:v1 -->",
+            "pr: #818",
+            "branch: feat/issue-638-webui-app-shell-phase1",
+            "worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1",
+            "profile: prgs-author",
+            "session_id: unknown",
+            "phase: claimed",
+            "head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb",
+            "expires_at: 2026-07-23T07:12:13Z",
+            "reviewer_active: no",
+        ])
+        comment_14730 = "\n".join([
+            "<!-- mcp-conflict-fix-lease:v1 -->",
+            "pr: #818",
+            "branch: feat/issue-638-webui-app-shell-phase1",
+            "worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1",
+            "profile: prgs-author",
+            "session_id: prgs-author-61241-e5129c60",
+            "phase: released",
+            "head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb",
+            "head_after: 64b6eb5d5402663098de5ded3b0617cc3b3df98f",
+            "expires_at: 2026-07-23T06:05:00Z",
+            "reviewer_active: no",
+        ])
+        comments = [{"body": comment_14696}, {"body": comment_14730}]
+        check_now = datetime(2026, 7, 23, 6, 30, tzinfo=timezone.utc)
+        lease = find_active_conflict_fix_lease(comments, pr_number=818, now=check_now)
+        self.assertIsNone(lease)
+        
+        reviewer_gate = assess_reviewer_mutation_blocked(
+            pr_number=818,
+            comments=comments,
+            reviewed_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f",
+            live_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f",
+            mutation="approve",
+            now=check_now,
+        )
+        self.assertTrue(reviewer_gate["mutation_allowed"])
+
+
 if __name__ == "__main__":
     unittest.main()