fix(restart): require both authorizations for apply, correct coordinator doc

Addresses the two blockers raised in the PR #886 review (comment 16559) for
issue #663.

B1 — apply_authorized ignored restart-class authorization.

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 this branch merged it. The union preserved both, but the apply
decision consulted only the drain gate:

    payload["apply_authorized"] = gate.allow

so a clean drain proof — or an authorized break-glass, which needs no proof
at all — reported apply_authorized: True for a class the least-privilege
matrix had just denied, in the same payload carrying allow_restart: False
and "role 'author' may not request full_mcp_restart". One environment
variable therefore collapsed the whole nine-class matrix for the apply
decision, including host_restart.

The apply decision is now the conjunction of both authorizations, and
apply_gate carries drain_gate_allow and restart_class_authorized so a denial
is attributable to the authorization that produced it. Break-glass keeps its
purpose — bypassing the drain proof — and never bypasses the class matrix.
No existing fail-closed behaviour is weakened: allow_restart, drain-proof
verification, fingerprint binding, and requester authorization are untouched.

B2 — docs/mcp-restart-coordinator.md described pre-#661 behaviour.

The document still called the drain proof "a separate child" and omitted
drain_proof_json and request_break_glass from the published signature, so a
safety document asserted there was no gate where a gate now exists. It now
documents both parameters, states that the gate executes inside this tool,
and records dry-run versus apply behaviour, authorization ordering, the
break-glass scope, and fail-closed conditions as implemented.

Regression coverage.

tests/test_issue_886_apply_authorization_conjunction.py exercises the MCP
tool itself, which previously had no test at all — that absence is why the
defect shipped. It pins both conjunction directions, proves a clean proof
cannot override a role, approval, unknown-class, or missing-target denial,
proves break-glass does not collapse the matrix for any worker role or
restricted class, and proves the existing scoped and unscoped paths and the
#661 denials still hold. Against the pre-fix tree 24 of these fail; against
this commit all 19 pass with 45 subtests.

tests/test_mcp_restart_governance_docs.py now binds the published signature
to inspect.signature() of the real tool and forbids the stale pre-#661
phrasing, so the drift that produced B2 cannot return unnoticed.

Verification: targeted restart/drain/governance/webui suites 194 passed,
113 subtests. Full suite 23 failed, 5230 passed, 6 skipped, 912 subtests —
the failure set is identical to the reviewed baseline at 9bc021e
(23 failed, 5201 passed), with +29 passing from the added tests and no new
or changed failure. Zero conflict markers; py_compile passes; the #882
union remains intact in both directions.

Refs #663, PR #886

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01V6xFqovhbArPv61j9KCGkL

allocator_service.py

@@ -133,10 +133,15 @@ 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). Matched case-insensitively against the
-# issue body. Title alone is never sufficient (ordinary issues may mention
-# "epic" incidentally).
+# 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.
 _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",
@@ -150,9 +155,26 @@ _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 labels (structured evidence preferred over title).
+# 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.
 _EPIC_LABELS: frozenset[str] = frozenset(
     {
         "type:epic",
@@ -161,6 +183,16 @@ _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",
     }
 )
 
