fix: remove safe worktrees before reassessing remote delete ownership (#851)

Post-merge cleanup previously continued past ownership-blocked remote
deletes, which skipped independently safe local worktree removal when the
only block was worktree_binding. Remove the clean owned worktree first,
reassess ownership, then delete the remote branch only if still safe.
Preserve fail-closed protection for dirty/foreign ownership categories.

Closes #851

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

allocator_service.py

@@ -53,6 +53,8 @@ OUTCOME_CANDIDATE_SET_DRIFT = "candidate_set_drift"
 SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session"
 # #776: controller-supplied pre-rank exclusion.
 SKIP_EXCLUDED_BY_CONTROLLER = "excluded_by_controller"
+# #844: epic / child-only implementation container (pre-rank).
+SKIP_EPIC_OR_CHILD_ONLY_CONTAINER = "epic_or_child_only_container"
 
 # Ownership verdicts for a live claim on a candidate (#765).
 OWNERSHIP_OWN = "own"
@@ -130,6 +132,39 @@ ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
 }
 
 
+# Body phrases that prove an issue is an implementation container, not a
+# unit of direct author work (#844). Matched case-insensitively against the
+# issue body. Title alone is never sufficient (ordinary issues may mention
+# "epic" incidentally).
+_CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
+    "implementation is delivered via child issues only",
+    "implementation is delivered through child issues only",
+    "implementation is delivered via child issues",
+    "implementation is delivered through child issues",
+    "do not implement product features in this epic",
+    "do not implement product features in this epic issue itself",
+    "no product feature implementation is claimed complete solely on this epic",
+    "implementable child issues remain independently eligible",
+    "owns the product roadmap and linkage",
+    "this epic owns the product roadmap",
+    "coordination container",
+    "child-only container",
+    "implementation is delegated to child",
+)
+
+# Explicit epic / umbrella labels (structured evidence preferred over title).
+_EPIC_LABELS: frozenset[str] = frozenset(
+    {
+        "type:epic",
+        "epic",
+        "kind:epic",
+        "scope:epic",
+        "type:umbrella",
+        "umbrella",
+    }
+)
+
+
 @dataclass
 class WorkCandidate:
     """One assignable Gitea issue or PR presented to the allocator."""
@@ -139,6 +174,7 @@ class WorkCandidate:
     state: str = "open"
     labels: tuple[str, ...] = ()
     title: str = ""
+    body: str = ""
     priority: int = 0
     head_sha: str | None = None
     # Routing signals (callers derive from Gitea / review feedback).
@@ -158,6 +194,7 @@ class WorkCandidate:
         self.labels = tuple(
             str(x).strip().lower() for x in (self.labels or ()) if str(x).strip()
         )
+        self.body = str(self.body or "")
         if self.kind not in WORK_KINDS:
             raise InvalidWorkKindError(
                 f"candidate kind '{self.kind}' is not assignable; only "
@@ -171,6 +208,7 @@ class WorkCandidate:
             "state": self.state,
             "labels": list(self.labels),
             "title": self.title,
+            "body": self.body,
             "priority": self.priority,
             "head_sha": self.head_sha,
             "request_changes_current_head": self.request_changes_current_head,
@@ -184,6 +222,51 @@ class WorkCandidate:
         }
 
 
