docs/llm-workflow-runbooks.md

@@ -738,6 +738,27 @@ merger handoff.
 - **Prompt (normal):** `After verifying master contains the merge of PR #N using post-merge file-presence verification, close issue #M and delete the merged branch. Include verification details in the report.`
 - **Prompt (reconcile):** `Reconcile closed-not-merged PR #N by verifying if its content landed on master.`
 
+### Post-merge merged cleanup ownership (#523)
+
+Post-merge **local worktree / remote branch cleanup** is **reconciler** work, not
+author work. Do not switch from `prgs-reconciler` to `prgs-author` only to run
+`gitea_reconcile_merged_cleanups`.
+
+- **Profile:** `prgs-reconciler` (task `reconcile_merged_cleanups` /
+  `reconciliation_cleanup`).
+- **Namespace:** reconciler MCP server; stable control checkout is allowed for
+  this role (branches-only author guard does not apply).
+- **Steps:**
+  1. `gitea_whoami` + `gitea_resolve_task_capability(task="reconcile_merged_cleanups")`.
+  2. Dry-run first: `gitea_reconcile_merged_cleanups(dry_run=True)`.
+  3. Execute only after audit/authorization gates when remote branch delete or
+     worktree removal is required (`dry_run=False`, `execute_confirmed=True`,
+     and `gitea.branch.delete` when deleting remotes).
+- **Fail closed:** unmerged/open heads, mismatched worktrees, and non-merged
+  closed PRs must not be cleaned.
+- **Reports:** label cleanup actions as reconciler cleanup (not author mutation).
+- **Prompt:** `As prgs-reconciler, dry-run then execute gitea_reconcile_merged_cleanups for recently merged PRs without switching to prgs-author.`
+
 ### Stop on blocker
 
 - **Any profile.** If a required gate cannot be satisfied — identity

task_capability_map.py

@@ -106,11 +106,11 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
     },
     "reconcile_merged_cleanups": {
         "permission": "gitea.read",
-        "role": "author",
+        "role": "reconciler",
     },
     "reconciliation_cleanup": {
         "permission": "gitea.branch.delete",
-        "role": "author",
+        "role": "reconciler",
     },
     "work_issue": {
         "permission": "gitea.pr.create",

tests/test_audit_reconciliation_mode.py

@@ -284,7 +284,7 @@ class TestTaskCapabilityMap(unittest.TestCase):
             required_permission("reconciliation_cleanup"),
             "gitea.branch.delete",
         )
-        self.assertEqual(required_role("reconciliation_cleanup"), "author")
+        self.assertEqual(required_role("reconciliation_cleanup"), "reconciler")
 
 
 if __name__ == "__main__":

tests/test_reconciler_cleanup_integration.py

