fix(gate): stop classifying stale-runtime blocks as permission denials (Closes #897)

Stale-runtime and runtime-mode mutation refusals previously shared the
permission-denial channel, so permission_report claimed a missing op the
active profile already held and recommended gitea_activate_profile.
Typed blocker_kind payloads report reconnect-only recovery for staleness,
omit permission_report for non-permission gates, and fail closed when a
permission_report would invent a missing permission the profile holds.

docs/mcp-restart-classes.md

@@ -1,53 +0,0 @@
-# MCP restart classes and blast-radius permissions (#663)
-
-This is the machine-enforced class matrix used by
-`restart_coordinator.RESTART_CLASS_POLICIES`. It implements the narrower-first
-recovery ladder from #655 and the authorization policy from #656, using the
-path inventory from #657 and the impact coordinator from #658. Product and
-delivery lineage: vision #652 and roadmap #653.
-
-Unknown class names are denied. The coordinator requires both the class
-permission and an eligible request role. Approval gates are additional: a
-caller cannot turn a request permission into execution authority.
-
-| Restart class | Required permission | Expected blast radius | Drain requirement | Approval requirement | Audit requirement | Recovery behavior |
-|---|---|---|---|---|---|---|
-| `client_reconnect` | `mcp.reconnect.client` | none | none | self service | class, actor, client namespace, reason, outcome | Reconnect only the caller's client transport. No daemon or peer work changes. |
-| `session_reconnect` | `mcp.reconnect.session` | low | requesting-session safe point | self service | class, actor, session, reason, outcome | Rebind identity, capability, and workspace state for one session. |
-| `worker_restart` | `mcp.restart.worker.request` | low | target worker | controller approval + automated gates | class, actor, worker, approval, scoped drain, outcome | Restart one worker after its own leases and mutations drain. |
-| `role_runtime_restart` | `mcp.restart.role_runtime.request` | medium | target role runtime | controller approval + automated gates | class, actor, role namespace, approval, scoped drain, outcome | Restart and re-probe one role runtime; unrelated roles remain available. |
-| `connector_restart` | `mcp.restart.connector.request` | medium | target connector | controller approval + automated gates | class, actor, connector, approval, scoped drain, outcome | Restart one connector while unrelated runtimes remain available. |
-| `configuration_reload` | `mcp.reload.configuration.request` | low | mutation quiesce | controller approval + automated gates | class, actor, configuration revision, approval, outcome | Gracefully reload configuration without replacing the daemon. |
-| `rolling_mcp_restart` | `mcp.restart.rolling.request` | medium | one instance at a time | controller approval + automated gates | class, actor, instance order, approval, per-instance drains, outcome | Drain, restart, verify, and restore each instance before advancing. |
-| `full_mcp_restart` | `mcp.restart.full.request` | high | all sessions and mutations | controller approval + automated gates | class, actor, full impact, approval, full drain proof, outcome | Replace the complete MCP runtime only after a verified full drain. |
-| `host_restart` | `mcp.restart.host.request` | high | all host work | controller approval + infrastructure operator | class, actor, host/change or incident id, approval, full drain proof, outcome | Hand off to infrastructure ownership and reconcile every runtime afterward. |
-
-## Drain boundary
-
-Only `full_mcp_restart` and `host_restart` set `full_drain_required=true`.
-Reconnects and configuration reloads do not disrupt peer sessions. Worker,
-role-runtime, and connector restarts evaluate only their explicitly named
-target. Rolling restart drains one instance at a time. Missing required target
-scope denies the request rather than silently widening it to a full restart.
-
-## Permission and approval boundary
-
-Author, reviewer, merger, and reconciler roles may self-request reconnects and
-request scoped worker/role/connector/reload recovery. They cannot request
-rolling, full, or host restart classes. Controller/operator/admin roles may
-request the broader classes, while execution remains operator/admin-owned.
-Controller approval is independently required for every class above a session
-reconnect. Host restart additionally requires infrastructure-operator proof.
-
-The MCP request tool derives class permissions from its authenticated runtime
-role. It does not accept caller-supplied permissions. Controller and operator
-authorization are read from the already-running daemon environment, never
-from a request argument.
-
-## Audit and failure behavior
-
-Every impact audit and every console restart/reload audit includes a
-`restart_class` field. The impact audit also includes the exact
-`required_permission`. Unknown classes, missing permissions, ineligible roles,
-missing approval, missing scoped targets, and incomplete inventory all deny
-fail closed. Manual process kills remain forbidden and contaminating (#630).

docs/mcp-restart-coordinator.md

@@ -6,23 +6,11 @@ console (#642 / #652) can see the blast radius *before* concurrent LLM work is
 disrupted. Uncoordinated restarts destroy in-flight author/reviewer/merger work
 and give operators no way to see what they are about to break.
 
-This lands the coordinator + impact DTO + the MCP tool. It is the single
+This lands the coordinator + impact DTO + a dry-run MCP tool. It is the single
 sanctioned entry point for restart evaluation post-#657 (which inventoried the
-restart/reload/kill paths).
+restart/reload/kill paths). The **mutative apply** path — actually performing a
-
+restart — is a later child gated by a drain proof and is explicitly out of
-The **drain-proof hard gate now executes inside this tool** (#661, via PR #882):
+scope here.
-an apply request (`dry_run=False`) is evaluated against a drain proof here and
-denied when that proof is missing, expired, unclean, tampered with, or stale.
-It is no longer a separate child operation. What remains a later child is only
-the **execution** step — actually stopping and restoring a process. This tool
-still never restarts anything: `apply_supported` is always `false` and
-`restart_performed` is always `false`.
-
-The coordinator now routes every request through the restart-class policy
-matrix defined for #663. See
-[`mcp-restart-classes.md`](./mcp-restart-classes.md) for permissions, expected
-blast radius, scoped drain and approval requirements, audit fields, and
-recovery behavior for all nine classes.
 
 ## Components
 
@@ -31,8 +19,7 @@ recovery behavior for all nine classes.
 | `restart_coordinator.evaluate_restart_impact` | `restart_coordinator.py` | Pure classification: inventory → impact report DTO. No I/O, no restart. |
 | `RestartImpactReport` / `SessionImpact` / `LeaseImpact` | `restart_coordinator.py` | Console-facing DTO (`.as_dict()` is JSON-serializable). |
 | `ControlPlaneDB.list_sessions` | `control_plane_db.py` | Read-only session inventory (the process-level unit a restart kills). |
-| `gitea_request_mcp_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 DB, calls the coordinator, returns the report, and on `dry_run=False` runs the #661 drain-proof hard gate. Never restarts a process. |
+| `gitea_request_mcp_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 DB, calls the coordinator, returns the report. Dry-run only. |
-| `drain_proof.gate_apply_restart` | `drain_proof.py` | The #661 hard gate: verifies a drain proof against the current impact fingerprint, or records an authorized break-glass bypass. |
 
 ## Dimensions evaluated
 
@@ -90,61 +77,17 @@ authorization is present.
 ```text
 gitea_request_mcp_restart(remote, host, org, repo,
                           dry_run=True, request_override=False,
-                          session_id=None, limit=200,
+                          session_id=None, limit=200)
-                          restart_class="full_mcp_restart",
-                          target_session_id=None, target_role=None,
-                          target_connector=None,
-                          drain_proof_json=None,
-                          request_break_glass=False)
 ```
 
-It **never restarts anything**: `apply_supported` is always `false` and
+Read-only, dry-run, and it **never restarts anything**. `apply_supported` is
-`restart_performed` is always `false`.
+always `false`; passing `dry_run=False` performs no restart and reports that
-
+apply is gated by a drain proof (a separate child).
-### Dry-run versus apply
-
-| Call | Behavior |
-|------|----------|
-| `dry_run=True` (default) | Read-only impact preview. No drain proof is required or consulted. |
-| `dry_run=False` | The #661 drain-proof hard gate runs **in this tool**. The outcome is reported under `apply_gate` / `apply_authorized`; a denial also returns a durable `incident` descriptor. Still no restart. |
-
-### Authorization ordering
-
-An apply requires **both** authorizations, and they are independent:
-
-1. **Restart-class authorization** (#663) — the requester's role and permissions
-   must allow the requested class, the class's approval requirement must be
-   satisfied, and any target-scoped class must name its target. Failing any of
-   these makes `allow_restart` `false`.
-2. **Drain-proof gate** (#661) — a valid, unexpired, clean proof bound to the
-   current impact fingerprint, or an authorized break-glass.
-
-`apply_authorized` is the conjunction: `gate.allow and allow_restart`. A clean
-drain proof therefore cannot override a class or requester-role denial, and a
-denied class never reports an authorized apply. `apply_gate` carries
-`drain_gate_allow` and `restart_class_authorized` so a denial is attributable to
-the authorization that produced it.
-
-### Break-glass
-
-Break-glass bypasses the **drain proof only** — never the restart-class matrix.
-It is honoured solely when `request_break_glass` is set *and* the environment
-carries `GITEA_BREAKGLASS_RESTART_AUTHORIZATION`; like operator override, the
-tool argument expresses caller intent and cannot be self-asserted by a worker
-session. `break_glass_requested` and `break_glass_authorized` are both reported,
-so a bypass is never silent.
-
-### Fail closed on apply
-
-A missing, malformed, expired, unclean, tampered, or fingerprint-stale drain
-proof denies the apply and returns an `incident` descriptor. An unknown restart
-class denies before any of this. Ambiguity always denies.
 
 ## Audit
 
 Every evaluation carries an `audit_record` (event, coordinator version, verdict,
-restart class, required permission, allow decision, blast radius, counts,
+allow decision, blast radius, counts, timestamp) so restart decisions are
-timestamp) so restart decisions are
 auditable. No secrets flow through the coordinator — session ids, pids, and
 profiles are operational metadata only.
 

gitea_mcp_server.py

@@ -9552,15 +9552,13 @@ def gitea_edit_pr(
     if closing:
         gate_reasons = _profile_operation_gate("gitea.pr.close")
         if gate_reasons:
-            return {
+            return _build_operation_gate_refusal(
-                "success": False,
+                "gitea.pr.close",
-                "performed": False,
+                gate_reasons,
-                "pr_number": pr_number,
+                pr_number=pr_number,
-                "requested_state": "closed",
+                requested_state="closed",
-                "required_permission": "gitea.pr.close",
+                required_permission="gitea.pr.close",
-                "reasons": gate_reasons,
+            )
-                "permission_report": _permission_block_report("gitea.pr.close"),
-            }
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -13823,13 +13821,17 @@ def gitea_view_issue(
 
 def _permission_block_report(required_operation: str,
                              identity: str | None = None) -> dict:
-    """Structured, LLM-safe explanation of a permission denial (#142).
+    """Structured, LLM-safe explanation of a permission denial (#142, #897).
 
     Built only after a gate has already refused; it adds guidance to the
     refusal and never widens any permission, performs network I/O, or
     raises (fail-soft: degrades to a minimal fail-closed report). Names
     configured profiles only — never auth references, tokens, endpoint
     URLs, or keychain IDs.
+
+    #897: never fabricate a missing permission when the active profile
+    already allows the operation. That path is a diagnostic defect (the
+    refusal was not a permission denial), not a cue to switch profiles.
     """
     report = {
         "requested_operation": required_operation,
@@ -13841,6 +13843,7 @@ def _permission_block_report(required_operation: str,
         "matching_configured_profiles": [],
         "runtime_switching_supported": False,
         "different_mcp_namespace_required": True,
+        "diagnostic_defect": False,
         "exact_safe_next_action": (
             "Ask the operator to fix GITEA_MCP_CONFIG/GITEA_MCP_PROFILE; "
             "the active profile could not be resolved (fail closed)."),
@@ -13855,6 +13858,32 @@ def _permission_block_report(required_operation: str,
     report["active_allowed_operations"] = (
         profile.get("allowed_operations") or [])
 
+    # #897: fail closed as a diagnostic defect when the active profile
+    # already holds the operation — callers must not invent "missing".
+    try:
+        holds, _hold_reason = gitea_config.check_operation(
+            required_operation,
+            profile.get("allowed_operations") or [],
+            profile.get("forbidden_operations") or [],
+        )
+    except Exception:
+        holds = False
+    if holds:
+        report["missing_permission"] = None
+        report["required_permission"] = required_operation
+        report["diagnostic_defect"] = True
+        report["different_mcp_namespace_required"] = False
+        report["exact_safe_next_action"] = (
+            "Diagnostic defect: the active profile already allows "
+            f"{required_operation}. This is not a permission denial — "
+            "inspect blocker_kind / reasons (stale-runtime or runtime-mode). "
+            "Do not call gitea_activate_profile or switch MCP sessions."
+        )
+        report["matching_configured_profiles"] = [
+            p for p in [profile.get("profile_name")] if p
+        ]
+        return report
+
     matching = []
     try:
         config = gitea_config.load_config() or {}
@@ -13903,6 +13932,205 @@ def _permission_block_report(required_operation: str,
     return report
 
 
+def _reason_is_stale_runtime(reason: str) -> bool:
+    """True when *reason* is a master-parity / stale-daemon refusal (#897)."""
+    r = (reason or "").lower()
+    if not r:
+        return False
+    if "stale relative to live master" in r:
+        return True
+    if "server code is stale" in r:
+        return True
+    if "daemon is stale" in r:
+        return True
+    if "started at commit" in r and "workspace master is now" in r:
+        return True
+    if "mcp server started at" in r and "stale" in r:
+        return True
+    if "restart the server to load the current capability gates" in r:
+        return True
+    if "restart/reconnect before mutating" in r:
+        return True
+    return False
+
+
+def _reason_is_runtime_mode(reason: str) -> bool:
+    """True when *reason* is a stable-control / runtime-mode refusal (#897)."""
+    r = (reason or "").lower()
+    if not r:
+        return False
+    if _reason_is_stale_runtime(reason):
+        return False
+    if "runtime mode could not be assessed" in r:
+        return True
+    if "runtime mode is" in r:
+        return True
+    if "stable control runtime" in r:
+        return True
+    if "dev-test" in r and ("runtime" in r or "production" in r):
+        return True
+    if "development worktree" in r or "dev worktree" in r:
+        return True
+    if "launched from a 'branches/" in r or "launched from a \"branches/" in r:
+        return True
+    if "process-root / active-workspace alignment" in r:
+        return True
+    if "namespace" in r and "reproof" in r:
+        return True
+    return False
+
+
+def _reason_is_permission(reason: str) -> bool:
+    """True when *reason* is a genuine profile-permission denial (#897)."""
+    r = (reason or "").lower()
+    if not r:
+        return False
+    if _reason_is_stale_runtime(reason) or _reason_is_runtime_mode(reason):
+        return False
+    if "profile could not be resolved" in r:
+        return True
+    if "profile has no configured allowed operations" in r:
+        return True
+    if "profile forbids" in r:
+        return True
+    if "profile is not allowed to" in r:
+        return True
+    if "unrecognized forbidden operation" in r:
+        return True
+    return False
+
+
+def _classify_operation_gate_reasons(reasons: list[str]) -> dict:
+    """Partition gate reasons into stale / runtime-mode / permission (#897)."""
+    stale: list[str] = []
+    runtime_mode: list[str] = []
+    permission: list[str] = []
+    other: list[str] = []
+    for reason in reasons or []:
+        if _reason_is_stale_runtime(reason):
+            stale.append(reason)
+        elif _reason_is_runtime_mode(reason):
+            runtime_mode.append(reason)
+        elif _reason_is_permission(reason):
+            permission.append(reason)
+        else:
+            other.append(reason)
+    return {
+        "stale_runtime": stale,
+        "runtime_mode": runtime_mode,
+        "permission": permission,
+        "other": other,
+    }
+
+
+def _stale_runtime_reconnect_action() -> str:
+    """Sanctioned recovery for a stale daemon — reconnect only (#685/#897)."""
+    return (
+        "Reconnect the IDE/client MCP session so the server reloads at the "
+        "current master head. Do not call gitea_activate_profile or switch "
+        "MCP role sessions — profile switching does not clear a stale daemon."
+    )
+
+
+def _build_operation_gate_refusal(
+    required_operation: str,
+    reasons: list[str],
+    **extra_fields,
+) -> dict:
+    """Structured gate refusal with typed blockers (#897).
+
+    Stale-runtime and runtime-mode refusals never attach a
+    ``permission_report`` and never recommend profile switching. True
+    permission denials still get ``permission_report``. When both apply,
+    causes are reported separately under distinct fields.
+    """
+    classified = _classify_operation_gate_reasons(reasons)
+    stale = classified["stale_runtime"]
+    runtime_mode = classified["runtime_mode"]
+    permission = classified["permission"]
+    other = classified["other"]
+
+    blocked: dict = {
+        "success": False,
+        "performed": False,
+        "reasons": list(reasons),
+        "mutation_performed": False,
+        "session_context_audit": session_ctx.mutation_context_audit_fields(),
+        "gate_reason_classes": {
+            "stale_runtime": list(stale),
+            "runtime_mode": list(runtime_mode),
+            "permission": list(permission),
+            "other": list(other),
+        },
+    }
+
+    if stale:
+        parity = _current_master_parity()
+        blocked["blocker_kind"] = "runtime_reconnect_required"
+        blocked["restart_required"] = True
+        blocked["stop_required"] = True
+        blocked["startup_head"] = parity.get("startup_head")
+        blocked["current_head"] = parity.get("current_head")
+        blocked["daemon_start_head"] = (
+            parity.get("daemon_start_head") or parity.get("startup_head")
+        )
+        blocked["local_head"] = (
+            parity.get("local_head") or parity.get("current_head")
+        )
+        blocked["live_remote_head"] = parity.get("live_remote_head")
+        blocked["live_stale"] = bool(parity.get("live_stale"))
+        blocked["live_known"] = bool(parity.get("live_known"))
+        blocked["exact_safe_next_action"] = _stale_runtime_reconnect_action()
+        if permission or other:
+            blocked["permission_block_reasons"] = list(permission) + list(other)
+            blocked["stale_runtime_reasons"] = list(stale)
+        # Never attach permission_report for a staleness refusal.
+        blocked.update(extra_fields)
+        return blocked
+
+    if runtime_mode:
+        blocked["blocker_kind"] = "runtime_mode_blocked"
+        blocked["restart_required"] = False
+        blocked["stop_required"] = True
+        blocked["exact_safe_next_action"] = (
+            "Real workflow mutations run only on the promoted stable control "
+            "runtime. Promote/reload the stable runtime; do not call "
+            "gitea_activate_profile or switch MCP role sessions to clear a "
+            "runtime-mode block."
+        )
+        if permission or other:
+            blocked["permission_block_reasons"] = list(permission) + list(other)
+            blocked["runtime_mode_reasons"] = list(runtime_mode)
+        blocked.update(extra_fields)
+        return blocked
+
+    # Pure permission (or unclassified-as-permission) denial.
+    blocked["blocker_kind"] = "permission_denied"
+    blocked["permission_report"] = _permission_block_report(required_operation)
+    blocked.update(extra_fields)
+    return blocked
+
+
+def _permission_report_for_gate_reasons(
+    required_operation: str,
+    reasons: list[str] | None,
+) -> dict | None:
+    """Attach ``permission_report`` only for true permission denials (#897).
+
+    Call sites that historically always attached a permission report after
+    ``_profile_operation_gate`` should use this so stale/runtime refusals
+    do not emit a fabricated missing-permission payload.
+    """
+    if not reasons:
+        return None
+    classified = _classify_operation_gate_reasons(reasons)
+    if classified["stale_runtime"] or classified["runtime_mode"]:
+        return None
+    if not (classified["permission"] or classified["other"]):
+        return None
+    return _permission_block_report(required_operation)
+
+
 def _role_for_operation(op: str) -> str | None:
     # Normalize op first
     try:
@@ -14079,7 +14307,7 @@ def _master_parity_block(op: str) -> list[str]:
 
 
 def _profile_operation_gate(op: str) -> list[str]:
-    """Profile permission check for a single gated operation (#126, #216, #420).
+    """Profile permission check for a single gated operation (#126, #216, #420, #897).
 
     Issue discussion comments are gated separately from the gitea.pr.*
     review/merge family: listing requires ``gitea.read``, creating requires
@@ -14092,21 +14320,26 @@ def _profile_operation_gate(op: str) -> list[str]:
     capability gate that has since been merged, and when the runtime itself is
     not the promoted stable control runtime (#615) -- a dev/test or unknown
     runtime holds production credentials but has not been promoted.
+
+    #897: collect *all* independent refusal classes (stale, runtime-mode,
+    permission) rather than short-circuiting after the first. Callers that
+    only need a boolean still treat any non-empty list as blocked; typed
+    consumers (``_build_operation_gate_refusal``) can separate causes.
     """
-    stale_reasons = _master_parity_block(op)
+    reasons: list[str] = []
-    if stale_reasons:
+    reasons.extend(_master_parity_block(op))
-        return stale_reasons
+    reasons.extend(_runtime_mode_block(op))
-    runtime_reasons = _runtime_mode_block(op)
-    if runtime_reasons:
-        return runtime_reasons
     try:
         profile = get_profile()
     except Exception as exc:
-        return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
+        reasons.append(
+            f"profile could not be resolved (fail closed): {_redact(str(exc))}"
+        )
+        return reasons
     op_ok, op_reason = gitea_config.check_operation(
         op, profile["allowed_operations"], profile["forbidden_operations"])
     if op_ok:
-        return []
+        return reasons
 
     if _try_auto_switch_for_operation(op):
         try:
@@ -14114,17 +14347,26 @@ def _profile_operation_gate(op: str) -> list[str]:
             op_ok, op_reason = gitea_config.check_operation(
                 op, profile["allowed_operations"], profile["forbidden_operations"])
             if op_ok:
-                return []
+                return reasons
         except Exception as exc:
-            return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
+            reasons.append(
+                f"profile could not be resolved (fail closed): {_redact(str(exc))}"
+            )
+            return reasons
 
     if op_reason == "no-allowed-operations":
-        return ["profile has no configured allowed operations (fail closed)"]
+        reasons.append(
-    if op_reason == "forbidden":
+            "profile has no configured allowed operations (fail closed)"
-        return [f"profile forbids '{op}'"]
+        )
-    if op_reason == "invalid-forbidden-entry":
+    elif op_reason == "forbidden":
-        return ["profile has an unrecognized forbidden operation entry (fail closed)"]
+        reasons.append(f"profile forbids '{op}'")
-    return [f"profile is not allowed to {op}"]
+    elif op_reason == "invalid-forbidden-entry":
+        reasons.append(
+            "profile has an unrecognized forbidden operation entry (fail closed)"
+        )
+    else:
+        reasons.append(f"profile is not allowed to {op}")
+    return reasons
 
 
 def _mutation_config_authority_block(required_operation: str) -> dict | None:
@@ -14344,10 +14586,14 @@ def _session_context_mutation_block(
 
 
 def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
-    """Structured permission denial for gated tools (#69, #142).
+    """Structured operation-gate denial for gated tools (#69, #142, #897).
 
     Returns a block dict when the active profile forbids *required_operation*,
-    or ``None`` when the gate passes. Never performs network I/O.
+    the daemon is stale, or the runtime mode is not mutation-safe — or
+    ``None`` when the gate passes. Never performs network I/O.
+
+    #897: stale-runtime and runtime-mode refusals are typed
+    (``blocker_kind``) and never carry a ``permission_report``.
     """
     req_role = "reviewer" if any(required_operation.startswith(p) for p in (
         "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review"
@@ -14357,15 +14603,9 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
 
     reasons = _profile_operation_gate(required_operation)
     if reasons:
-        blocked = {
+        return _build_operation_gate_refusal(
-            "success": False,
+            required_operation, reasons, **extra_fields
-            "performed": False,
+        )
-            "reasons": reasons,
-            "permission_report": _permission_block_report(required_operation),
-            "session_context_audit": session_ctx.mutation_context_audit_fields(),
-        }
-        blocked.update(extra_fields)
-        return blocked
 
     auth_block = _mutation_config_authority_block(required_operation)
     if auth_block is not None:
@@ -14494,20 +14734,14 @@ def gitea_acquire_reviewer_pr_lease(
     """Acquire a per-PR reviewer lease before review/merge mutations (#407)."""
     read_block = _profile_operation_gate("gitea.read")
     if read_block:
-        return {
+        return _build_operation_gate_refusal(
-            "success": False,
+            "gitea.read", read_block, acquired=False
-            "acquired": False,
+        )
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
     comment_block = _profile_operation_gate("gitea.pr.comment")
     if comment_block:
-        return {
+        return _build_operation_gate_refusal(
-            "success": False,
+            "gitea.pr.comment", comment_block, acquired=False
-            "acquired": False,
+        )
-            "reasons": comment_block,
-            "permission_report": _permission_block_report("gitea.pr.comment"),
-        }
 
     # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
     # anti-stomp for the declared lease-acquire mutation inventory entry.
@@ -14623,20 +14857,14 @@ def gitea_acquire_merger_pr_lease(
     """
     read_block = _profile_operation_gate("gitea.read")
     if read_block:
-        return {
+        return _build_operation_gate_refusal(
-            "success": False,
+            "gitea.read", read_block, acquired=False
-            "acquired": False,
+        )
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
     comment_block = _profile_operation_gate("gitea.pr.comment")
     if comment_block:
-        return {
+        return _build_operation_gate_refusal(
-            "success": False,
+            "gitea.pr.comment", comment_block, acquired=False
-            "acquired": False,
+        )
-            "reasons": comment_block,
-            "permission_report": _permission_block_report("gitea.pr.comment"),
-        }
     merge_block = _profile_operation_gate("gitea.pr.merge")
     if merge_block:
         return {
@@ -19375,14 +19603,12 @@ def gitea_update_pr_branch_by_merge(
     # Permission: author branch push / PR mutation surface.
     push_block = _profile_operation_gate("gitea.branch.push")
     if push_block:
-        return {
+        return _build_operation_gate_refusal(
-            "success": False,
+            "gitea.branch.push",
-            "performed": False,
+            push_block,
-            "mutation_allowed": False,
+            mutation_allowed=False,
-            "reasons": push_block,
+            role_kind=role,
-            "permission_report": _permission_block_report("gitea.branch.push"),
+        )
-            "role_kind": role,
-        }
 
     if role != "author":
         pre = pr_sync_status.assess_update_pr_branch_preflight(
@@ -22343,17 +22569,12 @@ def gitea_request_mcp_restart(
     request_override: bool = False,
     session_id: str | None = None,
     limit: int = 200,
-    restart_class: str = "full_mcp_restart",
-    target_session_id: str | None = None,
-    target_role: str | None = None,
-    target_connector: str | None = None,
     drain_proof_json: str | None = None,
     request_break_glass: bool = False,
 ) -> dict:
     """Evaluate a proposed MCP restart and return an impact preview (#658).
 
-    Central restart coordinator: resolves the requested restart class, gathers
+    Central restart coordinator: gathers live control-plane state (sessions,
-    live control-plane state (sessions,
     leases/locks, in-flight issue/PR work, mutations, worktrees) and returns a
     blast-radius impact report with a ``safe`` / ``unsafe`` / ``override``
     verdict, so the console (#642/#652) and operators can see what a restart
@@ -22367,16 +22588,7 @@ def gitea_request_mcp_restart(
     only when ``request_break_glass`` is set *and* the environment carries
     ``GITEA_BREAKGLASS_RESTART_AUTHORIZATION``. Even an authorized gate performs
     no restart here; actual execution is a further child. The gate outcome is
-    reported under ``apply_gate``.
+    reported under ``apply_gate`` / ``apply_authorized``.
-
-    ``apply_authorized`` requires **both** authorizations to pass: the #661 drain
-    gate *and* the #663 restart-class matrix (``allow_restart``). They are
-    independent — the drain gate proves the blast radius was drained and knows
-    nothing about whether this requester may request this class — so a class the
-    matrix denied never reports an authorized apply. Break-glass bypasses the
-    drain proof only; it never bypasses the class matrix. ``apply_gate`` carries
-    ``drain_gate_allow`` and ``restart_class_authorized`` so a denial is
-    attributable to the authorization that produced it.
 
     Operator override authority is read from the process environment
     (``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by
@@ -22461,9 +22673,6 @@ def gitea_request_mcp_restart(
 
     profile = get_profile()
     profile_name = (profile.get("profile_name") or "").strip() or "session"
-    requester_role = (
-        profile.get("role_kind") or profile.get("role") or ""
-    ).strip().lower()
     sid = (session_id or "").strip() or f"{profile_name}-{os.getpid()}"
 
     # Override authority is read from the environment only — a worker session
@@ -22473,15 +22682,6 @@ def gitea_request_mcp_restart(
         (os.environ.get("GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION") or "").strip()
     )
     operator_override = bool(request_override and operator_authorized)
-    controller_approved = bool(
-        (
-            os.environ.get("GITEA_CONTROLLER_RESTART_APPROVAL_AUTHORIZATION")
-            or ""
-        ).strip()
-    )
-    requester_permissions = restart_coordinator.permissions_for_role(
-        requester_role
-    )
 
     inventory = {
         "sessions": sessions,
@@ -22496,14 +22696,6 @@ def gitea_request_mcp_restart(
         operator_override=operator_override,
         requesting_session_id=sid,
         dry_run=True,  # coordinator is always analysis-only (#658)
-        restart_class=restart_class,
-        requester_role=requester_role,
-        requester_permissions=requester_permissions,
-        controller_approved=controller_approved,
-        operator_authorized=operator_authorized,
-        target_session_id=target_session_id,
-        target_role=target_role,
-        target_connector=target_connector,
     )
 
     payload = report.as_dict()
@@ -22515,9 +22707,6 @@ def gitea_request_mcp_restart(
     payload["requesting_session_id"] = sid
     payload["operator_override_requested"] = bool(request_override)
     payload["operator_override_authorized"] = operator_authorized
-    payload["controller_approval_authorized"] = controller_approved
-    payload["requester_role"] = requester_role
-    payload["requester_permissions"] = list(requester_permissions)
     # Actual restart execution remains a further child; this tool never restarts
     # a process. What #661 adds is the *hard gate*: an apply request (dry_run
     # False) must present a valid, unexpired, clean drain proof, or it is denied
@@ -22555,25 +22744,8 @@ def gitea_request_mcp_restart(
             gate_payload["reasons"] = [proof_parse_error] + list(
                 gate_payload.get("reasons") or []
             )
-        # The #663 restart-class matrix and the #661 drain gate are two
-        # independent authorizations, and an apply requires BOTH. ``gate.allow``
-        # proves only that the blast radius was drained — or that break-glass
-        # was authorized — and knows nothing about whether this requester may
-        # request this class at all. Conjoining them keeps a class the matrix
-        # denied from ever reporting an authorized apply, and keeps break-glass
-        # scoped to what it is for: bypassing the drain proof, never the
-        # least-privilege class matrix.
-        restart_class_authorized = bool(report.allow_restart)
-        gate_payload["drain_gate_allow"] = bool(gate.allow)
-        gate_payload["restart_class_authorized"] = restart_class_authorized
-        if not restart_class_authorized:
-            gate_payload["reasons"] = list(gate_payload.get("reasons") or []) + [
-                "restart class authorization denied; apply denied regardless of "
-                "drain proof or break-glass (fail closed, #663)",
-                *(report.authorization_reasons or []),
-            ]
         payload["apply_gate"] = gate_payload
-        payload["apply_authorized"] = bool(gate.allow and restart_class_authorized)
+        payload["apply_authorized"] = gate.allow
         payload["break_glass_requested"] = bool(request_break_glass)
         payload["break_glass_authorized"] = break_glass_authorized
         # Even an authorized gate performs no restart here: execution is a later

restart_coordinator.py

@@ -28,12 +28,11 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
-from enum import Enum
 from typing import Any, Mapping, Sequence
 
 import lease_lifecycle
 
-COORDINATOR_VERSION = "1.1.0-issue-663"
+COORDINATOR_VERSION = "1.0.0-issue-658"
 
 # Restart verdicts. Exactly the three the acceptance criteria name.
 VERDICT_SAFE = "safe"
@@ -55,194 +54,6 @@ LEASE_FRESHNESS_LIVE = "active"
 DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS = 900
 
 
-class RestartClass(str, Enum):
-    """The only restart/recovery classes accepted by the coordinator."""
-
-    CLIENT_RECONNECT = "client_reconnect"
-    SESSION_RECONNECT = "session_reconnect"
-    WORKER_RESTART = "worker_restart"
-    ROLE_RUNTIME_RESTART = "role_runtime_restart"
-    CONNECTOR_RESTART = "connector_restart"
-    CONFIGURATION_RELOAD = "configuration_reload"
-    ROLLING_MCP_RESTART = "rolling_mcp_restart"
-    FULL_MCP_RESTART = "full_mcp_restart"
-    HOST_RESTART = "host_restart"
-
-
-@dataclass(frozen=True)
-class RestartClassPolicy:
-    """Least-privilege policy for one :class:`RestartClass`."""
-
-    restart_class: RestartClass
-    required_permission: str
-    expected_blast_radius: str
-    drain_requirement: str
-    full_drain_required: bool
-    approval_requirement: str
-    audit_requirement: str
-    recovery_behavior: str
-    request_roles: tuple[str, ...]
-    execution_roles: tuple[str, ...]
-
-    def as_dict(self) -> dict[str, Any]:
-        return {
-            "restart_class": self.restart_class.value,
-            "required_permission": self.required_permission,
-            "expected_blast_radius": self.expected_blast_radius,
-            "drain_requirement": self.drain_requirement,
-            "full_drain_required": self.full_drain_required,
-            "approval_requirement": self.approval_requirement,
-            "audit_requirement": self.audit_requirement,
-            "recovery_behavior": self.recovery_behavior,
-            "request_roles": list(self.request_roles),
-            "execution_roles": list(self.execution_roles),
-        }
-
-
-WORKER_ROLES = ("author", "reviewer", "merger", "reconciler")
-CONTROL_ROLES = ("controller", "operator", "admin")
-ALL_REQUEST_ROLES = WORKER_ROLES + CONTROL_ROLES
-
-RESTART_CLASS_POLICIES: dict[RestartClass, RestartClassPolicy] = {
-    RestartClass.CLIENT_RECONNECT: RestartClassPolicy(
-        RestartClass.CLIENT_RECONNECT,
-        "mcp.reconnect.client",
-        BLAST_NONE,
-        "none",
-        False,
-        "self_service",
-        "record class, actor, client namespace, reason, and outcome",
-        "Reconnect only the caller's client transport; no daemon or peer session changes.",
-        ALL_REQUEST_ROLES,
-        ALL_REQUEST_ROLES,
-    ),
-    RestartClass.SESSION_RECONNECT: RestartClassPolicy(
-        RestartClass.SESSION_RECONNECT,
-        "mcp.reconnect.session",
-        BLAST_LOW,
-        "requesting_session_safe_point",
-        False,
-        "self_service",
-        "record class, actor, session id, reason, and outcome",
-        "Rebind identity, capability, and workspace state for one session.",
-        ALL_REQUEST_ROLES,
-        ALL_REQUEST_ROLES,
-    ),
-    RestartClass.WORKER_RESTART: RestartClassPolicy(
-        RestartClass.WORKER_RESTART,
-        "mcp.restart.worker.request",
-        BLAST_LOW,
-        "target_worker",
-        False,
-        "controller_approval_and_automated_gates",
-        "record class, actor, target worker, approval, drain proof, and outcome",
-        "Restart one worker after its own lease and mutation scope is drained.",
-        ALL_REQUEST_ROLES,
-        ("operator", "admin"),
-    ),
-    RestartClass.ROLE_RUNTIME_RESTART: RestartClassPolicy(
-        RestartClass.ROLE_RUNTIME_RESTART,
-        "mcp.restart.role_runtime.request",
-        BLAST_MEDIUM,
-        "target_role_runtime",
-        False,
-        "controller_approval_and_automated_gates",
-        "record class, actor, role namespace, approval, drain proof, and outcome",
-        "Restart only the selected role runtime and then re-probe that namespace.",
-        ALL_REQUEST_ROLES,
-        ("operator", "admin"),
-    ),
-    RestartClass.CONNECTOR_RESTART: RestartClassPolicy(
-        RestartClass.CONNECTOR_RESTART,
-        "mcp.restart.connector.request",
-        BLAST_MEDIUM,
-        "target_connector",
-        False,
-        "controller_approval_and_automated_gates",
-        "record class, actor, connector id, approval, drain proof, and outcome",
-        "Restart one connector while unrelated role runtimes remain available.",
-        ALL_REQUEST_ROLES,
-        ("operator", "admin"),
-    ),
-    RestartClass.CONFIGURATION_RELOAD: RestartClassPolicy(
-        RestartClass.CONFIGURATION_RELOAD,
-        "mcp.reload.configuration.request",
-        BLAST_LOW,
-        "mutation_quiesce",
-        False,
-        "controller_approval_and_automated_gates",
-        "record class, actor, configuration revision, approval, and outcome",
-        "Gracefully reload configuration without replacing the daemon process.",
-        ALL_REQUEST_ROLES,
-        ("operator", "admin"),
-    ),
-    RestartClass.ROLLING_MCP_RESTART: RestartClassPolicy(
-        RestartClass.ROLLING_MCP_RESTART,
-        "mcp.restart.rolling.request",
-        BLAST_MEDIUM,
-        "one_instance_at_a_time",
-        False,
-        "controller_approval_and_automated_gates",
-        "record class, actor, instance order, approval, per-instance drains, and outcome",
-        "Drain, restart, verify, and restore one instance before advancing to the next.",
-        CONTROL_ROLES,
-        ("operator", "admin"),
-    ),
-    RestartClass.FULL_MCP_RESTART: RestartClassPolicy(
-        RestartClass.FULL_MCP_RESTART,
-        "mcp.restart.full.request",
-        BLAST_HIGH,
-        "all_sessions_and_mutations",
-        True,
-        "controller_approval_and_automated_gates",
-        "record class, actor, full impact report, approval, drain proof, and outcome",
-        "Stop and restore the complete MCP runtime only after a verified full drain.",
-        CONTROL_ROLES,
-        ("operator", "admin"),
-    ),
-    RestartClass.HOST_RESTART: RestartClassPolicy(
-        RestartClass.HOST_RESTART,
-        "mcp.restart.host.request",
-        BLAST_HIGH,
-        "all_host_work",
-        True,
-        "controller_approval_plus_infrastructure_operator",
-        "record class, actor, host, incident or change id, approval, drain proof, and outcome",
-        "Hand off to infrastructure ownership; reconcile every runtime after the host returns.",
-        ("controller", "operator", "admin"),
-        ("operator", "admin"),
-    ),
-}
-
-
-def resolve_restart_class(value: RestartClass | str) -> RestartClass:
-    """Resolve a restart class or fail closed for an unknown value."""
-
-    if isinstance(value, RestartClass):
-        return value
-    try:
-        return RestartClass(str(value).strip())
-    except ValueError as exc:
-        raise ValueError(f"unknown restart class {value!r}; deny (fail closed)") from exc
-
-
-def restart_class_policy(value: RestartClass | str) -> RestartClassPolicy:
-    """Return the canonical policy for *value*."""
-
-    return RESTART_CLASS_POLICIES[resolve_restart_class(value)]
-
-
-def permissions_for_role(role: str | None) -> tuple[str, ...]:
-    """Return request permissions granted to a workflow role by this policy."""
-
-    normalized = str(role or "").strip().lower()
-    return tuple(
-        policy.required_permission
-        for policy in RESTART_CLASS_POLICIES.values()
-        if normalized in policy.request_roles
-    )
-
-
 def _utc_now() -> datetime:
     return datetime.now(timezone.utc)
 
@@ -264,7 +75,6 @@ class SessionImpact:
     heartbeat_stale: bool
     is_requester: bool
     live: bool
-    connector: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -277,7 +87,6 @@ class SessionImpact:
             "heartbeat_stale": self.heartbeat_stale,
             "is_requester": self.is_requester,
             "live": self.live,
-            "connector": self.connector,
         }
 
 
@@ -296,7 +105,6 @@ class LeaseImpact:
     disruptive: bool
     is_mutation: bool
     is_critical_section: bool
-    connector: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -311,7 +119,6 @@ class LeaseImpact:
             "disruptive": self.disruptive,
             "is_mutation": self.is_mutation,
             "is_critical_section": self.is_critical_section,
-            "connector": self.connector,
         }
 
 
@@ -320,13 +127,6 @@ class RestartImpactReport:
     """Impact preview DTO returned to the console / operator (#642/#652)."""
 
     coordinator_version: str
-    restart_class: str
-    restart_policy: dict[str, Any]
-    policy_enforced: bool
-    permission_authorized: bool
-    role_authorized: bool
-    approval_satisfied: bool
-    authorization_reasons: list[str]
     evaluated_at: str
     dry_run: bool
     restart_performed: bool
@@ -353,13 +153,6 @@ class RestartImpactReport:
     def as_dict(self) -> dict[str, Any]:
         return {
             "coordinator_version": self.coordinator_version,
-            "restart_class": self.restart_class,
-            "restart_policy": dict(self.restart_policy),
-            "policy_enforced": self.policy_enforced,
-            "permission_authorized": self.permission_authorized,
-            "role_authorized": self.role_authorized,
-            "approval_satisfied": self.approval_satisfied,
-            "authorization_reasons": list(self.authorization_reasons),
             "evaluated_at": self.evaluated_at,
             "dry_run": self.dry_run,
             "restart_performed": self.restart_performed,
@@ -413,7 +206,6 @@ def _classify_session(
             requesting_session_id and session_id == requesting_session_id
         ),
         live=live,
-        connector=(str(row.get("connector") or "").strip() or None),
     )
 
 
@@ -466,7 +258,6 @@ def _classify_lease(row: Mapping[str, Any]) -> LeaseImpact:
         disruptive=disruptive,
         is_mutation=is_mutation,
         is_critical_section=disruptive,
-        connector=(str(row.get("connector") or "").strip() or None),
     )
 
 
@@ -488,14 +279,6 @@ def evaluate_restart_impact(
     requesting_session_id: str | None = None,
     dry_run: bool = True,
     session_heartbeat_stale_seconds: int = DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS,
-    restart_class: RestartClass | str | None = None,
-    requester_role: str | None = None,
-    requester_permissions: Sequence[str] | None = None,
-    controller_approved: bool = False,
-    operator_authorized: bool = False,
-    target_session_id: str | None = None,
-    target_role: str | None = None,
-    target_connector: str | None = None,
 ) -> RestartImpactReport:
     """Evaluate a proposed MCP restart and return an impact preview.
 
@@ -518,61 +301,6 @@ def evaluate_restart_impact(
     """
     moment = now or _utc_now()
     reasons: list[str] = []
-    authorization_reasons: list[str] = []
-
-    # ``None`` preserves the pre-#663 impact-only API for callers that have not
-    # yet been migrated. All MCP requests pass an explicit class and therefore
-    # take the fail-closed policy path.
-    policy_enforced = restart_class is not None
-    try:
-        resolved_class = resolve_restart_class(
-            restart_class or RestartClass.FULL_MCP_RESTART
-        )
-        policy = RESTART_CLASS_POLICIES[resolved_class]
-        unknown_class = False
-    except ValueError as exc:
-        resolved_class = None
-        policy = None
-        unknown_class = True
-        authorization_reasons.append(str(exc))
-
-    normalized_role = str(requester_role or "").strip().lower()
-    granted = {str(p).strip() for p in (requester_permissions or ())}
-    if policy_enforced and policy is not None:
-        permission_authorized = policy.required_permission in granted
-        role_authorized = normalized_role in policy.request_roles
-        if not permission_authorized:
-            authorization_reasons.append(
-                f"missing required permission {policy.required_permission!r}"
-            )
-        if not role_authorized:
-            authorization_reasons.append(
-                f"role {normalized_role or 'unknown'!r} may not request "
-                f"{policy.restart_class.value}"
-            )
-    elif unknown_class:
-        permission_authorized = False
-        role_authorized = False
-    else:
-        permission_authorized = True
-        role_authorized = True
-
-    if policy_enforced and policy is not None:
-        approval = policy.approval_requirement
-        if approval == "self_service":
-            approval_satisfied = True
-        elif approval == "controller_approval_plus_infrastructure_operator":
-            approval_satisfied = bool(controller_approved and operator_authorized)
-        else:
-            approval_satisfied = bool(controller_approved)
-        if not approval_satisfied:
-            authorization_reasons.append(
-                f"approval requirement not satisfied: {approval}"
-            )
-    elif unknown_class:
-        approval_satisfied = False
-    else:
-        approval_satisfied = True
 
     inventory_complete = bool(inventory.get("inventory_complete", False))
     incomplete_reasons = [str(r) for r in (inventory.get("incomplete_reasons") or [])]
@@ -595,67 +323,15 @@ def evaluate_restart_impact(
     ]
     lease_impacts = [_classify_lease(l) for l in leases_raw]
 
-    # Route impact through the selected class. Narrow classes never inherit a
+    # Only *other* live sessions and live leases constitute blast radius: a
-    # full-runtime drain merely because unrelated work exists.
+    # restart that would kill only the requesting session with no other work in
-    target_complete = True
+    # flight is safe.
-    if resolved_class in {
-        RestartClass.CLIENT_RECONNECT,
-        RestartClass.SESSION_RECONNECT,
-        RestartClass.CONFIGURATION_RELOAD,
-    }:
-        scoped_sessions: list[SessionImpact] = []
-        scoped_leases: list[LeaseImpact] = []
-    elif resolved_class == RestartClass.WORKER_RESTART:
-        selected_session = (target_session_id or "").strip()
-        target_complete = bool(selected_session)
-        scoped_sessions = [
-            s for s in session_impacts if s.session_id == selected_session
-        ]
-        scoped_leases = [
-            l for l in lease_impacts if l.session_id == selected_session
-        ]
-    elif resolved_class == RestartClass.ROLE_RUNTIME_RESTART:
-        selected_role = (target_role or "").strip().lower()
-        target_complete = bool(selected_role)
-        scoped_sessions = [
-            s for s in session_impacts if str(s.role or "").lower() == selected_role
-        ]
-        scoped_leases = [
-            l for l in lease_impacts if str(l.role or "").lower() == selected_role
-        ]
-    elif resolved_class == RestartClass.CONNECTOR_RESTART:
-        selected_connector = (target_connector or "").strip()
-        target_complete = bool(selected_connector)
-        scoped_sessions = [
-            s for s in session_impacts if s.connector == selected_connector
-        ]
-        scoped_leases = [
-            l for l in lease_impacts if l.connector == selected_connector
-        ]
-    else:
-        scoped_sessions = list(session_impacts)
-        scoped_leases = list(lease_impacts)
-
-    if policy_enforced and not target_complete:
-        authorization_reasons.append(
-            f"target required for {resolved_class.value if resolved_class else 'unknown class'}"
-        )
-
     other_live_sessions = [
-        s for s in scoped_sessions if s.live and not s.is_requester
+        s for s in session_impacts if s.live and not s.is_requester
     ]
-    disruptive_leases = [l for l in scoped_leases if l.disruptive]
+    disruptive_leases = [l for l in lease_impacts if l.disruptive]
-    critical_sections = [l for l in scoped_leases if l.is_critical_section]
+    critical_sections = [l for l in lease_impacts if l.is_critical_section]
-    mutations = [l for l in scoped_leases if l.is_mutation]
+    mutations = [l for l in lease_impacts if l.is_mutation]
-    terminal_lock_in_scope = (
-        terminal_lock
-        if resolved_class
-        not in {
-            RestartClass.CLIENT_RECONNECT,
-            RestartClass.SESSION_RECONNECT,
-        }
-        else None
-    )
 
     affected_issues = sorted(
         {
@@ -672,24 +348,9 @@ def evaluate_restart_impact(
         }
     )
 
-    disruptive = bool(
+    disruptive = bool(disruptive_leases or other_live_sessions or terminal_lock)
-        disruptive_leases or other_live_sessions or terminal_lock_in_scope
-    )
 
-    authorization_ok = bool(
+    if not inventory_complete:
-        not unknown_class
-        and permission_authorized
-        and role_authorized
-        and approval_satisfied
-        and target_complete
-    )
-
-    if policy_enforced and not authorization_ok:
-        verdict = VERDICT_UNSAFE
-        allow_restart = False
-        reasons.append("restart class authorization denied (fail closed)")
-        reasons.extend(authorization_reasons)
-    elif not inventory_complete:
         verdict = VERDICT_UNSAFE
         allow_restart = False
         reasons.append(
@@ -720,7 +381,7 @@ def evaluate_restart_impact(
             f"{len(critical_sections)} critical section(s) in flight "
             "(active lease with a live owner)"
         )
-    if terminal_lock_in_scope:
+    if terminal_lock:
         reasons.append("active terminal (merge) lock present")
 
     override_would_allow = bool(inventory_complete and disruptive)
@@ -750,12 +411,6 @@ def evaluate_restart_impact(
     audit_record = {
         "event": "restart_impact_evaluated",
         "coordinator_version": COORDINATOR_VERSION,
-        "restart_class": (
-            resolved_class.value if resolved_class else str(restart_class or "")
-        ),
-        "required_permission": (
-            policy.required_permission if policy is not None else None
-        ),
         "evaluated_at": moment.isoformat(),
         "dry_run": dry_run,
         "operator_override": bool(operator_override),
@@ -769,15 +424,6 @@ def evaluate_restart_impact(
 
     return RestartImpactReport(
         coordinator_version=COORDINATOR_VERSION,
-        restart_class=(
-            resolved_class.value if resolved_class else str(restart_class or "")
-        ),
-        restart_policy=policy.as_dict() if policy is not None else {},
-        policy_enforced=policy_enforced,
-        permission_authorized=permission_authorized,
-        role_authorized=role_authorized,
-        approval_satisfied=approval_satisfied,
-        authorization_reasons=authorization_reasons,
         evaluated_at=moment.isoformat(),
         dry_run=dry_run,
         restart_performed=False,
@@ -794,11 +440,9 @@ def evaluate_restart_impact(
         affected_issues=affected_issues,
         affected_prs=affected_prs,
         mutations=mutations,
-        terminal_lock=(
+        terminal_lock=dict(terminal_lock)
-            dict(terminal_lock_in_scope)
+        if isinstance(terminal_lock, Mapping)
-            if isinstance(terminal_lock_in_scope, Mapping)
+        else terminal_lock,
-            else terminal_lock_in_scope
-        ),
         ack_state=ack_state,
         prior_recovery_attempts=prior_recovery_attempts,
         counts=counts,

tests/test_issue_886_apply_authorization_conjunction.py

@@ -1,381 +0,0 @@
-"""``apply_authorized`` requires BOTH authorizations (#886 review blocker B1).
-
-The #663 restart-class matrix and the #661 drain-proof hard gate are independent
-authorizations that first coexisted when PR #882 landed on master and PR #886
-merged it into the restart-class branch. The union preserved both, but the apply
-decision consulted only the drain gate::
-
-    payload["apply_authorized"] = gate.allow          # pre-fix
-
-so a clean drain proof — or an authorized break-glass, which needs no proof at
-all — reported ``apply_authorized: True`` for a restart class the least-privilege
-matrix had just denied, in the same payload that carried
-``allow_restart: False`` and "role 'author' may not request full_mcp_restart".
-
-These tests pin the conjunction and the properties that must survive it. They
-exercise the real MCP tool, which previously had no test coverage at all — that
-absence is why the defect shipped.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import unittest
-from unittest.mock import patch
-
-import drain_proof
-import gitea_mcp_server as srv
-
-CONTROLLER_APPROVAL_ENV = "GITEA_CONTROLLER_RESTART_APPROVAL_AUTHORIZATION"
-BREAK_GLASS_ENV = "GITEA_BREAKGLASS_RESTART_AUTHORIZATION"
-
-# A quiet control plane: nothing live, so the blast radius never masks the
-# authorization outcome under test.
-QUIET_SESSIONS: list[dict] = []
-QUIET_LEASES: list[dict] = []
-
-
-class _FakeDB:
-    """Minimal control-plane DB stand-in for the restart inventory."""
-
-    def __init__(self, sessions=QUIET_SESSIONS, terminal=None):
-        self._sessions = list(sessions)
-        self._terminal = terminal
-
-    def list_sessions(self, statuses=None, limit=None):
-        return list(self._sessions)
-
-    def get_active_terminal_lock(self, remote=None, org=None, repo=None):
-        return self._terminal
-
-
-def _profile(role: str) -> dict:
-    return {"profile_name": f"prgs-{role}", "role_kind": role, "role": role}
-
-
-class _RestartToolHarness(unittest.TestCase):
-    """Drives the real ``gitea_request_mcp_restart`` with a stubbed inventory."""
-
-    def _call(self, *, role: str, env: dict | None = None, **kwargs) -> dict:
-        environ = {k: v for k, v in os.environ.items()
-                   if k not in (CONTROLLER_APPROVAL_ENV, BREAK_GLASS_ENV)}
-        environ.update(env or {})
-        with patch.object(srv, "_profile_operation_gate", return_value=None), \
-                patch.object(srv, "_resolve",
-                             return_value=("gitea.prgs.cc",
-                                           "Scaled-Tech-Consulting",
-                                           "Gitea-Tools")), \
-                patch.object(srv, "get_profile", return_value=_profile(role)), \
-                patch.object(srv, "_control_plane_db_or_error",
-                             return_value=(_FakeDB(), [])), \
-                patch.object(srv.lease_lifecycle, "list_active_leases",
-                             return_value={"leases": list(QUIET_LEASES)}), \
-                patch.dict(os.environ, environ, clear=True):
-            return srv.gitea_request_mcp_restart(
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Gitea-Tools",
-                session_id="probe-session",
-                **kwargs,
-            )
-
-    def _clean_proof_for(self, preview: dict) -> str:
-        """Mint a genuinely clean, signature-valid proof bound to *preview*.
-
-        Built from the tool's own dry-run report, so the fingerprint matches and
-        the proof is rejected for authorization reasons only — never because it
-        was stale or forged.
-        """
-        proof = drain_proof.build_drain_proof(
-            impact_report=preview,
-            drain_state={
-                "assignments_stopped": True,
-                "checkpoints_complete": True,
-                "handoffs_verified": True,
-                "leases_handled": True,
-                "acks": {},
-            },
-            requesting_session_id="probe-session",
-        )
-        self.assertTrue(proof.clean, "harness must mint a clean proof")
-        return json.dumps(proof.as_dict())
-
-
-class TestConjunction(_RestartToolHarness):
-    """AC1/AC2 — the two authorizations are ANDed, in both directions."""
-
-    def test_gate_allow_with_class_denied_yields_apply_authorized_false(self):
-        # An author may not request full_mcp_restart (CONTROL_ROLES only).
-        preview = self._call(role="author", restart_class="full_mcp_restart")
-        self.assertFalse(preview["allow_restart"])
-
-        result = self._call(
-            role="author",
-            restart_class="full_mcp_restart",
-            dry_run=False,
-            drain_proof_json=self._clean_proof_for(preview),
-        )
-
-        self.assertTrue(result["apply_gate"]["drain_gate_allow"],
-                        "drain gate itself should have allowed this proof")
-        self.assertFalse(result["apply_gate"]["restart_class_authorized"])
-        self.assertFalse(result["apply_authorized"],
-                         "a clean proof must not authorize a denied class")
-        self.assertFalse(result["allow_restart"])
-
-    def test_gate_allow_with_class_allowed_can_yield_apply_authorized_true(self):
-        preview = self._call(
-            role="operator",
-            restart_class="full_mcp_restart",
-            env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
-        )
-        self.assertTrue(preview["allow_restart"],
-                        "operator + controller approval must authorize the class")
-
-        result = self._call(
-            role="operator",
-            restart_class="full_mcp_restart",
-            dry_run=False,
-            drain_proof_json=self._clean_proof_for(preview),
-            env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
-        )
-
-        self.assertTrue(result["apply_gate"]["drain_gate_allow"])
-        self.assertTrue(result["apply_gate"]["restart_class_authorized"])
-        self.assertTrue(result["apply_authorized"],
-                        "both authorizations pass; apply must be authorized")
-
-    def test_denial_is_attributable_to_the_authorization_that_caused_it(self):
-        preview = self._call(role="author", restart_class="full_mcp_restart")
-        result = self._call(
-            role="author",
-            restart_class="full_mcp_restart",
-            dry_run=False,
-            drain_proof_json=self._clean_proof_for(preview),
-        )
-        blob = " ".join(result["apply_gate"]["reasons"]).lower()
-        self.assertIn("restart class authorization denied", blob)
-        self.assertIn("full_mcp_restart", blob)
-
-
-class TestProofCannotOverrideAuthorization(_RestartToolHarness):
-    """AC3 — a clean proof never overrides a class or requester-role denial."""
-
-    def test_clean_proof_cannot_override_role_denial(self):
-        for role in ("author", "reviewer", "merger", "reconciler"):
-            with self.subTest(role=role):
-                preview = self._call(role=role, restart_class="full_mcp_restart")
-                result = self._call(
-                    role=role,
-                    restart_class="full_mcp_restart",
-                    dry_run=False,
-                    drain_proof_json=self._clean_proof_for(preview),
-                )
-                self.assertFalse(result["apply_authorized"])
-
-    def test_clean_proof_cannot_override_missing_controller_approval(self):
-        # Correct role, but the class demands controller approval and the
-        # environment carries none.
-        preview = self._call(role="operator", restart_class="full_mcp_restart")
-        self.assertFalse(preview["allow_restart"])
-        result = self._call(
-            role="operator",
-            restart_class="full_mcp_restart",
-            dry_run=False,
-            drain_proof_json=self._clean_proof_for(preview),
-        )
-        self.assertFalse(result["apply_authorized"])
-
-    def test_clean_proof_cannot_override_unknown_class(self):
-        preview = self._call(role="operator", restart_class="not_a_real_class",
-                             env={CONTROLLER_APPROVAL_ENV: "yes"})
-        self.assertFalse(preview["allow_restart"])
-        result = self._call(
-            role="operator",
-            restart_class="not_a_real_class",
-            dry_run=False,
-            drain_proof_json=self._clean_proof_for(preview),
-            env={CONTROLLER_APPROVAL_ENV: "yes"},
-        )
-        self.assertFalse(result["apply_authorized"])
-
-    def test_clean_proof_cannot_override_missing_scope_target(self):
-        # worker_restart without target_session_id fails closed on scoping.
-        preview = self._call(role="operator", restart_class="worker_restart",
-                             env={CONTROLLER_APPROVAL_ENV: "yes"})
-        self.assertFalse(preview["allow_restart"])
-        result = self._call(
-            role="operator",
-            restart_class="worker_restart",
-            dry_run=False,
-            drain_proof_json=self._clean_proof_for(preview),
-            env={CONTROLLER_APPROVAL_ENV: "yes"},
-        )
-        self.assertFalse(result["apply_authorized"])
-
-
-class TestBreakGlassDoesNotCollapseTheMatrix(_RestartToolHarness):
-    """AC4 — break-glass bypasses the drain proof only, never the class matrix."""
-
-    def test_break_glass_does_not_authorize_a_denied_class(self):
-        result = self._call(
-            role="author",
-            restart_class="host_restart",
-            dry_run=False,
-            request_break_glass=True,
-            env={BREAK_GLASS_ENV: "operator-issued"},
-        )
-        self.assertTrue(result["break_glass_authorized"])
-        self.assertTrue(result["apply_gate"]["drain_gate_allow"],
-                        "break-glass does satisfy the drain gate")
-        self.assertFalse(result["apply_gate"]["restart_class_authorized"])
-        self.assertFalse(result["apply_authorized"],
-                         "break-glass must not collapse the class matrix")
-
-    def test_break_glass_across_every_worker_role_and_restricted_class(self):
-        for role in ("author", "reviewer", "merger", "reconciler"):
-            for klass in ("rolling_mcp_restart", "full_mcp_restart",
-                          "host_restart"):
-                with self.subTest(role=role, restart_class=klass):
-                    result = self._call(
-                        role=role,
-                        restart_class=klass,
-                        dry_run=False,
-                        request_break_glass=True,
-                        env={BREAK_GLASS_ENV: "operator-issued"},
-                    )
-                    self.assertFalse(result["apply_authorized"])
-
-    def test_break_glass_still_works_when_the_class_is_authorized(self):
-        # Break-glass keeps its purpose: skipping the drain proof for a caller
-        # the matrix does allow.
-        result = self._call(
-            role="operator",
-            restart_class="full_mcp_restart",
-            dry_run=False,
-            request_break_glass=True,
-            env={BREAK_GLASS_ENV: "operator-issued",
-                 CONTROLLER_APPROVAL_ENV: "operator-approved"},
-        )
-        self.assertTrue(result["apply_authorized"])
-        self.assertEqual(result["apply_gate"]["verdict"], "break_glass")
-
-    def test_break_glass_is_not_self_assertable(self):
-        # Requested but no environment authorization -> no bypass, and the
-        # unproven apply is denied.
-        result = self._call(
-            role="operator",
-            restart_class="full_mcp_restart",
-            dry_run=False,
-            request_break_glass=True,
-            env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
-        )
-        self.assertTrue(result["break_glass_requested"])
-        self.assertFalse(result["break_glass_authorized"])
-        self.assertFalse(result["apply_authorized"])
-        self.assertIn("incident", result)
-
-
-class TestRestrictedClassesStayDenied(_RestartToolHarness):
-    """AC5 — restricted classes remain denied to unauthorized requesters."""
-
-    def test_restricted_classes_denied_for_worker_roles(self):
-        for role in ("author", "reviewer", "merger", "reconciler"):
-            for klass in ("rolling_mcp_restart", "full_mcp_restart",
-                          "host_restart"):
-                with self.subTest(role=role, restart_class=klass):
-                    preview = self._call(
-                        role=role,
-                        restart_class=klass,
-                        env={CONTROLLER_APPROVAL_ENV: "yes"},
-                    )
-                    self.assertFalse(preview["allow_restart"])
-                    self.assertFalse(preview["permission_authorized"])
-                    self.assertFalse(preview["role_authorized"])
-
-    def test_host_restart_needs_controller_and_infrastructure_operator(self):
-        # controller approval alone is not enough for host_restart.
-        preview = self._call(role="controller", restart_class="host_restart",
-                             env={CONTROLLER_APPROVAL_ENV: "yes"})
-        self.assertFalse(preview["approval_satisfied"])
-        self.assertFalse(preview["allow_restart"])
-
-
-class TestExistingPathsStillWork(_RestartToolHarness):
-    """AC6 — valid scoped and unscoped restart paths are unaffected."""
-
-    def test_dry_run_never_reports_apply_authorization(self):
-        result = self._call(role="operator", restart_class="full_mcp_restart",
-                            env={CONTROLLER_APPROVAL_ENV: "yes"})
-        self.assertNotIn("apply_authorized", result)
-        self.assertNotIn("apply_gate", result)
-        self.assertFalse(result["apply_supported"])
-        self.assertFalse(result["restart_performed"])
-
-    def test_self_service_unscoped_classes_authorize_for_every_role(self):
-        for role in ("author", "reviewer", "merger", "reconciler",
-                     "controller", "operator", "admin"):
-            for klass in ("client_reconnect", "session_reconnect"):
-                with self.subTest(role=role, restart_class=klass):
-                    preview = self._call(role=role, restart_class=klass)
-                    self.assertTrue(preview["allow_restart"])
-
-    def test_scoped_class_with_target_authorizes_and_applies(self):
-        env = {CONTROLLER_APPROVAL_ENV: "operator-approved"}
-        preview = self._call(role="operator", restart_class="worker_restart",
-                             target_session_id="worker-1", env=env)
-        self.assertTrue(preview["allow_restart"])
-
-        result = self._call(
-            role="operator",
-            restart_class="worker_restart",
-            target_session_id="worker-1",
-            dry_run=False,
-            drain_proof_json=self._clean_proof_for(preview),
-            env=env,
-        )
-        self.assertTrue(result["apply_authorized"])
-
-    def test_apply_still_denies_without_any_proof(self):
-        # The #661 hard gate is untouched by the conjunction.
-        result = self._call(
-            role="operator",
-            restart_class="full_mcp_restart",
-            dry_run=False,
-            env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
-        )
-        self.assertFalse(result["apply_gate"]["drain_gate_allow"])
-        self.assertTrue(result["apply_gate"]["restart_class_authorized"])
-        self.assertFalse(result["apply_authorized"])
-        self.assertEqual(result["incident"]["kind"], "restart_drain_gate_denied")
-
-    def test_apply_denies_on_malformed_proof(self):
-        result = self._call(
-            role="operator",
-            restart_class="full_mcp_restart",
-            dry_run=False,
-            drain_proof_json="{not valid json",
-            env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
-        )
-        self.assertFalse(result["apply_authorized"])
-        self.assertTrue(any("invalid drain_proof_json" in reason
-                            for reason in result["apply_gate"]["reasons"]))
-
-    def test_tool_never_restarts_on_any_path(self):
-        for kwargs in (
-            {"restart_class": "client_reconnect"},
-            {"restart_class": "full_mcp_restart", "dry_run": False},
-            {"restart_class": "host_restart", "dry_run": False,
-             "request_break_glass": True},
-        ):
-            with self.subTest(**kwargs):
-                result = self._call(role="operator", env={
-                    CONTROLLER_APPROVAL_ENV: "yes", BREAK_GLASS_ENV: "yes"},
-                    **kwargs)
-                self.assertFalse(result["restart_performed"])
-                self.assertFalse(result["apply_supported"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_issue_897_permission_stale_runtime_classification.py

@@ -0,0 +1,453 @@
+"""#897: stale-runtime / runtime-mode refusals must not look like permission denials.
+
+Acceptance criteria (issue #897):
+
+* Stale-runtime and runtime-mode refusals are typed distinctly from
+  profile-permission refusals (distinct ``blocker_kind``).
+* A refusal caused by staleness or runtime mode never emits a
+  ``permission_report`` and never names a permission the active profile holds.
+* ``_permission_block_report`` verifies the active profile actually lacks the
+  operation before reporting it missing.
+* A stale-runtime refusal reports reconnect-only recovery and never recommends
+  ``gitea_activate_profile`` or an MCP session switch.
+* The blocker payload states the observed heads (parity fields).
+* Matrix across author / reviewer / merger / reconciler profiles.
+* Regression: ``gitea_create_issue`` on a stale daemon under ``prgs-author``
+  never returns ``missing_permission: gitea.issue.create``.
+"""
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_config  # noqa: E402
+import gitea_mcp_server as mcp_server  # noqa: E402
+
+SHA_START = "7af40fb5ff7debd5e9165fe97d9c7c279358e175"
+SHA_LIVE = "2f4dec832327513118f2fe92b74da25d124a01cb"
+
+ROLE_MATRIX = (
+    (
+        "prgs-author",
+        "author",
+        "gitea.issue.create",
+        [
+            "gitea.read",
+            "gitea.issue.create",
+            "gitea.issue.comment",
+            "gitea.issue.close",
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.pr.create",
+            "gitea.pr.comment",
+            "gitea.repo.commit",
+        ],
+        ["gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes"],
+        "gitea.pr.merge",  # forbidden op for pure-permission case
+    ),
+    (
+        "prgs-reviewer",
+        "reviewer",
+        "gitea.pr.review",
+        [
+            "gitea.read",
+            "gitea.pr.review",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+        ],
+        ["gitea.branch.push", "gitea.pr.create"],
+        "gitea.branch.push",
+    ),
+    (
+        "prgs-merger",
+        "merger",
+        "gitea.pr.merge",
+        [
+            "gitea.read",
+            "gitea.pr.merge",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+        ],
+        ["gitea.pr.approve", "gitea.branch.push", "gitea.pr.create"],
+        "gitea.branch.push",
+    ),
+    (
+        "prgs-reconciler",
+        "reconciler",
+        "gitea.branch.delete",
+        [
+            "gitea.read",
+            "gitea.branch.delete",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+            "gitea.pr.close",
+            "gitea.issue.close",
+        ],
+        ["gitea.pr.approve", "gitea.pr.merge"],
+        "gitea.pr.merge",
+    ),
+)
+
+
+def _profile(name: str, role: str, allowed: list[str], forbidden: list[str]) -> dict:
+    return {
+        "profile_name": name,
+        "role": role,
+        "role_kind": role,
+        "allowed_operations": list(allowed),
+        "forbidden_operations": list(forbidden),
+        "identity": "test-user",
+    }
+
+
+def _config(profiles: dict) -> dict:
+    return {
+        "version": 2,
+        "profiles": {
+            name: {
+                "role": p["role"],
+                "allowed_operations": p["allowed_operations"],
+                "forbidden_operations": p["forbidden_operations"],
+            }
+            for name, p in profiles.items()
+        },
+        "rules": {"allow_runtime_switching": True},
+    }
+
+
+class Issue897Helpers(unittest.TestCase):
+    def test_classify_stale_reason_strings(self):
+        stale = (
+            f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
+            f"at {SHA_START[:12]}; the daemon is stale relative to live master "
+            "-- restart/reconnect before mutating"
+        )
+        classified = mcp_server._classify_operation_gate_reasons([stale])
+        self.assertEqual(classified["stale_runtime"], [stale])
+        self.assertEqual(classified["permission"], [])
+        self.assertEqual(classified["runtime_mode"], [])
+
+    def test_classify_permission_reason(self):
+        reason = "profile is not allowed to gitea.pr.merge"
+        classified = mcp_server._classify_operation_gate_reasons([reason])
+        self.assertEqual(classified["permission"], [reason])
+        self.assertEqual(classified["stale_runtime"], [])
+
+    def test_classify_runtime_mode_reason(self):
+        reason = (
+            "runtime mode is 'dev-test' and the mutation targets the "
+            "production repository; dev/test runtimes must not mutate real "
+            "issues or PRs (ADR: stable control runtime vs dev runtime)"
+        )
+        classified = mcp_server._classify_operation_gate_reasons([reason])
+        self.assertEqual(classified["runtime_mode"], [reason])
+        self.assertEqual(classified["stale_runtime"], [])
+
+
+class Issue897PermissionBlockReport(unittest.TestCase):
+    def test_holds_op_is_diagnostic_defect_not_missing_permission(self):
+        profile = _profile(
+            "prgs-author",
+            "author",
+            ["gitea.read", "gitea.issue.create", "gitea.issue.comment"],
+            [],
+        )
+        with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server.gitea_config, "load_config", return_value=_config({"prgs-author": profile})
+        ), patch.object(
+            mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=True
+        ):
+            report = mcp_server._permission_block_report("gitea.issue.create")
+        self.assertTrue(report.get("diagnostic_defect"), report)
+        self.assertIsNone(report.get("missing_permission"), report)
+        action = (report.get("exact_safe_next_action") or "").lower()
+        # Must not *recommend* profile switching; mentioning the forbidden
+        # action in a "do not call" instruction is fine.
+        self.assertNotIn("call gitea_activate_profile with", action)
+        self.assertNotIn("switch to the author mcp session", action)
+        self.assertNotIn("switch to the reviewer mcp session", action)
+        self.assertIn("diagnostic defect", action)
+
+    def test_true_missing_permission_still_reports(self):
+        profile = _profile(
+            "prgs-author",
+            "author",
+            ["gitea.read", "gitea.issue.create"],
+            ["gitea.pr.merge"],
+        )
+        reviewer = _profile(
+            "prgs-reviewer",
+            "reviewer",
+            ["gitea.read", "gitea.pr.merge", "gitea.pr.approve"],
+            [],
+        )
+        with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server.gitea_config,
+            "load_config",
+            return_value=_config({"prgs-author": profile, "prgs-reviewer": reviewer}),
+        ), patch.object(
+            mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=True
+        ):
+            report = mcp_server._permission_block_report("gitea.pr.merge")
+        self.assertFalse(report.get("diagnostic_defect"), report)
+        self.assertEqual(report.get("missing_permission"), "gitea.pr.merge")
+        self.assertIn("prgs-reviewer", report.get("matching_configured_profiles") or [])
+
+
+class Issue897GateRefusalMatrix(unittest.TestCase):
+    def _stale_parity(self) -> dict:
+        return {
+            "in_parity": True,
+            "stale": False,
+            "restart_required": True,
+            "determinable": True,
+            "startup_head": SHA_START,
+            "current_head": SHA_START,
+            "daemon_start_head": SHA_START,
+            "local_head": SHA_START,
+            "live_remote_head": SHA_LIVE,
+            "live_known": True,
+            "live_stale": True,
+            "mutation_safe": False,
+            "reasons": [
+                f"live remote master is {SHA_LIVE[:12]} but the MCP server "
+                f"started at {SHA_START[:12]}; the daemon is stale relative "
+                "to live master -- restart/reconnect before mutating"
+            ],
+        }
+
+    def test_stale_plus_permitted_op_all_roles(self):
+        for name, role, permitted_op, allowed, forbidden, _forbidden_op in ROLE_MATRIX:
+            with self.subTest(profile=name, op=permitted_op):
+                profile = _profile(name, role, allowed, forbidden)
+                parity = self._stale_parity()
+                with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+                    mcp_server, "_current_master_parity", return_value=parity
+                ), patch.object(
+                    mcp_server, "_master_parity_block", return_value=list(parity["reasons"])
+                ), patch.object(
+                    mcp_server, "_runtime_mode_block", return_value=[]
+                ), patch.object(
+                    mcp_server, "_ensure_matching_profile", return_value=None
+                ), patch.object(
+                    mcp_server.session_ctx,
+                    "mutation_context_audit_fields",
+                    return_value={"session_profile": name},
+                ):
+                    blocked = mcp_server._profile_permission_block(permitted_op)
+                self.assertIsNotNone(blocked, name)
+                assert blocked is not None
+                self.assertEqual(
+                    blocked.get("blocker_kind"),
+                    "runtime_reconnect_required",
+                    blocked,
+                )
+                self.assertNotIn("permission_report", blocked, blocked)
+                self.assertTrue(blocked.get("restart_required"), blocked)
+                self.assertEqual(blocked.get("startup_head"), SHA_START, blocked)
+                self.assertEqual(blocked.get("live_remote_head"), SHA_LIVE, blocked)
+                action = (blocked.get("exact_safe_next_action") or "").lower()
+                self.assertIn("reconnect", action)
+                self.assertNotIn("call gitea_activate_profile with", action)
+                self.assertNotIn("switch to the author mcp session", action)
+                self.assertNotIn("switch to the reviewer mcp session", action)
+
+    def test_fresh_plus_forbidden_op_all_roles(self):
+        for name, role, _permitted, allowed, forbidden, forbidden_op in ROLE_MATRIX:
+            with self.subTest(profile=name, op=forbidden_op):
+                profile = _profile(name, role, allowed, forbidden)
+                with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+                    mcp_server, "_master_parity_block", return_value=[]
+                ), patch.object(
+                    mcp_server, "_runtime_mode_block", return_value=[]
+                ), patch.object(
+                    mcp_server, "_ensure_matching_profile", return_value=None
+                ), patch.object(
+                    mcp_server.session_ctx,
+                    "mutation_context_audit_fields",
+                    return_value={"session_profile": name},
+                ), patch.object(
+                    mcp_server.gitea_config,
+                    "load_config",
+                    return_value=_config({name: profile}),
+                ), patch.object(
+                    mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=False
+                ):
+                    blocked = mcp_server._profile_permission_block(forbidden_op)
+                self.assertIsNotNone(blocked, name)
+                assert blocked is not None
+                self.assertEqual(blocked.get("blocker_kind"), "permission_denied", blocked)
+                self.assertIn("permission_report", blocked, blocked)
+                report = blocked["permission_report"]
+                self.assertEqual(report.get("missing_permission"), forbidden_op, report)
+                self.assertFalse(report.get("diagnostic_defect"), report)
+                # No runtime reconnect fields for pure permission denial
+                self.assertNotEqual(
+                    blocked.get("blocker_kind"), "runtime_reconnect_required"
+                )
+
+    def test_stale_plus_forbidden_op_both_causes_separated(self):
+        for name, role, _permitted, allowed, forbidden, forbidden_op in ROLE_MATRIX:
+            with self.subTest(profile=name, op=forbidden_op):
+                profile = _profile(name, role, allowed, forbidden)
+                parity = self._stale_parity()
+                stale_reason = parity["reasons"][0]
+                with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+                    mcp_server, "_current_master_parity", return_value=parity
+                ), patch.object(
+                    mcp_server, "_master_parity_block", return_value=[stale_reason]
+                ), patch.object(
+                    mcp_server, "_runtime_mode_block", return_value=[]
+                ), patch.object(
+                    mcp_server, "_ensure_matching_profile", return_value=None
+                ), patch.object(
+                    mcp_server.session_ctx,
+                    "mutation_context_audit_fields",
+                    return_value={"session_profile": name},
+                ):
+                    # Gate collects both classes; force permission reason too.
+                    with patch.object(
+                        mcp_server,
+                        "_profile_operation_gate",
+                        return_value=[
+                            stale_reason,
+                            f"profile is not allowed to {forbidden_op}",
+                        ],
+                    ):
+                        blocked = mcp_server._profile_permission_block(forbidden_op)
+                self.assertIsNotNone(blocked)
+                assert blocked is not None
+                self.assertEqual(
+                    blocked.get("blocker_kind"), "runtime_reconnect_required", blocked
+                )
+                self.assertNotIn("permission_report", blocked, blocked)
+                self.assertIn("permission_block_reasons", blocked, blocked)
+                self.assertIn("stale_runtime_reasons", blocked, blocked)
+                classes = blocked.get("gate_reason_classes") or {}
+                self.assertTrue(classes.get("stale_runtime"), classes)
+                self.assertTrue(classes.get("permission"), classes)
+
+    def test_runtime_mode_block_no_permission_report(self):
+        profile = _profile(
+            "prgs-author",
+            "author",
+            ["gitea.read", "gitea.issue.create"],
+            [],
+        )
+        runtime_reason = (
+            "runtime mode is 'dev-test' and the mutation targets the "
+            "production repository; dev/test runtimes must not mutate real "
+            "issues or PRs (ADR: stable control runtime vs dev runtime)"
+        )
+        with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server, "_master_parity_block", return_value=[]
+        ), patch.object(
+            mcp_server, "_runtime_mode_block", return_value=[runtime_reason]
+        ), patch.object(
+            mcp_server, "_ensure_matching_profile", return_value=None
+        ), patch.object(
+            mcp_server.session_ctx,
+            "mutation_context_audit_fields",
+            return_value={"session_profile": "prgs-author"},
+        ):
+            blocked = mcp_server._profile_permission_block("gitea.issue.create")
+        self.assertIsNotNone(blocked)
+        assert blocked is not None
+        self.assertEqual(blocked.get("blocker_kind"), "runtime_mode_blocked", blocked)
+        self.assertNotIn("permission_report", blocked, blocked)
+        action = (blocked.get("exact_safe_next_action") or "").lower()
+        self.assertNotIn("call gitea_activate_profile with", action)
+        self.assertIn("stable control runtime", action)
+
+
+class Issue897CreateIssueRegression(unittest.TestCase):
+    def test_create_issue_stale_daemon_never_missing_issue_create(self):
+        """Regression AC: stale prgs-author create_issue must not claim missing create."""
+        profile = _profile(
+            "prgs-author",
+            "author",
+            [
+                "gitea.read",
+                "gitea.issue.create",
+                "gitea.issue.comment",
+                "gitea.branch.create",
+                "gitea.branch.push",
+                "gitea.pr.create",
+                "gitea.pr.comment",
+                "gitea.repo.commit",
+            ],
+            [],
+        )
+        stale_reason = (
+            f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
+            f"at {SHA_START[:12]}; the daemon is stale relative to live master "
+            "-- restart/reconnect before mutating"
+        )
+        parity = {
+            "in_parity": True,
+            "stale": False,
+            "restart_required": True,
+            "determinable": True,
+            "startup_head": SHA_START,
+            "current_head": SHA_START,
+            "daemon_start_head": SHA_START,
+            "local_head": SHA_START,
+            "live_remote_head": SHA_LIVE,
+            "live_known": True,
+            "live_stale": True,
+            "mutation_safe": False,
+            "reasons": [stale_reason],
+        }
+
+        with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server, "_current_master_parity", return_value=parity
+        ), patch.object(
+            mcp_server, "_master_parity_block", return_value=[stale_reason]
+        ), patch.object(
+            mcp_server, "_runtime_mode_block", return_value=[]
+        ), patch.object(
+            mcp_server, "_ensure_matching_profile", return_value=None
+        ), patch.object(
+            mcp_server.session_ctx,
+            "mutation_context_audit_fields",
+            return_value={"session_profile": "prgs-author"},
+        ), patch.object(
+            mcp_server, "_mutation_config_authority_block", return_value=None
+        ), patch.object(
+            mcp_server, "_session_context_mutation_block", return_value=None
+        ):
+            blocked = mcp_server._profile_permission_block(
+                "gitea.issue.create", remote="prgs"
+            )
+
+        self.assertIsNotNone(blocked)
+        assert blocked is not None
+        self.assertEqual(blocked.get("blocker_kind"), "runtime_reconnect_required")
+        self.assertNotIn("permission_report", blocked)
+        # Even if a caller still built a raw report, holds-check must not claim missing.
+        with patch.object(mcp_server, "get_profile", return_value=profile):
+            raw = mcp_server._permission_block_report("gitea.issue.create")
+        self.assertIsNone(raw.get("missing_permission"), raw)
+        self.assertNotEqual(raw.get("missing_permission"), "gitea.issue.create")
+
+    def test_permission_report_for_gate_reasons_skips_stale(self):
+        stale = (
+            f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
+            f"at {SHA_START[:12]}; the daemon is stale relative to live master "
+            "-- restart/reconnect before mutating"
+        )
+        self.assertIsNone(
+            mcp_server._permission_report_for_gate_reasons(
+                "gitea.issue.comment", [stale]
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_restart_governance_docs.py

@@ -105,120 +105,3 @@ def test_cross_links_do_not_embed_secrets():
         text = _read(path)
         for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
             assert marker not in text, f"{path} contains {marker!r}"
-
-
-# --- Coordinator doc stays in lock-step with the tool (#886 review blocker B2) --
-#
-# PR #882 moved the #661 drain-proof hard gate *into* gitea_request_mcp_restart,
-# but the coordinator document still described the proof as "a separate child"
-# and omitted both new parameters. Nothing referenced that document, so nothing
-# caught the drift. These tests bind the prose to the real signature.
-
-COORDINATOR_DOC = REPO_ROOT / "docs" / "mcp-restart-coordinator.md"
-
-# Affirmative claims that were accurate before #661 landed and are now false.
-# Matched against whitespace-normalized text so re-wrapping cannot hide them.
-# Deliberately not the bare phrase "a separate child": the corrected prose uses
-# it in a negation ("no longer a separate child operation"), and a guard that
-# forbids naming the old behaviour would block explaining that it changed.
-STALE_PRE_661_PHRASES = (
-    "gated by a drain proof (a separate child)",
-    "is a later child gated by a drain proof",
-    "mutative apply path is explicitly out of scope",
-    "apply is gated by a drain proof (a separate child)",
-)
-
-
-def _documented_signature_block() -> str:
-    """The fenced signature block for the tool, as published in the doc."""
-    text = _read(COORDINATOR_DOC)
-    marker = "gitea_request_mcp_restart("
-    start = text.index(marker)
-    end = text.index("```", start)
-    return text[start:end]
-
-
-def test_documented_signature_matches_the_real_tool_signature():
-    import inspect
-
-    import gitea_mcp_server
-
-    block = _documented_signature_block()
-    real = inspect.signature(gitea_mcp_server.gitea_request_mcp_restart)
-    for name in real.parameters:
-        assert name in block, (
-            f"docs/mcp-restart-coordinator.md documents no {name!r} parameter; "
-            "the published signature has drifted from the tool"
-        )
-
-
-def test_drain_proof_and_break_glass_parameters_are_documented():
-    block = _documented_signature_block()
-    for name in ("drain_proof_json", "request_break_glass"):
-        assert name in block, f"signature block missing {name}"
-
-
-def test_restart_class_and_target_scoping_parameters_survive():
-    block = _documented_signature_block()
-    for name in ("restart_class", "target_session_id", "target_role",
-                 "target_connector"):
-        assert name in block, f"signature block lost #663 parameter {name}"
-
-
-def test_gate_is_documented_as_executing_inside_this_tool():
-    lower = _read(COORDINATOR_DOC).lower()
-    assert "inside this tool" in lower, (
-        "the coordinator doc must state that the drain-proof gate executes in "
-        "gitea_request_mcp_restart, not in a later child"
-    )
-    assert "no longer a separate child operation" in lower
-
-
-def test_stale_pre_661_wording_cannot_return():
-    normalized = " ".join(_read(COORDINATOR_DOC).split()).lower()
-    for phrase in STALE_PRE_661_PHRASES:
-        assert phrase not in normalized, (
-            f"stale pre-#661 wording returned to the coordinator doc: {phrase!r}"
-        )
-
-
-def test_dry_run_versus_apply_behavior_is_documented():
-    lower = _read(COORDINATOR_DOC).lower()
-    assert "dry_run=true" in lower and "dry_run=false" in lower
-    assert "apply_supported" in lower and "restart_performed" in lower
-    assert "never restarts anything" in lower
-
-
-def test_authorization_ordering_and_conjunction_are_documented():
-    text = _read(COORDINATOR_DOC)
-    lower = text.lower()
-    assert "authorization ordering" in lower
-    assert "allow_restart" in text
-    assert "apply_authorized" in text
-    # The conjunction itself, and the attribution fields behind it.
-    assert "gate.allow and allow_restart" in text
-    for field in ("drain_gate_allow", "restart_class_authorized"):
-        assert field in text, f"doc omits apply_gate.{field}"
-
-
-def test_break_glass_scope_is_documented_as_drain_proof_only():
-    text = _read(COORDINATOR_DOC)
-    lower = text.lower()
-    assert "break-glass" in lower
-    assert "drain proof only" in lower, (
-        "doc must state break-glass never bypasses the restart-class matrix"
-    )
-    assert "GITEA_BREAKGLASS_RESTART_AUTHORIZATION" in text
-
-
-def test_fail_closed_on_apply_is_documented():
-    lower = _read(COORDINATOR_DOC).lower()
-    assert "fail closed" in lower
-    for condition in ("expired", "unclean", "tampered", "stale"):
-        assert condition in lower, f"fail-closed list omits {condition!r}"
-
-
-def test_coordinator_doc_embeds_no_secrets():
-    text = _read(COORDINATOR_DOC)
-    for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
-        assert marker not in text, f"{COORDINATOR_DOC} contains {marker!r}"

tests/test_restart_classes.py

@@ -1,232 +0,0 @@
-"""Permission, drain, routing, and audit matrix for restart classes (#663)."""
-
-from __future__ import annotations
-
-import os
-from datetime import datetime, timezone
-
-import restart_coordinator as rc
-
-NOW = datetime(2026, 7, 24, 20, 0, tzinfo=timezone.utc)
-
-
-def _inventory() -> dict:
-    return {
-        "inventory_complete": True,
-        "sessions": [
-            {
-                "session_id": "requester",
-                "role": "author",
-                "profile": "prgs-author",
-                "pid": os.getpid(),
-                "status": "active",
-                "last_heartbeat_at": NOW.isoformat(),
-            },
-            {
-                "session_id": "reviewer",
-                "role": "reviewer",
-                "profile": "prgs-reviewer",
-                "pid": os.getpid(),
-                "status": "active",
-                "last_heartbeat_at": NOW.isoformat(),
-            },
-        ],
-        "leases": [
-            {
-                "lease_id": "review-lease",
-                "session_id": "reviewer",
-                "role": "reviewer",
-                "phase": "reviewing",
-                "work_kind": "pr",
-                "work_number": 900,
-                "worktree_path": "/tmp/review-900",
-                "freshness": {"freshness": "active"},
-            }
-        ],
-    }
-
-
-def _evaluate(
-    restart_class: rc.RestartClass,
-    *,
-    role: str = "controller",
-    permissions: tuple[str, ...] | None = None,
-    approved: bool = True,
-    operator: bool = True,
-    **targets,
-):
-    return rc.evaluate_restart_impact(
-        _inventory(),
-        now=NOW,
-        requesting_session_id="requester",
-        restart_class=restart_class,
-        requester_role=role,
-        requester_permissions=(
-            permissions if permissions is not None
-            else rc.permissions_for_role(role)
-        ),
-        controller_approved=approved,
-        operator_authorized=operator,
-        **targets,
-    )
-
-
-def test_policy_table_covers_exactly_all_nine_classes():
-    assert set(rc.RESTART_CLASS_POLICIES) == set(rc.RestartClass)
-    assert len(rc.RESTART_CLASS_POLICIES) == 9
-    for restart_class, policy in rc.RESTART_CLASS_POLICIES.items():
-        assert policy.restart_class is restart_class
-        assert policy.required_permission
-        assert policy.expected_blast_radius in {
-            rc.BLAST_NONE, rc.BLAST_LOW, rc.BLAST_MEDIUM, rc.BLAST_HIGH
-        }
-        assert policy.drain_requirement
-        assert policy.approval_requirement
-        assert policy.audit_requirement
-        assert policy.recovery_behavior
-
-
-def test_permission_matrix_allows_each_class_with_exact_permission():
-    targets = {
-        rc.RestartClass.WORKER_RESTART: {"target_session_id": "reviewer"},
-        rc.RestartClass.ROLE_RUNTIME_RESTART: {"target_role": "reviewer"},
-        rc.RestartClass.CONNECTOR_RESTART: {"target_connector": "github"},
-    }
-    for restart_class, policy in rc.RESTART_CLASS_POLICIES.items():
-        report = _evaluate(
-            restart_class,
-            permissions=(policy.required_permission,),
-            **targets.get(restart_class, {}),
-        )
-        assert report.permission_authorized, restart_class
-        assert report.role_authorized, restart_class
-        assert report.approval_satisfied, restart_class
-        assert report.audit_record["restart_class"] == restart_class.value
-        assert (
-            report.audit_record["required_permission"]
-            == policy.required_permission
-        )
-
-
-def test_missing_or_nearby_permission_denies():
-    report = _evaluate(
-        rc.RestartClass.ROLE_RUNTIME_RESTART,
-        permissions=("mcp.restart.worker.request",),
-        target_role="reviewer",
-    )
-    assert report.verdict == rc.VERDICT_UNSAFE
-    assert not report.allow_restart
-    assert not report.permission_authorized
-    assert any("missing required permission" in r for r in report.reasons)
-
-
-def test_unknown_restart_class_denies_fail_closed():
-    report = rc.evaluate_restart_impact(
-        _inventory(),
-        now=NOW,
-        restart_class="surprise_reboot",
-        requester_role="admin",
-        requester_permissions=("mcp.restart.host.request",),
-        controller_approved=True,
-        operator_authorized=True,
-    )
-    assert report.verdict == rc.VERDICT_UNSAFE
-    assert not report.allow_restart
-    assert report.restart_policy == {}
-    assert any("unknown restart class" in r for r in report.reasons)
-
-
-def test_worker_roles_cannot_request_full_or_host_restart():
-    for role in rc.WORKER_ROLES:
-        granted = rc.permissions_for_role(role)
-        assert "mcp.restart.full.request" not in granted
-        assert "mcp.restart.host.request" not in granted
-        report = _evaluate(
-            rc.RestartClass.FULL_MCP_RESTART,
-            role=role,
-            permissions=granted,
-        )
-        assert not report.role_authorized
-        assert not report.allow_restart
-
-
-def test_controller_approval_is_independent_of_permission():
-    report = _evaluate(
-        rc.RestartClass.WORKER_RESTART,
-        approved=False,
-        target_session_id="reviewer",
-    )
-    assert report.permission_authorized
-    assert not report.approval_satisfied
-    assert not report.allow_restart
-
-
-def test_narrow_classes_do_not_inherit_full_drain_or_peer_lease_block():
-    for restart_class in (
-        rc.RestartClass.CLIENT_RECONNECT,
-        rc.RestartClass.SESSION_RECONNECT,
-        rc.RestartClass.CONFIGURATION_RELOAD,
-    ):
-        report = _evaluate(restart_class)
-        assert not report.restart_policy["full_drain_required"]
-        assert report.counts["leases_disruptive"] == 0
-        assert report.counts["sessions_live_other"] == 0
-        assert report.counts["critical_sections"] == 0
-        assert report.counts["mutations"] == 0
-        assert report.allow_restart, (restart_class, report.reasons)
-
-
-def test_client_reconnect_does_not_wait_for_unrelated_terminal_lock():
-    inventory = _inventory()
-    inventory["terminal_lock"] = {"terminal_pr": 901}
-    report = rc.evaluate_restart_impact(
-        inventory,
-        now=NOW,
-        requesting_session_id="requester",
-        restart_class=rc.RestartClass.CLIENT_RECONNECT,
-        requester_role="author",
-        requester_permissions=rc.permissions_for_role("author"),
-    )
-    assert report.allow_restart
-    assert report.terminal_lock is None
-
-
-def test_scoped_restart_only_counts_named_target():
-    report = _evaluate(
-        rc.RestartClass.ROLE_RUNTIME_RESTART,
-        target_role="author",
-    )
-    assert report.counts["leases_disruptive"] == 0
-    assert report.affected_prs == []
-    assert report.allow_restart
-
-    reviewer = _evaluate(
-        rc.RestartClass.ROLE_RUNTIME_RESTART,
-        target_role="reviewer",
-    )
-    assert reviewer.counts["leases_disruptive"] == 1
-    assert reviewer.affected_prs == [900]
-    assert not reviewer.allow_restart
-
-
-def test_missing_scoped_target_denies_instead_of_widening():
-    for restart_class in (
-        rc.RestartClass.WORKER_RESTART,
-        rc.RestartClass.ROLE_RUNTIME_RESTART,
-        rc.RestartClass.CONNECTOR_RESTART,
-    ):
-        report = _evaluate(restart_class)
-        assert not report.allow_restart
-        assert any("target required" in r for r in report.reasons)
-
-
-def test_only_full_and_host_classes_require_full_drain():
-    requiring_full = {
-        restart_class
-        for restart_class, policy in rc.RESTART_CLASS_POLICIES.items()
-        if policy.full_drain_required
-    }
-    assert requiring_full == {
-        rc.RestartClass.FULL_MCP_RESTART,
-        rc.RestartClass.HOST_RESTART,
-    }

tests/test_webui_sanctioned_restart.py

@@ -444,12 +444,6 @@ class TestAuditEmission(unittest.TestCase):
         )
         self.assertEqual(record["target"]["namespace"], NAMESPACE)
         self.assertEqual(record["target"]["mode"], "restart")
-        self.assertEqual(
-            record["target"]["restart_class"], "role_runtime_restart"
-        )
-        self.assertEqual(
-            record["metadata"]["restart_class"], "role_runtime_restart"
-        )
         self.assertEqual(record["result"], console_audit.RESULT_ALLOWED)
         self.assertEqual(record["actor"]["subject"], "[email protected]")
         self.assertFalse(record["metadata"]["process_kill_executed"])

webui/sanctioned_restart.py

@@ -38,7 +38,6 @@ from dataclasses import asdict, dataclass
 from typing import Any
 
 import mcp_namespace_health
-import restart_coordinator
 import runtime_recovery_guard
 from webui import console_audit, console_authz
 
@@ -100,14 +99,6 @@ def _clean(value: Any) -> str:
     return str(value or "").strip()
 
 
-def restart_class_for_mode(mode: str) -> str:
-    """Map the existing namespace controls onto the #663 class taxonomy."""
-
-    if _clean(mode) == MODE_RELOAD:
-        return restart_coordinator.RestartClass.CONFIGURATION_RELOAD.value
-    return restart_coordinator.RestartClass.ROLE_RUNTIME_RESTART.value
-
-
 # --- Mutation ledger --------------------------------------------------------
 
 
@@ -265,7 +256,6 @@ def build_restart_preview(
 
     return {
         "action_id": action_id,
-        "restart_class": restart_class_for_mode(md),
         "namespace": ns,
         "mode": md,
         "scope_valid": scope_error is None,
@@ -319,7 +309,6 @@ def assess_restart_request(
             "reason_code": reason_code,
             "detail": detail,
             "action_id": action_id,
-            "restart_class": restart_class_for_mode(md),
             "namespace": ns,
             "mode": md,
             "preview": preview,
@@ -404,7 +393,6 @@ def assess_restart_request(
             "process."
         ),
         "action_id": action_id,
-        "restart_class": restart_class_for_mode(md),
         "namespace": ns,
         "mode": md,
         "preview": preview,
@@ -453,11 +441,7 @@ def execute_restart(
             else console_audit.RESULT_DENIED
         ),
         principal=principal,
-        target={
+        target={"namespace": assessment["namespace"], "mode": assessment["mode"]},
-            "namespace": assessment["namespace"],
-            "mode": assessment["mode"],
-            "restart_class": assessment["restart_class"],
-        },
         reason_code=assessment["reason_code"],
         detail=assessment["detail"],
         request_id=request_id,
@@ -466,7 +450,6 @@ def execute_restart(
             "gates_passed": assessment["gates_passed"],
             "process_kill_executed": False,
             "post_restart_verification_required": True,
-            "restart_class": assessment["restart_class"],
         },
     )
 
@@ -480,7 +463,6 @@ def execute_restart(
         "namespace": assessment["namespace"],
         "mode": assessment["mode"],
         "action_id": action_id,
-        "restart_class": assessment["restart_class"],
         "process_kill_executed": False,
         "host_hook": assessment["preview"]["restart_hook"],
         "next_action": (