fix(mcp): role-exclusive capability invariants and structured fail-closed submit (Closes #723)

Defect A - invariants:
- task_capability_map.ROLE_EXCLUSIVE_TASKS is now the single shared
  definition of role-exclusive tasks; the resolver's inline copy is gone
  (AC1/AC5 drift surface removed).
- tests: every role-exclusive task must exist in the capability map;
  formal-review tasks stay role-exclusive; canonical merger satisfies
  merge_pr; canonical reconciler satisfies branch-cleanup tasks
  (extends the #722 break-glass invariant suite).

Defect B - unactionable internal_error:
- AC3: gitea_resolve_task_capability records the capability purity
  baseline first but stamps _preflight_resolved_role/_task only for an
  ALLOWED resolve; a denied resolve clears any stale stamp
  (_clear_resolved_capability_stamp) so later mutation preflights key
  off the real profile role.
- AC4: _evaluate_pr_review_submission converts
  _verify_role_mutation_workspace failures (role binding, stale
  runtime) into result reasons with blocker_kind=workspace_role_binding
  instead of letting RuntimeError escape as a generic internal_error.
- AC5: _build_runtime_task_capabilities applies the resolver's
  role-exclusive filter when given the active role kind and labels each
  entry role_filtered/permission_only; matching_configured_profiles
  honors declared profile roles for role-exclusive tasks.

Validation: focused suites 22 passed (+48 subtests); adjacent resolver/
runtime/review suites 72 passed; full suite 2929 passed, 6 skipped,
221 subtests (single pre-existing Starlette warning, #682).

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

gitea_mcp_server.py

@@ -517,6 +517,19 @@ def _clear_preflight_capability_state() -> None:
     _preflight_reviewer_violation_files = []
 
 
+def _clear_resolved_capability_stamp() -> None:
+    """#723 AC3: drop only the resolved role/task stamp (denied resolve).
+
+    ``record_preflight_check("capability")`` without a role intentionally
+    preserves a prior stamp, so a DENIED resolve must clear it explicitly —
+    otherwise the stale role poisons :func:`_effective_workspace_role` and a
+    later mutation preflight raises instead of failing closed with reasons.
+    """
+    global _preflight_resolved_role, _preflight_resolved_task
+    _preflight_resolved_role = None
+    _preflight_resolved_task = None
+
+
 def record_preflight_check(
     type_name: str,
     resolved_role: str | None = None,
@@ -4112,14 +4125,21 @@ def _evaluate_pr_review_submission(
         "remote": remote if remote in REMOTES else None,
         "reasons": [],
     }
+    reasons = result["reasons"]
+    # #723 AC4: workspace/role-binding failures fail closed with structured
+    # reasons — never escape as a generic internal_error (PR #721: a stamped
+    # 'merger' role raised RuntimeError here and the client saw an
+    # unactionable internal_error after mark_final had already succeeded).
     try:
         _verify_role_mutation_workspace(
             remote, worktree_path=worktree_path, task="review_pr"
         )
-    except RuntimeError as e:
-        result["reasons"].append(str(e))
+    except RuntimeError as exc:
+        result["blocker_kind"] = "workspace_role_binding"
+        reasons.append(
+            "workspace/role binding failed (fail closed, #723): " f"{exc}"
+        )
         return result
-    reasons = result["reasons"]
     if workflow_blockers:
         reasons.extend(workflow_blockers)
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
@@ -10022,20 +10042,9 @@ def gitea_adopt_merger_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.merge"),
         }
 
-    try:
-        _verify_role_mutation_workspace(
-            remote, worktree=worktree, task="adopt_merger_pr_lease"
-        )
-    except RuntimeError as e:
-        return {
-            "success": False,
-            "adopted": False,
-            "pr_number": pr_number,
-            "reasons": [str(e)],
-            "active_lease": None,
-            "expected_head_sha": expected_head_sha,
-            "live_head_sha": None,
-        }
+    _verify_role_mutation_workspace(
+        remote, worktree=worktree, task="adopt_merger_pr_lease"
+    )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -11906,8 +11915,16 @@ _RUNTIME_CAPABILITY_TASKS = (
 def _matching_configured_profiles(
     config: dict | None,
     required_permission: str,
+    required_role_kind: str | None = None,
 ) -> list[str]:
-    """Profile names that allow *required_permission* (redacted metadata only)."""
+    """Profile names that allow *required_permission* (redacted metadata only).
+
+    #723 AC5: when *required_role_kind* is supplied (role-exclusive tasks),
+    a profile with a declared role must also match it — mirroring the
+    resolver's matching rule so the two lists cannot drift. Profiles without
+    a declared role still match on permission alone, exactly like the
+    resolver.
+    """
     if not config or "profiles" not in config:
         return []
     matches: list[str] = []
@@ -11931,7 +11948,13 @@ def _matching_configured_profiles(
         ok, _ = gitea_config.check_operation(
             required_permission, p_allowed_n, p_forbidden_n
         )
-        if ok:
+        p_role = (p_data.get("role") or "").strip()
+        role_ok = (
+            required_role_kind is None
+            or not p_role
+            or p_role == required_role_kind
+        )
+        if ok and role_ok:
             matches.append(p_name)
     return sorted(matches)
 
@@ -11940,8 +11963,16 @@ def _build_runtime_task_capabilities(
     allowed: list[str],
     forbidden: list[str],
     config: dict | None,
+    active_role_kind: str | None = None,
 ) -> dict:
-    """Per-task capability summary for role-aware runtime context (#139)."""
+    """Per-task capability summary for role-aware runtime context (#139).
+
+    #723 AC5: applies the same role-exclusive filter as
+    ``gitea_resolve_task_capability`` when *active_role_kind* is supplied, so
+    runtime context can never report a role-exclusive task as allowed while
+    the resolver fail-closes it in the same session (incident #722). Without
+    an *active_role_kind* the view is permission-only and each entry says so.
+    """
     task_entries = []
     flags: dict[str, bool] = {}
     flag_keys = {
@@ -11954,18 +11985,35 @@ def _build_runtime_task_capabilities(
         "close_issue": "can_close_issues",
         "reconcile_already_landed_pr": "can_reconcile_already_landed_prs",
     }
+    role_filter_applied = active_role_kind is not None
     for task in _RUNTIME_CAPABILITY_TASKS:
         permission = task_capability_map.required_permission(task)
-        allowed_here, _ = gitea_config.check_operation(
+        required_role_kind = task_capability_map.required_role(task)
+        role_exclusive = task in task_capability_map.ROLE_EXCLUSIVE_TASKS
+        permission_allowed, _ = gitea_config.check_operation(
             permission, allowed, forbidden
         )
+        role_ok = (
+            not role_exclusive
+            or not role_filter_applied
+            or active_role_kind == required_role_kind
+        )
+        allowed_here = permission_allowed and role_ok
         entry = {
             "task": task,
             "required_permission": permission,
-            "required_role_kind": task_capability_map.required_role(task),
+            "required_role_kind": required_role_kind,
+            "role_exclusive": role_exclusive,
+            "capability_view": (
+                "role_filtered" if role_filter_applied else "permission_only"
+            ),
             "allowed_in_current_session": allowed_here,
             "matching_configured_profiles": _matching_configured_profiles(
-                config, permission
+                config,
+                permission,
+                required_role_kind=(
+                    required_role_kind if role_exclusive else None
+                ),
             ),
         }
         task_entries.append(entry)
@@ -12418,7 +12466,10 @@ def gitea_get_runtime_context(
         )
 
     session_capabilities = _build_runtime_task_capabilities(
-        allowed, forbidden, config
+        allowed,
+        forbidden,
+        config,
+        active_role_kind=_profile_role_kind(profile),
     )
 
     preflight = assess_preflight_status(worktree_path)
@@ -13836,26 +13887,9 @@ def gitea_resolve_task_capability(
 
     required_permission = task_capability_map.required_permission(task)
     required_role = task_capability_map.required_role(task)
-    role_exclusive_tasks = {
-        "review_pr",
-        "approve_pr",
-        "request_changes_pr",
-        "blind_pr_queue_review",
-        "pr_queue_cleanup",
-        "pr-queue-cleanup",
-        "merge_pr",
-        "create_branch",
-        "push_branch",
-        "create_pr",
-        "commit_files",
-        "gitea_commit_files",
-        "address_pr_change_requests",
-        "delete_branch",
-        "cleanup_merged_pr_branch",
-        "reconciliation_cleanup",
-        "work_issue",
-        "work-issue",
-    }
+    # #723 AC5: single shared definition; the runtime-context capability
+    # report applies the same set so the two views cannot drift.
+    role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
 
     infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
     if required_role == "reviewer":
@@ -13936,6 +13970,13 @@ def gitea_resolve_task_capability(
                         "exact_safe_next_action": next_safe_action,
                     }
 
+    # #723 AC3: record the capability purity baseline now, but DEFER the
+    # resolved-role/task stamp until after the allow/deny decision below. A
+    # denied resolve previously stamped ``_preflight_resolved_role`` anyway
+    # (PR #721: a denied review_pr resolve stamped 'merger', and the later
+    # submit escaped as a generic internal_error instead of failing closed).
+    record_preflight_check("capability")
+
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)
 
@@ -13969,8 +14010,13 @@ def gitea_resolve_task_capability(
         permission_allowed_in_current_session and role_matches_current_session
     )
 
+    # #723 AC3: stamp the resolved role/task only for an ALLOWED resolve; a
+    # denied resolve clears any stale stamp so later mutation preflights key
+    # off the real profile role and fail closed with structured reasons.
     if allowed_in_current_session:
         record_preflight_check("capability", required_role, resolved_task=task)
+    else:
+        _clear_resolved_capability_stamp()
 
     switching = gitea_config.is_runtime_switching_enabled()
     available_in_session = allowed_in_current_session

task_capability_map.py

@@ -330,6 +330,32 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
     },
 }
 
+# #723 AC1/AC5: tasks where permission alone is NOT enough — the profile's
+# role kind must also match. Shared by ``gitea_resolve_task_capability`` and
+# the runtime-context capability report so the two can never disagree about
+# which tasks are role-exclusive (incident #722: runtime context said
+# review_pr was allowed while the resolver fail-closed the same session).
+ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset({
+    "review_pr",
+    "approve_pr",
+    "request_changes_pr",
+    "blind_pr_queue_review",
+    "pr_queue_cleanup",
+    "pr-queue-cleanup",
+    "merge_pr",
+    "create_branch",
+    "push_branch",
+    "create_pr",
+    "commit_files",
+    "gitea_commit_files",
+    "address_pr_change_requests",
+    "delete_branch",
+    "cleanup_merged_pr_branch",
+    "reconciliation_cleanup",
+    "work_issue",
+    "work-issue",
+})
+
 # Issue-mutating MCP tools and their resolver task keys.
 ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
     "gitea_create_issue": "create_issue",

tests/test_issue_723_role_stamp_and_submission.py

@@ -0,0 +1,337 @@
+"""#723 AC3/AC4/AC5: denied resolves must not poison the session role stamp,
+review-submission workspace failures must fail closed with structured
+reasons, and runtime-context capabilities must apply the resolver's
+role-exclusive filter.
+
+Reproduces the PR #721 sequence: a reviewer session resolved a task whose
+capability-map role had drifted to ``merger``; the denied resolve stamped
+``_preflight_resolved_role = "merger"`` anyway, ``mark_final`` succeeded, and
+``gitea_submit_pr_review`` raised RuntimeError out of the tool as a generic
+``internal_error``.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = str(Path(__file__).resolve().parent.parent)
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+import gitea_mcp_server as mcp_server
+import task_capability_map
+
+
+REVIEWER_PROFILE = {
+    "profile_name": "prgs-reviewer",
+    "role": "reviewer",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.review",
+        "gitea.pr.approve",
+        "gitea.pr.request_changes",
+        "gitea.pr.comment",
+        "gitea.issue.comment",
+    ],
+    "forbidden_operations": [
+        "gitea.branch.create",
+        "gitea.branch.push",
+        "gitea.repo.commit",
+        "gitea.pr.create",
+        "gitea.pr.merge",
+    ],
+}
+
+CONFIG = {
+    "profiles": {
+        "prgs-reviewer": {
+            "role": "reviewer",
+            "allowed_operations": REVIEWER_PROFILE["allowed_operations"],
+            "forbidden_operations": REVIEWER_PROFILE["forbidden_operations"],
+        },
+        "prgs-merger": {
+            "role": "merger",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.merge",
+                "gitea.pr.comment",
+                "gitea.issue.comment",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve",
+                "gitea.pr.review",
+                "gitea.pr.request_changes",
+            ],
+        },
+    }
+}
+
+
+def _reset_preflight():
+    mcp_server._clear_preflight_capability_state()
+    mcp_server._preflight_whoami_called = False
+    mcp_server._preflight_whoami_violation = False
+
+
+class _ResolveHarness(unittest.TestCase):
+    """Shared patched-resolver harness."""
+
+    def setUp(self):
+        _reset_preflight()
+
+    def tearDown(self):
+        _reset_preflight()
+
+    def _resolve(self, task, profile=REVIEWER_PROFILE, required_role=None):
+        """Run gitea_resolve_task_capability with a fixed profile/config."""
+        patches = [
+            patch.object(mcp_server, "get_profile", return_value=profile),
+            patch.object(
+                mcp_server.gitea_config, "load_config", return_value=CONFIG
+            ),
+            patch.object(
+                mcp_server, "_authenticated_username", return_value="tester"
+            ),
+            patch.object(
+                mcp_server, "_ensure_matching_profile", return_value=None
+            ),
+            patch.object(
+                mcp_server, "init_review_decision_lock", return_value=None
+            ),
+        ]
+        if required_role is not None:
+            patches.append(
+                patch.object(
+                    mcp_server.task_capability_map,
+                    "required_role",
+                    side_effect=lambda t: (
+                        required_role
+                        if t == task
+                        else task_capability_map.TASK_CAPABILITY_MAP[t]["role"]
+                    ),
+                )
+            )
+        for p in patches:
+            p.__enter__()
+        try:
+            return mcp_server.gitea_resolve_task_capability(
+                task=task, remote="prgs"
+            )
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+
+class TestAC3DeniedResolveDoesNotStampRole(_ResolveHarness):
+    def test_allowed_resolve_stamps_role(self):
+        result = self._resolve("review_pr")
+        self.assertTrue(result["allowed_in_current_session"], result)
+        self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
+        self.assertEqual(mcp_server._preflight_resolved_task, "review_pr")
+
+    def test_denied_resolve_does_not_stamp_required_role(self):
+        """The PR #721 poison: review_pr requiring merger under a reviewer."""
+        result = self._resolve("review_pr", required_role="merger")
+        self.assertFalse(result["allowed_in_current_session"], result)
+        self.assertIsNone(
+            mcp_server._preflight_resolved_role,
+            "denied resolve must not stamp the required role (#723 AC3)",
+        )
+        self.assertIsNone(mcp_server._preflight_resolved_task)
+
+    def test_denied_resolve_clears_prior_stale_stamp(self):
+        allowed = self._resolve("review_pr")
+        self.assertTrue(allowed["allowed_in_current_session"])
+        self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
+
+        denied = self._resolve("merge_pr")  # reviewer profile cannot merge
+        self.assertFalse(denied["allowed_in_current_session"], denied)
+        self.assertIsNone(
+            mcp_server._preflight_resolved_role,
+            "a denied resolve must clear the previous stamp, not keep it",
+        )
+
+    def test_denied_resolve_keeps_effective_role_on_actual_profile(self):
+        with patch.object(
+            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
+        ):
+            self._resolve("review_pr", required_role="merger")
+            self.assertEqual(
+                mcp_server._effective_workspace_role(), "reviewer"
+            )
+
+
+class TestAC4SubmissionFailsClosedStructured(unittest.TestCase):
+    def setUp(self):
+        _reset_preflight()
+
+    def tearDown(self):
+        _reset_preflight()
+
+    def test_workspace_binding_failure_returns_reasons_not_raise(self):
+        boom = RuntimeError(
+            "namespace workspace binding blocked: role 'merger' cannot "
+            "mutate from reviewer session workspace"
+        )
+        with patch.object(
+            mcp_server,
+            "_verify_role_mutation_workspace",
+            side_effect=boom,
+        ), patch.object(
+            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
+        ):
+            result = mcp_server._evaluate_pr_review_submission(
+                pr_number=721,
+                action="approve",
+                expected_head_sha="80f59b334e6671b08006725292c08a8e8b6c823f",
+                remote="prgs",
+                live=True,
+                final_review_decision_ready=True,
+            )
+        self.assertFalse(result["performed"])
+        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
+        self.assertTrue(
+            any("workspace/role binding failed" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertTrue(
+            any("cannot mutate from reviewer session" in r for r in result["reasons"]),
+            "the underlying binding error text must be preserved",
+        )
+
+    def test_stale_runtime_failure_also_structured(self):
+        boom = RuntimeError(
+            "stale-runtime: The active Gitea MCP server process is stale"
+        )
+        with patch.object(
+            mcp_server,
+            "_verify_role_mutation_workspace",
+            side_effect=boom,
+        ), patch.object(
+            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
+        ):
+            result = mcp_server._evaluate_pr_review_submission(
+                pr_number=721,
+                action="approve",
+                remote="prgs",
+                live=True,
+            )
+        self.assertFalse(result["performed"])
+        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
+        self.assertTrue(
+            any("stale-runtime" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_dry_run_also_fails_closed_structured(self):
+        boom = RuntimeError("binding blocked")
+        with patch.object(
+            mcp_server,
+            "_verify_role_mutation_workspace",
+            side_effect=boom,
+        ), patch.object(
+            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
+        ):
+            result = mcp_server._evaluate_pr_review_submission(
+                pr_number=721,
+                action="approve",
+                remote="prgs",
+                live=False,
+            )
+        self.assertFalse(result["would_perform"])
+        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
+
+
+class TestAC5RuntimeContextRoleFilter(unittest.TestCase):
+    def test_role_filtered_view_matches_resolver_denial(self):
+        """Reviewer session: merge_pr stays denied under the role filter even
+        when the permission is (pathologically) present."""
+        allowed_ops = [
+            "gitea.read",
+            "gitea.pr.review",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+            "gitea.pr.merge",
+        ]
+        caps = mcp_server._build_runtime_task_capabilities(
+            allowed_ops, [], CONFIG, active_role_kind="reviewer"
+        )
+        merge_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
+        )
+        self.assertTrue(merge_entry["role_exclusive"])
+        self.assertEqual(merge_entry["capability_view"], "role_filtered")
+        self.assertFalse(
+            merge_entry["allowed_in_current_session"],
+            "role-exclusive merge_pr must stay denied for a reviewer role "
+            "even when the permission is present (#723 AC5)",
+        )
+        self.assertFalse(caps["can_merge_prs"])
+
+    def test_role_filter_restricts_matching_profiles(self):
+        caps = mcp_server._build_runtime_task_capabilities(
+            ["gitea.read"], [], CONFIG, active_role_kind="author"
+        )
+        review_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "review_pr"
+        )
+        self.assertEqual(
+            review_entry["matching_configured_profiles"],
+            ["prgs-reviewer"],
+            "role-exclusive review_pr must not list the merger profile",
+        )
+        merge_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
+        )
+        self.assertEqual(
+            merge_entry["matching_configured_profiles"], ["prgs-merger"]
+        )
+
+    def test_permission_only_view_is_labeled(self):
+        caps = mcp_server._build_runtime_task_capabilities(
+            ["gitea.pr.merge", "gitea.read"], [], CONFIG
+        )
+        merge_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
+        )
+        self.assertEqual(merge_entry["capability_view"], "permission_only")
+        # Legacy permission-only semantics preserved when no role supplied.
+        self.assertTrue(merge_entry["allowed_in_current_session"])
+
+    def test_incident_722_shape_runtime_context_agrees_with_resolver(self):
+        """Post-970e68b shape: review_pr requires merger; a reviewer session
+        must see review_pr denied in runtime context, matching the resolver."""
+        with patch.object(
+            mcp_server.task_capability_map,
+            "required_role",
+            side_effect=lambda t: (
+                "merger"
+                if t == "review_pr"
+                else task_capability_map.TASK_CAPABILITY_MAP[t]["role"]
+            ),
+        ):
+            caps = mcp_server._build_runtime_task_capabilities(
+                REVIEWER_PROFILE["allowed_operations"],
+                REVIEWER_PROFILE["forbidden_operations"],
+                CONFIG,
+                active_role_kind="reviewer",
+            )
+        review_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "review_pr"
+        )
+        self.assertFalse(
+            review_entry["allowed_in_current_session"],
+            "runtime context must not report review_pr allowed when the "
+            "resolver would fail-close it (incident #722)",
+        )
+        self.assertFalse(caps["can_review_prs"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -442,23 +442,5 @@ class TestResolveTaskCapability(unittest.TestCase):
                 with self.assertRaises(ValueError):
                     mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
 
-    @patch("mcp_server.api_request", return_value={"login": "author-user"})
-    @patch("mcp_server.get_auth_header", return_value="token author-pass")
-    def test_resolve_task_capability_does_not_poison_role_stamp_on_denial(self, _auth, _api):
-        # Issue #723: attempting to acquire a reviewer lease from a merger profile
-        # should fail without poisoning the session role stamp as 'reviewer'.
-        with patch.dict(os.environ, self._env("merger-profile")):
-            # Initially no preflight check is recorded
-            self.assertEqual(mcp_server._preflight_resolved_role, None)
-            
-            # Request reviewer lease task which should be denied for merger
-            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
-            
-            # The task is denied
-            self.assertFalse(res["allowed_in_current_session"])
-            
-            # The session role stamp should NOT be poisoned
-            self.assertNotEqual(mcp_server._preflight_resolved_role, "reviewer")
-
 if __name__ == "__main__":
     unittest.main()

tests/test_task_capability_role_invariants.py

@@ -19,7 +19,12 @@ import unittest
 
 import gitea_config
 from role_session_router import MERGER_TASKS, REVIEWER_TASKS
-from task_capability_map import required_permission, required_role
+from task_capability_map import (
+    ROLE_EXCLUSIVE_TASKS,
+    TASK_CAPABILITY_MAP,
+    required_permission,
+    required_role,
+)
 
 # Canonical role-profile permission shape. Mirrors the configured
 # author/reviewer/merger/reconciler profiles (profiles.json v2 role split):
@@ -201,5 +206,44 @@ class TestMergerBoundary(unittest.TestCase):
                 )
 
 
+class TestRoleExclusiveSetIntegrity(unittest.TestCase):
+    """#723 AC1/AC5: the shared role-exclusive set stays coherent."""
+
+    def test_every_role_exclusive_task_exists_in_capability_map(self):
+        for task in sorted(ROLE_EXCLUSIVE_TASKS):
+            with self.subTest(task=task):
+                self.assertIn(
+                    task,
+                    TASK_CAPABILITY_MAP,
+                    f"role-exclusive task {task!r} missing from the "
+                    f"capability map — required_role() would raise and the "
+                    f"resolver would 500 instead of failing closed",
+                )
+
+    def test_formal_review_tasks_are_role_exclusive(self):
+        for task in FORMAL_REVIEW_TASKS:
+            with self.subTest(task=task):
+                self.assertIn(task, ROLE_EXCLUSIVE_TASKS)
+
+    def test_merge_pr_is_role_exclusive_and_merger_satisfiable(self):
+        """AC1 merger equivalent: merging must stay possible for a canonical
+        merger profile (permission AND role together)."""
+        self.assertIn("merge_pr", ROLE_EXCLUSIVE_TASKS)
+        self.assertTrue(
+            _profile_satisfies("merger", "merge_pr"),
+            "no canonical merger profile satisfies merge_pr — merging would "
+            "be impossible for every configured profile",
+        )
+
+    def test_reconciler_cleanup_tasks_stay_reconciler_satisfiable(self):
+        for task in ("cleanup_merged_pr_branch", "reconciliation_cleanup"):
+            with self.subTest(task=task):
+                self.assertIn(task, ROLE_EXCLUSIVE_TASKS)
+                self.assertTrue(
+                    _profile_satisfies("reconciler", task),
+                    f"no canonical reconciler profile satisfies {task!r}",
+                )
+
+
 if __name__ == "__main__":
     unittest.main()