@@ -0,0 +1,263 @@
+"""Integration tests for reconciler merged cleanup (#523)."""
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+import task_capability_map
+from audit_reconciliation_mode import clear_phase
+from task_capability_map import required_role
+
+RECONCILER_PROFILE = {
+    "profile_name": "prgs-reconciler",
+    "context": "prgs",
+    "role": "reconciler",
+    "username": "sysadmin",
+    "base_url": "https://gitea.prgs.cc",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.close",
+        "gitea.pr.comment",
+        "gitea.issue.comment",
+    ],
+    "forbidden_operations": [],
+}
+
+AUTHOR_PROFILE = {
+    "profile_name": "prgs-author",
+    "context": "prgs",
+    "role": "author",
+    "username": "jcwalker3",
+    "base_url": "https://gitea.prgs.cc",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.create",
+        "gitea.branch.create",
+        "gitea.branch.push",
+        "gitea.repo.commit",
+    ],
+    "forbidden_operations": [],
+}
+
+CONTROL_ROOT = "/Users/jasonwalker/Development/Gitea-Tools"
+
+
+class TestReconcilerCleanupIntegration(unittest.TestCase):
+    def setUp(self):
+        clear_phase()
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        mcp_server._preflight_resolved_role = None
+        mcp_server._preflight_resolved_task = None
+        mcp_server._preflight_whoami_violation = False
+        mcp_server._preflight_capability_violation = False
+        mcp_server._preflight_whoami_violation_files = []
+        mcp_server._preflight_capability_violation_files = []
+        mcp_server._preflight_capability_baseline_porcelain = ""
+        self.mock_api = patch("gitea_auth.api_request").start()
+        self.mock_auth = patch(
+            "mcp_server.get_auth_header", return_value="token test"
+        ).start()
+
+    def tearDown(self):
+        patch.stopall()
+        clear_phase()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def test_capability_map_routes_merged_cleanup_to_reconciler(self):
+        self.assertEqual(required_role("reconcile_merged_cleanups"), "reconciler")
+        self.assertEqual(required_role("reconciliation_cleanup"), "reconciler")
+        self.assertEqual(
+            task_capability_map.required_permission("reconcile_merged_cleanups"),
+            "gitea.read",
+        )
+
+    @patch.dict(os.environ, {"GITEA_PROFILE_NAME": "prgs-reconciler"}, clear=True)
+    @patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
+    def test_reconciler_can_run_reconcile_merged_cleanups_directly(self, _profile):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability",
+            resolved_role="reconciler",
+            resolved_task="reconcile_merged_cleanups",
+        )
+
+        self.mock_api.side_effect = [
+            [  # closed pulls page
+                {
+                    "number": 100,
+                    "title": "merged feature",
+                    "body": "Closes #100",
+                    "merged": True,
+                    "merged_at": "2026-07-06T12:00:00Z",
+                    "merge_commit_sha": "deadbeef",
+                    "head": {"ref": "feat/issue-100", "sha": "cafebabe"},
+                },
+                {
+                    # closed but NOT merged — must not become a cleanup entry
+                    "number": 101,
+                    "title": "abandoned",
+                    "body": "Closes #101",
+                    "merged": False,
+                    "merged_at": None,
+                    "head": {"ref": "feat/issue-101", "sha": "badcafe"},
+                },
+            ],
+            [],  # open pulls
+        ]
+
+        with patch("mcp_server._remote_branch_exists", return_value=True):
+            with patch(
+                "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
+                return_value=True,
+            ):
+                result = mcp_server.gitea_reconcile_merged_cleanups(
+                    dry_run=True,
+                    remote="prgs",
+                )
+        self.assertTrue(result["success"])
+        self.assertTrue(result["dry_run"])
+        entry_numbers = [e["issue_number"] for e in result["entries"]]
+        self.assertEqual(entry_numbers, [100])
+        self.assertNotIn(101, entry_numbers)
+
+    @patch.dict(
+        os.environ,
+        {
+            "GITEA_PROFILE_NAME": "prgs-reconciler",
+            # Force preflight purity to evaluate (disable test short-circuit).
+            "GITEA_TEST_PORCELAIN": "",
+        },
+        clear=True,
+    )
+    @patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
+    def test_reconciler_role_bypasses_branches_only_mutation_guard(self, _profile):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability",
+            resolved_role="reconciler",
+            resolved_task="reconcile_merged_cleanups",
+        )
+
+        with patch("mcp_server.PROJECT_ROOT", CONTROL_ROOT):
+            with patch(
+                "mcp_server.root_checkout_guard.resolve_remote_master_sha",
+                return_value="abc123",
+            ):
+                with patch(
+                    "mcp_server.issue_lock_worktree.read_worktree_git_state",
+                    return_value={
+                        "current_branch": "master",
+                        "head_sha": "abc123",
+                        "porcelain_status": "",
+                    },
+                ):
+                    try:
+                        mcp_server.verify_preflight_purity(
+                            remote="prgs",
+                            task="reconcile_merged_cleanups",
+                        )
+                    except RuntimeError as e:
+                        self.fail(
+                            "verify_preflight_purity raised RuntimeError "
+                            f"unexpectedly for reconciler: {e}"
+                        )
+
+    @patch.dict(
+        os.environ,
+        {"GITEA_PROFILE_NAME": "prgs-author", "GITEA_TEST_PORCELAIN": ""},
+        clear=True,
+    )
+    @patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE)
+    def test_author_role_remains_blocked_on_root_checkout(self, _profile):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability",
+            resolved_role="author",
+            resolved_task="reconcile_merged_cleanups",
+        )
+
+        with patch("mcp_server.PROJECT_ROOT", CONTROL_ROOT):
+            with patch(
+                "mcp_server.root_checkout_guard.resolve_remote_master_sha",
+                return_value="abc123",
+            ):
+                with patch(
+                    "mcp_server.issue_lock_worktree.read_worktree_git_state",
+                    return_value={
+                        "current_branch": "master",
+                        "head_sha": "abc123",
+                        "porcelain_status": "",
+                    },
+                ):
+                    with self.assertRaises(RuntimeError) as ctx:
+                        mcp_server.verify_preflight_purity(
+                            remote="prgs",
+                            task="reconcile_merged_cleanups",
+                        )
+        message = str(ctx.exception)
+        self.assertTrue(
+            "stable control checkout" in message
+            or "author mutation blocked" in message
+            or "branches/" in message,
+            msg=message,
+        )
+
+    @patch.dict(os.environ, {"GITEA_PROFILE_NAME": "prgs-reconciler"}, clear=True)
+    @patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
+    def test_open_unmerged_pr_heads_are_not_safe_cleanup_targets(self, _profile):
+        """AC5: active/unmerged author work must remain blocked from cleanup."""
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability",
+            resolved_role="reconciler",
+            resolved_task="reconcile_merged_cleanups",
+        )
+
+        open_pr = {
+            "number": 200,
+            "title": "active work",
+            "body": "Closes #200",
+            "merged": False,
+            "state": "open",
+            "head": {"ref": "feat/issue-200", "sha": "livehead1"},
+        }
+        # Same head branch still open on another PR while a closed/merged PR
+        # used the same branch name historically — open head must block delete.
+        closed_merged = {
+            "number": 199,
+            "title": "older merge same branch name",
+            "body": "Closes #199",
+            "merged": True,
+            "merged_at": "2026-07-01T12:00:00Z",
+            "merge_commit_sha": "deadbeef",
+            "head": {"ref": "feat/issue-200", "sha": "oldhead99"},
+        }
+        self.mock_api.side_effect = [
+            [closed_merged],  # closed
+            [open_pr],  # open
+        ]
+
+        with patch("mcp_server._remote_branch_exists", return_value=True):
+            with patch(
+                "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
+                return_value=True,
+            ):
+                result = mcp_server.gitea_reconcile_merged_cleanups(
+                    dry_run=True,
+                    remote="prgs",
+                )
+        self.assertTrue(result["success"])
+        self.assertEqual(len(result["entries"]), 1)
+        remote_assessment = result["entries"][0].get("remote_branch") or {}
+        self.assertFalse(
+            remote_assessment.get("safe_to_delete_remote"),
+            msg=f"open head must block remote delete: {remote_assessment}",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()