+def classify_epic_or_child_only_container(
+    c: WorkCandidate,
+) -> tuple[bool, str | None]:
+    """Return whether *c* is an epic / child-only implementation container (#844).
+
+    Exclusion uses structured evidence first (labels, body scope language).
+    A bare title containing the word "epic" is **not** enough — ordinary
+    implementable issues may mention epics incidentally. A title that is
+    explicitly prefixed ``Epic:`` only counts when the body also proves
+    child-only / no-direct-implementation scope (or an epic label is present).
+
+    PRs are never classified as containers here (they already have a head).
+    """
+    if c.kind != "issue":
+        return False, None
+
+    labels = set(c.labels)
+    epic_label = sorted(labels & _EPIC_LABELS)
+    body_l = (c.body or "").lower()
+    title = (c.title or "").strip()
+    title_l = title.lower()
+
+    body_hits = [m for m in _CHILD_ONLY_BODY_MARKERS if m in body_l]
+    title_epic_prefix = title_l.startswith("epic:") or title_l.startswith("epic ")
+
+    if epic_label:
+        detail = f"label={epic_label[0]}"
+        if body_hits:
+            detail = f"{detail}; body_marker={body_hits[0]!r}"
+        return True, detail
+
+    if body_hits:
+        # Body proves child-only / umbrella scope. Title "Epic:" is corroborating
+        # but not required — containers without the word still exclude.
+        detail = f"body_marker={body_hits[0]!r}"
+        if title_epic_prefix:
+            detail = f"title_epic_prefix; {detail}"
+        return True, detail
+
+    # Title-only "Epic:" without body scope evidence is insufficient (#844 AC:
+    # eligibility does not rely solely on the word "Epic" in a title).
+    # Similarly, incidental "epic" mid-title without markers stays eligible.
+    return False, None
+
+
 @dataclass
 class SkipRecord:
     kind: str
@@ -850,7 +933,8 @@ def allocate_next_work(
     ownership_defects: list[dict[str, Any]] = []
     controller_excluded: list[dict[str, Any]] = []
 
-    # #776 AC2: remove excluded numbers *before* ranking / selection / lease.
+    # #776 AC2 + #844: remove excluded numbers *and* epic/child-only containers
+    # *before* ranking / selection / lease so they never receive assignments.
     rankable: list[WorkCandidate] = []
     for c in candidates:
         if int(c.number) in exclude_set:
@@ -929,6 +1013,23 @@ def allocate_next_work(
                     },
                 }
             continue
+        # #844: epics / child-only containers are never direct implement targets.
+        is_container, container_detail = classify_epic_or_child_only_container(c)
+        if is_container:
+            detail = container_detail or "epic or child-only container"
+            reason = (
+                f"{c.kind}#{c.number} {SKIP_EPIC_OR_CHILD_ONLY_CONTAINER}: "
+                f"{detail}; implementation is delegated to child issues"
+            )
+            skipped.append(
+                SkipRecord(
+                    c.kind,
+                    c.number,
+                    reason,
+                    SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
+                )
+            )
+            continue
         rankable.append(c)
 
     ordered = sort_candidates(rankable)
@@ -1410,6 +1511,7 @@ def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate:
         state=str(data.get("state") or "open"),
         labels=tuple(data.get("labels") or ()),
         title=str(data.get("title") or ""),
+        body=str(data.get("body") or ""),
         priority=priority,
         head_sha=data.get("head_sha"),
         request_changes_current_head=bool(data.get("request_changes_current_head")),

gitea_mcp_server.py

@@ -11242,25 +11242,22 @@ 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] = []
-    for entry in report.get("entries") or []:
+    project_root = _canonical_local_git_root()
-        head_branch = entry.get("head_branch") or ""
-        remote_assessment = entry.get("remote_branch") or {}
-        local_assessment = entry.get("local_worktree") or {}
 
-        if remote_assessment.get("safe_to_delete_remote"):
+    def _ownership_records_for_branch(
-            import urllib.parse
+        head_branch: str, pr_num_int: int | None
-
+    ) -> list[dict]:
-            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,
@@ -11268,7 +11265,7 @@ def gitea_reconcile_merged_cleanups(
             repo=r,
             branch=head_branch,
             pr_number=pr_num_int,
-                project_root=_canonical_local_git_root(),
+            project_root=project_root,
             auth=auth,
             base_api=base,
         )
@@ -11289,6 +11286,18 @@ 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,
@@ -11298,8 +11307,7 @@ def gitea_reconcile_merged_cleanups(
             records=ownership_records,
         )
         if ownership.get("block"):