@@ -222,16 +254,45 @@ 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 an epic / child-only implementation container (#844).
+    """Return whether *c* is a non-implementable coordination container (#844/#854).
 
     Exclusion uses structured evidence first (labels, body scope language).
-    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).
+    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.
 
     PRs are never classified as containers here (they already have a head).
     """
@@ -245,25 +306,28 @@ 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_epic_prefix = title_l.startswith("epic:") or title_l.startswith("epic ")
+    title_prefix = _title_container_prefix(title_l)
 
     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 / umbrella scope. Title "Epic:" is corroborating
-        # but not required — containers without the word still exclude.
+        # Body proves child-only / vision / roadmap / umbrella scope. Title
+        # prefixes are corroborating but not required — containers without the
+        # title word still exclude.
         detail = f"body_marker={body_hits[0]!r}"
-        if title_epic_prefix:
-            detail = f"title_epic_prefix; {detail}"
+        if title_prefix:
+            detail = f"{title_prefix}; {detail}"
         return True, detail
 
-    # 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.
+    # 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.
     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

@@ -6,11 +6,23 @@ 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 + a dry-run MCP tool. It is the single
+This lands the coordinator + impact DTO + the MCP tool. It is the single
 sanctioned entry point for restart evaluation post-#657 (which inventoried the
-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
-scope here.
+restart/reload/kill paths).
+
+The **drain-proof hard gate now executes inside this tool** (#661, via PR #882):
+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
 
@@ -19,7 +31,8 @@ scope here.
 | `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. Dry-run only. |
+| `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. |
+| `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
 
@@ -77,17 +90,61 @@ 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)
 ```
 
-Read-only, dry-run, and it **never restarts anything**. `apply_supported` is
-always `false`; passing `dry_run=False` performs no restart and reports that
-apply is gated by a drain proof (a separate child).
+It **never restarts anything**: `apply_supported` is always `false` and
+`restart_performed` is always `false`.
+
+### 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,
-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

@@ -2068,6 +2068,7 @@ import lease_lifecycle  # noqa: E402
 import lease_policy  # noqa: E402
 import workflow_dashboard  # noqa: E402  # #605 live queue/lease dashboard
 import restart_coordinator  # noqa: E402  # #658 MCP restart coordinator/impact
+import drain_proof  # noqa: E402  # #661 pre-restart drain proof and hard gate
 import incident_bridge  # noqa: E402
 import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
 import sentry_incident_bridge  # noqa: E402  (#607 Sentry→Gitea incident bridge)
@@ -22342,19 +22343,40 @@ 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: 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
     would disrupt *before* any concurrent LLM work is destroyed.
 
-    This tool is **dry-run and never restarts anything.** The mutative apply
-    path is a separate child gated by a drain proof (non-goal here); calling
-    with ``dry_run=False`` still performs no restart and reports that apply is
-    not yet available.
+    This tool **never restarts a process.** In dry-run (the default) it returns
+    only the impact preview. With ``dry_run=False`` it enforces the #661 hard
+    gate: the apply request must present a valid, unexpired, clean drain proof
+    (``drain_proof_json``) or it is denied and a durable incident descriptor is
+    returned under ``incident``. Break-glass is the only bypass and is honoured
+    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``.
+
+    ``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
@@ -22439,6 +22461,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 +22473,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 +22496,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,12 +22515,74 @@ 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
+    # and a durable incident is raised. Break-glass is the only bypass and its
+    # authorization is read from the environment, never self-asserted.
     payload["apply_supported"] = False
     if not dry_run:
-        payload["reasons"] = list(payload.get("reasons") or []) + [
-            "apply requested but not supported: sanctioned restart apply is "
-            "gated by a drain proof (separate child); no restart performed (#658)"
-        ]
+        proof_obj: dict | None = None
+        proof_parse_error: str | None = None
+        if drain_proof_json:
+            try:
+                parsed = json.loads(drain_proof_json)
+                proof_obj = parsed if isinstance(parsed, dict) else None
+                if proof_obj is None:
+                    proof_parse_error = "drain_proof_json is not a JSON object"
+            except (ValueError, TypeError) as exc:
+                proof_parse_error = f"invalid drain_proof_json: {_redact(str(exc))}"
+
+        break_glass_authorized = bool(
+            (
+                os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or ""
+            ).strip()
+        )
+        break_glass = bool(request_break_glass and break_glass_authorized)
+
+        expected_fp = drain_proof.impact_fingerprint(report.as_dict())
+        gate = drain_proof.gate_apply_restart(
+            proof=proof_obj,
+            break_glass=break_glass,
+            expected_impact_fingerprint=expected_fp,
+            requesting_session_id=sid,
+        )
+        gate_payload = gate.as_dict()
+        if proof_parse_error and not break_glass:
+            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["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
+        # child. The gate proves the apply path *would* be permitted.
+        payload["reasons"] = list(payload.get("reasons") or []) + list(
+            gate_payload.get("reasons") or []
+        )
+        if not gate.allow and gate.incident is not None:
+            payload["incident"] = gate.incident
     return payload
 
 

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

@@ -0,0 +1,907 @@
+"""Tests for the pre-restart drain proof and hard gate (#661).
+
+Covers the acceptance criteria:
+
+1. Restart apply without a proof fails closed.
+2. A successful drain produces a verifiable proof.
+3. An open unsafe mutation makes the proof fail (multi-session fixture).
+4. Pass / fail / expired verification paths.
+
+Plus the security posture: forged/tampered proofs are rejected, break-glass is
+the only bypass and is never silent, a stale blast-radius fingerprint rejects a
+proof, and no per-process secret ever leaks into a serialized artifact.
+"""
+
+from __future__ import annotations
+
+import os
+import unittest
+from datetime import datetime, timedelta, timezone
+
+import drain_proof as dp
+import restart_coordinator as rc
+
+
+NOW = datetime(2026, 7, 24, 6, 0, 0, tzinfo=timezone.utc)
+SECRET = b"unit-test-drain-proof-secret-0123456789abcdef"
+
+
+def _live_pid() -> int:
+    return os.getpid()
+
+
+def _clean_drain_state() -> dict:
+    """Every drain action succeeded, no sessions outstanding."""
+
+    return {
+        "assignments_stopped": True,
+        "checkpoints_complete": True,
+        "handoffs_verified": True,
+        "leases_handled": True,
+        "acks": {},  # no other live sessions to acknowledge
+        "ack_timeout_policy_applied": False,
+    }
+
+
+def _safe_report() -> dict:
+    """Impact report with no other live work: a restart here is safe."""
+
+    report = rc.evaluate_restart_impact(
+        {"sessions": [], "leases": [], "inventory_complete": True},
+        now=NOW,
+        requesting_session_id="prgs-controller-1-req",
+    )
+    return report.as_dict()
+
+
+def _unsafe_mutation_report() -> dict:
+    """Multi-session report: a second session holds a live author mutation."""
+
+    sessions = [
+        {
+            "session_id": "prgs-controller-1-req",
+            "role": "controller",
+            "profile": "prgs-controller",
+            "pid": _live_pid(),
+            "status": "active",
+            "last_heartbeat_at": NOW.isoformat(),
+        },
+        {
+            "session_id": "prgs-author-99",
+            "role": "author",
+            "profile": "prgs-author",
+            "pid": _live_pid(),
+            "status": "active",
+            "last_heartbeat_at": NOW.isoformat(),
+        },
+    ]
+    leases = [
+        {
+            "lease_id": "lease-mut",
+            "session_id": "prgs-author-99",
+            "role": "author",
+            "phase": "implementing",
+            "work_kind": "issue",
+            "work_number": 661,
+            "worktree_path": "branches/issue-661",
+            "freshness": {"freshness": "active"},
+        }
+    ]
+    report = rc.evaluate_restart_impact(
+        {"sessions": sessions, "leases": leases, "inventory_complete": True},
+        now=NOW,
+        requesting_session_id="prgs-controller-1-req",
+    )
+    return report.as_dict()
+
+
+class BuildDrainProofTests(unittest.TestCase):
+    def test_clean_drain_produces_verifiable_clean_proof(self):
+        """AC#2: a successful drain produces a verifiable proof."""
+
+        proof = dp.build_drain_proof(
+            impact_report=_safe_report(),
+            drain_state=_clean_drain_state(),
+            requesting_session_id="prgs-controller-1-req",
+            now=NOW,
+            secret=SECRET,
+        )
+        self.assertTrue(proof.clean)
+        self.assertEqual(proof.failed_checks, [])
+        self.assertEqual(
+            {c.name for c in proof.checks}, set(dp.REQUIRED_CHECKS)
+        )
+        result = dp.verify_drain_proof(
+            proof.as_dict(), now=NOW, secret=SECRET
+        )
+        self.assertTrue(result.valid, result.reasons)
+        self.assertFalse(result.expired)
+        self.assertFalse(result.tampered)
+
+    def test_open_mutation_makes_proof_unclean(self):
+        """AC#3: an unsafe mutation still in flight fails the proof."""
+
+        proof = dp.build_drain_proof(
+            impact_report=_unsafe_mutation_report(),
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        )
+        self.assertFalse(proof.clean)
+        self.assertIn(dp.CHECK_NO_INFLIGHT_MUTATIONS, proof.failed_checks)
+        # Leases-handled also fails: the report still shows a disruptive lease.
+        self.assertIn(dp.CHECK_LEASES_HANDLED, proof.failed_checks)
+        result = dp.verify_drain_proof(proof.as_dict(), now=NOW, secret=SECRET)
+        self.assertFalse(result.valid)
+
+    def test_incomplete_inventory_fails_no_mutations_check(self):
+        proof = dp.build_drain_proof(
+            impact_report={"inventory_complete": False},
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        )
+        self.assertFalse(proof.clean)
+        self.assertIn(dp.CHECK_NO_INFLIGHT_MUTATIONS, proof.failed_checks)
+
+    def test_missing_checkpoint_flag_fails_closed(self):
+        state = _clean_drain_state()
+        del state["checkpoints_complete"]
+        proof = dp.build_drain_proof(
+            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
+        )
+        self.assertFalse(proof.clean)
+        self.assertIn(dp.CHECK_CHECKPOINTS_COMPLETE, proof.failed_checks)
+
+    def test_non_true_flags_fail_closed(self):
+        """A truthy-but-not-True value (e.g. the string 'yes') must not pass."""
+
+        state = _clean_drain_state()
+        state["assignments_stopped"] = "yes"
+        proof = dp.build_drain_proof(
+            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
+        )
+        self.assertIn(dp.CHECK_ASSIGNMENTS_STOPPED, proof.failed_checks)
+
+    def test_ack_timeout_policy_satisfies_ack_check(self):
+        state = _clean_drain_state()
+        state["acks"] = {"prgs-author-99": "pending"}
+        state["ack_timeout_policy_applied"] = True
+        proof = dp.build_drain_proof(
+            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
+        )
+        names = {c.name: c.passed for c in proof.checks}
+        self.assertTrue(names[dp.CHECK_ACKS_OR_TIMEOUT])
+
+    def test_outstanding_acks_without_timeout_fail(self):
+        state = _clean_drain_state()
+        state["acks"] = {"prgs-author-99": "pending"}
+        state["ack_timeout_policy_applied"] = False
+        proof = dp.build_drain_proof(
+            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
+        )
+        self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
+
+    def test_all_acked_satisfies_ack_check(self):
+        state = _clean_drain_state()
+        state["acks"] = {"prgs-author-99": "acked", "prgs-author-2": "acknowledged"}
+        proof = dp.build_drain_proof(
+            impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
+        )
+        names = {c.name: c.passed for c in proof.checks}
+        self.assertTrue(names[dp.CHECK_ACKS_OR_TIMEOUT])
+
+
+class VerifyDrainProofTests(unittest.TestCase):
+    def _clean_proof_dict(self) -> dict:
+        return dp.build_drain_proof(
+            impact_report=_safe_report(),
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        ).as_dict()
+
+    def test_missing_proof_is_invalid(self):
+        result = dp.verify_drain_proof(None, now=NOW, secret=SECRET)
+        self.assertFalse(result.valid)
+        self.assertIsNone(result.proof_id)
+
+    def test_expired_proof_is_invalid(self):
+        """AC#4: an expired proof fails verification."""
+
+        proof = self._clean_proof_dict()
+        later = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS + 1)
+        result = dp.verify_drain_proof(proof, now=later, secret=SECRET)
+        self.assertFalse(result.valid)
+        self.assertTrue(result.expired)
+
+    def test_proof_valid_just_before_expiry(self):
+        proof = self._clean_proof_dict()
+        almost = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS - 1)
+        result = dp.verify_drain_proof(proof, now=almost, secret=SECRET)
+        self.assertTrue(result.valid, result.reasons)
+
+    def test_wrong_secret_rejected(self):
+        """A proof minted in a prior process (different secret) will not verify."""
+
+        proof = self._clean_proof_dict()
+        result = dp.verify_drain_proof(proof, now=NOW, secret=b"other-secret")
+        self.assertFalse(result.valid)
+        self.assertTrue(result.tampered)
+
+    def test_flipping_clean_flag_is_detected(self):
+        """Forging clean=True on an unclean proof breaks the signature."""
+
+        unclean = dp.build_drain_proof(
+            impact_report=_unsafe_mutation_report(),
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        ).as_dict()
+        self.assertFalse(unclean["clean"])
+        unclean["clean"] = True  # forge
+        result = dp.verify_drain_proof(unclean, now=NOW, secret=SECRET)
+        self.assertFalse(result.valid)
+        self.assertTrue(result.tampered)
+
+    def test_tampering_a_check_is_detected(self):
+        unclean = dp.build_drain_proof(
+            impact_report=_unsafe_mutation_report(),
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        ).as_dict()
+        for c in unclean["checks"]:
+            if c["name"] == dp.CHECK_NO_INFLIGHT_MUTATIONS:
+                c["passed"] = True  # forge the failing check to pass
+        result = dp.verify_drain_proof(unclean, now=NOW, secret=SECRET)
+        self.assertFalse(result.valid)
+        self.assertTrue(result.tampered)
+
+    def test_missing_required_check_rejected(self):
+        proof = self._clean_proof_dict()
+        proof["checks"] = [
+            c for c in proof["checks"] if c["name"] != dp.CHECK_HANDOFFS_OK
+        ]
+        result = dp.verify_drain_proof(proof, now=NOW, secret=SECRET)
+        self.assertFalse(result.valid)
+
+    def test_stale_fingerprint_rejected(self):
+        proof = self._clean_proof_dict()
+        result = dp.verify_drain_proof(
+            proof,
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint="deadbeef",
+        )
+        self.assertFalse(result.valid)
+
+    def test_matching_fingerprint_accepted(self):
+        report = _safe_report()
+        proof = dp.build_drain_proof(
+            impact_report=report,
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        ).as_dict()
+        fp = dp.impact_fingerprint(report)
+        result = dp.verify_drain_proof(
+            proof, now=NOW, secret=SECRET, expected_impact_fingerprint=fp
+        )
+        self.assertTrue(result.valid, result.reasons)
+
+
+class GateApplyRestartTests(unittest.TestCase):
+    def _clean_proof_dict(self) -> dict:
+        return dp.build_drain_proof(
+            impact_report=_safe_report(),
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        ).as_dict()
+
+    def test_apply_without_proof_denied(self):
+        """AC#1: restart apply without a proof fails closed + raises incident."""
+
+        decision = dp.gate_apply_restart(proof=None, now=NOW, secret=SECRET)
+        self.assertFalse(decision.allow)
+        self.assertEqual(decision.verdict, dp.GATE_DENY)
+        self.assertIsNotNone(decision.incident)
+        self.assertEqual(
+            decision.incident["kind"], "restart_drain_gate_denied"
+        )
+
+    def test_apply_with_valid_proof_allowed(self):
+        decision = dp.gate_apply_restart(
+            proof=self._clean_proof_dict(), now=NOW, secret=SECRET
+        )
+        self.assertTrue(decision.allow)
+        self.assertEqual(decision.verdict, dp.GATE_ALLOW)
+        self.assertIsNone(decision.incident)
+
+    def test_apply_with_expired_proof_denied_with_incident(self):
+        later = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS + 5)
+        decision = dp.gate_apply_restart(
+            proof=self._clean_proof_dict(), now=later, secret=SECRET
+        )
+        self.assertFalse(decision.allow)
+        self.assertIsNotNone(decision.incident)
+
+    def test_apply_with_unclean_proof_denied(self):
+        """AC#3 at the gate: an unsafe-mutation proof is denied."""
+
+        unclean = dp.build_drain_proof(
+            impact_report=_unsafe_mutation_report(),
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        ).as_dict()
+        decision = dp.gate_apply_restart(proof=unclean, now=NOW, secret=SECRET)
+        self.assertFalse(decision.allow)
+        self.assertIsNotNone(decision.incident)
+
+    def test_break_glass_allows_without_proof_but_records_bypass(self):
+        decision = dp.gate_apply_restart(
+            proof=None, now=NOW, secret=SECRET, break_glass=True
+        )
+        self.assertTrue(decision.allow)
+        self.assertEqual(decision.verdict, dp.GATE_BREAK_GLASS)
+        self.assertTrue(decision.break_glass)
+        self.assertIsNone(decision.incident)
+        self.assertTrue(decision.audit_record["break_glass"])
+
+    def test_denied_gate_carries_stale_fingerprint_reason(self):
+        decision = dp.gate_apply_restart(
+            proof=self._clean_proof_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint="not-the-fingerprint",
+        )
+        self.assertFalse(decision.allow)
+
+
+class SecretHygieneTests(unittest.TestCase):
+    def test_secret_never_serialized(self):
+        proof = dp.build_drain_proof(
+            impact_report=_safe_report(),
+            drain_state=_clean_drain_state(),
+            now=NOW,
+            secret=SECRET,
+        )
+        blob = dp._canonical(proof.as_dict())
+        self.assertNotIn(SECRET.decode(), blob)
+        # The signature is a hex digest, not the raw secret.
+        self.assertNotIn(SECRET.hex(), blob)
+
+    def test_incident_descriptor_has_no_secret(self):
+        decision = dp.gate_apply_restart(proof=None, now=NOW, secret=SECRET)
+        blob = dp._canonical(decision.incident)
+        self.assertNotIn(SECRET.decode(), blob)
+
+
+def _drained_report_with_live_sessions(count: int) -> dict:
+    """Report with ``count`` other live sessions but nothing in flight.
+
+    Every other checklist item passes against this report, so a failure
+    isolates the acknowledgement check rather than tripping on mutations.
+    """
+
+    sessions = [
+        {
+            "session_id": "prgs-controller-1-req",
+            "role": "controller",
+            "profile": "prgs-controller",
+            "pid": _live_pid(),
+            "status": "active",
+            "last_heartbeat_at": NOW.isoformat(),
+        }
+    ]
+    for index in range(count):
+        sessions.append(
+            {
+                "session_id": f"prgs-author-{index}",
+                "role": "author",
+                "profile": "prgs-author",
+                "pid": _live_pid(),
+                "status": "active",
+                "last_heartbeat_at": NOW.isoformat(),
+            }
+        )
+    report = rc.evaluate_restart_impact(
+        {"sessions": sessions, "leases": [], "inventory_complete": True},
+        now=NOW,
+        requesting_session_id="prgs-controller-1-req",
+    )
+    return report.as_dict()
+
+
+class AcknowledgementFailClosedTests(unittest.TestCase):
+    """Acknowledgement evidence must fail closed unless explicitly verified.
+
+    Regression cover for the reviewed fail-open on PR #882: an absent ``acks``
+    key collapsed to ``{}`` and was read as "no other live sessions required to
+    acknowledge", so a proof minted clean and the restart gate allowed while the
+    impact report still showed other live sessions.
+    """
+
+    def _state(self, **overrides) -> dict:
+        state = _clean_drain_state()
+        state.pop("acks", None)
+        state["ack_timeout_policy_applied"] = False
+        state.update(overrides)
+        return state
+
+    def _acks_check(self, proof) -> dp.DrainCheck:
+        return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT)
+
+    def _build(self, report: dict, state: dict):
+        return dp.build_drain_proof(
+            impact_report=report, drain_state=state, now=NOW, secret=SECRET
+        )
+
+    def assertAcksFailClosed(self, report: dict, state: dict) -> None:
+        proof = self._build(report, state)
+        self.assertFalse(self._acks_check(proof).passed)
+        self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
+        self.assertFalse(proof.clean)
+
+    # --- missing / null / empty / malformed ------------------------------
+
+    def test_missing_acks_key_with_live_sessions_fails_closed(self):
+        """The exact reviewed defect: absent key, three other live sessions."""
+        report = _drained_report_with_live_sessions(3)
+        self.assertEqual(report["counts"]["sessions_live_other"], 3)
+        state = self._state()
+        self.assertNotIn("acks", state)
+        proof = self._build(report, state)
+        check = self._acks_check(proof)
+        self.assertFalse(check.passed)
+        self.assertNotIn("no other live sessions", check.detail)
+        self.assertIn("fail closed", check.detail)
+        self.assertFalse(proof.clean)
+        self.assertEqual(proof.failed_checks, [dp.CHECK_ACKS_OR_TIMEOUT])
+
+    def test_none_acks_with_live_sessions_fails_closed(self):
+        self.assertAcksFailClosed(
+            _drained_report_with_live_sessions(2), self._state(acks=None)
+        )
+
+    def test_empty_acks_with_live_sessions_fails_closed(self):
+        self.assertAcksFailClosed(
+            _drained_report_with_live_sessions(1), self._state(acks={})
+        )
+
+    def test_malformed_acks_fail_closed(self):
+        for malformed in ([], "ack", 7, ("ack",), True):
+            with self.subTest(malformed=malformed):
+                self.assertAcksFailClosed(
+                    _drained_report_with_live_sessions(1),
+                    self._state(acks=malformed),
+                )
+
+    # --- stale / unproven values -----------------------------------------
+
+    def test_stale_or_unproven_ack_values_fail_closed(self):
+        for value in ("pending", "stale", "unknown", "", None, True, 1, NOW):
+            with self.subTest(value=value):
+                self.assertAcksFailClosed(
+                    _drained_report_with_live_sessions(1),
+                    self._state(acks={"prgs-author-0": value}),
+                )
+
+    def test_partial_coverage_fails_closed(self):
+        """Fewer acknowledgements than the report's live-session count."""
+        self.assertAcksFailClosed(
+            _drained_report_with_live_sessions(3),
+            self._state(acks={"prgs-author-0": "ack"}),
+        )
+
+    def test_one_unacked_entry_among_many_fails_closed(self):
+        self.assertAcksFailClosed(
+            _drained_report_with_live_sessions(2),
+            self._state(acks={"prgs-author-0": "ack", "prgs-author-1": "pending"}),
+        )
+
+    def test_unproven_live_session_count_fails_closed(self):
+        """A missing or malformed count cannot prove nobody had to acknowledge."""
+        malformed_counts = (
+            None,
+            {},
+            {"sessions_live_other": None},
+            {"sessions_live_other": "3"},
+            {"sessions_live_other": -1},
+            {"sessions_live_other": True},
+        )
+        for counts in malformed_counts:
+            with self.subTest(counts=counts):
+                report = _drained_report_with_live_sessions(0)
+                if counts is None:
+                    report.pop("counts", None)
+                else:
+                    report["counts"] = counts
+                self.assertAcksFailClosed(report, self._state())
+
+    # --- valid evidence still passes -------------------------------------
+
+    def test_complete_valid_acks_pass(self):
+        report = _drained_report_with_live_sessions(2)
+        state = self._state(
+            acks={"prgs-author-0": "ack", "prgs-author-1": "acknowledged"}
+        )
+        proof = self._build(report, state)
+        self.assertTrue(self._acks_check(proof).passed)
+        self.assertTrue(proof.clean)
+        self.assertEqual(proof.failed_checks, [])
+
+    def test_no_other_live_sessions_still_passes(self):
+        """Intended behavior retained: zero live sessions needs no acks."""
+        report = _drained_report_with_live_sessions(0)
+        self.assertEqual(report["counts"]["sessions_live_other"], 0)
+        proof = self._build(report, self._state())
+        check = self._acks_check(proof)
+        self.assertTrue(check.passed)
+        self.assertIn("sessions_live_other=0", check.detail)
+        self.assertTrue(proof.clean)
+
+    # --- timeout policy cannot become a second fail-open ------------------
+
+    def test_unproven_timeout_policy_cannot_open_the_gate(self):
+        for value in (None, "true", "yes", 1, "True", [], {}):
+            with self.subTest(value=value):
+                self.assertAcksFailClosed(
+                    _drained_report_with_live_sessions(2),
+                    self._state(ack_timeout_policy_applied=value),
+                )
+
+    def test_explicit_timeout_policy_permits(self):
+        proof = self._build(
+            _drained_report_with_live_sessions(2),
+            self._state(ack_timeout_policy_applied=True),
+        )
+        check = self._acks_check(proof)
+        self.assertTrue(check.passed)
+        self.assertIn("timeout policy", check.detail)
+        self.assertTrue(proof.clean)
+
+    # --- the gate itself must deny ---------------------------------------
+
+    def test_failed_ack_check_denies_the_restart_gate(self):
+        report = _drained_report_with_live_sessions(3)
+        proof = self._build(report, self._state())
+        self.assertFalse(proof.clean)
+        decision = dp.gate_apply_restart(
+            proof=proof.as_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint=dp.impact_fingerprint(report),
+        )
+        self.assertFalse(decision.allow)
+        self.assertEqual(decision.verdict, dp.GATE_DENY)
+        self.assertIsNotNone(decision.incident)
+
+    def test_unclean_ack_proof_fails_verification(self):
+        report = _drained_report_with_live_sessions(3)
+        proof = self._build(report, self._state())
+        result = dp.verify_drain_proof(
+            proof.as_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint=dp.impact_fingerprint(report),
+        )
+        self.assertFalse(result.valid)
+        self.assertFalse(result.clean)
+
+
+def _identity_report(*, requester: str, others: tuple[str, ...]) -> dict:
+    """Report with explicitly named requester and other live sessions.
+
+    Unlike :func:`_drained_report_with_live_sessions`, the session ids are
+    chosen by the caller so a test can supply acknowledgements for the *wrong*
+    identities while keeping the count correct.
+    """
+
+    sessions = [
+        {
+            "session_id": requester,
+            "role": "controller",
+            "profile": "prgs-controller",
+            "pid": _live_pid(),
+            "status": "active",
+            "last_heartbeat_at": NOW.isoformat(),
+        }
+    ]
+    for session_id in others:
+        sessions.append(
+            {
+                "session_id": session_id,
+                "role": "author",
+                "profile": "prgs-author",
+                "pid": _live_pid(),
+                "status": "active",
+                "last_heartbeat_at": NOW.isoformat(),
+            }
+        )
+    report = rc.evaluate_restart_impact(
+        {"sessions": sessions, "leases": [], "inventory_complete": True},
+        now=NOW,
+        requesting_session_id=requester,
+    )
+    return report.as_dict()
+
+
+class AcknowledgementIdentityBindingTests(unittest.TestCase):
+    """Acknowledgement coverage must be bound to session identity, not counted.
+
+    Regression cover for the second reviewed fail-open on PR #882 (review 582,
+    blocker B1): coverage compared ``acked_count`` against
+    ``counts.sessions_live_other``, so acknowledgements supplied for the
+    requesting session and for ids that do not exist satisfied the obligations
+    of the live sessions that never answered. The required identities are
+    carried by the report itself — ``ack_state`` keys and ``affected_sessions``
+    filtered on ``live and not is_requester`` — and only an acknowledgement
+    keyed by one of those ids may count for it.
+    """
+
+    def _state(self, **overrides) -> dict:
+        state = _clean_drain_state()
+        state.pop("acks", None)
+        state["ack_timeout_policy_applied"] = False
+        state.update(overrides)
+        return state
+
+    def _acks_check(self, proof) -> dp.DrainCheck:
+        return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT)
+
+    def _build(self, report: dict, state: dict):
+        return dp.build_drain_proof(
+            impact_report=report, drain_state=state, now=NOW, secret=SECRET
+        )
+
+    def assertAcksFailClosed(self, report: dict, state: dict) -> dp.DrainCheck:
+        """Failure must propagate through the check, the proof, and the gate."""
+
+        proof = self._build(report, state)
+        check = self._acks_check(proof)
+        self.assertFalse(check.passed)
+        self.assertFalse(proof.clean)
+        self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
+        decision = dp.gate_apply_restart(
+            proof=proof.as_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint=dp.impact_fingerprint(report),
+        )
+        self.assertEqual(decision.verdict, dp.GATE_DENY)
+        self.assertFalse(decision.allow)
+        return check
+
+    # --- the reviewer's exact reproduction --------------------------------
+
+    def test_requester_plus_unknown_id_cannot_satisfy_two_live_sessions(self):
+        """Review 582 B1 verbatim: requester + a nonexistent session.
+
+        ``sessions_live_other=2`` with ``ack_state`` naming ``other-0`` and
+        ``other-1``; the drain state supplies an acknowledgement from the
+        requesting session itself and from a session that does not exist. The
+        count matches, the identities do not.
+        """
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        self.assertEqual(report["counts"]["sessions_live_other"], 2)
+        self.assertEqual(
+            report["ack_state"], {"other-0": "pending", "other-1": "pending"}
+        )
+        state = self._state(acks={"req": "ack", "totally-bogus-session": "ack"})
+        check = self.assertAcksFailClosed(report, state)
+        self.assertIn("other-0", check.detail)
+        self.assertIn("other-1", check.detail)
+        self.assertIn("fail closed", check.detail)
+
+    # --- wrong / unknown / requester identities ---------------------------
+
+    def test_sufficient_count_of_wrong_ids_fails_closed(self):
+        """Right cardinality, wrong identities: two acks, neither required."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        state = self._state(acks={"ghost-a": "ack", "ghost-b": "ack"})
+        check = self.assertAcksFailClosed(report, state)
+        self.assertIn("do not count", check.detail)
+
+    def test_more_acks_than_required_still_fails_on_wrong_ids(self):
+        """Coverage cannot be bought with volume: five acks, none required."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        state = self._state(acks={f"ghost-{i}": "acknowledged" for i in range(5)})
+        self.assertAcksFailClosed(report, state)
+
+    def test_partial_identity_match_fails_closed(self):
+        """One required id acknowledged, the rest padded with unknown ids."""
+
+        report = _identity_report(
+            requester="req", others=("other-0", "other-1", "other-2")
+        )
+        state = self._state(
+            acks={"other-0": "ack", "ghost-1": "ack", "ghost-2": "ack"}
+        )
+        check = self.assertAcksFailClosed(report, state)
+        self.assertIn("other-1", check.detail)
+        self.assertIn("other-2", check.detail)
+
+    def test_requester_ack_never_satisfies_another_sessions_obligation(self):
+        """The requester is excluded from the required set and stays excluded."""
+
+        report = _identity_report(requester="req", others=("other-0",))
+        requester_rows = [s for s in report["affected_sessions"] if s["is_requester"]]
+        self.assertEqual([s["session_id"] for s in requester_rows], ["req"])
+        self.assertNotIn("req", report["ack_state"])
+        check = self.assertAcksFailClosed(report, self._state(acks={"req": "ack"}))
+        self.assertIn("other-0", check.detail)
+
+    def test_fabricated_ids_do_not_count_toward_coverage(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        for bogus in ("", " ", "other-0 extra", "OTHER-0", "other-01", "0"):
+            with self.subTest(bogus=bogus):
+                self.assertAcksFailClosed(report, self._state(acks={bogus: "ack"}))
+
+    # --- per-session state must be explicitly valid ------------------------
+
+    def test_unproven_per_session_states_fail_closed(self):
+        """A required id present but not explicitly acknowledged fails closed."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        for value in ("pending", "stale", "unknown", "", None, True, 1, NOW):
+            with self.subTest(value=value):
+                self.assertAcksFailClosed(
+                    report,
+                    self._state(acks={"other-0": "ack", "other-1": value}),
+                )
+
+    def test_report_ack_state_placeholder_is_never_read_as_an_ack(self):
+        """``ack_state`` values are the report's own placeholders, not evidence."""
+
+        report = _identity_report(requester="req", others=("other-0",))
+        report["ack_state"] = {"other-0": "ack"}
+        self.assertAcksFailClosed(report, self._state())
+
+    # --- missing / malformed / contradictory identity evidence -------------
+
+    def test_missing_identity_evidence_fails_closed(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        report.pop("ack_state", None)
+        report.pop("affected_sessions", None)
+        check = self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
+        self.assertIn("no session-identity evidence", check.detail)
+
+    def test_malformed_ack_state_fails_closed(self):
+        for malformed in ([], "other-0", 7, None, ("other-0",)):
+            with self.subTest(malformed=malformed):
+                report = _identity_report(requester="req", others=("other-0",))
+                report["ack_state"] = malformed
+                self.assertAcksFailClosed(
+                    report, self._state(acks={"other-0": "ack"})
+                )
+
+    def test_non_string_ack_state_key_fails_closed(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        report["ack_state"] = {7: "pending"}
+        self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
+
+    def test_malformed_affected_sessions_fails_closed(self):
+        for malformed in ("sessions", 7, {"session_id": "other-0"}, [None], [7]):
+            with self.subTest(malformed=malformed):
+                report = _identity_report(requester="req", others=("other-0",))
+                report.pop("ack_state", None)
+                report["affected_sessions"] = malformed
+                self.assertAcksFailClosed(
+                    report, self._state(acks={"other-0": "ack"})
+                )
+
+    def test_affected_sessions_without_explicit_booleans_fails_closed(self):
+        """``live``/``is_requester`` must be real booleans, never inferred."""
+
+        report = _identity_report(requester="req", others=("other-0",))
+        report.pop("ack_state", None)
+        for row in report["affected_sessions"]:
+            if row["session_id"] == "other-0":
+                row["is_requester"] = "false"
+        self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
+
+    def test_affected_sessions_missing_live_flag_fails_closed(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        report.pop("ack_state", None)
+        for row in report["affected_sessions"]:
+            row.pop("live", None)
+        self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
+
+    def test_contradictory_ack_state_and_affected_sessions_fails_closed(self):
+        """Both views present and disagreeing is unresolvable, not a tie-break."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        report["ack_state"] = {"other-0": "pending", "other-9": "pending"}
+        check = self.assertAcksFailClosed(
+            report, self._state(acks={"other-0": "ack", "other-9": "ack"})
+        )
+        self.assertIn("contradicts itself", check.detail)
+
+    def test_identity_count_mismatch_fails_closed(self):
+        """Identity evidence that cannot be reconciled with the count denies."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        report["counts"] = dict(report["counts"], sessions_live_other=1)
+        check = self.assertAcksFailClosed(
+            report, self._state(acks={"other-0": "ack", "other-1": "ack"})
+        )
+        self.assertIn("cannot be reconciled", check.detail)
+
+    def test_broken_identity_evidence_outranks_timeout_policy(self):
+        """The sanctioned timeout path cannot paper over an unreadable report."""
+
+        report = _identity_report(requester="req", others=("other-0",))
+        report["ack_state"] = "not-a-mapping"
+        self.assertAcksFailClosed(report, self._state(ack_timeout_policy_applied=True))
+
+    # --- legitimate success is preserved -----------------------------------
+
+    def test_every_required_session_acknowledged_passes(self):
+        report = _identity_report(
+            requester="req", others=("other-0", "other-1", "other-2")
+        )
+        state = self._state(
+            acks={
+                "other-0": "ack",
+                "other-1": "acked",
+                "other-2": "acknowledged",
+            }
+        )
+        proof = self._build(report, state)
+        check = self._acks_check(proof)
+        self.assertTrue(check.passed)
+        self.assertTrue(proof.clean)
+        self.assertEqual(proof.failed_checks, [])
+        self.assertIn("acknowledged by identity", check.detail)
+        decision = dp.gate_apply_restart(
+            proof=proof.as_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint=dp.impact_fingerprint(report),
+        )
+        self.assertEqual(decision.verdict, dp.GATE_ALLOW)
+        self.assertTrue(decision.allow)
+
+    def test_required_session_ack_tolerates_surrounding_whitespace(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        proof = self._build(report, self._state(acks={" other-0 ": " ACK "}))
+        self.assertTrue(self._acks_check(proof).passed)
+        self.assertTrue(proof.clean)
+
+    def test_no_other_live_sessions_still_passes_with_identity_evidence(self):
+        report = _identity_report(requester="req", others=())
+        self.assertEqual(report["counts"]["sessions_live_other"], 0)
+        self.assertEqual(report["ack_state"], {})
+        proof = self._build(report, self._state())
+        check = self._acks_check(proof)
+        self.assertTrue(check.passed)
+        self.assertIn("sessions_live_other=0", check.detail)
+        self.assertTrue(proof.clean)
+
+    def test_explicit_timeout_policy_retains_intended_behavior(self):
+        """Valid, correctly typed timeout evidence still permits the check."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        proof = self._build(report, self._state(ack_timeout_policy_applied=True))
+        check = self._acks_check(proof)
+        self.assertTrue(check.passed)
+        self.assertIn("timeout policy", check.detail)
+        self.assertTrue(proof.clean)
+
+    def test_timeout_policy_still_strictly_typed_under_identity_binding(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        for value in (None, "true", "True", 1, [], {}):
+            with self.subTest(value=value):
+                self.assertAcksFailClosed(
+                    report, self._state(ack_timeout_policy_applied=value)
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_854_semantic_container_exclusion.py

@@ -0,0 +1,378 @@
+"""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_issue_886_apply_authorization_conjunction.py

@@ -0,0 +1,381 @@
+"""``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_mcp_restart_governance_docs.py

@@ -105,3 +105,120 @@ 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

@@ -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": (