feat: enforce MCP restart class permissions (#663)

allocator_service.py

@@ -133,15 +133,10 @@ ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
 
 
 # Body phrases that prove an issue is an implementation container, not a
-# unit of direct author work (#844 / #854). Matched case-insensitively against
-# the issue body. Title alone is never sufficient (ordinary issues may mention
-# "epic", "roadmap", "vision", or "umbrella" incidentally).
-#
-# #854 extends the #844 marker set so product-vision (#652), phased-roadmap
-# (#653), and umbrella (#655) coordination records — which do not use the word
-# "epic" — are classified with the same semantic exclusion as epic containers.
+# unit of direct author work (#844). Matched case-insensitively against the
+# issue body. Title alone is never sufficient (ordinary issues may mention
+# "epic" incidentally).
 _CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
-    # Epic / child-only (#844, live #631)
     "implementation is delivered via child issues only",
     "implementation is delivered through child issues only",
     "implementation is delivered via child issues",
@@ -155,26 +150,9 @@ _CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
     "coordination container",
     "child-only container",
     "implementation is delegated to child",
-    # Vision / roadmap / umbrella coordination (#854, live #652/#653/#655).
-    # Prefer authoritative non-implementation / child-only scope language over
-    # bare words like "roadmap" so ordinary implementable issues that mention
-    # a parent vision or roadmap stay eligible.
-    "do not implement features on this issue",
-    "implementing features on this roadmap issue",
-    "implementation is via linked children only",
-    "no product feature claimed complete on this issue alone",
-    "phased delivery roadmap and epic sequencing",
-    "this issue is the enduring source of truth",
-    "enduring source of truth for the",
-    "canonical product vision — enduring source of truth",
-    "canonical product vision - enduring source of truth",
-    "state: vision-active",
-    "state: roadmap-active",
 )
 
-# Explicit epic / umbrella / vision / roadmap labels (structured evidence
-# preferred over title). Tracker alone is *not* included — ordinary issues
-# may carry a tracker label without being non-implementable containers.
+# Explicit epic / umbrella labels (structured evidence preferred over title).
 _EPIC_LABELS: frozenset[str] = frozenset(
     {
         "type:epic",
@@ -183,16 +161,6 @@ _EPIC_LABELS: frozenset[str] = frozenset(
         "scope:epic",
         "type:umbrella",
         "umbrella",
-        "kind:umbrella",
-        "scope:umbrella",
-        "type:vision",
-        "vision",
-        "kind:vision",
-        "scope:vision",
-        "type:roadmap",
-        "roadmap",
-        "kind:roadmap",
-        "scope:roadmap",
     }
 )
 
@@ -254,45 +222,16 @@ class WorkCandidate:
         }
 
 
-def _title_container_prefix(title_l: str) -> str | None:
-    """Return a coordination-title prefix token if *title_l* uses one (#854).
-
-    Title prefixes alone never exclude; they only corroborate body/label
-    evidence. Ordinary issues may say "roadmap" or "vision" mid-title.
-    """
-    for prefix, token in (
-        ("epic:", "title_epic_prefix"),
-        ("epic ", "title_epic_prefix"),
-        ("umbrella:", "title_umbrella_prefix"),
-        ("umbrella ", "title_umbrella_prefix"),
-        ("roadmap:", "title_roadmap_prefix"),
-        ("roadmap ", "title_roadmap_prefix"),
-        ("product vision:", "title_vision_prefix"),
-        ("product vision ", "title_vision_prefix"),
-        ("vision:", "title_vision_prefix"),
-        ("vision ", "title_vision_prefix"),
-    ):
-        if title_l.startswith(prefix):
-            return token
-    return None
-
-
 def classify_epic_or_child_only_container(
     c: WorkCandidate,
 ) -> tuple[bool, str | None]:
-    """Return whether *c* is a non-implementable coordination container (#844/#854).
+    """Return whether *c* is an epic / child-only implementation container (#844).
 
     Exclusion uses structured evidence first (labels, body scope language).
-    A bare title containing the words "epic", "roadmap", "vision", or
-    "umbrella" is **not** enough — ordinary implementable issues may mention
-    those terms incidentally. Explicit title prefixes (``Epic:``, ``Roadmap:``,
-    ``Product vision:``, ``Umbrella:``) only count when the body also proves
-    child-only / no-direct-implementation scope (or a container label is
-    present).
-
-    Covers epic, product-vision, phased-roadmap, umbrella, and child-only
-    records so the allocator never assigns coordination containers as direct
-    author work.
+    A bare title containing the word "epic" is **not** enough — ordinary
+    implementable issues may mention epics incidentally. A title that is
+    explicitly prefixed ``Epic:`` only counts when the body also proves
+    child-only / no-direct-implementation scope (or an epic label is present).
 
     PRs are never classified as containers here (they already have a head).
     """
@@ -306,28 +245,25 @@ def classify_epic_or_child_only_container(
     title_l = title.lower()
 
     body_hits = [m for m in _CHILD_ONLY_BODY_MARKERS if m in body_l]
-    title_prefix = _title_container_prefix(title_l)
+    title_epic_prefix = title_l.startswith("epic:") or title_l.startswith("epic ")
 
     if epic_label:
         detail = f"label={epic_label[0]}"
         if body_hits:
             detail = f"{detail}; body_marker={body_hits[0]!r}"
-        if title_prefix:
-            detail = f"{title_prefix}; {detail}"
         return True, detail
 
     if body_hits:
-        # Body proves child-only / vision / roadmap / umbrella scope. Title
-        # prefixes are corroborating but not required — containers without the
-        # title word still exclude.
+        # Body proves child-only / umbrella scope. Title "Epic:" is corroborating
+        # but not required — containers without the word still exclude.
         detail = f"body_marker={body_hits[0]!r}"
-        if title_prefix:
-            detail = f"{title_prefix}; {detail}"
+        if title_epic_prefix:
+            detail = f"title_epic_prefix; {detail}"
         return True, detail
 
-    # Title-only coordination prefix without body scope evidence is
-    # insufficient (#844/#854 AC: eligibility does not rely solely on a title
-    # word). Incidental mid-title mentions without markers stay eligible.
+    # Title-only "Epic:" without body scope evidence is insufficient (#844 AC:
+    # eligibility does not rely solely on the word "Epic" in a title).
+    # Similarly, incidental "epic" mid-title without markers stays eligible.
     return False, None
 
 

docs/mcp-restart-classes.md

@@ -0,0 +1,53 @@
+# 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

@@ -12,6 +12,12 @@ restart/reload/kill paths). The **mutative apply** path — actually performing
 restart — is a later child gated by a drain proof and is explicitly out of
 scope here.
 
+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
 
 | Piece | Where | Responsibility |
@@ -77,7 +83,10 @@ 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)
 ```
 
 Read-only, dry-run, and it **never restarts anything**. `apply_supported` is
@@ -87,7 +96,8 @@ apply is gated by a drain proof (a separate child).
 ## Audit
 
 Every evaluation carries an `audit_record` (event, coordinator version, verdict,
-allow decision, blast radius, counts, timestamp) so restart decisions are
+restart class, required permission, allow decision, blast radius, counts,
+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

@@ -22342,10 +22342,15 @@ 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,
 ) -> dict:
     """Evaluate a proposed MCP restart and return an impact preview (#658).
 
-    Central restart coordinator: gathers live control-plane state (sessions,
+    Central restart coordinator: resolves the requested restart class, gathers
+    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
@@ -22439,6 +22444,9 @@ 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
@@ -22448,6 +22456,15 @@ 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,
@@ -22462,6 +22479,14 @@ 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()
@@ -22473,6 +22498,9 @@ 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)
     payload["apply_supported"] = False
     if not dry_run:
         payload["reasons"] = list(payload.get("reasons") or []) + [

restart_coordinator.py

@@ -28,11 +28,12 @@ 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.0.0-issue-658"
+COORDINATOR_VERSION = "1.1.0-issue-663"
 
 # Restart verdicts. Exactly the three the acceptance criteria name.
 VERDICT_SAFE = "safe"
@@ -54,6 +55,194 @@ 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)
 
@@ -75,6 +264,7 @@ class SessionImpact:
     heartbeat_stale: bool
     is_requester: bool
     live: bool
+    connector: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -87,6 +277,7 @@ class SessionImpact:
             "heartbeat_stale": self.heartbeat_stale,
             "is_requester": self.is_requester,
             "live": self.live,
+            "connector": self.connector,
         }
 
 
@@ -105,6 +296,7 @@ class LeaseImpact:
     disruptive: bool
     is_mutation: bool
     is_critical_section: bool
+    connector: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -119,6 +311,7 @@ class LeaseImpact:
             "disruptive": self.disruptive,
             "is_mutation": self.is_mutation,
             "is_critical_section": self.is_critical_section,
+            "connector": self.connector,
         }
 
 
