diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 8b2f72d..02ab1df 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -701,6 +701,19 @@ def _clear_preflight_capability_state() -> None: _preflight_reviewer_violation_files = [] +def _clear_resolved_capability_stamp() -> None: + """Clear the task/role stamp without manufacturing capability proof. + + A fresh resolver attempt invalidates the prior stamp immediately. The + new role/task is recorded only after the current attempt is permitted, so + denial or an unexpected exception cannot leave stale authority behind. + """ + global _preflight_resolved_role, _preflight_resolved_task + + _preflight_resolved_role = None + _preflight_resolved_task = None + + def _invalidate_preflight_identity_state() -> None: """Fail closed when whoami cannot prove the configured identity.""" global _preflight_whoami_called, _preflight_whoami_violation @@ -5239,13 +5252,6 @@ def _evaluate_pr_review_submission( worktree_path: str | None = None, ) -> dict: """Shared gate chain for live submit and dry-run review tools.""" - _verify_role_mutation_workspace( - remote, - worktree_path=worktree_path, - task="review_pr", - org=org, - repo=repo, - ) action = (action or "").strip().lower() workflow_blockers = _review_workflow_load_gate_reasons() if live else [] result = { @@ -5263,6 +5269,21 @@ def _evaluate_pr_review_submission( "reasons": [], } reasons = result["reasons"] + try: + _verify_role_mutation_workspace( + remote, + worktree_path=worktree_path, + task="review_pr", + org=org, + repo=repo, + ) + except RuntimeError as exc: + result["blocker_kind"] = "workspace_role_binding" + reasons.append( + "workspace/role binding failed (fail closed, #723): " + f"{_redact(str(exc))}" + ) + return result if workflow_blockers: reasons.extend(workflow_blockers) reasons.extend(review_workflow_load.recovery_handoff_without_replay()) @@ -12084,14 +12105,31 @@ def gitea_adopt_merger_pr_lease( "permission_report": _permission_block_report("gitea.pr.merge"), } - # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp. - _verify_role_mutation_workspace( - remote, - worktree=worktree, - task="adopt_merger_pr_lease", - org=org, - repo=repo, - ) + # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 + # anti-stomp. A role/workspace rejection is an expected guarded outcome, + # not an opaque MCP internal_error (#723). + try: + _verify_role_mutation_workspace( + remote, + worktree=worktree, + task="adopt_merger_pr_lease", + org=org, + repo=repo, + ) + except RuntimeError as exc: + return { + "success": False, + "adopted": False, + "pr_number": pr_number, + "blocker_kind": "workspace_role_binding", + "reasons": [ + "workspace/role binding failed (fail closed, #723): " + f"{_redact(str(exc))}" + ], + "active_lease": None, + "expected_head_sha": expected_head_sha, + "live_head_sha": None, + } h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) profile = get_profile() @@ -14306,11 +14344,15 @@ def _matching_configured_profiles( config: dict | None, required_permission: str, remote: str | None = None, + required_role_kind: str | None = None, ) -> list[str]: """Profile names that allow *required_permission* (redacted metadata only). #714: when *remote* is provided, only profiles bound to that remote's host - are returned — a dadeschools request never lists prgs profiles. + are returned — a dadeschools request never lists prgs profiles. For a + role-exclusive task, a profile that declares a role must also declare the + required role (#723); undeclared legacy profiles retain permission-only + compatibility. """ if not config or "profiles" not in config: return [] @@ -14348,7 +14390,15 @@ def _matching_configured_profiles( ok, _ = gitea_config.check_operation( required_permission, p_allowed_n, p_forbidden_n ) - if ok: + declared_role = ( + p_data.get("role") or p_data.get("role_kind") or "" + ).strip() + role_allowed = ( + required_role_kind is None + or not declared_role + or declared_role == required_role_kind + ) + if ok and role_allowed: matches.append(p_name) return sorted(matches) @@ -14358,8 +14408,15 @@ def _build_runtime_task_capabilities( forbidden: list[str], config: dict | None, remote: str | None = None, + active_role_kind: str | None = None, ) -> dict: - """Per-task capability summary for role-aware runtime context (#139, #714).""" + """Per-task capability summary for role-aware runtime context. + + A declared active role produces a ``role_filtered`` view consistent with + the resolver. Callers that deliberately omit the role receive the legacy + ``permission_only`` view, labeled as such rather than implying stronger + authority (#723). + """ task_entries = [] flags: dict[str, bool] = {} flag_keys = { @@ -14372,18 +14429,36 @@ 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_allowed = ( + not role_exclusive + or not role_filter_applied + or active_role_kind == required_role_kind + ) + allowed_here = permission_allowed and role_allowed 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, remote=remote + config, + permission, + remote=remote, + required_role_kind=( + required_role_kind if role_exclusive else None + ), ), } task_entries.append(entry) @@ -14847,7 +14922,11 @@ def gitea_get_runtime_context( # #714: filter matching profiles by requested remote when building capability summary session_capabilities = _build_runtime_task_capabilities( - allowed, forbidden, config, remote=remote if remote in REMOTES else None + allowed, + forbidden, + config, + remote=remote if remote in REMOTES else None, + active_role_kind=_profile_role_kind(profile), ) preflight = assess_preflight_status(worktree_path) @@ -17307,6 +17386,10 @@ def gitea_resolve_task_capability( host: Optional override for the Gitea host. """ TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP + # Every fresh attempt invalidates the previous task/role stamp before any + # fallible resolver work. Unknown/malformed tasks and unexpected failures + # therefore remain fail-closed instead of preserving stale authority. + _clear_resolved_capability_stamp() if task not in TASK_MAP: # #723: structured fail-closed unknown_task (never raise into internal_error). @@ -17356,36 +17439,7 @@ def gitea_resolve_task_capability( required_permission = task_capability_map.required_permission(task) required_role = task_capability_map.required_role(task) - role_exclusive_tasks = { - "acquire_reviewer_pr_lease", - "gitea_acquire_reviewer_pr_lease", - "review_pr", - "approve_pr", - "request_changes_pr", - "blind_pr_queue_review", - "pr_queue_cleanup", - "pr-queue-cleanup", - "merge_pr", - "acquire_merger_pr_lease", - "gitea_acquire_merger_pr_lease", - "adopt_merger_pr_lease", - "gitea_adopt_merger_pr_lease", - "release_merger_pr_lease", - "gitea_release_merger_pr_lease", - "create_branch", - "push_branch", - "create_pr", - "commit_files", - "gitea_commit_files", - "address_pr_change_requests", - "update_pr_branch_by_merge", - "gitea_update_pr_branch_by_merge", - "delete_branch", - "cleanup_merged_pr_branch", - "reconciliation_cleanup", - "work_issue", - "work-issue", - } + role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT) if required_role == "reviewer": @@ -17483,7 +17537,9 @@ def gitea_resolve_task_capability( blocked_result["stale_binding_recovery"] = stale_binding return blocked_result - record_preflight_check("capability", required_role, resolved_task=task) + # Record the purity baseline now, but do not record a role/task authority + # stamp until the final allow decision is known (#723). + record_preflight_check("capability") # #714: never auto-switch profiles during capability resolution. profile = get_profile() @@ -17556,17 +17612,13 @@ def gitea_resolve_task_capability( # Matching profiles: same remote/host only (#714) — advisory, never activated. matching_profiles = _matching_configured_profiles( - config, required_permission, remote=remote + config, + required_permission, + remote=remote, + required_role_kind=( + required_role if task in role_exclusive_tasks else None + ), ) - # Role filter for exclusive tasks - if config and "profiles" in config and task in role_exclusive_tasks: - filtered = [] - for p_name in matching_profiles: - p_data = (config.get("profiles") or {}).get(p_name) or {} - p_role = (p_data.get("role") or "").strip() - if not p_role or p_role == required_role: - filtered.append(p_name) - matching_profiles = filtered configured = len(matching_profiles) > 0 available_in_session = allowed_in_current_session @@ -17781,6 +17833,13 @@ def gitea_resolve_task_capability( # A denied or stale resolver result is diagnostic evidence, never a # consumable capability token for a later mutation (#763). _clear_preflight_capability_state() + else: + # Stamp only after every resolver gate and fallible bookkeeping step + # has completed. A denied, stale, malformed, or unexpectedly failing + # attempt therefore cannot transiently authorize a later mutation. + record_preflight_check( + "capability", required_role, resolved_task=task + ) return result diff --git a/task_capability_map.py b/task_capability_map.py index f912917..0b450ae 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -462,6 +462,44 @@ def preflight_task_matches( return False return resolved == mutation or (resolved, mutation) in _PREFLIGHT_TASK_TRANSITIONS + +# Tasks for which permission alone is insufficient: the active/configured +# profile's declared role must also match the task role. This is the complete +# resolver set from master at the #723 reconstruction point, shared with +# runtime reporting so those two authorities cannot drift again. +ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset( + { + "acquire_reviewer_pr_lease", + "gitea_acquire_reviewer_pr_lease", + "review_pr", + "approve_pr", + "request_changes_pr", + "blind_pr_queue_review", + "pr_queue_cleanup", + "pr-queue-cleanup", + "merge_pr", + "acquire_merger_pr_lease", + "gitea_acquire_merger_pr_lease", + "adopt_merger_pr_lease", + "gitea_adopt_merger_pr_lease", + "release_merger_pr_lease", + "gitea_release_merger_pr_lease", + "create_branch", + "push_branch", + "create_pr", + "commit_files", + "gitea_commit_files", + "address_pr_change_requests", + "update_pr_branch_by_merge", + "gitea_update_pr_branch_by_merge", + "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", diff --git a/tests/test_issue_723_role_stamp_and_submission.py b/tests/test_issue_723_role_stamp_and_submission.py new file mode 100644 index 0000000..048b8bf --- /dev/null +++ b/tests/test_issue_723_role_stamp_and_submission.py @@ -0,0 +1,308 @@ +"""Regression coverage for issue #723 role and capability invariants.""" + +from __future__ import annotations + +import unittest +from unittest.mock import patch + +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() -> None: + mcp_server._clear_preflight_capability_state() + mcp_server._preflight_whoami_called = False + mcp_server._preflight_whoami_violation = False + mcp_server.capability_stop_terminal.clear() + mcp_server.role_session_router.clear_route_state() + + +class _ResolveHarness(unittest.TestCase): + def setUp(self): + _reset_preflight() + + def tearDown(self): + _reset_preflight() + + def _resolve( + self, + task, + profile=REVIEWER_PROFILE, + required_role=None, + init_side_effect=None, + ): + 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, + "init_review_decision_lock", + return_value=None, + side_effect=init_side_effect, + ), + patch.object( + mcp_server, "record_mutation_authority", return_value=None + ), + patch.object( + mcp_server, "_check_mcp_runtimes_diagnostics", return_value=[] + ), + ] + if required_role is not None: + patches.append( + patch.object( + mcp_server.task_capability_map, + "required_role", + side_effect=lambda candidate: ( + required_role + if candidate == task + else task_capability_map.TASK_CAPABILITY_MAP[candidate][ + "role" + ] + ), + ) + ) + for context in patches: + context.__enter__() + try: + return mcp_server.gitea_resolve_task_capability( + task=task, remote="prgs" + ) + finally: + for context in reversed(patches): + context.__exit__(None, None, None) + + +class TestCapabilityRoleStampSafety(_ResolveHarness): + def test_allowed_resolution_records_the_correct_stamp(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_resolution_records_no_stamp(self): + with patch.object( + mcp_server, + "record_preflight_check", + wraps=mcp_server.record_preflight_check, + ) as record: + result = self._resolve("review_pr", required_role="merger") + + self.assertFalse(result["allowed_in_current_session"], result) + stamped_calls = [ + call + for call in record.call_args_list + if len(call.args) > 1 and call.args[1] is not None + ] + self.assertEqual( + stamped_calls, + [], + "a denied resolution must never transiently record a role stamp", + ) + self.assertIsNone(mcp_server._preflight_resolved_role) + self.assertIsNone(mcp_server._preflight_resolved_task) + + def test_denied_resolution_clears_an_existing_stamp(self): + allowed = self._resolve("review_pr") + self.assertTrue(allowed["allowed_in_current_session"], allowed) + self.assertEqual(mcp_server._preflight_resolved_role, "reviewer") + + denied = self._resolve("merge_pr") + self.assertFalse(denied["allowed_in_current_session"], denied) + self.assertIsNone(mcp_server._preflight_resolved_role) + self.assertIsNone(mcp_server._preflight_resolved_task) + + def test_denial_cannot_poison_a_later_allowed_task(self): + denied = self._resolve("merge_pr") + self.assertFalse(denied["allowed_in_current_session"], denied) + + allowed = self._resolve("review_pr") + self.assertTrue(allowed["allowed_in_current_session"], allowed) + self.assertEqual(mcp_server._preflight_resolved_role, "reviewer") + self.assertEqual(mcp_server._preflight_resolved_task, "review_pr") + + def test_unexpected_resolver_failure_leaves_no_stamp(self): + with self.assertRaisesRegex(RuntimeError, "malformed decision state"): + self._resolve( + "review_pr", + init_side_effect=RuntimeError("malformed decision state"), + ) + self.assertIsNone(mcp_server._preflight_resolved_role) + self.assertIsNone(mcp_server._preflight_resolved_task) + + +class TestStructuredWorkspaceRoleFailures(unittest.TestCase): + def test_review_submission_returns_workspace_role_binding_failure(self): + error = RuntimeError( + "namespace workspace binding blocked: merger role in reviewer workspace" + ) + with patch.object( + mcp_server, "_verify_role_mutation_workspace", side_effect=error + ): + result = mcp_server._evaluate_pr_review_submission( + pr_number=721, + action="approve", + expected_head_sha="8" * 40, + 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 reason for reason in result["reasons"]), + result, + ) + self.assertTrue(any("merger role" in reason for reason in result["reasons"])) + + def test_adopt_merger_lease_returns_workspace_role_binding_failure(self): + error = RuntimeError("merger workspace binding rejected") + with patch.object( + mcp_server, "_profile_operation_gate", return_value=[] + ), patch.object( + mcp_server, "_verify_role_mutation_workspace", side_effect=error + ), patch.object(mcp_server, "_resolve") as resolve: + result = mcp_server.gitea_adopt_merger_pr_lease( + pr_number=718, + worktree="branches/merge-pr-718", + expected_head_sha="7" * 40, + remote="prgs", + ) + + self.assertFalse(result["success"]) + self.assertFalse(result["adopted"]) + self.assertEqual(result["blocker_kind"], "workspace_role_binding") + self.assertEqual(result["pr_number"], 718) + self.assertEqual(result["expected_head_sha"], "7" * 40) + self.assertIsNone(result["live_head_sha"]) + self.assertTrue(any("binding rejected" in reason for reason in result["reasons"])) + resolve.assert_not_called() + + def test_unexpected_verifier_failure_remains_fail_closed(self): + with patch.object( + mcp_server, + "_verify_role_mutation_workspace", + side_effect=ValueError("unexpected verifier state"), + ), patch.object(mcp_server, "_resolve") as resolve: + with self.assertRaisesRegex(ValueError, "unexpected verifier state"): + mcp_server._evaluate_pr_review_submission( + pr_number=721, + action="approve", + remote="prgs", + live=True, + ) + resolve.assert_not_called() + + +class TestRuntimeCapabilityRoleFiltering(unittest.TestCase): + def test_runtime_role_filter_denies_permission_bearing_wrong_role(self): + allowed = REVIEWER_PROFILE["allowed_operations"] + ["gitea.pr.merge"] + capabilities = mcp_server._build_runtime_task_capabilities( + allowed, + [], + CONFIG, + remote="prgs", + active_role_kind="reviewer", + ) + merge_entry = next( + item + for item in capabilities["task_capabilities"] + if item["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"]) + self.assertFalse(capabilities["can_merge_prs"]) + + def test_permission_only_view_is_explicit(self): + capabilities = mcp_server._build_runtime_task_capabilities( + ["gitea.read", "gitea.pr.merge"], + [], + CONFIG, + active_role_kind=None, + ) + merge_entry = next( + item + for item in capabilities["task_capabilities"] + if item["task"] == "merge_pr" + ) + self.assertEqual(merge_entry["capability_view"], "permission_only") + self.assertTrue(merge_entry["allowed_in_current_session"]) + + def test_matching_profiles_honor_declared_roles(self): + capabilities = mcp_server._build_runtime_task_capabilities( + ["gitea.read"], + [], + CONFIG, + active_role_kind="author", + ) + review_entry = next( + item + for item in capabilities["task_capabilities"] + if item["task"] == "review_pr" + ) + merge_entry = next( + item + for item in capabilities["task_capabilities"] + if item["task"] == "merge_pr" + ) + self.assertEqual( + review_entry["matching_configured_profiles"], ["prgs-reviewer"] + ) + self.assertEqual( + merge_entry["matching_configured_profiles"], ["prgs-merger"] + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 10e6e25..1b1020c 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -242,6 +242,32 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertTrue(res.get("stop_required")) self.assertIs(res.get("mutation_performed"), False) + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_denied_role_exclusive_resolution_does_not_stamp_role( + self, _auth, _api + ): + with patch.dict(os.environ, self._env("author-profile")): + with patch.object( + mcp_server, + "record_preflight_check", + wraps=mcp_server.record_preflight_check, + ) as record: + result = mcp_server.gitea_resolve_task_capability( + task="review_pr", remote="prgs" + ) + + self.assertFalse(result["allowed_in_current_session"], result) + self.assertFalse( + any( + len(call.args) > 1 and call.args[1] == "reviewer" + for call in record.call_args_list + ), + "denied reviewer resolution must never record a reviewer stamp", + ) + self.assertIsNone(mcp_server._preflight_resolved_role) + self.assertIsNone(mcp_server._preflight_resolved_task) + # Additional regression tests per #145 for permission boundaries and structured guidance def test_issue_comment_does_not_imply_close(self): # Author profile has issue.comment but not issue.close diff --git a/tests/test_task_capability_role_invariants.py b/tests/test_task_capability_role_invariants.py index a77f570..df69e9e 100644 --- a/tests/test_task_capability_role_invariants.py +++ b/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): @@ -112,6 +117,42 @@ FORMAL_REVIEW_TASKS = ( "pr-queue-cleanup", ) +# Complete resolver role-exclusive set on master when #723 was reconstructed. +# The shared constant must replace this exact inline authority without dropping +# later lease and PR-sync aliases added after the preserved source commits. +EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset( + { + "acquire_reviewer_pr_lease", + "gitea_acquire_reviewer_pr_lease", + "review_pr", + "approve_pr", + "request_changes_pr", + "blind_pr_queue_review", + "pr_queue_cleanup", + "pr-queue-cleanup", + "merge_pr", + "acquire_merger_pr_lease", + "gitea_acquire_merger_pr_lease", + "adopt_merger_pr_lease", + "gitea_adopt_merger_pr_lease", + "release_merger_pr_lease", + "gitea_release_merger_pr_lease", + "create_branch", + "push_branch", + "create_pr", + "commit_files", + "gitea_commit_files", + "address_pr_change_requests", + "update_pr_branch_by_merge", + "gitea_update_pr_branch_by_merge", + "delete_branch", + "cleanup_merged_pr_branch", + "reconciliation_cleanup", + "work_issue", + "work-issue", + } +) + def _profile_satisfies(role_name, task): """True when the canonical *role_name* profile can perform *task*.""" @@ -201,5 +242,30 @@ class TestMergerBoundary(unittest.TestCase): ) +class TestRoleExclusiveSetIntegrity(unittest.TestCase): + """#723: the shared set is complete, mapped, and role-satisfiable.""" + + def test_complete_current_role_exclusive_set(self): + self.assertEqual(ROLE_EXCLUSIVE_TASKS, EXPECTED_ROLE_EXCLUSIVE_TASKS) + + 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) + + def test_formal_review_tasks_are_role_exclusive(self): + self.assertTrue(set(FORMAL_REVIEW_TASKS) <= ROLE_EXCLUSIVE_TASKS) + + def test_every_role_exclusive_task_has_a_satisfying_profile(self): + for task in sorted(ROLE_EXCLUSIVE_TASKS): + with self.subTest(task=task): + role = required_role(task) + self.assertIn(role, CANONICAL_ROLE_PROFILES) + self.assertTrue( + _profile_satisfies(role, task), + f"canonical {role!r} profile cannot satisfy {task!r}", + ) + + if __name__ == "__main__": unittest.main()