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

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.
 

docs/webui-authz-audit.md

@@ -94,7 +94,6 @@ already define, and a regression test asserts each mapping matches.
 | `record_analytics_usage` | operator | gated_write | `runtime.record_analytics_usage` | Yes | No | No | 2 |
 | `system.reload_namespace` | controller | privileged | `runtime.reload_namespace` | Yes | No | No | 2 |
 | `system.restart_namespace` | admin | destructive | `runtime.restart_namespace` | Yes | **Yes** | **Yes** | 2 |
-| `initiate_workflow` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
 
 **Dual control** means the acting principal may not be the sole authority: a
 second distinct principal must confirm. **Break-glass** means the action is
@@ -113,12 +112,6 @@ by the console — both hand off to a host supervisor, and neither exposes a raw
 process kill. See
 [`sanctioned-restart-controls.md`](sanctioned-restart-controls.md) (#642).
 
-`initiate_workflow` (#643) is operator-class because its outcome is a *claim*,
-not a Gitea verdict. Requesting reviewer or merger work reserves that work
-through the allocator; it does not grant the right to approve or merge, which
-stays with the MCP role profile and its own capability gates. See
-[`webui-requests.md`](webui-requests.md).
-
 ### Authorization decision
 
 `authorize(action_id, principal, for_execution=False)` returns a decision
@@ -133,24 +126,9 @@ record and **denies by default**. The deny reasons are closed and enumerated:
 | `phase_not_active` | Execution requested for an action whose phase is not open. |
 | `allowed_preview_only` | Authorized — preview only, execution still disabled. |
 
-There is no implicit allow branch.
-
-`execution_enabled` on the decision reports whether the action has a live
-execution path at all, and is computed by `execution_wired(action)`. There are
-exactly two ways to be wired:
-
-1. the action's `phase` is at or below `ACTIVE_PHASE`; or
-2. the action declares an `execution_env_flag` **and** that variable is set.
-
-Every action that declares no flag therefore reports `execution_enabled: false`
-while the console is in Phase 1, so no caller can read an allow as permission
-to mutate. The per-action flag exists because raising `ACTIVE_PHASE` would
-enable execution for every action of that phase at once, including ones whose
-execution path is not implemented. One implemented action goes live on its own
-flag instead of dragging its unimplemented phase-mates with it.
-
-`initiate_workflow` is the only action that currently declares a flag
-(`WEBUI_REQUESTS_EXECUTION`), and it stays denied until an operator sets it.
+There is no implicit allow branch. Even the allow result reports
+`execution_enabled: false` while the console is in Phase 1, so no caller can
+read an allow as permission to mutate.
 
 ## Secret redaction
 
@@ -257,22 +235,13 @@ second one. The integration points are already wired and observable:
   instead of adding a parallel check.
 - **`GET /api/console/security-model`** publishes the RBAC matrix, redaction
   policy, and audit policy as JSON for operators and tests.
-- **`POST /api/v1/requests/preview` and `.../apply`** (#643) are the first
-  actions to use this model for a real execution path. Preview always returns a
-  decision and an audited `previewed` record; apply requires `confirm=true`,
-  emits `succeeded` or `denied`, and reserves work only through the allocator.
-  See [`webui-requests.md`](webui-requests.md).
 
-A Phase 2 action must: use `execution_wired` rather than a private enable flag,
-implement the confirmation and dual-control flow the matrix already declares,
-emit a `succeeded` or `failed` record alongside the `gitea_audit` mutation
-record, and keep `viewer` unable to reach any of it. Turning on execution
-without the confirmation flow contradicts a declared requirement and is a
-review failure, not a shortcut.
-
-Raising `ACTIVE_PHASE` remains the way to open a whole phase at once, and is
-deliberately *not* what #643 did: an action-scoped opt-in cannot enable an
-action whose execution path nobody wrote.
+To open Phase 2, a child issue must: raise `ACTIVE_PHASE`, implement the
+confirmation and dual-control flow the matrix already declares, emit a
+`succeeded` or `failed` record alongside the `gitea_audit` mutation record, and
+keep `viewer` unable to reach any of it. Turning on execution without the
+confirmation flow contradicts a declared requirement and is a review failure,
+not a shortcut.
 
 ## Local-dev mode
 
@@ -325,7 +294,6 @@ Until Phase 2 wires it, probe protection rests on network placement alone, as
 | `WEBUI_ROLE_MAP` | unset | JSON subject → role map |
 | `WEBUI_REQUIRE_PROBE_AUTH` | unset | Require auth for non-public probes |
 | `WEBUI_CONSOLE_AUDIT_LOG` | unset | Append-only audit sink path |
-| `WEBUI_REQUESTS_EXECUTION` | unset | Opt in to `initiate_workflow` execution (#643) |
 
 All are read server-side only. None is ever rendered into a page or returned by
 an API.

docs/webui-requests.md

@@ -1,160 +0,0 @@
-# Web console requests: intent preview and workflow initiation (#643)
-
-**Phase 2. Preview is always live and always read-only. Initiation is wired but
-denied until an operator opts in.**
-
-Before this surface, starting role work meant pasting a prompt into a terminal
-and trusting the operator to have checked the allocator first. Nothing enforced
-that check, so two sessions could reach for the same issue and each believe it
-was theirs. This page replaces the paste with a *request*: a desired role, an
-issue or PR, and a stated intent, answered by an authorization decision and —
-on confirmation — an exclusive assignment from the allocator.
-
-| Concern | Module |
-|---------|--------|
-| Request model, preview, initiation | `webui/request_service.py` |
-| Form and preview rendering | `webui/request_views.py` |
-| Authorization | `webui/console_authz.py` (`initiate_workflow`) |
-| Audit | `webui/console_audit.py` |
-| Ownership substrate | `allocator_service.py` + `control_plane_db.py` |
-
-## Surfaces
-
-| Path | Method | Purpose |
-|------|--------|---------|
-| `/requests` | GET | Request form |
-| `/requests` | POST | Render an intent preview. **Never assigns.** |
-| `/api/v1/requests/preview` | POST | Intent preview as JSON |
-| `/api/v1/requests/apply` | POST | Initiate — confirmed, audited, allocator-owned |
-
-The HTML form has no initiate button on purpose. Initiating requires a
-confirmed POST to `/api/v1/requests/apply`, so a stray form submission cannot
-reserve work as a side effect.
-
-## The request
-
-```json
-{
-  "desired_role": "author",
-  "work_kind": "issue",
-  "work_number": 643,
-  "intent_summary": "implement request preview and initiation",
-  "remote": "prgs",
-  "org": "Scaled-Tech-Consulting",
-  "repo": "Gitea-Tools",
-  "expected_head_sha": null
-}
-```
-
-`desired_role` is one of `author`, `reviewer`, `merger`, `reconciler`,
-`controller`. `work_kind` is `issue` or `pr`. `remote`/`org`/`repo` default to
-the first project in the registry when omitted; when neither the request nor
-the registry resolves them, the request is rejected rather than pointed at some
-other repository. `intent_summary` is required — it is what the audit record
-states as the reason — and is truncated to 500 characters.
-
-Parsing rejects rather than corrects. An unknown role, an unknown work kind, a
-non-positive number, or a missing intent each return `400` with a `reason_code`
-and the offending `field`.
-
-## Preview
-
-Five checks, each with its own verdict, reason code, and detail:
-
-| Check | Passes when |
-|-------|-------------|
-| `authorization` | The console principal holds `operator` or above |
-| `capability` | The desired role maps to a declared profile and MCP namespace |
-| `lease_availability` | No active claim holds the work unit |
-| `next_safe_action` | The allocator would independently select this exact work unit |
-| `head_pin` | PR work resolves to a head SHA, and a supplied SHA still matches |
-
-A preview also returns the role's `allowed_actions` and `prohibited_actions`
-(from `allocator_service.ROLE_ACTIONS`), the `required_profile` and
-`required_namespace` the work must run under, and a `correlation_id` that ties
-the preview to its audit record and to any assignment that follows.
-
-Preview is read-only in the strict sense: it calls the allocator with
-`apply=false` and writes nothing but an audit line. An unauthorized principal
-never reaches the allocator or the control-plane DB at all, so a denial cannot
-be used to enumerate the queue.
-
-## Initiation
-
-`POST /api/v1/requests/apply` refuses in this order, and every refusal returns
-before any assignment is attempted:
-
-| Condition | Outcome | Status |
-|-----------|---------|--------|
-| Unparseable request | `invalid_request` | 400 |
-| Not authorized, or execution not wired | `denied` | 403 |
-| `confirm` not set | `denied` / `confirmation_required` | 409 |
-| Work unit already claimed | `blocked` / `duplicate_assignment` | 409 |
-| Allocator would select other work | `wait` / `not_next_safe_work` | 409 |
-| Allocator declines on apply | `blocked` or `wait` | 409 |
-| Evidence unavailable | `wait` / `evidence_unavailable` | 503 |
-| Assigned | `assigned_work` | 201 |
-
-A success returns the assignment plus a `handoff` block naming the profile, the
-namespace, and the actions that stay forbidden — enough for the operator to
-continue in the right MCP namespace without guessing.
-
-### Why apply runs the allocator twice
-
-The allocator is the only source of exclusive ownership (#600 / #613), and it
-selects work; it does not take orders. So `apply` runs a dry-run first and
-proceeds only when the allocator would independently pick the requested work
-unit. If it would not, the request reports `wait` and mutates nothing.
-
-A request is therefore a *confirmation* of the allocator's decision, never an
-override of it. The apply call carries the dry-run's
-`candidate_set_fingerprint` as a CAS pin (#776), so a queue that changed
-between the two calls fails closed rather than assigning against a stale view.
-The result is checked again on the way out: an assignment naming a different
-work unit is not read as success.
-
-### Fail-closed defaults
-
-- An unreadable control-plane DB denies. It is never treated as "nothing holds
-  this work unit".
-- An incomplete queue inventory denies (#758). Ranking a partial candidate set
-  can select the wrong work.
-- An allocator that raises denies.
-- PR work with no resolvable head SHA denies; a supplied SHA that no longer
-  matches denies with `head_moved`.
-
-## Enabling initiation
-
-Execution is wired off. Set `WEBUI_REQUESTS_EXECUTION=1` to enable it for the
-`initiate_workflow` action only — see
-[`webui-authz-audit.md`](webui-authz-audit.md) for why this is an
-action-scoped flag rather than a phase bump. With the variable unset, `apply`
-returns `403` with `reason_code: unauthorized` no matter who asks.
-
-Enabling execution does **not** enable approvals or merges. Those are phase 3
-console actions and remain forbidden in every path here; the console reserves
-work and hands off, and the MCP role profile enforces what that role may then
-do.
-
-## Audit
-
-Every preview and every apply emits a console audit record (schema in
-[`webui-authz-audit.md`](webui-authz-audit.md)):
-
-| Event | `result` |
-|-------|----------|
-| Preview | `previewed` |
-| Refusal at any stage | `denied` |
-| Assignment created | `succeeded` |
-
-`correlation.request_id` carries the request's `correlation_id`, and a
-successful record's `metadata` carries `assignment_id` and `lease_id`, so an
-assignment can be traced back to the intent that produced it. The operator's
-`intent_summary` travels in `metadata` and passes through the standard
-redaction pass before persistence like every other field.
-
-## Non-goals
-
-- No browser-initiated approve or merge, in this phase or any other.
-- No bypass of allocator exclusive ownership; no self-selection of work.
-- No auto-start from raw monitoring incidents (#612 stays downstream).

gitea_mcp_server.py

@@ -22343,12 +22343,17 @@ 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
@@ -22362,7 +22367,16 @@ def gitea_request_mcp_restart(
     only when ``request_break_glass`` is set *and* the environment carries
     ``GITEA_BREAKGLASS_RESTART_AUTHORIZATION``. Even an authorized gate performs
     no restart here; actual execution is a further child. The gate outcome is
-    reported under ``apply_gate`` / ``apply_authorized``.
+    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
@@ -22447,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
@@ -22456,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,
@@ -22470,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()
@@ -22481,6 +22515,9 @@ def gitea_request_mcp_restart(
     payload["requesting_session_id"] = sid
     payload["operator_override_requested"] = bool(request_override)
     payload["operator_override_authorized"] = operator_authorized
+    payload["controller_approval_authorized"] = controller_approved
+    payload["requester_role"] = requester_role
+    payload["requester_permissions"] = list(requester_permissions)
     # 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
@@ -22518,8 +22555,25 @@ def gitea_request_mcp_restart(
             gate_payload["reasons"] = [proof_parse_error] + list(
                 gate_payload.get("reasons") or []
             )
+        # The #663 restart-class matrix and the #661 drain gate are two
+        # independent authorizations, and an apply requires BOTH. ``gate.allow``
+        # proves only that the blast radius was drained — or that break-glass
+        # was authorized — and knows nothing about whether this requester may
+        # request this class at all. Conjoining them keeps a class the matrix
+        # denied from ever reporting an authorized apply, and keeps break-glass
+        # scoped to what it is for: bypassing the drain proof, never the
+        # least-privilege class matrix.
+        restart_class_authorized = bool(report.allow_restart)
+        gate_payload["drain_gate_allow"] = bool(gate.allow)
+        gate_payload["restart_class_authorized"] = restart_class_authorized
+        if not restart_class_authorized:
+            gate_payload["reasons"] = list(gate_payload.get("reasons") or []) + [
+                "restart class authorization denied; apply denied regardless of "
+                "drain proof or break-glass (fail closed, #663)",
+                *(report.authorization_reasons or []),
+            ]
         payload["apply_gate"] = gate_payload
-        payload["apply_authorized"] = gate.allow
+        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

restart_coordinator.py

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

tests/test_issue_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_request_initiation.py

@@ -1,917 +0,0 @@
-"""Request preview, authorization, and workflow initiation tests (#643).
-
-Covers each acceptance criterion:
-
-* AC1 — preview shows authorize/deny with reasons.
-* AC2 — apply creates an exclusive assignment or returns wait/blocked.
-* AC3 — duplicate assign rejected.
-* AC4 — preview / apply / deny / collision are all exercised.
-* AC5 — the UI never renders a secret, and messaging stays brief.
-
-Required tests named in the issue: allocator integration with fakes, and
-gated-action tests. The allocator is injected as a fake throughout so no test
-touches Gitea or reserves real work; one class asserts the *real* default
-allocator refuses an incomplete inventory rather than ranking a partial set.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import pathlib
-import sys
-import tempfile
-import unittest
-from typing import Any
-from unittest import mock
-
-from tests.webui_testclient import TestClient
-
-sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
-
-import allocator_service  # noqa: E402
-from webui import console_audit, console_authz, request_service  # noqa: E402
-from webui.app import create_app  # noqa: E402
-from webui.console_redaction import scan_for_secrets  # noqa: E402
-from webui.request_views import render_requests_page  # noqa: E402
-
-EXEC_FLAG = "WEBUI_REQUESTS_EXECUTION"
-
-SCOPE = {
-    "remote": "prgs",
-    "org": "Scaled-Tech-Consulting",
-    "repo": "Gitea-Tools",
-}
-
-
-def _principal(role: str) -> console_authz.Principal:
-    return console_authz.Principal(
-        subject=f"{role}@example.com",
-        role=role,
-        identity_source=console_authz.IDENTITY_ACCESS_PROXY,
-        authenticated=True,
-    )
-
-
-def _request(
-    *,
-    role: str = "author",
-    kind: str = "issue",
-    number: int = 643,
-    intent: str = "implement request preview and initiation",
-    head: str | None = None,
-) -> request_service.WorkRequest:
-    parsed, error = request_service.parse_request(
-        {
-            "desired_role": role,
-            "work_kind": kind,
-            "work_number": number,
-            "intent_summary": intent,
-            "expected_head_sha": head,
-            **SCOPE,
-        }
-    )
-    assert error is None, error
-    assert parsed is not None
-    return parsed
-
-
-def _selection(
-    *, kind: str = "issue", number: int = 643, head_sha: str | None = None
-) -> dict[str, Any]:
-    return {
-        "kind": kind,
-        "number": number,
-        "title": "Web Console: Requests, intent preview, authorization",
-        "head_sha": head_sha,
-        "selected_action": "implement",
-        "expected_role_next": "author",
-    }
-
-
-def _fake_allocator(
-    *,
-    selection: dict[str, Any] | None = None,
-    preview_outcome: str = allocator_service.OUTCOME_PREVIEW,
-    apply_outcome: str = allocator_service.OUTCOME_ASSIGNED,
-    assignment: dict[str, Any] | None = None,
-    calls: list[dict[str, Any]] | None = None,
-):
-    """Build an allocator double that records how it was called."""
-    chosen = selection if selection is not None else _selection()
-    made = (
-        assignment
-        if assignment is not None
-        else {
-            "assignment_id": "asn-test-0001",
-            "lease_id": "lease-test-0001",
-            "session_id": "webui-request-test",
-            "expected_head_sha": chosen.get("head_sha"),
-        }
-    )
-
-    def _allocator(*, request, apply, expected_candidate_set_fingerprint=None):
-        if calls is not None:
-            calls.append(
-                {
-                    "apply": apply,
-                    "role": request.desired_role,
-                    "fingerprint": expected_candidate_set_fingerprint,
-                }
-            )
-        return {
-            "outcome": apply_outcome if apply else preview_outcome,
-            "selected": dict(chosen),
-            "reasons": ["fake allocator"],
-            "candidate_set_fingerprint": "fp-test",
-            "candidate_count": 3,
-            "inventory_complete": True,
-            "selection_policy": allocator_service.SELECTION_POLICY,
-            "substrate": "control_plane_db",
-            "assignment": dict(made) if apply else None,
-        }
-
-    return _allocator
-
-
-def _no_claims(_request):
-    return {}
-
-
-def _claimed(role: str = "author"):
-    def _source(request):
-        return {
-            request.work_key: {
-                "lease_id": "lease-foreign-9999",
-                "session_id": "prgs-author-999-foreign",
-                "role": role,
-                "expires_at": "2026-07-25T09:10:37Z",
-            }
-        }
-
-    return _source
-
-
-class TestRequestParsing(unittest.TestCase):
-    """The request model rejects rather than guesses."""
-
-    def test_valid_request_round_trips(self):
-        req = _request()
-        self.assertEqual(req.work_key, ("issue", 643))
-        self.assertEqual(req.display_ref, "#643")
-        self.assertEqual(req.to_dict()["desired_role"], "author")
-
-    def test_unknown_role_rejected(self):
-        parsed, error = request_service.parse_request(
-            {
-                "desired_role": "admin",
-                "work_kind": "issue",
-                "work_number": 1,
-                "intent_summary": "x",
-                **SCOPE,
-            }
-        )
-        self.assertIsNone(parsed)
-        self.assertEqual(error.reason_code, "unknown_role")
-        self.assertEqual(error.field_name, "desired_role")
-
-    def test_unknown_work_kind_rejected(self):
-        parsed, error = request_service.parse_request(
-            {
-                "desired_role": "author",
-                "work_kind": "branch",
-                "work_number": 1,
-                "intent_summary": "x",
-                **SCOPE,
-            }
-        )
-        self.assertIsNone(parsed)
-        self.assertEqual(error.reason_code, "unknown_work_kind")
-
-    def test_non_positive_number_rejected(self):
-        for value in (0, -3):
-            with self.subTest(value=value):
-                parsed, error = request_service.parse_request(
-                    {
-                        "desired_role": "author",
-                        "work_kind": "issue",
-                        "work_number": value,
-                        "intent_summary": "x",
-                        **SCOPE,
-                    }
-                )
-                self.assertIsNone(parsed)
-                self.assertEqual(error.reason_code, "invalid_work_number")
-
-    def test_missing_intent_rejected(self):
-        parsed, error = request_service.parse_request(
-            {
-                "desired_role": "author",
-                "work_kind": "issue",
-                "work_number": 1,
-                **SCOPE,
-            }
-        )
-        self.assertIsNone(parsed)
-        self.assertEqual(error.reason_code, "missing_intent")
-
-    def test_intent_is_bounded(self):
-        req = _request(intent="x" * 5000)
-        self.assertEqual(len(req.intent_summary), request_service.MAX_INTENT_CHARS)
-
-    def test_unresolved_scope_rejected(self):
-        parsed, error = request_service.parse_request(
-            {
-                "desired_role": "author",
-                "work_kind": "issue",
-                "work_number": 1,
-                "intent_summary": "x",
-            }
-        )
-        self.assertIsNone(parsed)
-        self.assertEqual(error.reason_code, "scope_unresolved")
-
-    def test_default_scope_fills_missing_fields(self):
-        parsed, error = request_service.parse_request(
-            {
-                "desired_role": "author",
-                "work_kind": "issue",
-                "work_number": 7,
-                "intent_summary": "x",
-            },
-            default_scope=SCOPE,
-        )
-        self.assertIsNone(error)
-        self.assertEqual(parsed.repo, "Gitea-Tools")
-
-
-class TestPreviewAuthorizeDeny(unittest.TestCase):
-    """AC1 — preview shows authorize/deny with reasons."""
-
-    def test_authorized_preview_names_every_check(self):
-        preview = request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertTrue(preview.authorized)
-        self.assertEqual(
-            {c.name for c in preview.checks},
-            {
-                request_service.CHECK_AUTHORIZATION,
-                request_service.CHECK_CAPABILITY,
-                request_service.CHECK_LEASE_AVAILABILITY,
-                request_service.CHECK_NEXT_SAFE_ACTION,
-                request_service.CHECK_HEAD_PIN,
-            },
-        )
-        self.assertEqual(preview.required_profile, "prgs-author")
-        self.assertEqual(preview.required_namespace, "gitea-author")
-
-    def test_every_check_carries_a_reason(self):
-        preview = request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        for check in preview.checks:
-            with self.subTest(check=check.name):
-                self.assertTrue(check.reason_code.strip())
-                self.assertTrue(check.detail.strip())
-
-    def test_anonymous_preview_denied_with_reason(self):
-        preview = request_service.preview_request(
-            _request(),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertFalse(preview.authorized)
-        self.assertEqual(preview.reason_code, console_authz.DENY_UNAUTHENTICATED)
-
-    def test_viewer_preview_denied_for_insufficient_role(self):
-        preview = request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.VIEWER),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertFalse(preview.authorized)
-        self.assertEqual(preview.reason_code, console_authz.DENY_INSUFFICIENT_ROLE)
-
-    def test_denied_preview_never_reaches_the_allocator(self):
-        """A denial must not double as a queue oracle."""
-        calls: list[dict[str, Any]] = []
-        request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.VIEWER),
-            allocator=_fake_allocator(calls=calls),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertEqual(calls, [])
-
-    def test_preview_lists_prohibited_actions(self):
-        preview = request_service.preview_request(
-            _request(role="author"),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertIn("merge", preview.prohibited_actions)
-        self.assertIn("approve", preview.prohibited_actions)
-
-    def test_preview_reports_next_safe_action(self):
-        preview = request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertIn("issue #643", preview.next_safe_action)
-
-    def test_preview_never_mutates(self):
-        calls: list[dict[str, Any]] = []
-        request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(calls=calls),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertEqual([c["apply"] for c in calls], [False])
-
-
-class TestPreviewFailClosed(unittest.TestCase):
-    """Missing evidence denies; it never reads as an absence of obstacles."""
-
-    def test_unreadable_claim_inventory_denies(self):
-        def _boom(_request):
-            raise RuntimeError("db unavailable")
-
-        preview = request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_boom,
-            audit=False,
-        )
-        self.assertFalse(preview.authorized)
-        self.assertEqual(
-            preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
-        )
-
-    def test_allocator_failure_denies(self):
-        def _boom(**_kwargs):
-            raise RuntimeError("allocator exploded")
-
-        preview = request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_boom,
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertFalse(preview.authorized)
-        self.assertEqual(
-            preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
-        )
-
-    def test_allocator_selecting_other_work_denies(self):
-        preview = request_service.preview_request(
-            _request(number=643),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(selection=_selection(number=999)),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertFalse(preview.authorized)
-        self.assertEqual(preview.reason_code, request_service.REASON_NOT_NEXT_SAFE)
-        self.assertIn("#999", preview.detail)
-
-    def test_pr_without_head_sha_denies(self):
-        preview = request_service.preview_request(
-            _request(role="reviewer", kind="pr", number=898),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(
-                selection=_selection(kind="pr", number=898, head_sha=None)
-            ),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertFalse(preview.authorized)
-        self.assertEqual(
-            preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
-        )
-
-    def test_pr_head_moved_denies(self):
-        preview = request_service.preview_request(
-            _request(role="reviewer", kind="pr", number=898, head="a" * 40),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(
-                selection=_selection(kind="pr", number=898, head_sha="b" * 40)
-            ),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertFalse(preview.authorized)
-        self.assertEqual(preview.reason_code, "head_moved")
-
-    def test_pr_head_matching_passes(self):
-        preview = request_service.preview_request(
-            _request(role="reviewer", kind="pr", number=898, head="b" * 40),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(
-                selection=_selection(kind="pr", number=898, head_sha="b" * 40)
-            ),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        self.assertTrue(preview.authorized)
-
-
-class TestApplyExecutionGate(unittest.TestCase):
-    """Execution stays wired off unless an operator opts in explicitly."""
-
-    def test_action_is_registered_and_unwired_by_default(self):
-        action = console_authz.get_action(request_service.ACTION_ID)
-        self.assertIsNotNone(action)
-        self.assertEqual(action.phase, 2)
-        self.assertEqual(action.minimum_role, console_authz.OPERATOR)
-        self.assertTrue(action.requires_confirmation)
-        self.assertFalse(console_authz.execution_wired(action, env={}))
-
-    def test_flag_named_but_unset_does_not_wire(self):
-        action = console_authz.get_action(request_service.ACTION_ID)
-        self.assertFalse(console_authz.execution_wired(action, env={EXEC_FLAG: "no"}))
-        self.assertTrue(console_authz.execution_wired(action, env={EXEC_FLAG: "1"}))
-
-    def test_opting_in_wires_only_this_action(self):
-        env = {EXEC_FLAG: "1"}
-        for action_id, action in console_authz.ACTIONS.items():
-            with self.subTest(action=action_id):
-                self.assertEqual(
-                    console_authz.execution_wired(action, env=env),
-                    action_id == request_service.ACTION_ID,
-                )
-
-    def test_apply_denied_while_unwired(self):
-        with mock.patch.dict(os.environ, {EXEC_FLAG: ""}):
-            result = request_service.apply_request(
-                _request(),
-                principal=_principal(console_authz.OPERATOR),
-                confirm=True,
-                allocator=_fake_allocator(),
-                claims_source=_no_claims,
-            )
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["outcome"], request_service.OUTCOME_DENIED)
-        self.assertEqual(result["reason_code"], request_service.REASON_UNAUTHORIZED)
-        self.assertFalse(result["mutation_performed"])
-
-
-class TestApplyOutcomes(unittest.TestCase):
-    """AC2/AC3/AC4 — assignment, wait, blocked, and duplicate rejection."""
-
-    def setUp(self):
-        patcher = mock.patch.dict(os.environ, {EXEC_FLAG: "1"})
-        patcher.start()
-        self.addCleanup(patcher.stop)
-
-    def test_apply_creates_exclusive_assignment(self):
-        calls: list[dict[str, Any]] = []
-        result = request_service.apply_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            confirm=True,
-            allocator=_fake_allocator(calls=calls),
-            claims_source=_no_claims,
-        )
-        self.assertTrue(result["ok"])
-        self.assertEqual(result["outcome"], allocator_service.OUTCOME_ASSIGNED)
-        self.assertEqual(result["assignment"]["assignment_id"], "asn-test-0001")
-        self.assertTrue(result["mutation_performed"])
-        self.assertEqual(result["status_code"], 201)
-        # Dry-run first, then apply — never apply alone.
-        self.assertEqual([c["apply"] for c in calls], [False, True])
-        # The apply call carries the fingerprint the dry-run produced.
-        self.assertEqual(calls[1]["fingerprint"], "fp-test")
-
-    def test_assignment_returns_a_role_handoff(self):
-        result = request_service.apply_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            confirm=True,
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-        )
-        handoff = result["handoff"]
-        self.assertEqual(handoff["required_profile"], "prgs-author")
-        self.assertEqual(handoff["required_namespace"], "gitea-author")
-        self.assertEqual(handoff["assignment_id"], "asn-test-0001")
-        self.assertIn("merge", handoff["forbidden_actions"])
-
-    def test_unconfirmed_apply_refuses_before_the_allocator(self):
-        calls: list[dict[str, Any]] = []
-        result = request_service.apply_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            confirm=False,
-            allocator=_fake_allocator(calls=calls),
-            claims_source=_no_claims,
-        )
-        self.assertFalse(result["ok"])
-        self.assertEqual(
-            result["reason_code"], request_service.REASON_CONFIRMATION_REQUIRED
-        )
-        self.assertEqual(calls, [])
-
-    def test_duplicate_assignment_rejected(self):
-        """AC3 — an active lease on the work unit blocks a second assign."""
-        calls: list[dict[str, Any]] = []
-        result = request_service.apply_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            confirm=True,
-            allocator=_fake_allocator(calls=calls),
-            claims_source=_claimed(),
-        )
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["outcome"], request_service.OUTCOME_BLOCKED)
-        self.assertEqual(
-            result["reason_code"], request_service.REASON_DUPLICATE_ASSIGNMENT
-        )
-        self.assertFalse(result["mutation_performed"])
-        # The dry-run ran; the apply never did.
-        self.assertEqual([c["apply"] for c in calls], [False])
-
-    def test_not_next_safe_work_returns_wait_without_applying(self):
-        calls: list[dict[str, Any]] = []
-        result = request_service.apply_request(
-            _request(number=643),
-            principal=_principal(console_authz.OPERATOR),
-            confirm=True,
-            allocator=_fake_allocator(
-                selection=_selection(number=999), calls=calls
-            ),
-            claims_source=_no_claims,
-        )
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["outcome"], request_service.OUTCOME_WAIT)
-        self.assertEqual(result["reason_code"], request_service.REASON_NOT_NEXT_SAFE)
-        self.assertEqual([c["apply"] for c in calls], [False])
-
-    def test_allocator_declining_on_apply_returns_blocked(self):
-        result = request_service.apply_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            confirm=True,
-            allocator=_fake_allocator(
-                apply_outcome=allocator_service.OUTCOME_BLOCKED_LEASE,
-                assignment={},
-            ),
-            claims_source=_no_claims,
-        )
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["outcome"], request_service.OUTCOME_BLOCKED)
-        self.assertEqual(
-            result["reason_code"], request_service.REASON_ALLOCATOR_OUTCOME
-        )
-        self.assertFalse(result["mutation_performed"])
-
-    def test_allocator_drift_on_apply_is_not_read_as_an_assignment(self):
-        """The apply call must return *this* work unit, not a substitute."""
-
-        def _drifting(*, request, apply, expected_candidate_set_fingerprint=None):
-            return {
-                "outcome": (
-                    allocator_service.OUTCOME_ASSIGNED
-                    if apply
-                    else allocator_service.OUTCOME_PREVIEW
-                ),
-                "selected": _selection(number=999 if apply else 643),
-                "assignment": {"assignment_id": "asn-wrong"} if apply else None,
-                "candidate_set_fingerprint": "fp-test",
-            }
-
-        result = request_service.apply_request(
-            _request(number=643),
-            principal=_principal(console_authz.OPERATOR),
-            confirm=True,
-            allocator=_drifting,
-            claims_source=_no_claims,
-        )
-        self.assertFalse(result["ok"])
-        self.assertIsNone(result["assignment"])
-        self.assertFalse(result["mutation_performed"])
-
-    def test_viewer_cannot_apply(self):
-        result = request_service.apply_request(
-            _request(),
-            principal=_principal(console_authz.VIEWER),
-            confirm=True,
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-        )
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["reason_code"], request_service.REASON_UNAUTHORIZED)
-
-    def test_anonymous_cannot_apply(self):
-        result = request_service.apply_request(
-            _request(),
-            confirm=True,
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-        )
-        self.assertFalse(result["ok"])
-        self.assertFalse(result["mutation_performed"])
-
-
-class TestAllocatorIntegrationFakes(unittest.TestCase):
-    """The real default allocator refuses a partial inventory (#758)."""
-
-    def test_incomplete_inventory_returns_none(self):
-        from webui.queue_loader import PaginationMeta, QueueSnapshot
-
-        snapshot = QueueSnapshot(
-            project_id="p",
-            repo_label="r",
-            prs=(),
-            issues=(),
-            pr_pagination=PaginationMeta(
-                page=1,
-                per_page=50,
-                returned_count=50,
-                has_more=True,
-                is_final_page=False,
-                inventory_complete=False,
-                pages_fetched=1,
-            ),
-            issue_pagination=None,
-        )
-        with mock.patch(
-            "webui.queue_loader.load_queue_snapshot", return_value=snapshot
-        ):
-            result = request_service.default_allocator(
-                request=_request(), apply=False
-            )
-        self.assertIsNone(result)
-
-    def test_fetch_error_returns_none(self):
-        from webui.queue_loader import QueueSnapshot
-
-        snapshot = QueueSnapshot(
-            project_id="p",
-            repo_label="r",
-            prs=(),
-            issues=(),
-            pr_pagination=None,
-            issue_pagination=None,
-            fetch_error="no credentials",
-        )
-        with mock.patch(
-            "webui.queue_loader.load_queue_snapshot", return_value=snapshot
-        ):
-            result = request_service.default_allocator(
-                request=_request(), apply=True
-            )
-        self.assertIsNone(result)
-
-
-class TestAuditRecords(unittest.TestCase):
-    """Every preview and apply is auditable, correlated, and redacted."""
-
-    def setUp(self):
-        handle = tempfile.NamedTemporaryFile(
-            mode="w", suffix=".jsonl", delete=False
-        )
-        handle.close()
-        self.sink = handle.name
-        self.addCleanup(
-            lambda: os.path.exists(self.sink) and os.remove(self.sink)
-        )
-        patcher = mock.patch.dict(
-            os.environ, {console_audit.AUDIT_LOG_ENV: self.sink}
-        )
-        patcher.start()
-        self.addCleanup(patcher.stop)
-
-    def _records(self) -> list[dict[str, Any]]:
-        with open(self.sink, encoding="utf-8") as handle:
-            return [json.loads(line) for line in handle if line.strip()]
-
-    def test_preview_is_audited_with_a_correlation_id(self):
-        preview = request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-        )
-        records = self._records()
-        self.assertEqual(len(records), 1)
-        record = records[0]
-        self.assertEqual(record["action"], request_service.ACTION_ID)
-        self.assertEqual(record["result"], console_audit.RESULT_PREVIEWED)
-        self.assertEqual(
-            record["correlation"]["request_id"], preview.correlation_id
-        )
-        self.assertEqual(record["target"]["ref"], "#643")
-
-    def test_denied_apply_is_audited(self):
-        with mock.patch.dict(os.environ, {EXEC_FLAG: ""}):
-            request_service.apply_request(
-                _request(),
-                principal=_principal(console_authz.VIEWER),
-                confirm=True,
-                allocator=_fake_allocator(),
-                claims_source=_no_claims,
-            )
-        self.assertEqual(self._records()[-1]["result"], console_audit.RESULT_DENIED)
-
-    def test_assignment_is_audited_and_correlated(self):
-        with mock.patch.dict(os.environ, {EXEC_FLAG: "1"}):
-            result = request_service.apply_request(
-                _request(),
-                principal=_principal(console_authz.OPERATOR),
-                confirm=True,
-                allocator=_fake_allocator(),
-                claims_source=_no_claims,
-            )
-        record = self._records()[-1]
-        self.assertEqual(record["result"], console_audit.RESULT_SUCCEEDED)
-        self.assertEqual(
-            record["correlation"]["request_id"], result["correlation_id"]
-        )
-        self.assertEqual(
-            record["metadata"]["assignment_id"],
-            result["assignment"]["assignment_id"],
-        )
-
-    def test_intent_bearing_a_secret_is_not_persisted_raw(self):
-        request_service.preview_request(
-            _request(intent="use token=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-        )
-        for record in self._records():
-            with self.subTest(event=record.get("event_id")):
-                self.assertFalse(scan_for_secrets(record))
-
-
-class TestRequestRoutes(unittest.TestCase):
-    """The HTTP surface: form page, preview API, apply API."""
-
-    def setUp(self):
-        self.client = TestClient(create_app())
-
-    def test_requests_page_renders_form(self):
-        response = self.client.get("/requests")
-        self.assertEqual(response.status_code, 200)
-        body = response.text
-        self.assertIn("Requests", body)
-        self.assertIn("desired_role", body)
-        self.assertIn("intent_summary", body)
-
-    def test_requests_page_is_linked_from_nav(self):
-        from webui.nav import nav_hrefs
-
-        self.assertIn("/requests", nav_hrefs())
-
-    def test_preview_api_rejects_an_invalid_request(self):
-        response = self.client.post(
-            "/api/v1/requests/preview",
-            json={
-                "desired_role": "wizard",
-                "work_kind": "issue",
-                "work_number": 1,
-                "intent_summary": "x",
-                **SCOPE,
-            },
-        )
-        self.assertEqual(response.status_code, 400)
-        self.assertEqual(response.json()["reason_code"], "unknown_role")
-
-    def test_preview_api_denies_anonymous(self):
-        response = self.client.post(
-            "/api/v1/requests/preview",
-            json={
-                "desired_role": "author",
-                "work_kind": "issue",
-                "work_number": 643,
-                "intent_summary": "x",
-                **SCOPE,
-            },
-        )
-        self.assertEqual(response.status_code, 403)
-        payload = response.json()
-        self.assertFalse(payload["authorized"])
-        self.assertFalse(payload["mutation_performed"])
-
-    def test_apply_api_denies_anonymous(self):
-        response = self.client.post(
-            "/api/v1/requests/apply",
-            json={
-                "desired_role": "author",
-                "work_kind": "issue",
-                "work_number": 643,
-                "intent_summary": "x",
-                "confirm": True,
-                **SCOPE,
-            },
-        )
-        self.assertEqual(response.status_code, 403)
-        payload = response.json()
-        self.assertFalse(payload["ok"])
-        self.assertFalse(payload["mutation_performed"])
-        self.assertIsNone(payload["assignment"])
-
-    def test_apply_api_rejects_an_invalid_request(self):
-        response = self.client.post(
-            "/api/v1/requests/apply",
-            json={
-                "desired_role": "author",
-                "work_kind": "issue",
-                "work_number": -1,
-                "intent_summary": "x",
-                **SCOPE,
-            },
-        )
-        self.assertEqual(response.status_code, 400)
-
-    def test_request_apis_are_post_only(self):
-        """GET is not a way in. The app's 405 handler renders a read-only
-        method against a write route as 404, so that is what is asserted."""
-        for path in ("/api/v1/requests/preview", "/api/v1/requests/apply"):
-            with self.subTest(path=path):
-                self.assertEqual(self.client.get(path).status_code, 404)
-
-    def test_form_post_previews_and_never_assigns(self):
-        response = self.client.post(
-            "/requests",
-            data={
-                "desired_role": "author",
-                "work_kind": "issue",
-                "work_number": "643",
-                "intent_summary": "implement the request surface",
-                "remote": "prgs",
-                "org": "Scaled-Tech-Consulting",
-                "repo": "Gitea-Tools",
-            },
-        )
-        self.assertEqual(response.status_code, 200)
-        self.assertIn("Intent preview", response.text)
-
-
-class TestRenderingSafety(unittest.TestCase):
-    """AC5 — the page escapes hostile input and shows no secret."""
-
-    def test_intent_is_escaped(self):
-        preview = request_service.preview_request(
-            _request(intent="<script>alert(1)</script>"),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        html = render_requests_page(preview=preview)
-        self.assertNotIn("<script>alert(1)</script>", html)
-        self.assertIn("&lt;script&gt;", html)
-
-    def test_page_renders_a_denial_without_a_preview(self):
-        _, error = request_service.parse_request(
-            {
-                "desired_role": "wizard",
-                "work_kind": "issue",
-                "work_number": 1,
-                "intent_summary": "x",
-                **SCOPE,
-            }
-        )
-        html = render_requests_page(error=error)
-        self.assertIn("Request rejected", html)
-        self.assertIn("unknown_role", html)
-
-    def test_page_shows_no_credential_material(self):
-        preview = request_service.preview_request(
-            _request(),
-            principal=_principal(console_authz.OPERATOR),
-            allocator=_fake_allocator(),
-            claims_source=_no_claims,
-            audit=False,
-        )
-        html = render_requests_page(preview=preview)
-        for needle in ("token=", "Bearer ", "password"):
-            with self.subTest(needle=needle):
-                self.assertNotIn(needle, html)
-
-
-if __name__ == "__main__":  # pragma: no cover
-    unittest.main()

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/app.py

@@ -67,8 +67,6 @@ from webui.system_health import (
     snapshot_to_dict as system_health_to_dict,
 )
 from webui.system_health_views import render_system_health_page
-from webui import request_service
-from webui.request_views import render_requests_page
 
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -724,109 +722,6 @@ async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
     )
 
 
-def _default_request_scope() -> dict[str, str]:
-    """Resolve remote/org/repo from the project registry for request forms.
-
-    Returns an empty mapping when the registry cannot be read, which makes
-    ``parse_request`` reject a request that did not name its own scope rather
-    than letting it default to some other repository.
-    """
-    from webui.queue_loader import _host_from_url  # host normalisation helper
-
-    registry, error = _load_project_registry()
-    if error is not None or not registry.projects:
-        return {}
-    project = registry.projects[0]
-    host = _host_from_url(project.remote_host)
-    return {
-        "remote": _derive_remote(host),
-        "org": project.gitea_owner or "",
-        "repo": project.repo_name or "",
-    }
-
-
-async def _request_payload(request: Request) -> dict[str, object]:
-    """Read a request body as JSON or form-encoded. Never raises."""
-    content_type = (request.headers.get("content-type") or "").lower()
-    if "application/json" in content_type:
-        try:
-            body = await request.json()
-        except Exception:
-            return {}
-        return dict(body) if isinstance(body, dict) else {}
-    try:
-        form = await request.form()
-    except Exception:
-        return {}
-    return {key: form[key] for key in form}
-
-
-async def requests_page(request: Request) -> HTMLResponse:
-    """Operator request form and intent preview (#643).
-
-    POST here only ever *previews*. Initiation is a separate confirmed call to
-    ``/api/v1/requests/apply`` so that submitting this form cannot reserve
-    work as a side effect.
-    """
-    submitted: dict[str, object] = {}
-    preview = None
-    error = None
-    if request.method == "POST":
-        submitted = await _request_payload(request)
-        work_request, error = request_service.parse_request(
-            submitted, default_scope=_default_request_scope()
-        )
-        if work_request is not None:
-            preview = request_service.preview_request(
-                work_request,
-                principal=resolve_principal(headers=dict(request.headers)),
-            )
-    return HTMLResponse(
-        render_requests_page(
-            preview=preview, error=error, submitted=submitted
-        )
-    )
-
-
-async def api_v1_request_preview(request: Request) -> JSONResponse:
-    """Dry-run authorization and intent preview for a work request (#643)."""
-    payload = await _request_payload(request)
-    work_request, error = request_service.parse_request(
-        payload, default_scope=_default_request_scope()
-    )
-    if work_request is None:
-        return JSONResponse(error.to_dict(), status_code=400)
-    preview = request_service.preview_request(
-        work_request,
-        principal=resolve_principal(headers=dict(request.headers)),
-    )
-    return JSONResponse(
-        preview.to_dict(), status_code=200 if preview.authorized else 403
-    )
-
-
-async def api_v1_request_apply(request: Request) -> JSONResponse:
-    """Initiate a previewed work request through the allocator (#643).
-
-    Fail-closed at every step: unauthorized, unconfirmed, not-next-safe, and
-    already-claimed all return without attempting an assignment.
-    """
-    payload = await _request_payload(request)
-    work_request, error = request_service.parse_request(
-        payload, default_scope=_default_request_scope()
-    )
-    if work_request is None:
-        return JSONResponse(error.to_dict(), status_code=400)
-    confirm = _truthy_flag(str(payload.get("confirm") or ""))
-    result = request_service.apply_request(
-        work_request,
-        principal=resolve_principal(headers=dict(request.headers)),
-        confirm=confirm,
-    )
-    status = int(result.pop("status_code", 403))
-    return JSONResponse(result, status_code=status)
-
-
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
     path = request.url.path
     if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -891,17 +786,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 api_action_attempt,
                 methods=["POST"],
             ),
-            Route("/requests", requests_page, methods=["GET", "POST"]),
-            Route(
-                "/api/v1/requests/preview",
-                api_v1_request_preview,
-                methods=["POST"],
-            ),
-            Route(
-                "/api/v1/requests/apply",
-                api_v1_request_apply,
-                methods=["POST"],
-            ),
             Route("/api/leases", api_leases, methods=["GET"]),
             Route("/api/v1/inventory", api_inventory, methods=["GET"]),
             Route(

webui/console_authz.py

@@ -115,12 +115,6 @@ class ConsoleAction:
     break_glass: bool
     phase: int
     summary: str
-    # Opt-in switch for an action whose execution path is genuinely wired
-    # ahead of its phase becoming globally active (#643). Naming a variable
-    # here enables nothing on its own: the variable must also be set in the
-    # environment. An action that leaves this ``None`` can only execute once
-    # ACTIVE_PHASE reaches its phase, exactly as before.
-    execution_env_flag: str | None = None
 
     @property
     def mcp_permission(self) -> str:
@@ -283,27 +277,6 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
         phase=2,
         summary="Restart one MCP namespace via the host supervisor.",
     ),
-    # #643: submit a work request — desired role, issue/PR, intent — and let
-    # the allocator reserve it. This is the one Phase 2 action whose execution
-    # path is actually implemented (``webui.request_service``), so it carries
-    # the opt-in flag; it stays denied until an operator sets that variable.
-    # Authority is operator-class because the outcome is a claim, not a Gitea
-    # verdict: initiating reviewer or merger *work* does not grant the right
-    # to approve or merge, which stays with the MCP role profile.
-    ConsoleAction(
-        action_id="initiate_workflow",
-        task_key="allocate_next_work",
-        action_class=CLASS_WRITE,
-        minimum_role=OPERATOR,
-        requires_confirmation=True,
-        dual_control=False,
-        break_glass=False,
-        phase=2,
-        summary=(
-            "Preview and initiate allocator-owned workflow work for a role."
-        ),
-        execution_env_flag="WEBUI_REQUESTS_EXECUTION",
-    ),
 )
 
 ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
@@ -457,33 +430,6 @@ ALLOW_PREVIEW = "allowed_preview_only"
 # gated on this model landing; nothing here enables it.
 ACTIVE_PHASE = 1
 
-_TRUTHY = frozenset({"1", "true", "yes", "on"})
-
-
-def execution_wired(
-    action: ConsoleAction | None, env: dict[str, str] | None = None
-) -> bool:
-    """Whether *action* has a live execution path right now.
-
-    Two ways to be wired, and only two. The action's phase is active, or the
-    action declares an opt-in environment variable *and* that variable is set.
-    Everything else — including every action that never declares a flag — is
-    unwired, so the default across the registry stays deny.
-
-    Bumping ``ACTIVE_PHASE`` would enable execution for every action of that
-    phase at once. The per-action flag exists so a single implemented action
-    can go live without dragging its unimplemented phase-mates with it.
-    """
-    if action is None:
-        return False
-    if action.phase <= ACTIVE_PHASE:
-        return True
-    flag = (action.execution_env_flag or "").strip()
-    if not flag:
-        return False
-    source = env if env is not None else os.environ
-    return (source.get(flag) or "").strip().lower() in _TRUTHY
-
 
 @dataclass(frozen=True)
 class AuthorizationDecision:
@@ -523,19 +469,16 @@ def authorize(
     principal: Principal | None = None,
     *,
     for_execution: bool = False,
-    env: dict[str, str] | None = None,
 ) -> AuthorizationDecision:
     """Decide whether *principal* may invoke *action_id*. Deny by default.
 
     ``for_execution`` distinguishes a read-only preview from a real invocation.
-    ``execution_enabled`` reports whether the action has a live execution path
-    at all (:func:`execution_wired`) — for every action without an explicit
-    opt-in flag that stays ``False`` while the console is in Phase 1, so no
-    caller can read an allow as permission to mutate.
+    Even an allowed decision reports ``execution_enabled=False`` while the
+    console is in Phase 1, so no caller can read an allow as permission to
+    mutate.
     """
     who = principal if principal is not None else ANONYMOUS
     action = get_action(action_id)
-    wired = execution_wired(action, env)
 
     if action is None:
         return AuthorizationDecision(
@@ -554,7 +497,7 @@ def authorize(
         "requires_confirmation": action.requires_confirmation,
         "dual_control": action.dual_control,
         "break_glass": action.break_glass,
-        "execution_enabled": wired,
+        "execution_enabled": False,
     }
 
     if not who.authenticated:
@@ -587,19 +530,13 @@ def authorize(
             **base,
         )
 
-    if for_execution and not wired:
+    if for_execution and action.phase > ACTIVE_PHASE:
         return AuthorizationDecision(
             allowed=False,
             reason_code=DENY_PHASE_NOT_ACTIVE,
             detail=(
                 f"Action {action_id!r} belongs to phase {action.phase}; the "
-                f"console is in phase {ACTIVE_PHASE}"
-                + (
-                    f" and {action.execution_env_flag} is not set"
-                    if action.execution_env_flag
-                    else ""
-                )
-                + ". Execution is not wired."
+                f"console is in phase {ACTIVE_PHASE}. Execution is not wired."
             ),
             **base,
         )
@@ -608,8 +545,8 @@ def authorize(
         allowed=True,
         reason_code=ALLOW_PREVIEW,
         detail=(
-            "Principal holds the required role. Execution proceeds only for an "
-            "action with a wired execution path; everything else is preview."
+            "Principal holds the required role. Preview only — execution "
+            "remains disabled until the Phase 2 action framework ships."
         ),
         **base,
     )

webui/nav.py

@@ -45,7 +45,6 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
         NavItem("/queue", "Queue"),
         NavItem("/leases", "Leases"),
         NavItem("/actions", "Actions"),
-        NavItem("/requests", "Requests"),
     )),
     NavGroup("Runtime/Sessions", (
         NavItem("/runtime", "Runtime health"),

webui/request_service.py

@@ -1,967 +0,0 @@
-"""Operator work-request preview and initiation (#643, Phase 2).
-
-An operator's alternative to pasting a role prompt into a terminal. A
-*request* names three things — the role to run as, the issue or PR to run
-against, and what the operator intends — and this module answers two questions
-about it:
-
-* **Preview** (:func:`preview_request`) — would that request be authorized,
-  is the work unit actually free, is it the next safe thing that role should
-  touch, and which actions stay prohibited? Read-only, always. It creates no
-  assignment and never mutates.
-* **Initiate** (:func:`apply_request`) — turn an authorized request into an
-  *exclusive assignment*, and only ever through the allocator.
-
-Three invariants hold and are the reason this module exists rather than a
-direct call to :func:`allocator_service.allocate_next_work` from a route:
-
-1. **The allocator remains the only source of exclusive ownership** (#600 /
-   #613). ``apply`` never assigns the requested item directly. It runs a
-   dry-run first and proceeds only when the allocator would independently pick
-   that exact item; otherwise it reports ``wait`` and mutates nothing. A
-   request is therefore a *confirmation* of the allocator's decision, never an
-   override of it.
-2. **Duplicate assignment is rejected before it is attempted.** An active
-   claim on the work unit — held by any session, this one included — blocks.
-3. **Fail closed at every unknown.** An unparseable request, an unavailable
-   control-plane DB, an incomplete queue inventory, or an unresolved
-   authorization all deny. There is no branch that proceeds on missing
-   evidence.
-
-Authorization comes from :mod:`webui.console_authz` (``initiate_workflow``)
-and every outcome is audited through :mod:`webui.console_audit`, correlated to
-the resulting assignment by ``correlation_id``.
-"""
-
-from __future__ import annotations
-
-import uuid
-from dataclasses import dataclass, field
-from typing import Any, Callable, Mapping, Sequence
-
-import allocator_service
-from task_capability_map import required_permission, required_role
-from webui import console_audit, console_authz
-
-# The console action this module is gated by. Registered in console_authz.
-ACTION_ID = "initiate_workflow"
-
-KIND_ISSUE = "issue"
-KIND_PR = "pr"
-WORK_KINDS: tuple[str, ...] = (KIND_ISSUE, KIND_PR)
-
-REQUESTABLE_ROLES: tuple[str, ...] = (
-    allocator_service.ROLE_AUTHOR,
-    allocator_service.ROLE_REVIEWER,
-    allocator_service.ROLE_MERGER,
-    allocator_service.ROLE_RECONCILER,
-    allocator_service.ROLE_CONTROLLER,
-)
-
-# Intent is operator prose echoed back into an audit record. Bounded so a
-# pasted transcript cannot bloat the append-only log.
-MAX_INTENT_CHARS = 500
-
-# --- Outcomes ---------------------------------------------------------------
-OUTCOME_ASSIGNED = allocator_service.OUTCOME_ASSIGNED
-OUTCOME_WAIT = allocator_service.OUTCOME_WAIT
-OUTCOME_BLOCKED = "blocked"
-OUTCOME_DENIED = "denied"
-OUTCOME_INVALID = "invalid_request"
-OUTCOME_PREVIEW = allocator_service.OUTCOME_PREVIEW
-
-# --- Reason codes -----------------------------------------------------------
-REASON_AUTHORIZED = "request_authorized"
-REASON_PREVIEW_OK = "preview_authorized"
-REASON_UNAUTHORIZED = "unauthorized"
-REASON_NOT_NEXT_SAFE = "not_next_safe_work"
-REASON_DUPLICATE_ASSIGNMENT = "duplicate_assignment"
-REASON_CONFIRMATION_REQUIRED = "confirmation_required"
-REASON_EVIDENCE_UNAVAILABLE = "evidence_unavailable"
-REASON_ALLOCATOR_OUTCOME = "allocator_declined"
-
-# --- Check names ------------------------------------------------------------
-CHECK_AUTHORIZATION = "authorization"
-CHECK_CAPABILITY = "capability"
-CHECK_LEASE_AVAILABILITY = "lease_availability"
-CHECK_NEXT_SAFE_ACTION = "next_safe_action"
-CHECK_HEAD_PIN = "head_pin"
-
-
-# --- Request model ----------------------------------------------------------
-
-
-@dataclass(frozen=True)
-class WorkRequest:
-    """One operator request: a role, a work unit, and a stated intent."""
-
-    desired_role: str
-    work_kind: str
-    work_number: int
-    intent_summary: str
-    remote: str
-    org: str
-    repo: str
-    expected_head_sha: str | None = None
-
-    @property
-    def work_key(self) -> tuple[str, int]:
-        return (self.work_kind, self.work_number)
-
-    @property
-    def display_ref(self) -> str:
-        return f"#{self.work_number}"
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "desired_role": self.desired_role,
-            "work_kind": self.work_kind,
-            "work_number": self.work_number,
-            "intent_summary": self.intent_summary,
-            "remote": self.remote,
-            "org": self.org,
-            "repo": self.repo,
-            "expected_head_sha": self.expected_head_sha,
-        }
-
-
-@dataclass(frozen=True)
-class RequestError:
-    """A rejected request, with the field that caused the rejection."""
-
-    reason_code: str
-    detail: str
-    field_name: str | None = None
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "ok": False,
-            "outcome": OUTCOME_INVALID,
-            "reason_code": self.reason_code,
-            "detail": self.detail,
-            "field": self.field_name,
-        }
-
-
-def _clean(value: Any) -> str:
-    return str(value or "").strip()
-
-
-def parse_request(
-    payload: Mapping[str, Any] | None,
-    *,
-    default_scope: Mapping[str, str] | None = None,
-) -> tuple[WorkRequest | None, RequestError | None]:
-    """Validate an operator payload into a :class:`WorkRequest`.
-
-    Returns ``(request, None)`` or ``(None, error)``. Never raises and never
-    guesses: an unknown role, an unknown work kind, or a non-positive number is
-    an error rather than a silently corrected value.
-    """
-    body = dict(payload or {})
-    scope = dict(default_scope or {})
-
-    role = _clean(body.get("desired_role") or body.get("role")).lower()
-    if role not in REQUESTABLE_ROLES:
-        return None, RequestError(
-            reason_code="unknown_role",
-            detail=(
-                f"desired_role must be one of {', '.join(REQUESTABLE_ROLES)}; "
-                f"got {role or '(empty)'!r}."
-            ),
-            field_name="desired_role",
-        )
-
-    kind = _clean(body.get("work_kind") or body.get("kind")).lower()
-    if kind not in WORK_KINDS:
-        return None, RequestError(
-            reason_code="unknown_work_kind",
-            detail=(
-                f"work_kind must be 'issue' or 'pr'; got {kind or '(empty)'!r}."
-            ),
-            field_name="work_kind",
-        )
-
-    raw_number = body.get("work_number")
-    if raw_number is None:
-        raw_number = (
-            body.get("pr_number") if kind == KIND_PR else body.get("issue_number")
-        )
-    if raw_number is None:
-        raw_number = body.get("number")
-    try:
-        number = int(str(raw_number).strip())
-    except (TypeError, ValueError):
-        return None, RequestError(
-            reason_code="invalid_work_number",
-            detail=f"work_number must be an integer; got {raw_number!r}.",
-            field_name="work_number",
-        )
-    if number <= 0:
-        return None, RequestError(
-            reason_code="invalid_work_number",
-            detail="work_number must be a positive issue or PR number.",
-            field_name="work_number",
-        )
-
-    intent = _clean(body.get("intent_summary") or body.get("intent"))
-    if not intent:
-        return None, RequestError(
-            reason_code="missing_intent",
-            detail="intent_summary is required so the audit record states why.",
-            field_name="intent_summary",
-        )
-    intent = intent[:MAX_INTENT_CHARS]
-
-    remote = _clean(body.get("remote")) or _clean(scope.get("remote"))
-    org = _clean(body.get("org")) or _clean(scope.get("org"))
-    repo = _clean(body.get("repo")) or _clean(scope.get("repo"))
-    if not (remote and org and repo):
-        return None, RequestError(
-            reason_code="scope_unresolved",
-            detail=(
-                "remote, org, and repo could not be resolved from the request "
-                "or the project registry."
-            ),
-            field_name="repo",
-        )
-
-    head = _clean(body.get("expected_head_sha")) or None
-
-    return (
-        WorkRequest(
-            desired_role=role,
-            work_kind=kind,
-            work_number=number,
-            intent_summary=intent,
-            remote=remote,
-            org=org,
-            repo=repo,
-            expected_head_sha=head,
-        ),
-        None,
-    )
-
-
-# --- Preview ----------------------------------------------------------------
-
-
-@dataclass(frozen=True)
-class RequestCheck:
-    """One named precondition and its verdict."""
-
-    name: str
-    ok: bool
-    reason_code: str
-    detail: str
-    evidence: dict[str, Any] = field(default_factory=dict)
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "name": self.name,
-            "ok": self.ok,
-            "reason_code": self.reason_code,
-            "detail": self.detail,
-            "evidence": dict(self.evidence),
-        }
-
-
-@dataclass(frozen=True)
-class RequestPreview:
-    """The full intent preview for one request. Read-only in every field."""
-
-    request: WorkRequest
-    authorized: bool
-    reason_code: str
-    detail: str
-    authorization: dict[str, Any]
-    checks: tuple[RequestCheck, ...]
-    prohibited_actions: tuple[str, ...]
-    allowed_actions: tuple[str, ...]
-    next_safe_action: str
-    required_profile: str
-    required_namespace: str
-    required_permission: str
-    correlation_id: str
-    allocator_evidence: dict[str, Any] = field(default_factory=dict)
-
-    @property
-    def failed_checks(self) -> tuple[RequestCheck, ...]:
-        return tuple(c for c in self.checks if not c.ok)
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "ok": self.authorized,
-            "outcome": OUTCOME_PREVIEW,
-            "dry_run": True,
-            "mutation_performed": False,
-            "authorized": self.authorized,
-            "reason_code": self.reason_code,
-            "detail": self.detail,
-            "request": self.request.to_dict(),
-            "authorization": dict(self.authorization),
-            "checks": [c.to_dict() for c in self.checks],
-            "failed_checks": [c.name for c in self.failed_checks],
-            "prohibited_actions": list(self.prohibited_actions),
-            "allowed_actions": list(self.allowed_actions),
-            "next_safe_action": self.next_safe_action,
-            "required_profile": self.required_profile,
-            "required_namespace": self.required_namespace,
-            "required_permission": self.required_permission,
-            "correlation_id": self.correlation_id,
-            "allocator_evidence": dict(self.allocator_evidence),
-        }
-
-
-AllocatorFn = Callable[..., dict[str, Any] | None]
-ClaimsFn = Callable[["WorkRequest"], Mapping[tuple[str, int], dict[str, Any]]]
-
-
-def _correlation_id() -> str:
-    return f"req-{uuid.uuid4().hex}"
-
-
-def _selection_matches(
-    selection: Mapping[str, Any] | None, request: WorkRequest
-) -> bool:
-    if not selection:
-        return False
-    kind = _clean(selection.get("kind")).lower()
-    try:
-        number_int = int(selection.get("number"))
-    except (TypeError, ValueError):
-        return False
-    return (kind, number_int) == request.work_key
-
-
-def _authorization_check(
-    decision: console_authz.AuthorizationDecision,
-) -> RequestCheck:
-    return RequestCheck(
-        name=CHECK_AUTHORIZATION,
-        ok=bool(decision.allowed),
-        reason_code=decision.reason_code,
-        detail=decision.detail,
-        evidence={
-            "subject": decision.principal.subject,
-            "role": decision.principal.role,
-            "required_role": decision.required_role,
-            "identity_source": decision.principal.identity_source,
-        },
-    )
-
-
-def _capability_check(request: WorkRequest) -> RequestCheck:
-    """Whether the requested role maps to a declared MCP capability.
-
-    The console never invents an authority: the permission and role come from
-    ``task_capability_map`` via the same ``allocate_next_work`` task the MCP
-    allocator gates on.
-    """
-    # The remote-prefixed hint keeps a dadeschools request from being told to
-    # run under a prgs profile; ``required_profile_for_role`` preserves the
-    # prefix when one is present and falls back to its own default otherwise.
-    profile_hint = f"{request.remote}-{request.desired_role}"
-    try:
-        profile = allocator_service.required_profile_for_role(
-            request.desired_role, profile_name=profile_hint
-        )
-        namespace = allocator_service.required_namespace_for_role(
-            request.desired_role, profile_name=profile_hint
-        )
-    except Exception as exc:  # noqa: BLE001 — an unresolved role is a denial
-        return RequestCheck(
-            name=CHECK_CAPABILITY,
-            ok=False,
-            reason_code="capability_unresolved",
-            detail=(
-                f"no profile/namespace maps to role {request.desired_role!r}: "
-                f"{exc}"
-            ),
-        )
-    resolved = bool(profile and namespace)
-    return RequestCheck(
-        name=CHECK_CAPABILITY,
-        ok=resolved,
-        reason_code="capability_resolved" if resolved else "capability_unresolved",
-        detail=(
-            f"role {request.desired_role!r} runs under profile {profile!r} in "
-            f"MCP namespace {namespace!r}."
-        ),
-        evidence={
-            "required_profile": profile,
-            "required_namespace": namespace,
-            "required_permission": required_permission("allocate_next_work"),
-            "capability_role": required_role("allocate_next_work"),
-        },
-    )
-
-
-def _lease_check(
-    request: WorkRequest,
-    claims: Mapping[tuple[str, int], dict[str, Any]] | None,
-) -> RequestCheck:
-    """Whether the work unit is free of an active claim.
-
-    ``claims is None`` means the control-plane DB could not be read. That is a
-    failure, not an absence of claims: an unreadable substrate must never read
-    as "nothing holds this".
-    """
-    if claims is None:
-        return RequestCheck(
-            name=CHECK_LEASE_AVAILABILITY,
-            ok=False,
-            reason_code=REASON_EVIDENCE_UNAVAILABLE,
-            detail=(
-                "active-claim inventory is unavailable; refusing to treat an "
-                "unreadable control-plane DB as an unclaimed work unit."
-            ),
-        )
-    claim = claims.get(request.work_key)
-    if claim:
-        return RequestCheck(
-            name=CHECK_LEASE_AVAILABILITY,
-            ok=False,
-            reason_code=REASON_DUPLICATE_ASSIGNMENT,
-            detail=(
-                f"{request.work_kind} {request.display_ref} already carries an "
-                f"active {claim.get('role') or 'unknown'} lease."
-            ),
-            evidence={
-                "lease_id": claim.get("lease_id"),
-                "session_id": claim.get("session_id"),
-                "role": claim.get("role"),
-                "expires_at": claim.get("expires_at"),
-            },
-        )
-    return RequestCheck(
-        name=CHECK_LEASE_AVAILABILITY,
-        ok=True,
-        reason_code="lease_available",
-        detail=f"no active lease holds {request.work_kind} {request.display_ref}.",
-    )
-
-
-def _next_safe_action_check(
-    request: WorkRequest, allocation: Mapping[str, Any] | None
-) -> RequestCheck:
-    """Whether the allocator would independently select this exact work unit."""
-    if not allocation:
-        return RequestCheck(
-            name=CHECK_NEXT_SAFE_ACTION,
-            ok=False,
-            reason_code=REASON_EVIDENCE_UNAVAILABLE,
-            detail="allocator dry-run produced no result; refusing to proceed.",
-        )
-    selection = allocation.get("selected") or {}
-    outcome = _clean(allocation.get("outcome"))
-    if not _selection_matches(selection, request):
-        chosen = (
-            f"{_clean(selection.get('kind')) or 'unknown'} #{selection.get('number')}"
-            if selection
-            else "nothing"
-        )
-        return RequestCheck(
-            name=CHECK_NEXT_SAFE_ACTION,
-            ok=False,
-            reason_code=REASON_NOT_NEXT_SAFE,
-            detail=(
-                f"the allocator would select {chosen} for role "
-                f"{request.desired_role!r}, not {request.work_kind} "
-                f"{request.display_ref}. Requests confirm the allocator's "
-                "decision; they never override it."
-            ),
-            evidence={
-                "allocator_outcome": outcome,
-                "allocator_selection": dict(selection),
-                "reasons": list(allocation.get("reasons") or ()),
-            },
-        )
-    return RequestCheck(
-        name=CHECK_NEXT_SAFE_ACTION,
-        ok=True,
-        reason_code="next_safe_work",
-        detail=(
-            f"the allocator selects {request.work_kind} {request.display_ref} "
-            f"for role {request.desired_role!r}."
-        ),
-        evidence={
-            "allocator_outcome": outcome,
-            "selected_action": _clean(selection.get("selected_action")),
-            "expected_role_next": _clean(selection.get("expected_role_next")),
-        },
-    )
-
-
-def _head_pin_check(
-    request: WorkRequest, allocation: Mapping[str, Any] | None
-) -> RequestCheck:
-    """PR work must be pinned to a head SHA; issue work has nothing to pin."""
-    if request.work_kind != KIND_PR:
-        return RequestCheck(
-            name=CHECK_HEAD_PIN,
-            ok=True,
-            reason_code="head_pin_not_applicable",
-            detail="issue work carries no head SHA to pin.",
-        )
-    selection = (allocation or {}).get("selected") or {}
-    allocator_head = _clean(selection.get("head_sha")) or None
-    if not allocator_head:
-        return RequestCheck(
-            name=CHECK_HEAD_PIN,
-            ok=False,
-            reason_code=REASON_EVIDENCE_UNAVAILABLE,
-            detail=(
-                "the allocator reported no head SHA for this PR; PR work "
-                "cannot be initiated unpinned."
-            ),
-        )
-    if request.expected_head_sha and request.expected_head_sha != allocator_head:
-        return RequestCheck(
-            name=CHECK_HEAD_PIN,
-            ok=False,
-            reason_code="head_moved",
-            detail=(
-                "the requested head SHA does not match the PR's current head; "
-                "re-preview against the live head before initiating."
-            ),
-            evidence={
-                "requested_head_sha": request.expected_head_sha,
-                "current_head_sha": allocator_head,
-            },
-        )
-    return RequestCheck(
-        name=CHECK_HEAD_PIN,
-        ok=True,
-        reason_code="head_pinned",
-        detail=f"PR {request.display_ref} is pinned at {allocator_head}.",
-        evidence={"head_sha": allocator_head},
-    )
-
-
-def _next_safe_action_text(
-    request: WorkRequest, checks: Sequence[RequestCheck], authorized: bool
-) -> str:
-    if authorized:
-        return (
-            f"Confirm and initiate {request.desired_role} work on "
-            f"{request.work_kind} {request.display_ref} via the allocator."
-        )
-    for check in checks:
-        if not check.ok:
-            return f"Resolve {check.name}: {check.detail}"
-    return "No safe action; the request is not authorized."
-
-
-def preview_request(
-    request: WorkRequest,
-    *,
-    principal: console_authz.Principal | None = None,
-    allocator: AllocatorFn | None = None,
-    claims_source: ClaimsFn | None = None,
-    correlation_id: str | None = None,
-    audit: bool = True,
-) -> RequestPreview:
-    """Build the read-only intent preview for *request*. Never mutates."""
-    who = principal or console_authz.ANONYMOUS
-    corr = correlation_id or _correlation_id()
-    decision = console_authz.authorize(ACTION_ID, who, for_execution=False)
-
-    allocation: dict[str, Any] | None = None
-    claims: Mapping[tuple[str, int], dict[str, Any]] | None = None
-    checks: list[RequestCheck] = [_authorization_check(decision)]
-    if decision.allowed:
-        # An unauthorized principal never reaches the allocator or the
-        # control-plane DB: a denial must not double as a queue oracle.
-        allocation = _run_allocator(request, allocator, apply=False)
-        claims = _load_claims(request, claims_source)
-        checks.append(_capability_check(request))
-        checks.append(_lease_check(request, claims))
-        checks.append(_next_safe_action_check(request, allocation))
-        checks.append(_head_pin_check(request, allocation))
-
-    authorized = all(c.ok for c in checks)
-    allowed_actions, prohibited_actions = allocator_service.role_actions(
-        request.desired_role
-    )
-    capability = next((c for c in checks if c.name == CHECK_CAPABILITY), None)
-    evidence = capability.evidence if capability else {}
-
-    if authorized:
-        reason_code = REASON_PREVIEW_OK
-        detail = (
-            "Request is authorized. Preview only — nothing has been assigned."
-        )
-    else:
-        first_failure = next(c for c in checks if not c.ok)
-        reason_code, detail = first_failure.reason_code, first_failure.detail
-
-    preview = RequestPreview(
-        request=request,
-        authorized=authorized,
-        reason_code=reason_code,
-        detail=detail,
-        authorization=decision.to_dict(),
-        checks=tuple(checks),
-        prohibited_actions=tuple(prohibited_actions),
-        allowed_actions=tuple(allowed_actions),
-        next_safe_action=_next_safe_action_text(request, checks, authorized),
-        required_profile=str(evidence.get("required_profile") or ""),
-        required_namespace=str(evidence.get("required_namespace") or ""),
-        required_permission=str(evidence.get("required_permission") or ""),
-        correlation_id=corr,
-        allocator_evidence=_allocator_evidence(allocation),
-    )
-
-    if audit:
-        _audit(
-            request,
-            result=console_audit.RESULT_PREVIEWED,
-            decision=decision,
-            principal=who,
-            reason_code=reason_code,
-            detail=detail,
-            correlation_id=corr,
-            metadata={
-                "intent_summary": request.intent_summary,
-                "desired_role": request.desired_role,
-                "authorized": authorized,
-                "failed_checks": [c.name for c in preview.failed_checks],
-                "phase": "preview",
-            },
-        )
-    return preview
-
-
-# --- Initiation -------------------------------------------------------------
-
-
-def apply_request(
-    request: WorkRequest,
-    *,
-    principal: console_authz.Principal | None = None,
-    confirm: bool = False,
-    allocator: AllocatorFn | None = None,
-    claims_source: ClaimsFn | None = None,
-    correlation_id: str | None = None,
-) -> dict[str, Any]:
-    """Initiate *request* as an exclusive assignment, or refuse.
-
-    The only path to an assignment is the allocator agreeing, on a dry-run,
-    that this work unit is what the requested role should take next. Every
-    refusal returns before any mutation is attempted.
-    """
-    who = principal or console_authz.ANONYMOUS
-    corr = correlation_id or _correlation_id()
-
-    execution_decision = console_authz.authorize(ACTION_ID, who, for_execution=True)
-    authorization = execution_decision.to_dict()
-
-    def _refuse(
-        outcome: str,
-        reason_code: str,
-        detail: str,
-        *,
-        status: int,
-        extra: dict[str, Any] | None = None,
-    ) -> dict[str, Any]:
-        _audit(
-            request,
-            result=console_audit.RESULT_DENIED,
-            decision=execution_decision,
-            principal=who,
-            reason_code=reason_code,
-            detail=detail,
-            correlation_id=corr,
-            metadata={
-                "intent_summary": request.intent_summary,
-                "desired_role": request.desired_role,
-                "phase": "apply",
-                "outcome": outcome,
-            },
-        )
-        payload: dict[str, Any] = {
-            "ok": False,
-            "outcome": outcome,
-            "reason_code": reason_code,
-            "detail": detail,
-            "request": request.to_dict(),
-            "authorization": authorization,
-            "assignment": None,
-            "correlation_id": corr,
-            "mutation_performed": False,
-            "status_code": status,
-        }
-        payload.update(extra or {})
-        return payload
-
-    if not (execution_decision.allowed and execution_decision.execution_enabled):
-        return _refuse(
-            OUTCOME_DENIED,
-            REASON_UNAUTHORIZED,
-            execution_decision.detail,
-            status=403,
-        )
-
-    # Confirmation is a property of the action in the RBAC model, so it is read
-    # from there rather than assumed here.
-    action = console_authz.get_action(ACTION_ID)
-    if action is not None and action.requires_confirmation and not confirm:
-        return _refuse(
-            OUTCOME_DENIED,
-            REASON_CONFIRMATION_REQUIRED,
-            (
-                "This action requires explicit confirmation. Re-submit with "
-                "confirm=true after reviewing the preview."
-            ),
-            status=409,
-        )
-
-    preview = preview_request(
-        request,
-        principal=who,
-        allocator=allocator,
-        claims_source=claims_source,
-        correlation_id=corr,
-        audit=False,
-    )
-    if not preview.authorized:
-        outcome = (
-            OUTCOME_BLOCKED
-            if preview.reason_code == REASON_DUPLICATE_ASSIGNMENT
-            else OUTCOME_WAIT
-        )
-        return _refuse(
-            outcome,
-            preview.reason_code,
-            preview.detail,
-            status=409,
-            extra={"preview": preview.to_dict()},
-        )
-
-    fingerprint = (
-        _clean(preview.allocator_evidence.get("candidate_set_fingerprint")) or None
-    )
-    allocation = _run_allocator(
-        request,
-        allocator,
-        apply=True,
-        expected_candidate_set_fingerprint=fingerprint,
-    )
-    if not allocation:
-        return _refuse(
-            OUTCOME_WAIT,
-            REASON_EVIDENCE_UNAVAILABLE,
-            "the allocator returned no result; nothing was assigned.",
-            status=503,
-        )
-
-    assignment = allocation.get("assignment") or None
-    outcome = _clean(allocation.get("outcome"))
-    assigned = bool(
-        outcome == allocator_service.OUTCOME_ASSIGNED
-        and assignment
-        and _selection_matches(allocation.get("selected"), request)
-    )
-    if not assigned:
-        blocked = outcome in {
-            allocator_service.OUTCOME_BLOCKED_LEASE,
-            allocator_service.OUTCOME_BLOCKED_TERMINAL,
-            allocator_service.OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE,
-        }
-        return _refuse(
-            OUTCOME_BLOCKED if blocked else OUTCOME_WAIT,
-            REASON_ALLOCATOR_OUTCOME,
-            (
-                f"the allocator returned {outcome or 'no outcome'} rather than "
-                "an assignment for this work unit; nothing was assigned."
-            ),
-            status=409,
-            extra={"allocator_evidence": _allocator_evidence(allocation)},
-        )
-
-    _audit(
-        request,
-        result=console_audit.RESULT_SUCCEEDED,
-        decision=execution_decision,
-        principal=who,
-        reason_code=REASON_AUTHORIZED,
-        detail=(
-            f"assigned {request.work_kind} {request.display_ref} to role "
-            f"{request.desired_role}."
-        ),
-        correlation_id=corr,
-        metadata={
-            "intent_summary": request.intent_summary,
-            "desired_role": request.desired_role,
-            "phase": "apply",
-            "outcome": OUTCOME_ASSIGNED,
-            "assignment_id": assignment.get("assignment_id"),
-            "lease_id": assignment.get("lease_id"),
-        },
-    )
-    return {
-        "ok": True,
-        "outcome": OUTCOME_ASSIGNED,
-        "reason_code": REASON_AUTHORIZED,
-        "detail": (
-            "Exclusive assignment created via the allocator. Continue in the "
-            f"{preview.required_namespace or 'assigned'} MCP namespace."
-        ),
-        "request": request.to_dict(),
-        "authorization": authorization,
-        "assignment": dict(assignment),
-        "handoff": {
-            "assignment_id": assignment.get("assignment_id"),
-            "lease_id": assignment.get("lease_id"),
-            "session_id": assignment.get("session_id"),
-            "required_profile": preview.required_profile,
-            "required_namespace": preview.required_namespace,
-            "allowed_actions": list(preview.allowed_actions),
-            "forbidden_actions": list(preview.prohibited_actions),
-            "expected_head_sha": assignment.get("expected_head_sha"),
-        },
-        "correlation_id": corr,
-        "mutation_performed": True,
-        "status_code": 201,
-        "allocator_evidence": _allocator_evidence(allocation),
-    }
-
-
-# --- Adapters ---------------------------------------------------------------
-
-
-def _allocator_evidence(allocation: Mapping[str, Any] | None) -> dict[str, Any]:
-    """Reduce an allocator result to the non-secret fields worth surfacing."""
-    if not allocation:
-        return {}
-    return {
-        "outcome": allocation.get("outcome"),
-        "selected": allocation.get("selected"),
-        "reasons": list(allocation.get("reasons") or ()),
-        "candidate_set_fingerprint": allocation.get("candidate_set_fingerprint"),
-        "candidate_count": allocation.get("candidate_count"),
-        "inventory_complete": allocation.get("inventory_complete"),
-        "selection_policy": allocation.get("selection_policy"),
-        "substrate": allocation.get("substrate"),
-    }
-
-
-def _run_allocator(
-    request: WorkRequest,
-    allocator: AllocatorFn | None,
-    *,
-    apply: bool,
-    expected_candidate_set_fingerprint: str | None = None,
-) -> dict[str, Any] | None:
-    fn = allocator or default_allocator
-    try:
-        result = fn(
-            request=request,
-            apply=apply,
-            expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
-        )
-    except Exception:  # noqa: BLE001 — an allocator failure denies, never proceeds
-        return None
-    return result if isinstance(result, dict) else None
-
-
-def _load_claims(
-    request: WorkRequest, claims_source: ClaimsFn | None
-) -> Mapping[tuple[str, int], dict[str, Any]] | None:
-    fn = claims_source or default_claims_source
-    try:
-        claims = fn(request)
-    except Exception:  # noqa: BLE001 — an unreadable substrate is a denial
-        return None
-    return claims if isinstance(claims, Mapping) else None
-
-
-def default_claims_source(
-    request: WorkRequest,
-) -> Mapping[tuple[str, int], dict[str, Any]]:
-    """Live active-claim inventory from the #613 control-plane DB."""
-    import control_plane_db
-
-    db = control_plane_db.ControlPlaneDB()
-    return db.list_active_claims(
-        remote=request.remote, org=request.org, repo=request.repo
-    )
-
-
-def default_allocator(
-    *,
-    request: WorkRequest,
-    apply: bool,
-    expected_candidate_set_fingerprint: str | None = None,
-) -> dict[str, Any] | None:
-    """Run the real allocator over the live queue for *request*'s scope.
-
-    An incomplete candidate inventory returns ``None`` rather than a ranking
-    over a partial set (#758): selecting from a short list can pick the wrong
-    work unit, so the request denies instead.
-    """
-    import control_plane_db
-
-    from webui.queue_loader import load_queue_snapshot
-    from webui.traffic_loader import candidates_from_queue_snapshot
-
-    snapshot = load_queue_snapshot()
-    if snapshot.fetch_error:
-        return None
-    for pagination in (snapshot.pr_pagination, snapshot.issue_pagination):
-        if pagination is not None and not pagination.inventory_complete:
-            return None
-
-    candidates = candidates_from_queue_snapshot(snapshot)
-    db = control_plane_db.ControlPlaneDB()
-    result = allocator_service.allocate_next_work(
-        db,
-        session_id=f"webui-request-{uuid.uuid4().hex[:12]}",
-        role=request.desired_role,
-        remote=request.remote,
-        org=request.org,
-        repo=request.repo,
-        candidates=candidates,
-        apply=bool(apply),
-        allocation_mode="role_scoped",
-        expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
-    )
-    if isinstance(result, dict):
-        result.setdefault("candidate_count", len(candidates))
-        result.setdefault("inventory_complete", True)
-        result.setdefault("selection_policy", allocator_service.SELECTION_POLICY)
-    return result
-
-
-# --- Audit ------------------------------------------------------------------
-
-
-def _audit(
-    request: WorkRequest,
-    *,
-    result: str,
-    decision: console_authz.AuthorizationDecision,
-    principal: console_authz.Principal,
-    reason_code: str,
-    detail: str,
-    correlation_id: str,
-    metadata: dict[str, Any],
-) -> dict[str, Any]:
-    return console_audit.record_event(
-        action_id=ACTION_ID,
-        result=result,
-        decision=decision,
-        principal=principal,
-        target={
-            "kind": request.work_kind,
-            "ref": request.display_ref,
-            "remote": request.remote,
-            "org": request.org,
-            "repo": request.repo,
-        },
-        reason_code=reason_code,
-        request_id=correlation_id,
-        detail=detail,
-        metadata=metadata,
-    )

webui/request_views.py

@@ -1,164 +0,0 @@
-"""HTML views for the operator request surface (#643).
-
-The form is deliberately a *preview* form. It has no initiate button, because
-initiating requires a confirmed POST to ``/api/v1/requests/apply`` and a stray
-form submission must not be able to produce one by accident.
-
-Nothing rendered here is trusted input: every interpolated value is escaped,
-and the page renders only values the service already produced rather than
-echoing a raw request body back.
-"""
-
-from __future__ import annotations
-
-import html
-import json
-from typing import Any
-
-from webui.layout import render_page
-from webui.request_service import (
-    REQUESTABLE_ROLES,
-    WORK_KINDS,
-    RequestError,
-    RequestPreview,
-)
-
-REQUESTS_PATH = "/requests"
-PREVIEW_API_PATH = "/api/v1/requests/preview"
-APPLY_API_PATH = "/api/v1/requests/apply"
-
-
-def _escape(text: Any) -> str:
-    return html.escape(str(text if text is not None else ""), quote=True)
-
-
-REQUEST_PAGE_STYLES = """
-<style>
-  .request-form { display: grid; gap: 0.75rem; max-width: 44rem; }
-  .request-form label { display: grid; gap: 0.25rem; font-size: 0.9rem; }
-  .request-check { margin: 0.35rem 0; }
-  .request-check .verdict-ok { color: var(--accent); }
-  .request-check .verdict-fail { color: #d14; }
-  .request-prohibited code { margin-right: 0.4rem; }
-</style>
-"""
-
-
-def _options(values: tuple[str, ...], selected: Any) -> str:
-    return "".join(
-        f"<option value='{_escape(value)}'"
-        + (" selected" if selected == value else "")
-        + f">{_escape(value)}</option>"
-        for value in values
-    )
-
-
-def _form(values: dict[str, Any] | None = None) -> str:
-    current = dict(values or {})
-    number = current.get("work_number")
-    return (
-        f"<form class='request-form' method='post' action='{REQUESTS_PATH}'>"
-        "<label>Desired role<select name='desired_role'>"
-        f"{_options(REQUESTABLE_ROLES, current.get('desired_role'))}"
-        "</select></label>"
-        "<label>Work kind<select name='work_kind'>"
-        f"{_options(WORK_KINDS, current.get('work_kind'))}"
-        "</select></label>"
-        "<label>Issue or PR number"
-        "<input type='number' name='work_number' min='1' required "
-        f"value='{_escape(number) if number else ''}'></label>"
-        "<label>Intent summary"
-        "<input type='text' name='intent_summary' maxlength='500' required "
-        f"value='{_escape(current.get('intent_summary'))}'></label>"
-        "<label>Expected head SHA <span class='muted'>(PR work only)</span>"
-        "<input type='text' name='expected_head_sha' "
-        f"value='{_escape(current.get('expected_head_sha'))}'></label>"
-        "<button type='submit' class='copy-btn'>Preview request</button>"
-        "<p class='muted meta'>Preview is read-only and creates no assignment. "
-        f"Initiating requires a confirmed POST to <code>{APPLY_API_PATH}</code>."
-        "</p>"
-        "</form>"
-    )
-
-
-def _checks_block(preview: RequestPreview) -> str:
-    rows = []
-    for check in preview.checks:
-        verdict = "PASS" if check.ok else "FAIL"
-        css = "verdict-ok" if check.ok else "verdict-fail"
-        rows.append(
-            "<li class='request-check'>"
-            f"<span class='{css}'><strong>{verdict}</strong></span> "
-            f"<code>{_escape(check.name)}</code> — {_escape(check.detail)} "
-            f"<span class='muted meta'>({_escape(check.reason_code)})</span>"
-            "</li>"
-        )
-    return "<ul>" + "".join(rows) + "</ul>"
-
-
-def _preview_block(preview: RequestPreview) -> str:
-    verdict = "AUTHORIZED" if preview.authorized else "DENIED"
-    prohibited = "".join(
-        f"<code>{_escape(action)}</code>" for action in preview.prohibited_actions
-    )
-    request = preview.request
-    evidence = json.dumps(preview.allocator_evidence, indent=2, default=str)
-    return (
-        "<h3>Intent preview</h3>"
-        f"<p><strong>{verdict}</strong> — {_escape(preview.detail)}</p>"
-        "<p class='meta'>"
-        f"Role <code>{_escape(request.desired_role)}</code> · "
-        f"{_escape(request.work_kind)} <code>{_escape(request.display_ref)}</code>"
-        f" · profile <code>{_escape(preview.required_profile)}</code> · "
-        f"namespace <code>{_escape(preview.required_namespace)}</code> · "
-        f"permission <code>{_escape(preview.required_permission)}</code>"
-        "</p>"
-        f"<p>Intent: {_escape(request.intent_summary)}</p>"
-        f"{_checks_block(preview)}"
-        f"<p><strong>Next safe action:</strong> "
-        f"{_escape(preview.next_safe_action)}</p>"
-        "<p class='request-prohibited'><strong>Prohibited for this role:</strong> "
-        + (prohibited or "<span class='muted'>none declared</span>")
-        + "</p>"
-        "<p class='muted meta'>Correlation id "
-        f"<code>{_escape(preview.correlation_id)}</code></p>"
-        "<details><summary>Allocator evidence</summary>"
-        f"<pre class='prompt-text'>{_escape(evidence)}</pre>"
-        "</details>"
-    )
-
-
-def _error_block(error: RequestError) -> str:
-    field = (
-        f"<p class='meta'>Field: <code>{_escape(error.field_name)}</code></p>"
-        if error.field_name
-        else ""
-    )
-    return (
-        "<h3>Request rejected</h3>"
-        f"<p><strong>{_escape(error.reason_code)}</strong> — "
-        f"{_escape(error.detail)}</p>{field}"
-    )
-
-
-def render_requests_page(
-    *,
-    preview: RequestPreview | None = None,
-    error: RequestError | None = None,
-    submitted: dict[str, Any] | None = None,
-) -> str:
-    """Render the request form, plus a preview or rejection when one exists."""
-    body = (
-        "<h2>Requests</h2>"
-        "<p>Submit a work request — desired role, issue or PR, and intent — "
-        "and see whether it would be authorized before anything is reserved. "
-        "Initiation goes through the allocator (#600/#613); this console never "
-        "self-selects work, never approves, and never merges.</p>"
-        + _form(submitted)
-        + (_error_block(error) if error is not None else "")
-        + (_preview_block(preview) if preview is not None else "")
-        + f"<p class='meta'><a href='{PREVIEW_API_PATH}'>Preview API</a> · "
-        "<a href='/api/console/security-model'>RBAC model</a></p>"
-        + REQUEST_PAGE_STYLES
-    )
-    return render_page(title="Requests", body_html=body)

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

webui/traffic_loader.py

@@ -201,16 +201,6 @@ def _candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate
     return candidates
 
 
-def candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate]:
-    """Public alias for :func:`_candidates_from_queue_snapshot` (#643).
-
-    The request-initiation service ranks the same candidate set this view
-    renders, so both must agree on how a queue row becomes a candidate. One
-    construction, two callers — not two that can drift apart.
-    """
-    return _candidates_from_queue_snapshot(q_snap)
-
-
 def _claim_lease_records(inventory: dict[str, Any] | None) -> list[dict[str, Any]]:
     """Normalize ``build_claim_inventory`` entries into lease records.