@@ -127,6 +320,13 @@ 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
@@ -153,6 +353,13 @@ 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,
@@ -206,6 +413,7 @@ def _classify_session(
             requesting_session_id and session_id == requesting_session_id
         ),
         live=live,
+        connector=(str(row.get("connector") or "").strip() or None),
     )
 
 
@@ -258,6 +466,7 @@ 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),
     )
 
 
@@ -279,6 +488,14 @@ 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.
 
@@ -301,6 +518,61 @@ 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 [])]
@@ -323,15 +595,67 @@ def evaluate_restart_impact(
     ]
     lease_impacts = [_classify_lease(l) for l in leases_raw]
 
-    # Only *other* live sessions and live leases constitute blast radius: a
-    # restart that would kill only the requesting session with no other work in
-    # flight is safe.
+    # Route impact through the selected class. Narrow classes never inherit a
+    # full-runtime drain merely because unrelated work exists.
+    target_complete = True
+    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 session_impacts if s.live and not s.is_requester
+        s for s in scoped_sessions if s.live and not s.is_requester
     ]
-    disruptive_leases = [l for l in lease_impacts if l.disruptive]
-    critical_sections = [l for l in lease_impacts if l.is_critical_section]
-    mutations = [l for l in lease_impacts if l.is_mutation]
+    disruptive_leases = [l for l in scoped_leases if l.disruptive]
+    critical_sections = [l for l in scoped_leases if l.is_critical_section]
+    mutations = [l for l in scoped_leases 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(
         {
@@ -348,9 +672,24 @@ def evaluate_restart_impact(
         }
     )
 
-    disruptive = bool(disruptive_leases or other_live_sessions or terminal_lock)
+    disruptive = bool(
+        disruptive_leases or other_live_sessions or terminal_lock_in_scope
+    )
 
-    if not inventory_complete:
+    authorization_ok = bool(
+        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(
@@ -381,7 +720,7 @@ def evaluate_restart_impact(
             f"{len(critical_sections)} critical section(s) in flight "
             "(active lease with a live owner)"
         )
-    if terminal_lock:
+    if terminal_lock_in_scope:
         reasons.append("active terminal (merge) lock present")
 
     override_would_allow = bool(inventory_complete and disruptive)
@@ -411,6 +750,12 @@ 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),
@@ -424,6 +769,15 @@ 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,
@@ -440,9 +794,11 @@ def evaluate_restart_impact(
         affected_issues=affected_issues,
         affected_prs=affected_prs,
         mutations=mutations,
-        terminal_lock=dict(terminal_lock)
-        if isinstance(terminal_lock, Mapping)
-        else terminal_lock,
+        terminal_lock=(
+            dict(terminal_lock_in_scope)
+            if isinstance(terminal_lock_in_scope, Mapping)
+            else terminal_lock_in_scope
+        ),
         ack_state=ack_state,
         prior_recovery_attempts=prior_recovery_attempts,
         counts=counts,

tests/test_issue_854_semantic_container_exclusion.py

@@ -1,378 +0,0 @@
-"""Allocator semantic container exclusion for vision/roadmap/umbrella (#854).
-
-#844 excluded epic / child-only containers (live #631) but product-vision
-(#652), phased-roadmap (#653), and umbrella (#655) coordination records still
-ranked as implementable work. This module is the live-equivalent canary:
-
-* #631 / #652 / #653 / #655-shaped records are all excluded in one inventory.
-* Independently executable children remain eligible and can be selected.
-* Ordinary issues that merely mention vision / roadmap / umbrella stay eligible.
-* Excluded containers never receive assignments or workflow leases.
-* Structured skip reason ``epic_or_child_only_container`` is reported.
-* Candidate-set fingerprint remains stable after exclusions.
-"""
-
-from __future__ import annotations
-
-import os
-import tempfile
-import unittest
-
-from allocator_service import (
-    OUTCOME_ASSIGNED,
-    OUTCOME_PREVIEW,
-    SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
-    WorkCandidate,
-    allocate_next_work,
-    candidate_set_fingerprint,
-    classify_epic_or_child_only_container,
-)
-from control_plane_db import ControlPlaneDB
-
-REMOTE = "prgs"
-ORG = "Scaled-Tech-Consulting"
-REPO = "Gitea-Tools"
-
-# Minimal bodies mirroring live coordination records (not full issue text).
-_EPIC_631_BODY = """
-## Scope (umbrella)
-
-This epic owns the **product roadmap and linkage** for the Web Console.
-Implementation is delivered via child issues only.
-
-## Explicit non-goals
-
-* Do not implement product features in this epic issue itself.
-* No product feature implementation is claimed complete solely on this epic.
-"""
-
-_VISION_652_BODY = """
-## Canonical product vision — enduring source of truth
-
-**This issue is the enduring source of truth for the MCP Control Plane Web Console product vision.**
-
-## Implementation linkage
-
-* **Do not implement features on this issue.**
-* Sequencing: roadmap issue + #631 children.
-
-## Canonical issue state
-
-```text
-STATE: vision-active
-WHO_IS_NEXT: controller (triage/ordering) / author (implementation of linked children only)
-```
-"""
-
-_ROADMAP_653_BODY = """
-## Purpose
-
-This issue is the **phased delivery roadmap and epic sequencing** for the MCP Control Plane Web Console.
-
-## Non-goals
-
-* Implementing features on this roadmap issue.
-* Deleting vision items by omitting them from phases without #652 change log.
-
-## Canonical issue state
-
-```text
-STATE: roadmap-active
-WHO_IS_NEXT: author
-```
-"""
-
-_UMBRELLA_655_BODY = """
-## Scope (umbrella)
-
-This issue owns the **canonical restart-governance program**. Implementation is via linked children only.
-
-## Acceptance criteria (umbrella)
-
-6. No product feature claimed complete on this issue alone.
-"""
-
-_CHILD_BODY = """
-## Problem
-
-Operators need a workflow-event timeline model for Phase 1.
-
-## Acceptance criteria
-
-- [ ] Timeline model API exists
-"""
-
-
-def _issue(
-    number: int,
-    *,
-    title: str = "",
-    body: str = "",
-    labels: tuple[str, ...] = ("status:ready", "type:feature"),
-    priority: int = 20,
-) -> WorkCandidate:
-    return WorkCandidate(
-        kind="issue",
-        number=number,
-        state="open",
-        labels=labels,
-        title=title or f"issue {number}",
-        body=body,
-        priority=priority,
-    )
-
-
-def _live_shaped_containers() -> list[WorkCandidate]:
-    return [
-        _issue(
-            631,
-            title="Epic: MCP Control Plane Web Console",
-            body=_EPIC_631_BODY,
-        ),
-        _issue(
-            652,
-            title="Product vision: MCP Control Plane Web Console (canonical)",
-            body=_VISION_652_BODY,
-        ),
-        _issue(
-            653,
-            title="Roadmap: MCP Control Plane Web Console (phased delivery)",
-            body=_ROADMAP_653_BODY,
-        ),
-        _issue(
-            655,
-            title="Umbrella: Governed MCP restart coordination and zero-disruption recovery",
-            body=_UMBRELLA_655_BODY,
-        ),
-    ]
-
-
-class ClassifySemanticContainersTest(unittest.TestCase):
-    def test_652_vision_is_container(self) -> None:
-        c = _issue(
-            652,
-            title="Product vision: MCP Control Plane Web Console (canonical)",
-            body=_VISION_652_BODY,
-        )
-        is_c, detail = classify_epic_or_child_only_container(c)
-        self.assertTrue(is_c)
-        self.assertIsNotNone(detail)
-        self.assertIn("body_marker", detail or "")
-
-    def test_653_roadmap_is_container(self) -> None:
-        c = _issue(
-            653,
-            title="Roadmap: MCP Control Plane Web Console (phased delivery)",
-            body=_ROADMAP_653_BODY,
-        )
-        is_c, detail = classify_epic_or_child_only_container(c)
-        self.assertTrue(is_c)
-        self.assertIn("body_marker", detail or "")
-
-    def test_655_umbrella_is_container(self) -> None:
-        c = _issue(
-            655,
-            title="Umbrella: Governed MCP restart coordination and zero-disruption recovery",
-            body=_UMBRELLA_655_BODY,
-        )
-        is_c, detail = classify_epic_or_child_only_container(c)
-        self.assertTrue(is_c)
-        self.assertIn("body_marker", detail or "")
-
-    def test_631_still_container_after_854(self) -> None:
-        c = _issue(
-            631,
-            title="Epic: MCP Control Plane Web Console",
-            body=_EPIC_631_BODY,
-        )
-        is_c, detail = classify_epic_or_child_only_container(c)
-        self.assertTrue(is_c)
-        self.assertIn("body_marker", detail or "")
-
-    def test_incidental_vision_roadmap_umbrella_words_not_container(self) -> None:
-        cases = (
-            (
-                "Document vision handoff conventions",
-                "Update docs so implementable issues that mention a vision "
-                "remain independently executable.",
-            ),
-            (
-                "Clarify roadmap sequencing notes",
-                "Write a short note about how the roadmap issue relates to children.",
-            ),
-            (
-                "Umbrella recovery checklist for authors",
-                "Authors should still implement the concrete recovery fix here.",
-            ),
-            (
-                "Product vision wording in the help text",
-                "Fix a typo in the operator-facing help string that says product vision.",
-            ),
-        )
-        for title, body in cases:
-            with self.subTest(title=title):
-                c = _issue(900, title=title, body=body)
-                is_c, detail = classify_epic_or_child_only_container(c)
-                self.assertFalse(is_c)
-                self.assertIsNone(detail)
-
-    def test_title_prefix_alone_not_container(self) -> None:
-        for title in (
-            "Epic: something mentioned only in title",
-            "Roadmap: title only without body scope",
-            "Product vision: title only without body scope",
-            "Umbrella: title only without body scope",
-        ):
-            with self.subTest(title=title):
-                c = _issue(
-                    901,
-                    title=title,
-                    body="Implement a concrete fix for the allocator skip list.",
-                )
-                is_c, detail = classify_epic_or_child_only_container(c)
-                self.assertFalse(is_c)
-                self.assertIsNone(detail)
-
-    def test_roadmap_label_alone_is_container(self) -> None:
-        c = _issue(
-            902,
-            title="Console delivery sequencing",
-            body="Track phased delivery only.",
-            labels=("status:ready", "type:roadmap"),
-        )
-        is_c, detail = classify_epic_or_child_only_container(c)
-        self.assertTrue(is_c)
-        self.assertIn("type:roadmap", detail or "")
-
-    def test_child_referencing_parent_policy_stays_eligible(self) -> None:
-        """Children may quote parent policy without becoming containers."""
-        c = _issue(
-            637,
-            title="Web Console: Workflow-event timeline model (Phase 1)",
-            body=(
-                _CHILD_BODY
-                + "\n\nParent #652 says do not implement on the vision issue; "
-                "this child is the implementable unit."
-            ),
-        )
-        is_c, _ = classify_epic_or_child_only_container(c)
-        self.assertFalse(is_c)
-
-
-class AllocateSemanticContainerExclusionTest(unittest.TestCase):
-    def setUp(self) -> None:
-        self._tmp = tempfile.TemporaryDirectory()
-        self.addCleanup(self._tmp.cleanup)
-        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
-
-    def _alloc(self, candidates, **kwargs):
-        defaults = dict(
-            session_id="sess-854",
-            role="author",
-            remote=REMOTE,
-            org=ORG,
-            repo=REPO,
-            profile_name="prgs-author",
-            username="jcwalker3",
-            claims={},
-            apply=False,
-        )
-        defaults.update(kwargs)
-        return allocate_next_work(self.db, candidates=candidates, **defaults)
-
-    def test_live_equivalent_canary_excludes_all_containers_selects_child(self) -> None:
-        containers = _live_shaped_containers()
-        child = _issue(
-            637,
-            title="Web Console: Workflow-event timeline model (Phase 1)",
-            body=_CHILD_BODY,
-        )
-        inventory = containers + [child]
-        res = self._alloc(inventory, apply=False)
-        self.assertTrue(res["success"], res)
-        self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
-        self.assertEqual(res["selected"]["number"], 637)
-
-        skipped = {s["number"]: s for s in res["skipped"]}
-        for number in (631, 652, 653, 655):
-            self.assertIn(number, skipped, res["skipped"])
-            self.assertEqual(
-                skipped[number]["reason_code"],
-                SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
-            )
-            self.assertIn(
-                SKIP_EPIC_OR_CHILD_ONLY_CONTAINER, skipped[number]["reason"]
-            )
-
-    def test_containers_cannot_receive_assignment_or_lease(self) -> None:
-        containers = _live_shaped_containers()
-        res = self._alloc(containers, apply=True)
-        self.assertTrue(res["success"], res)
-        self.assertNotEqual(res["outcome"], OUTCOME_ASSIGNED)
-        self.assertIsNone(res.get("assignment"))
-        self.assertIsNone(res.get("selected"))
-        skipped = {s["number"]: s for s in res["skipped"]}
-        for number in (631, 652, 653, 655):
-            self.assertEqual(
-                skipped[number]["reason_code"],
-                SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
-            )
-
-        leases = []
-        if hasattr(self.db, "list_active_leases"):
-            leases = self.db.list_active_leases(
-                remote=REMOTE, org=ORG, repo=REPO
-            )
-        if not leases and hasattr(self.db, "list_leases"):
-            leases = self.db.list_leases(remote=REMOTE, org=ORG, repo=REPO)
-        for lease in leases or []:
-            work_number = (
-                lease.get("work_number") if isinstance(lease, dict) else None
-            )
-            self.assertNotIn(work_number, {631, 652, 653, 655})
-
-    def test_apply_selects_child_not_container(self) -> None:
-        containers = _live_shaped_containers()
-        child = _issue(
-            637,
-            title="Web Console: Workflow-event timeline model (Phase 1)",
-            body=_CHILD_BODY,
-        )
-        res = self._alloc(containers + [child], apply=True)
-        self.assertTrue(res["success"], res)
-        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
-        self.assertEqual(res["selected"]["number"], 637)
-        self.assertEqual(res["assignment"]["work_number"], 637)
-
-    def test_fingerprint_stable_with_containers_present(self) -> None:
-        containers = _live_shaped_containers()
-        child = _issue(
-            637,
-            title="Web Console: Workflow-event timeline model (Phase 1)",
-            body=_CHILD_BODY,
-        )
-        inventory = containers + [child]
-        fp_before = candidate_set_fingerprint(inventory)
-        res = self._alloc(inventory, apply=False)
-        self.assertTrue(res["success"], res)
-        self.assertEqual(res["selected"]["number"], 637)
-        # Allocator reports the same CAS fingerprint for the full candidate set.
-        reported = res.get("candidate_set_fingerprint")
-        self.assertEqual(reported, fp_before)
-        # Re-fingerprint of the same inventory is byte-stable.
-        self.assertEqual(candidate_set_fingerprint(inventory), fp_before)
-
-    def test_incidental_mentions_remain_eligible(self) -> None:
-        ordinary = _issue(
-            700,
-            title="Document roadmap handoff conventions",
-            body="Write runbook text about vision vs roadmap vs child issues.",
-        )
-        res = self._alloc([ordinary], apply=False)
-        self.assertTrue(res["success"], res)
-        self.assertEqual(res["selected"]["number"], 700)
-        self.assertEqual(res["skipped"], [])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_restart_classes.py

@@ -0,0 +1,232 @@
+"""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,6 +444,12 @@ 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,6 +38,7 @@ 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
 
@@ -99,6 +100,14 @@ 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 --------------------------------------------------------
 
 
@@ -256,6 +265,7 @@ 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,
@@ -309,6 +319,7 @@ 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,
@@ -393,6 +404,7 @@ def assess_restart_request(
             "process."
         ),
         "action_id": action_id,
+        "restart_class": restart_class_for_mode(md),
         "namespace": ns,
         "mode": md,
         "preview": preview,
@@ -441,7 +453,11 @@ def execute_restart(
             else console_audit.RESULT_DENIED
         ),
         principal=principal,
-        target={"namespace": assessment["namespace"], "mode": assessment["mode"]},
+        target={
+            "namespace": assessment["namespace"],
+            "mode": assessment["mode"],
+            "restart_class": assessment["restart_class"],
+        },
         reason_code=assessment["reason_code"],
         detail=assessment["detail"],
         request_id=request_id,
@@ -450,6 +466,7 @@ def execute_restart(
             "gates_passed": assessment["gates_passed"],
             "process_kill_executed": False,
             "post_restart_verification_required": True,
+            "restart_class": assessment["restart_class"],
         },
     )
 
@@ -463,6 +480,7 @@ 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": (