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,
@@ -3344,6 +3357,11 @@ def _save_review_decision_lock(data):
     )
     payload["session_profile_lock"] = binding["session_profile_lock"]
     payload["profile_identity"] = binding["profile_identity"]
+    # #720: durable decision locks are recovery-critical terminal provenance,
+    # not generic TTL session cache. Stamp kind + recovery_critical for
+    # pre-existing readers and identity_match_reasons flag-based exempt.
+    payload["kind"] = mcp_session_state.KIND_DECISION_LOCK
+    payload["recovery_critical"] = True
     if binding.get("remote") and not payload.get("remote"):
         payload["remote"] = binding["remote"]
     # #695 AC6: stamp native transport provenance on durable decision locks.
@@ -4091,9 +4109,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"
-    )
     action = (action or "").strip().lower()
     workflow_blockers = _review_workflow_load_gate_reasons() if live else []
     result = {
@@ -4111,6 +4126,20 @@ def _evaluate_pr_review_submission(
         "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 exc:
+        result["blocker_kind"] = "workspace_role_binding"
+        reasons.append(
+            "workspace/role binding failed (fail closed, #723): " f"{exc}"
+        )
+        return result
     if workflow_blockers:
         reasons.extend(workflow_blockers)
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
@@ -5027,6 +5056,12 @@ def gitea_cleanup_stale_review_decision_lock(
     binding = _decision_lock_binding()
     active_identity = binding.get("profile_identity")
     lock = _load_review_decision_lock()
+    # #720: when normal load yields no lock, inspect disk so assessment does not
+    # silently report "absent" while an expired/non-critical envelope remains.
+    disk_inspect = mcp_session_state.inspect_state_envelope(
+        kind=mcp_session_state.KIND_DECISION_LOCK,
+        profile_identity=active_identity,
+    )
     last = stale_review_decision_lock.last_terminal_mutation(lock)
     pr_live = None
     pr_lookup_error = None
@@ -5048,6 +5083,26 @@ def gitea_cleanup_stale_review_decision_lock(
         pr_lookup_error=pr_lookup_error,
         active_profile_identity=active_identity,
     )
+    if lock is None and disk_inspect.get("on_disk"):
+        assessment = dict(assessment)
+        assessment["reasons"] = list(assessment.get("reasons") or []) + [
+            "decision-lock file is present on disk but not loadable via normal "
+            f"TTL/identity gates: {disk_inspect.get('summary')} "
+            "(do not rm session-state files; #720)"
+        ]
+        assessment["disk_inspect"] = {
+            k: disk_inspect.get(k)
+            for k in (
+                "on_disk",
+                "has_payload",
+                "age_hours",
+                "age_exceeds_default_ttl",
+                "recovery_critical",
+                "ttl_exempt",
+                "would_ttl_reject",
+                "summary",
+            )
+        }
 
     # Optional pin: refuse apply against a different terminal PR than expected.
     if (
@@ -5102,6 +5157,19 @@ def gitea_cleanup_stale_review_decision_lock(
         "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
         "merge_commit_sha": assessment.get("merge_commit_sha"),
         "lock_summary": assessment.get("lock_summary"),
+        "disk_inspect": assessment.get("disk_inspect") or {
+            k: disk_inspect.get(k)
+            for k in (
+                "on_disk",
+                "has_payload",
+                "age_hours",
+                "age_exceeds_default_ttl",
+                "recovery_critical",
+                "ttl_exempt",
+                "would_ttl_reject",
+                "summary",
+            )
+        },
         "audit": audit,
         "audit_comment_id": None,
         "reasons": list(assessment.get("reasons") or []),
@@ -11847,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] = []
@@ -11872,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)
 
@@ -11881,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 = {
@@ -11895,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)
@@ -12359,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)
@@ -13777,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":
@@ -13877,7 +13970,12 @@ def gitea_resolve_task_capability(
                         "exact_safe_next_action": next_safe_action,
                     }
 
-    record_preflight_check("capability", required_role, resolved_task=task)
+    # #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)
@@ -13912,6 +14010,14 @@ 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
     configured = False

mcp_session_state.py

@@ -46,8 +46,15 @@ KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
 KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"
 
 # Kinds that must survive the default session-state TTL (forensic / recovery).
+#
+# KIND_DECISION_LOCK is recovery-critical (#720): terminal review provenance is
+# not disposable cache. A generic four-hour TTL must not drop old-head evidence
+# or make ``fresh_review_on_current_head_allowed`` unreachable. Same-head / same-
+# run #332 protections still apply once the ledger is loadable. Other session
+# kinds (workflow load, drafts, etc.) remain TTL-bound.
 RECOVERY_CRITICAL_KINDS = frozenset(
     {
+        KIND_DECISION_LOCK,
         KIND_DECISION_LOCK_ARCHIVE,
         KIND_POST_MERGE_DECISION_RECOVERY,
         KIND_IRRECOVERABLE_DECISION_PROVENANCE,
@@ -245,6 +252,16 @@ def _write_json(path: str, data: dict[str, Any]) -> None:
                 pass
 
 
+def is_recovery_critical_record(record: dict[str, Any] | None, kind: str | None = None) -> bool:
+    """True when a durable record must outlive the generic session-state TTL."""
+    if not record and not kind:
+        return False
+    record_kind = ((record or {}).get("kind") or kind or "").strip()
+    if record_kind in RECOVERY_CRITICAL_KINDS:
+        return True
+    return bool((record or {}).get("recovery_critical"))
+
+
 def identity_match_reasons(
     record: dict[str, Any] | None,
     *,
@@ -288,9 +305,7 @@ def identity_match_reasons(
     else:
         age = _now_utc() - recorded_at
         kind = (record.get("kind") or "").strip()
-        ttl_exempt = kind in RECOVERY_CRITICAL_KINDS or bool(
-            record.get("recovery_critical")
-        )
+        ttl_exempt = is_recovery_critical_record(record, kind=kind)
         if age > timedelta(hours=ttl_hours()) and not ttl_exempt:
             reasons.append(
                 f"session state expired after {ttl_hours():g}h (fail closed)"
@@ -300,6 +315,113 @@ def identity_match_reasons(
     return reasons
 
 
+def inspect_state_envelope(
+    *,
+    kind: str,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    profile_identity: str | None = None,
+    state_dir: str | None = None,
+) -> dict[str, Any]:
+    """Read-only disk inspection for assessment when TTL would otherwise hide state (#720).
+
+    Does **not** apply identity/TTL rejection to the returned presence flags.
+    Callers use this to distinguish:
+      * no file on disk
+      * file present but TTL would reject a non-critical kind
+      * recovery-critical ledger (e.g. KIND_DECISION_LOCK) still loadable
+    Never mutates files. Never returns secrets.
+    """
+    profile = current_profile_identity(profile_identity=profile_identity)
+    path = state_file_path(
+        kind=kind,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+        state_dir=state_dir,
+    )
+    result: dict[str, Any] = {
+        "kind": kind,
+        "profile_identity": profile,
+        "path_basename": os.path.basename(path),
+        "on_disk": False,
+        "has_payload": False,
+        "recorded_at": None,
+        "updated_at": None,
+        "age_hours": None,
+        "ttl_hours": ttl_hours(),
+        "age_exceeds_default_ttl": False,
+        "recovery_critical": kind in RECOVERY_CRITICAL_KINDS,
+        "ttl_exempt": kind in RECOVERY_CRITICAL_KINDS,
+        "would_ttl_reject": False,
+        "identity_reasons": [],
+        "summary": "no session-state file on disk",
+    }
+    if not path or not os.path.exists(path):
+        return result
+    result["on_disk"] = True
+    envelope = _read_json(path)
+    if not envelope:
+        result["summary"] = "session-state file present but unreadable or empty"
+        return result
+    payload = envelope.get("payload")
+    merged: dict[str, Any] = dict(payload) if isinstance(payload, dict) else {}
+    result["has_payload"] = isinstance(payload, dict)
+    for key in (
+        "kind",
+        "remote",
+        "org",
+        "repo",
+        "profile_identity",
+        "session_profile_lock",
+        "recorded_at",
+        "updated_at",
+        "writer_pid",
+        "recovery_critical",
+    ):
+        if key in envelope and key not in merged:
+            merged[key] = envelope[key]
+    if not merged.get("kind"):
+        merged["kind"] = kind
+    recorded_at = _parse_iso(merged.get("recorded_at") or merged.get("updated_at"))
+    result["recorded_at"] = merged.get("recorded_at") or merged.get("updated_at")
+    result["updated_at"] = merged.get("updated_at") or merged.get("recorded_at")
+    if recorded_at is not None:
+        age = _now_utc() - recorded_at
+        age_hours = age.total_seconds() / 3600.0
+        result["age_hours"] = age_hours
+        result["age_exceeds_default_ttl"] = age > timedelta(hours=ttl_hours())
+    ttl_exempt = is_recovery_critical_record(merged, kind=kind)
+    result["recovery_critical"] = ttl_exempt
+    result["ttl_exempt"] = ttl_exempt
+    identity_reasons = identity_match_reasons(
+        merged,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+    )
+    result["identity_reasons"] = list(identity_reasons)
+    result["would_ttl_reject"] = any("expired" in r for r in identity_reasons)
+    if result["has_payload"] and ttl_exempt:
+        result["summary"] = (
+            "recovery-critical session-state present on disk and TTL-exempt; "
+            "load via load_state for full payload"
+        )
+    elif result["has_payload"] and result["would_ttl_reject"]:
+        result["summary"] = (
+            "session-state file present on disk but generic TTL would reject load "
+            f"(age_hours={result.get('age_hours')!r}, ttl={ttl_hours():g}h)"
+        )
+    elif result["has_payload"]:
+        result["summary"] = "session-state file present and within TTL / identity gates"
+    else:
+        result["summary"] = "session-state file present without a dict payload"
+    return result
+
+
 def load_state(
     *,
     kind: str,

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_720_expired_decision_lock.py

@@ -0,0 +1,499 @@
+"""#720: Expired old-head KIND_DECISION_LOCK must not block fresh review.
+
+Reproduction shape (PR #616):
+  * REQUEST_CHANGES terminal at head A
+  * open PR advanced to head B
+  * durable decision lock age > default 4h TTL
+  * fresh_review_on_current_head_allowed is true but unreachable under TTL-first reject
+  * mark_final fails with "session state expired after 4h"
+  * assessment may report "no lock" while the file remains on disk
+
+Security invariants preserved:
+  * same-head second terminal remains fail-closed (#332/#620)
+  * REQUEST_CHANGES is never treated as approval
+  * historical mutations remain on the ledger
+  * merged/closed moot cleanup still allowed (#594)
+  * irrecoverable provenance still requires #709 authorization
+  * ordinary profiles do not gain gitea.decision_lock.irrecoverable_recovery
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+import sys
+
+ROOT = str(Path(__file__).resolve().parent.parent)
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+import mcp_session_state as ss
+import mcp_server
+import stale_review_decision_lock as srdl
+import task_capability_map
+
+HEAD_A = "a0fffae576673ba7df7456b32e6aec916581bdfb"
+HEAD_B = "a6a2243aad9c3e385fc70509f4941a0f0ec33162"
+HEAD_SAME = HEAD_A
+
+RC_616_A = {
+    "pr_number": 616,
+    "action": "request_changes",
+    "review_id": 443,
+    "review_state": "request_changes",
+    "head_sha": HEAD_A,
+}
+
+
+def _lock(mutations=None, **kwargs):
+    base = {
+        "task": "review_pr",
+        "kind": ss.KIND_DECISION_LOCK,
+        "remote": "prgs",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "session_pid": os.getpid(),
+        "session_profile": "prgs-reviewer",
+        "session_profile_lock": "prgs-reviewer",
+        "profile_identity": "prgs-reviewer",
+        "final_review_decision_ready": False,
+        "ready_pr_number": kwargs.get("ready_pr"),
+        "ready_action": kwargs.get("ready_action"),
+        "ready_expected_head_sha": kwargs.get("ready_head"),
+        "ready_remote": "prgs" if kwargs.get("ready_pr") else None,
+        "ready_org": "Scaled-Tech-Consulting" if kwargs.get("ready_pr") else None,
+        "ready_repo": "Gitea-Tools" if kwargs.get("ready_pr") else None,
+        "live_mutations": list(mutations or []),
+        "correction_authorized": False,
+        "correction_reason": None,
+    }
+    return base
+
+
+def _open_pr(pr_number=616, head=HEAD_B, merged=False, closed=False):
+    state = "closed" if closed or merged else "open"
+    return {
+        "number": pr_number,
+        "state": state,
+        "merged": merged,
+        "merged_at": "2026-07-16T12:00:00Z" if merged else None,
+        "merge_commit_sha": "m" * 40 if merged else None,
+        "head": {"sha": head},
+    }
+
+
+def _no_lease():
+    return {"block": False, "reasons": [], "mutation_allowed": True}
+
+
+def _feedback(blocking=False, stale=True):
+    return {
+        "success": True,
+        "has_blocking_change_requests": blocking,
+        "review_feedback_stale": stale,
+        "current_head_sha": HEAD_B,
+    }
+
+
+def _age_lock_payload(lock: dict, hours: float = 5.0) -> dict:
+    """Rewrite timestamps to *hours* ago (simulates durable age without touching prod)."""
+    aged = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat().replace(
+        "+00:00", "Z"
+    )
+    out = dict(lock)
+    out["recorded_at"] = aged
+    out["updated_at"] = aged
+    return out
+
+
+class TestIssue720ExpiredDecisionLockLifecycle(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.env = patch.dict(
+            os.environ,
+            {
+                ss.STATE_DIR_ENV: self._tmp.name,
+                ss.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+                "GITEA_MCP_PROFILE": "prgs-reviewer",
+                "GITEA_PROFILE_NAME": "prgs-reviewer",
+            },
+            clear=False,
+        )
+        self.env.start()
+        mcp_server._REVIEW_DECISION_LOCK = None
+        import review_workflow_load
+
+        review_workflow_load.clear_review_workflow_load()
+        review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
+        mcp_server.gitea_load_review_workflow()
+
+    def tearDown(self):
+        mcp_server._REVIEW_DECISION_LOCK = None
+        import review_workflow_load
+
+        review_workflow_load.clear_review_workflow_load()
+        self.env.stop()
+        self._tmp.cleanup()
+
+    def _persist_aged_lock(self, mutations, hours: float = 5.0, **kwargs):
+        """Write decision lock aged > TTL to the temp durable store (test-only)."""
+        payload = _age_lock_payload(_lock(mutations, **kwargs), hours=hours)
+        saved = ss.save_state(
+            kind=ss.KIND_DECISION_LOCK,
+            payload=payload,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            profile_identity="prgs-reviewer",
+        )
+        # save_state refreshes updated_at; re-age the on-disk envelope for TTL tests.
+        path = ss.state_file_path(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        aged = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat().replace(
+            "+00:00", "Z"
+        )
+        import json
+
+        with open(path, encoding="utf-8") as fh:
+            envelope = json.load(fh)
+        envelope["recorded_at"] = aged
+        envelope["updated_at"] = aged
+        if isinstance(envelope.get("payload"), dict):
+            envelope["payload"]["recorded_at"] = aged
+            envelope["payload"]["updated_at"] = aged
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(envelope, fh, indent=2, sort_keys=True)
+            fh.write("\n")
+        # Drop memory so subsequent loads hit durable store.
+        mcp_server._REVIEW_DECISION_LOCK = None
+        return saved
+
+    def _mark(self, pr, action, head, feedback=None):
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]), patch(
+            "mcp_server._pr_work_lease_reviewer_block", return_value=_no_lease()
+        ), patch.object(
+            mcp_server,
+            "gitea_get_pr_review_feedback",
+            return_value=feedback or _feedback(blocking=False, stale=True),
+        ), patch.object(
+            mcp_server,
+            "gitea_check_pr_eligibility",
+            return_value={"eligible": True, "head_sha": head},
+        ), patch.object(
+            mcp_server.mcp_daemon_guard,
+            "assert_sanctioned_mutation_runtime",
+            return_value=None,
+        ), patch.object(
+            mcp_server.mcp_daemon_guard,
+            "assert_no_direct_import_bypass",
+            return_value=None,
+        ):
+            return mcp_server.gitea_mark_final_review_decision(
+                pr_number=pr,
+                action=action,
+                expected_head_sha=head,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+
+    # --- AC regression matrix ---
+
+    def test_under_four_hours_fresh_review_at_b_allowed(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=1.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_B)
+        self.assertTrue(res.get("marked_ready"), res)
+
+    def test_over_four_hours_fresh_review_at_b_still_allowed(self):
+        """Primary #720 defect: age > TTL must not block head-B mark_final."""
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        # Prove durable load is possible for recovery-critical decision locks.
+        loaded = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+        )
+        self.assertIsNotNone(
+            loaded,
+            "expired KIND_DECISION_LOCK must remain loadable (not generic TTL cache)",
+        )
+        res = self._mark(616, "approve", HEAD_B)
+        self.assertTrue(
+            res.get("marked_ready"),
+            f"expected mark_ready on head B after >4h; got {res}",
+        )
+        reasons = " ".join(res.get("reasons") or [])
+        self.assertNotIn("session state expired", reasons)
+
+    def test_historical_review_at_a_preserved_and_stale(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_B)
+        self.assertTrue(res.get("marked_ready"), res)
+        lock = mcp_server._load_review_decision_lock()
+        self.assertIsNotNone(lock)
+        hist = [m for m in lock["live_mutations"] if m.get("review_id") == 443]
+        self.assertEqual(len(hist), 1)
+        self.assertEqual(hist[0]["head_sha"], HEAD_A)
+        self.assertEqual(hist[0]["action"], "request_changes")
+        a = srdl.assess_stale_review_decision_lock(
+            lock, pr_live=_open_pr(616, HEAD_B)
+        )
+        self.assertTrue(a["stale_by_head"])
+        self.assertTrue(a["fresh_review_on_current_head_allowed"])
+        self.assertEqual(a["locked_head_sha"], HEAD_A)
+
+    def test_no_reuse_approval_from_head_a(self):
+        """REQUEST_CHANGES at A is never an approval credential for B."""
+        self._persist_aged_lock([RC_616_A], hours=5.0)
+        lock = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
+        )
+        last = srdl.last_terminal_mutation(lock)
+        self.assertEqual(last.get("action"), "request_changes")
+        self.assertNotEqual(last.get("action"), "approve")
+        # Gate must still require a new mark/submit for head B; prior RC is not approve.
+        reasons = mcp_server.terminal_review_hard_stop_reasons(
+            616, "mark_ready", expected_head_sha=HEAD_B
+        )
+        self.assertEqual(reasons, [])
+
+    def test_second_terminal_mutation_at_b_same_run_blocked(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_B)
+        self.assertTrue(res.get("marked_ready"), res)
+        # Record terminal approve at B (same run).
+        lock = mcp_server._load_review_decision_lock()
+        lock["final_review_decision_ready"] = True
+        lock["ready_pr_number"] = 616
+        lock["ready_action"] = "approve"
+        lock["ready_expected_head_sha"] = HEAD_B
+        mcp_server._save_review_decision_lock(lock)
+        mcp_server.record_live_review_mutation(616, "approve", review_id=999)
+        # Second terminal at same head B must hard-stop.
+        hard = mcp_server.terminal_review_hard_stop_reasons(
+            616, "mark_ready", expected_head_sha=HEAD_B
+        )
+        self.assertTrue(hard)
+        res2 = self._mark(616, "approve", HEAD_B)
+        self.assertFalse(res2.get("marked_ready"))
+
+    def test_open_pr_same_head_expired_still_fail_closed(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_SAME)
+        self.assertFalse(res.get("marked_ready"), res)
+        self.assertTrue(
+            any("#332" in r or "already consumed" in r for r in (res.get("reasons") or [])),
+            res,
+        )
+
+    def test_merged_pr_moot_cleanup_still_allowed(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        lock = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
+        )
+        a = srdl.assess_stale_review_decision_lock(
+            lock, pr_live=_open_pr(616, HEAD_A, merged=True)
+        )
+        self.assertTrue(a["is_moot"])
+        self.assertTrue(a["cleanup_allowed"])
+        self.assertFalse(a["fresh_review_on_current_head_allowed"])
+
+    def test_assessment_reports_expired_old_head_not_absent(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        # Disk presence + load after lifecycle fix.
+        path = ss.state_file_path(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        self.assertTrue(os.path.exists(path))
+        loaded = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
+        )
+        self.assertIsNotNone(loaded)
+        inspect = ss.inspect_state_envelope(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        self.assertTrue(inspect.get("on_disk"))
+        self.assertTrue(inspect.get("has_payload"))
+        self.assertTrue(inspect.get("recovery_critical") or inspect.get("ttl_exempt"))
+        self.assertTrue(inspect.get("age_hours", 0) >= 4.0)
+        a = srdl.assess_stale_review_decision_lock(
+            loaded, pr_live=_open_pr(616, HEAD_B)
+        )
+        self.assertTrue(a["has_lock"])
+        self.assertNotIn("no review decision lock present", " ".join(a["reasons"]))
+        self.assertTrue(a["stale_by_head"])
+        self.assertEqual(a["last_terminal_action"], "request_changes")
+
+    def test_existing_serialized_records_compatible(self):
+        """Pre-fix ledgers without recovery_critical flag still load via kind."""
+        payload = _age_lock_payload(_lock([RC_616_A]), hours=8.0)
+        payload.pop("recovery_critical", None)
+        # Manually write pre-#720-shaped envelope (kind only on envelope).
+        path = ss.state_file_path(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        import json
+
+        os.makedirs(self._tmp.name, exist_ok=True)
+        aged = payload["recorded_at"]
+        envelope = {
+            "kind": ss.KIND_DECISION_LOCK,
+            "remote": "prgs",
+            "org": "Scaled-Tech-Consulting",
+            "repo": "Gitea-Tools",
+            "profile_identity": "prgs-reviewer",
+            "session_profile_lock": "prgs-reviewer",
+            "recorded_at": aged,
+            "updated_at": aged,
+            "writer_pid": os.getpid(),
+            "payload": payload,
+        }
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(envelope, fh, indent=2, sort_keys=True)
+            fh.write("\n")
+        loaded = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
+        )
+        self.assertIsNotNone(loaded)
+        self.assertEqual(loaded["live_mutations"][0]["review_id"], 443)
+
+    def test_decision_lock_is_recovery_critical_kind(self):
+        self.assertIn(ss.KIND_DECISION_LOCK, ss.RECOVERY_CRITICAL_KINDS)
+
+    def test_workflow_load_still_ttl_expires(self):
+        """Do not make every session-state kind permanently TTL-exempt."""
+        aged = (datetime.now(timezone.utc) - timedelta(hours=5)).isoformat().replace(
+            "+00:00", "Z"
+        )
+        rec = {
+            "kind": ss.KIND_WORKFLOW_LOAD,
+            "recorded_at": aged,
+            "updated_at": aged,
+            "profile_identity": "prgs-reviewer",
+            "session_profile_lock": "prgs-reviewer",
+        }
+        reasons = ss.identity_match_reasons(
+            rec, profile_identity="prgs-reviewer"
+        )
+        self.assertTrue(any("expired" in r for r in reasons), reasons)
+
+    def test_irrecoverable_permission_not_on_ordinary_profiles(self):
+        perm = "gitea.decision_lock.irrecoverable_recovery"
+        # Capability map must not map ordinary author/reviewer/merger tasks to it.
+        for task in (
+            "create_issue",
+            "comment_issue",
+            "review_pr",
+            "merge_pr",
+            "lock_issue",
+            "create_pr",
+        ):
+            req = task_capability_map.required_permission(task)
+            self.assertNotEqual(req, perm, msg=task)
+
+    def test_structured_error_when_same_head_expired(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_A)
+        self.assertFalse(res.get("marked_ready"))
+        reasons = res.get("reasons") or []
+        self.assertTrue(reasons)
+        blob = " ".join(reasons)
+        # Actionable recovery text from hard-stop (not generic internal_error).
+        self.assertIn("#332", blob)
+        self.assertTrue(
+            "#620" in blob or "head moved" in blob or "new expected_head_sha" in blob
+            or "already consumed" in blob
+        )
+        self.assertNotIn("internal_error", blob.lower())
+
+
+class TestIssue720InspectEnvelope(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.env = patch.dict(
+            os.environ,
+            {
+                ss.STATE_DIR_ENV: self._tmp.name,
+                ss.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+            },
+            clear=False,
+        )
+        self.env.start()
+
+    def tearDown(self):
+        self.env.stop()
+        self._tmp.cleanup()
+
+    def test_inspect_missing_file(self):
+        info = ss.inspect_state_envelope(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        self.assertFalse(info["on_disk"])
+        self.assertFalse(info["has_payload"])
+
+
+if __name__ == "__main__":
+    unittest.main()

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_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()