-                actions.append(
+            return {
-                    {
                 "action": "delete_remote_branch",
                 "branch": head_branch,
                 "success": False,
@@ -11308,13 +11316,10 @@ def gitea_reconcile_merged_cleanups(
                 "verified_absent": False,
                 "blocker_kind": "active_branch_ownership",
                 "reasons": ownership.get("reasons") or [],
-                        "blocking_categories": ownership.get(
+                "blocking_categories": ownership.get("blocking_categories") or [],
-                            "blocking_categories"
+                "after_worktree_removal": after_worktree_removal,
-                        )
+                "ownership_reassessed": after_worktree_removal,
-                        or [],
             }
-                )
-                continue
 
         encoded = urllib.parse.quote(head_branch, safe="")
         url = f"{base}/branches/{encoded}"
@@ -11329,6 +11334,7 @@ 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)
@@ -11337,8 +11343,7 @@ def gitea_reconcile_merged_cleanups(
             readback
         )
         verified = bool(readback_assessment.get("verified_absent"))
-            actions.append(
+        return {
-                {
             "action": "delete_remote_branch",
             "branch": head_branch,
             "success": bool(readback_assessment.get("ok")),
@@ -11347,22 +11352,53 @@ 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(
-                _canonical_local_git_root(),
+                project_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(
-            _canonical_local_git_root(), scratch.get("worktree_path") or ""
+            project_root, scratch.get("worktree_path") or ""
         )
         actions.append({"action": "remove_reviewer_scratch_worktree", **result})
 
@@ -19912,6 +19948,7 @@ def _allocator_candidates_from_gitea(
                         state="open",
                         labels=tuple(labels),
                         title=title,
+                        body=body,
                         priority=20 if "status:ready" in labels else 1,
                         blocked=blocked,
                         dependency_unmet=dep_unmet,

merged_cleanup_reconcile.py

@@ -566,6 +566,10 @@ 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,
@@ -576,9 +580,63 @@ 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,

tests/test_allocator_epic_container_exclusion.py

@@ -0,0 +1,243 @@
+"""Allocator epic / child-only container pre-rank exclusion (#844).
+
+Covers:
+* Issue #631-shaped child-only epic is excluded before ranking.
+* Implementable child issues remain eligible and can be selected.
+* Ordinary issues that merely mention "epic" in title/body are not excluded.
+* Excluded containers never receive assignments or workflow leases.
+* Structured skip reason ``epic_or_child_only_container`` is reported.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+
+from allocator_service import (
+    OUTCOME_ASSIGNED,
+    OUTCOME_PREVIEW,
+    SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
+    WorkCandidate,
+    allocate_next_work,
+    classify_epic_or_child_only_container,
+)
+from control_plane_db import ControlPlaneDB
+
+REMOTE = "prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+# Minimal body mirroring issue #631 authoritative scope language.
+_EPIC_631_BODY = """
+## Scope (umbrella)
+
+This epic owns the **product roadmap and linkage** for the Web Console.
+Implementation is delivered via child issues only.
+
+## Explicit non-goals
+
+* Do not implement product features in this epic issue itself.
+* No product feature implementation is claimed complete solely on this epic.
+"""
+
+_CHILD_BODY = """
+## Problem
+
+Operators need a workflow-event timeline model for Phase 1.
+
+## Acceptance criteria
+
+- [ ] Timeline model API exists
+"""
+
+
+def _issue(
+    number: int,
+    *,
+    title: str = "",
+    body: str = "",
+    labels: tuple[str, ...] = ("status:ready", "type:feature"),
+    priority: int = 20,
+) -> WorkCandidate:
+    return WorkCandidate(
+        kind="issue",
+        number=number,
+        state="open",
+        labels=labels,
+        title=title or f"issue {number}",
+        body=body,
+        priority=priority,
+    )
+
+
+class ClassifyEpicContainerTest(unittest.TestCase):
+    def test_631_shaped_body_and_title_is_container(self) -> None:
+        c = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIsNotNone(detail)
+        self.assertIn("body_marker", detail or "")
+
+    def test_body_markers_without_epic_title(self) -> None:
+        c = _issue(
+            900,
+            title="Control plane roadmap tracker",
+            body="Implementation is delivered via child issues only.",
+        )
+        is_c, _ = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+
+    def test_epic_label_alone_is_container(self) -> None:
+        c = _issue(
+            901,
+            title="Roadmap linkage",
+            body="Track children.",
+            labels=("status:ready", "type:epic"),
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIn("type:epic", detail or "")
+
+    def test_title_epic_prefix_alone_not_container(self) -> None:
+        """Title-only 'Epic:' without body scope evidence stays eligible (#844)."""
+        c = _issue(
+            902,
+            title="Epic: something mentioned only in title",
+            body="Implement a concrete fix for the allocator skip list.",
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertFalse(is_c)
+        self.assertIsNone(detail)
+
+    def test_incidental_epic_word_not_container(self) -> None:
+        c = _issue(
+            903,
+            title="Document epic handoff conventions",
+            body=(
+                "Update the docs so implementable issues that mention an epic "
+                "remain independently executable."
+            ),
+        )
+        is_c, _ = classify_epic_or_child_only_container(c)
+        self.assertFalse(is_c)
+
+    def test_prs_never_classified(self) -> None:
+        pr = WorkCandidate(
+            kind="pr",
+            number=10,
+            state="open",
+            title="Epic: fake",
+            body="Implementation is delivered via child issues only.",
+            head_sha="a" * 40,
+            priority=5,
+        )
+        is_c, _ = classify_epic_or_child_only_container(pr)
+        self.assertFalse(is_c)
+
+
+class AllocateEpicContainerExclusionTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def _alloc(self, candidates, **kwargs):
+        defaults = dict(
+            session_id="sess-844",
+            role="author",
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            profile_name="prgs-author",
+            username="jcwalker3",
+            claims={},
+            apply=False,
+        )
+        defaults.update(kwargs)
+        return allocate_next_work(self.db, candidates=candidates, **defaults)
+
+    def test_631_shaped_epic_excluded_child_selected(self) -> None:
+        epic = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        child = _issue(
+            637,
+            title="Web Console: Workflow-event timeline model (Phase 1)",
+            body=_CHILD_BODY,
+        )
+        res = self._alloc([epic, child], apply=False)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(res["selected"]["number"], 637)
+        skipped = {s["number"]: s for s in res["skipped"]}
+        self.assertIn(631, skipped)
+        self.assertEqual(
+            skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
+        )
+        self.assertIn(SKIP_EPIC_OR_CHILD_ONLY_CONTAINER, skipped[631]["reason"])
+
+    def test_container_cannot_receive_assignment_or_lease(self) -> None:
+        epic = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        res = self._alloc([epic], apply=True)
+        self.assertTrue(res["success"], res)
+        # Only container present → no safe work; never assigned_work.
+        self.assertNotEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertIsNone(res.get("assignment"))
+        self.assertIsNone(res.get("selected"))
+        skipped = {s["number"]: s for s in res["skipped"]}
+        self.assertEqual(
+            skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
+        )
+        # No lease row for the epic.
+        leases = self.db.list_active_leases(
+            remote=REMOTE, org=ORG, repo=REPO
+        ) if hasattr(self.db, "list_active_leases") else []
+        # Prefer generic inventory if available.
+        if not leases and hasattr(self.db, "list_leases"):
+            leases = self.db.list_leases(remote=REMOTE, org=ORG, repo=REPO)
+        for lease in leases or []:
+            work_number = lease.get("work_number") if isinstance(lease, dict) else None
+            self.assertNotEqual(work_number, 631)
+
+    def test_incidental_epic_title_remains_eligible(self) -> None:
+        ordinary = _issue(
+            700,
+            title="Document epic handoff conventions",
+            body="Write runbook text about epic vs child issues.",
+        )
+        res = self._alloc([ordinary], apply=False)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["selected"]["number"], 700)
+        self.assertEqual(res["skipped"], [])
+
+    def test_apply_selects_child_not_epic(self) -> None:
+        epic = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        child = _issue(
+            637,
+            title="Web Console: Workflow-event timeline model (Phase 1)",
+            body=_CHILD_BODY,
+        )
+        res = self._alloc([epic, child], apply=True)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertEqual(res["selected"]["number"], 637)
+        self.assertEqual(res["assignment"]["work_number"], 637)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_branch_cleanup_guard.py

@@ -1266,6 +1266,378 @@ 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,6 +12,59 @@ 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)",