Merge branch 'master' into feat/issue-643-request-preview-initiate

Addresses both blockers from the PR #902 review (review 589) for issue #643.

B1 — an allocator-created assignment could be orphaned and reported as no
mutation.

apply_request re-previews, then re-runs the allocator with apply=True. The CAS
fingerprint hashes only {kind, number} plus exclusions, so a competing lease
taken on the requested unit inside the window leaves the fingerprint identical:
the pin passes, the selection loop skips the now-claimed unit and commits an
assignment on the *next* one, and _selection_matches then fails on egress. The
old code returned mutation_performed False with that lease still committed and
owned by a synthetic session nothing heartbeats. The window contains a second
full load_queue_snapshot(), so it is seconds wide, and foreign sessions acting
on this repo concurrently are an observed condition.

The egress mismatch now releases the assignment the allocator created before
refusing. When the release succeeds the refusal reports mutation_performed
False and a compensation record; when it fails the response carries
mutation_performed True, an explicit orphaned_assignment, and a
gitea_release_workflow_lease reclaim action, because claiming nothing changed
while a lease is live is the defect rather than a report of it.

The same state was reachable through _run_allocator's bare except Exception:
allocate_next_work only catches InvalidWorkKindError, LeaseRequiredError and
ControlPlaneError, so anything raised after assign_and_lease committed arrived
as "no result" with a durable lease. A None result on the apply path now sweeps
and releases whatever this flow's session owns. That sweep is only possible
because of the B2 fix below — the session id is now stable across the flow, so
the lease is findable.

B2 — the "read-only" preview wrote to the control-plane DB.

allocate_next_work called db.upsert_session and db.expire_stale_leases
unconditionally, before the apply branch was consulted, and default_allocator
minted a fresh webui-request-<hex> per call. Every preview therefore appended a
never-reused session row and mutated global lease state while the payload said
dry_run True / mutation_performed False, driven by an operator refreshing a
form. One apply wrote two rows and bound the lease to the second, which is why
B1's orphan had no reclaimable owner.

allocate_next_work gains a keyword-only side_effect_free flag, default False so
every existing caller is byte-for-byte unchanged. Under the flag both writes are
suppressed and expired leases are instead filtered out of the claim map in
memory, which reaches the same selection the sweep would have produced without
persisting anything; a claim whose expiry cannot be parsed is kept, since an
unreadable expiry is not evidence that work is free. side_effect_free with
apply=True fails closed rather than silently reserving. request_service mints
one session id per request flow and threads it through both the dry-run and the
apply, and the dry-run now routes through the side-effect-free path.

Coverage.

test_allocator_drift_on_apply_is_not_read_as_an_assignment asserted the defect —
it built the orphan state and then required mutation_performed to be False, which
a leak satisfies. It now requires the compensating release. Added: release-failure
surfacing a reclaim action, an assignment with no lease id, the post-commit
exception route, and session-id identity across the flow. The two areas the review
named as having zero coverage now have it: default_allocator past its two
fail-closed early returns (side_effect_free routing, session-id pass-through and
minting, scope and fingerprint propagation) and default_claims_source (scoped
read, and an unreadable substrate denying rather than reading as "nothing
claimed"). allocator_service gains side-effect-free tests against a real
temp-file DB plus the expiry-filter unit tests.

Every new guard was mutation-tested by reverting it one at a time: B1
compensation removed → 3 failures; post-commit sweep removed → 1; session id
re-minted per call → 2; side_effect_free ignored → 3; in-memory expiry filter
removed → 1.

Verification: WEBUI_TEST_OFFLINE=1 python -m pytest tests/ -q from this
branches/ worktree gives 23 failed, 5263 passed, 6 skipped, 899 subtests. The
sorted FAILED set is identical to the reviewer's clean-master baseline at
2f4dec83 (23 failed, 5190 passed) — no new, changed, or disappeared failure —
and 21 tests were added over the reviewed head's 5242. Zero conflict markers;
py_compile passes; git diff --check clean.

Refs #643, PR #902

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

allocator_service.py

@@ -23,6 +23,7 @@ import json
 import os
 import uuid
 from dataclasses import dataclass, field
+from datetime import datetime, timezone
 from typing import Any, Mapping, Sequence
 
 from control_plane_db import (
@@ -738,6 +739,46 @@ def normalize_exclude_issue_numbers(
     return sorted(out)
 
 
+def _claim_expires_at(claim: Any) -> datetime | None:
+    """Parse a claim's ``expires_at``, or ``None`` when it is absent/malformed."""
+    if not isinstance(claim, Mapping):
+        return None
+    text = str(claim.get("expires_at") or "").strip()
+    if not text:
+        return None
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _drop_expired_claims(
+    claims: Mapping[tuple[str, int], dict[str, Any]],
+    *,
+    now: datetime | None = None,
+) -> dict[tuple[str, int], dict[str, Any]]:
+    """Claims minus those whose lease has already expired (#643).
+
+    The read-only mirror of ``expire_stale_leases``: the sweep marks such rows
+    ``expired`` so they stop being returned as claims, and this reaches the same
+    view without writing. A claim with no parseable ``expires_at`` is **kept** —
+    an unreadable expiry is not evidence that work is free.
+    """
+    moment = now or datetime.now(timezone.utc)
+    kept: dict[tuple[str, int], dict[str, Any]] = {}
+    for key, claim in (claims or {}).items():
+        expires_at = _claim_expires_at(claim)
+        if expires_at is not None and expires_at <= moment:
+            continue
+        kept[key] = claim
+    return kept
+
+
 def candidate_set_fingerprint(
     candidates: Sequence[WorkCandidate],
     *,
@@ -826,12 +867,22 @@ def allocate_next_work(
     exclude_issue_numbers: Sequence[int] | None = None,
     expected_candidate_set_fingerprint: str | None = None,
     allocation_mode: str | None = None,
+    side_effect_free: bool = False,
 ) -> dict[str, Any]:
     """Select and optionally reserve the next work unit via control-plane DB.
 
     *apply=False* (default): dry-run selection only — no lease/assignment.
     *apply=True*: atomic ``assign_and_lease`` for the selected candidate.
 
+    *side_effect_free* (#643): a dry run that writes **nothing** to the
+    control-plane DB. A plain ``apply=False`` still registered a session row and
+    swept stale leases globally, so a caller advertising a read-only preview was
+    mutating on every call. Under this flag both writes are suppressed and stale
+    leases are instead filtered out of the claim map in memory, which yields the
+    same selection the sweep would have produced without persisting anything.
+    Incompatible with *apply* — the combination fails closed rather than
+    silently reserving.
+
     *allocation_mode* (#840): ``cross_role`` (default for controller) inspects
     the complete queue and returns one authoritative selection naming the
     required downstream role/profile/action. ``role_scoped`` keeps prior
@@ -885,7 +936,24 @@ def allocate_next_work(
             "allocation_mode": (allocation_mode or "").strip() or None,
         }
 
+    # A side-effect-free run may never reserve: reserving is a write, and the
+    # flag is the caller's assertion that this call writes nothing (#643).
+    if side_effect_free and apply:
+        return {
+            "success": False,
+            "outcome": OUTCOME_NO_SAFE,
+            "apply": True,
+            "reasons": [
+                "side_effect_free is incompatible with apply=True; an "
+                "assignment is a write (fail closed, #643)"
+            ],
+            "skipped": [],
+            "assignment": None,
+            "substrate": "control_plane_db",
+        }
+
     session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
+    if not side_effect_free:
         try:
             db.upsert_session(
                 session_id=session_id,
@@ -953,6 +1021,12 @@ def allocate_next_work(
                 "assignment": None,
                 "substrate": "control_plane_db",
             }
+        if side_effect_free:
+            # ``list_active_claims`` filters on status alone, so without the
+            # global sweep an already-expired lease would still read as a live
+            # claim and the preview would report work as taken that is free.
+            # Drop those in memory: same view the sweep produces, no write.
+            claims = _drop_expired_claims(claims)
 
     try:
         exclude_nums = normalize_exclude_issue_numbers(exclude_issue_numbers)

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,6 +94,7 @@ 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
@@ -112,6 +113,12 @@ 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
@@ -126,9 +133,24 @@ 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. 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.
+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.
 
 ## Secret redaction
 
@@ -235,13 +257,22 @@ 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).
 
-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.
+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.
 
 ## Local-dev mode
 
@@ -294,6 +325,7 @@ 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-local-dev.md

@@ -77,7 +77,9 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
-| `/sessions` | Phase 1 shell stub — session inventory (backed by #636) |
+| `/sessions` | Runtime and session view (#641) — health + inventory sessions/namespaces/worktrees |
+| `/api/sessions` | JSON export for the runtime/session view |
+| `/api/v1/sessions` | Versioned alias of `/api/sessions` |
 | `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
 | `/timeline` | Phase 1 shell stub — workflow event timeline |
 | `/policy` | Phase 1 shell stub — capability/role policy placeholder |
@@ -284,11 +286,46 @@ The header carries two read-only status badges — an **environment** badge
 a **mode: read-only** badge — plus a **Docs** link to this document. No
 privileged action controls are present in the Phase 1 shell.
 
-Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
-`/policy`, `/insights`) resolve to graceful read-only stub pages instead of
-404s; their backing views land in later child issues of #631 (the inventory
-surfaces are backed by #636). Mutating methods on stub routes still fail closed
-with `read-only-mvp`.
+Not-yet-implemented surfaces (`/inventory`, `/timeline`, `/policy`,
+`/insights`) resolve to graceful read-only stub pages instead of 404s; their
+backing views land in later child issues of #631 (the inventory surfaces are
+backed by #636). Mutating methods on stub routes still fail closed with
+`read-only-mvp`.
+
+### Runtime and sessions (#641)
+
+`/sessions` is a live Phase 1 read-only view that composes:
+
+* runtime health from `#430` (profile, role, identity, master parity, stale warning)
+* control-plane sessions / leases and filesystem locks / worktrees / namespaces from `#636`
+* durable contamination markers when detectable (`#630` runtime recovery, `#671` stable-branch push)
+
+It surfaces stale indicators (dead PID, expired lease) and never silences an
+active contamination marker. Recovery links point only at sanctioned
+reconnect/operator restart docs (`docs/mcp-namespace-eof-recovery.md`,
+`docs/mcp-namespace-health.md`, `docs/mcp-restart-path-inventory.md`, this
+document). The page does **not** restart, kill, or take over sessions; manual
+`pkill` of MCP daemons is contamination, not recovery.
+
+Honesty rules specific to this view:
+
+* **Ownership columns never assert absence they cannot prove.** When the
+  `leases` or `locks` section is degraded or unavailable, the Leases and
+  Worktree-binding cells render `unknown (inventory <status>)` with an
+  *authority unproven* badge instead of `none` / `unbound`, and a caveat names
+  the unreadable sections. A worktree binding is correlated through lease work
+  numbers, so it is unproven when *either* section fails to read.
+  `/api/sessions` carries the same facts as `ownership_authority_complete`,
+  `ownership_section_status`, and per-row `lease_authority` /
+  `worktree_authority`, so a JSON consumer can tell "holds none" from "could
+  not be read".
+* **Contamination text is redacted at the display boundary.** Marker payloads
+  (`command_summary`, `reason_class`, `session_id`, `role`) are
+  operator-supplied free text that does not arrive through inventory scrubbing,
+  so they pass through `webui.inventory.scrub_text`, which collapses `$HOME` and
+  redacts credential-shaped tokens and URL userinfo *anywhere* in the string.
+  The write-time redactor is a narrow denylist and is not relied on. The field
+  itself is kept — it is the `#630` evidence naming which daemon was killed.
 
 ## System-health dashboard (#639)
 

docs/webui-requests.md

@@ -0,0 +1,160 @@
+# 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.
-    other_live_sessions = [
-        s for s in session_impacts if s.live and not s.is_requester
+    # 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
         ]
-    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]
+        scoped_leases = [
+            l for l in lease_impacts if l.session_id == selected_session
+        ]
+    elif resolved_class == RestartClass.ROLE_RUNTIME_RESTART:
+        selected_role = (target_role or "").strip().lower()
+        target_complete = bool(selected_role)
+        scoped_sessions = [
+            s for s in session_impacts if str(s.role or "").lower() == selected_role
+        ]
+        scoped_leases = [
+            l for l in lease_impacts if str(l.role or "").lower() == selected_role
+        ]
+    elif resolved_class == RestartClass.CONNECTOR_RESTART:
+        selected_connector = (target_connector or "").strip()
+        target_complete = bool(selected_connector)
+        scoped_sessions = [
+            s for s in session_impacts if s.connector == selected_connector
+        ]
+        scoped_leases = [
+            l for l in lease_impacts if l.connector == selected_connector
+        ]
+    else:
+        scoped_sessions = list(session_impacts)
+        scoped_leases = list(lease_impacts)
+
+    if policy_enforced and not target_complete:
+        authorization_reasons.append(
+            f"target required for {resolved_class.value if resolved_class else 'unknown class'}"
+        )
+
+    other_live_sessions = [
+        s for s in scoped_sessions if s.live and not s.is_requester
+    ]
+    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_allocator_service.py

@@ -7,6 +7,7 @@ import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime, timezone
 
 from allocator_service import (
     OUTCOME_ASSIGNED,
@@ -15,6 +16,7 @@ from allocator_service import (
     OUTCOME_PREVIEW,
     OUTCOME_WAIT,
     WorkCandidate,
+    _drop_expired_claims,
     allocate_next_work,
     candidate_from_dict,
     classify_skip,
@@ -362,5 +364,161 @@ class AllocatorServiceTest(unittest.TestCase):
         self.assertIn("unavailable", res["reasons"][0].lower())
 
 
+class SideEffectFreeAllocationTest(unittest.TestCase):
+    """``side_effect_free`` dry runs write nothing to the control plane (#643).
+
+    A plain ``apply=False`` still called ``upsert_session`` and
+    ``expire_stale_leases`` before the apply branch was consulted, so a caller
+    advertising a read-only preview mutated on every call — one unreferenced
+    session row per preview, plus a global lease sweep.
+    """
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _alloc(self, **kwargs):
+        defaults = dict(
+            db=self.db,
+            session_id="s-preview",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            candidates=[
+                WorkCandidate(kind="issue", number=643, labels=("status:ready",))
+            ],
+            apply=False,
+            profile_name="prgs-author",
+            username="jcwalker3",
+        )
+        defaults.update(kwargs)
+        return allocate_next_work(**defaults)
+
+    def _session_ids(self) -> set[str]:
+        return {str(r.get("session_id")) for r in self.db.list_sessions()}
+
+    def test_side_effect_free_preview_writes_no_session_row(self):
+        before = self._session_ids()
+        result = self._alloc(side_effect_free=True)
+        self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(self._session_ids(), before)
+        self.assertNotIn("s-preview", self._session_ids())
+
+    def test_plain_dry_run_still_registers_a_session(self):
+        # The default is unchanged for every existing caller.
+        self._alloc()
+        self.assertIn("s-preview", self._session_ids())
+
+    def test_repeated_previews_do_not_accumulate_rows(self):
+        for index in range(5):
+            self._alloc(side_effect_free=True, session_id=f"s-{index}")
+        self.assertEqual(self._session_ids(), set())
+
+    def test_side_effect_free_does_not_sweep_stale_leases(self):
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        assigned = self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=999,
+            lease_ttl_seconds=-60,  # already expired
+        )
+        self.assertEqual(assigned.outcome, "assigned")
+
+        self._alloc(side_effect_free=True)
+
+        # The expired row is still 'active' in the DB: nothing swept it.
+        statuses = {
+            r["lease_id"]: r["status"]
+            for r in self.db.list_leases(
+                remote="prgs", org="org", repo="repo",
+                statuses=("active", "expired"),
+            )
+        }
+        self.assertEqual(statuses.get(assigned.lease_id), "active")
+
+    def test_expired_claims_are_filtered_in_memory_so_work_stays_selectable(self):
+        """The read-only mirror of the sweep: expired claims must not block."""
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=643,
+            lease_ttl_seconds=-60,  # expired: must not withhold #643
+        )
+        result = self._alloc(side_effect_free=True)
+        self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(result["selected"]["number"], 643)
+
+    def test_a_live_claim_still_withholds_the_work(self):
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=643,
+            lease_ttl_seconds=3600,
+        )
+        result = self._alloc(side_effect_free=True)
+        self.assertNotEqual(result["outcome"], OUTCOME_ASSIGNED)
+        self.assertNotEqual((result.get("selected") or {}).get("number"), 643)
+
+    def test_side_effect_free_with_apply_fails_closed(self):
+        result = self._alloc(side_effect_free=True, apply=True)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["outcome"], OUTCOME_NO_SAFE)
+        self.assertIsNone(result["assignment"])
+        self.assertIn("incompatible with apply", result["reasons"][0])
+        # And it reserved nothing.
+        self.assertEqual(
+            self.db.list_leases(remote="prgs", org="org", repo="repo"), []
+        )
+
+
+class DropExpiredClaimsTest(unittest.TestCase):
+    """The in-memory expiry filter behind side-effect-free previews (#643)."""
+
+    def test_unparseable_expiry_is_kept_rather_than_assumed_free(self):
+        claims = {
+            ("issue", 1): {"lease_id": "l1", "expires_at": "not-a-date"},
+            ("issue", 2): {"lease_id": "l2"},
+            ("issue", 3): {"lease_id": "l3", "expires_at": None},
+        }
+        self.assertEqual(_drop_expired_claims(claims), claims)
+
+    def test_expired_dropped_and_future_kept(self):
+        now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc)
+        claims = {
+            ("issue", 1): {"expires_at": "2026-07-25T11:59:59+00:00"},
+            ("issue", 2): {"expires_at": "2026-07-25T12:00:01+00:00"},
+            ("issue", 3): {"expires_at": "2026-07-25T12:00:00+00:00"},  # boundary
+        }
+        kept = _drop_expired_claims(claims, now=now)
+        self.assertEqual(set(kept), {("issue", 2)})
+
+    def test_naive_and_zulu_timestamps_are_treated_as_utc(self):
+        now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc)
+        claims = {
+            ("issue", 1): {"expires_at": "2026-07-25T11:00:00"},  # naive, past
+            ("issue", 2): {"expires_at": "2026-07-25T13:00:00Z"},  # zulu, future
+        }
+        kept = _drop_expired_claims(claims, now=now)
+        self.assertEqual(set(kept), {("issue", 2)})
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_issue_886_apply_authorization_conjunction.py

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

tests/test_mcp_restart_governance_docs.py

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

tests/test_restart_classes.py

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

tests/test_webui_sanctioned_restart.py

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

tests/test_webui_sessions_view.py

@@ -0,0 +1,739 @@
+"""Tests for the Runtime and session view (Phase 1, #641).
+
+Covers clean and stale session rendering, contamination marker surfacing,
+worktree binding display, sanctioned recovery links (no pkill), nav/live
+status, and the JSON API export.
+
+Also pins the two invariants a reviewer found violated at head a81db754:
+degraded ownership sections must render as *unknown* rather than as an
+affirmative "none"/"unbound", and contamination payload text must be redacted
+at the display boundary rather than trusted from the write-time denylist.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from tests.webui_testclient import TestClient
+
+from webui.app import create_app
+from webui.inventory import (
+    AUTHORITY_CONTROL_PLANE_DB,
+    AUTHORITY_FILESYSTEM,
+    InventorySection,
+    InventorySnapshot,
+    STATUS_DEGRADED,
+    STATUS_OK,
+    STATUS_UNAVAILABLE,
+)
+from webui.nav import NAV_GROUPS, STUB_PAGES, iter_nav_items
+from webui.runtime_health import FileHash, RuntimeSnapshot
+from webui.session_loader import (
+    ContaminationMarker,
+    SessionRow,
+    SessionViewSnapshot,
+    _build_session_rows,
+    _inspect_contamination,
+    load_session_view_snapshot,
+    snapshot_to_dict,
+)
+from webui.session_views import render_sessions_page
+
+
+def _runtime(
+    *,
+    stale: str | None = None,
+    profile: str = "prgs-author",
+    role: str = "author",
+) -> RuntimeSnapshot:
+    return RuntimeSnapshot(
+        project_id="gitea-tools",
+        repo_root="/tmp/repo",
+        remote="prgs",
+        host="gitea.prgs.cc",
+        profile_name=profile,
+        role_kind=role,
+        config_model="v2-contexts",
+        profile_mode="dynamic-profile",
+        profile_source="config file profile",
+        authenticated_username="jcwalker3",
+        identity_error=None,
+        repo_sha="a" * 40,
+        remote_master_sha="a" * 40,
+        commits_behind_master=0,
+        stale_runtime_warning=stale,
+        shell_health={"shell_use_allowed": True, "consecutive_spawn_failures": 0},
+        workflow_hashes=(
+            FileHash(label="SKILL.md", path="skills/llm-project-workflow/SKILL.md", sha256="abc"),
+        ),
+        schema_hashes=(),
+        restart_guidance="docs/mcp-namespace-eof-recovery.md",
+        fetch_error=None,
+    )
+
+
+def _inventory(
+    *,
+    sessions: tuple[dict, ...] = (),
+    leases: tuple[dict, ...] = (),
+    locks: tuple[dict, ...] = (),
+    worktrees: tuple[dict, ...] = (),
+    namespaces: tuple[dict, ...] = (),
+    statuses: dict[str, str] | None = None,
+) -> InventorySnapshot:
+    """Build a snapshot; ``statuses`` degrades named sections (default all ok)."""
+    status_of = statuses or {}
+
+    def _status(name: str) -> str:
+        return status_of.get(name, STATUS_OK)
+
+    sections = (
+        InventorySection(
+            name="sessions",
+            authority=AUTHORITY_CONTROL_PLANE_DB,
+            status=_status("sessions"),
+            items=sessions,
+        ),
+        InventorySection(
+            name="leases",
+            authority=AUTHORITY_CONTROL_PLANE_DB,
+            status=_status("leases"),
+            items=leases,
+        ),
+        InventorySection(
+            name="locks",
+            authority=AUTHORITY_FILESYSTEM,
+            status=_status("locks"),
+            items=locks,
+        ),
+        InventorySection(
+            name="worktrees",
+            authority=AUTHORITY_FILESYSTEM,
+            status=_status("worktrees"),
+            items=worktrees,
+        ),
+        InventorySection(
+            name="namespaces",
+            authority=AUTHORITY_FILESYSTEM,
+            status=_status("namespaces"),
+            items=namespaces
+            or (
+                {
+                    "profile_name": "prgs-author",
+                    "role": "author",
+                    "mcp_namespace": "gitea-author",
+                    "capability_summary": {
+                        "can_author": True,
+                        "can_review": False,
+                        "can_merge": False,
+                    },
+                    "active": True,
+                },
+            ),
+            reason="only the profile serving this web process is observable",
+        ),
+    )
+    index = {section.name: section for section in sections}
+    return InventorySnapshot(
+        generated_at="2026-07-25T00:00:00+00:00",
+        sections=sections,
+        collisions=(),
+        correlations=(),
+        scan_ms=1.0,
+        _section_index=index,
+    )
+
+
+def _clean_session() -> dict:
+    return {
+        "session_id": "prgs-author-111-clean",
+        "role": "author",
+        "profile": "prgs-author",
+        "namespace": "gitea-author",
+        "pid": 1111,
+        "pid_alive": True,
+        "status": "active",
+        "started_at": "2026-07-25T00:00:00Z",
+        "last_heartbeat_at": "2026-07-25T01:00:00Z",
+    }
+
+
+def _stale_session() -> dict:
+    return {
+        "session_id": "prgs-author-222-stale",
+        "role": "author",
+        "profile": "prgs-author",
+        "namespace": "gitea-author",
+        "pid": 2222,
+        "pid_alive": False,
+        "status": "active",
+        "started_at": "2026-07-24T00:00:00Z",
+        "last_heartbeat_at": "2026-07-24T01:00:00Z",
+    }
+
+
+class TestBuildSessionRows(unittest.TestCase):
+    def test_clean_session_has_no_stale_or_contamination_flags(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            leases=(
+                {
+                    "lease_id": "lease-clean",
+                    "session_id": "prgs-author-111-clean",
+                    "status": "active",
+                    "expired": False,
+                    "work_kind": "issue",
+                    "work_number": 641,
+                },
+            ),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "branch_name": "feat/issue-641-runtime-session-view",
+                    "worktree_path": "~/Development/Gitea-Tools/branches/feat-issue-641",
+                    "live": True,
+                },
+            ),
+        )
+        rows = _build_session_rows(inventory, contamination=())
+        self.assertEqual(len(rows), 1)
+        row = rows[0]
+        self.assertEqual(row.session_id, "prgs-author-111-clean")
+        self.assertEqual(row.role, "author")
+        self.assertEqual(row.namespace, "gitea-author")
+        self.assertEqual(row.pid_alive, True)
+        self.assertEqual(row.lease_ids, ("lease-clean",))
+        self.assertEqual(row.work_refs, ("issue#641",))
+        self.assertTrue(row.worktree_paths)
+        self.assertEqual(row.stale_flags, ())
+        self.assertEqual(row.contamination_flags, ())
+
+    def test_stale_session_flags_dead_pid(self):
+        inventory = _inventory(sessions=(_stale_session(),))
+        rows = _build_session_rows(inventory, contamination=())
+        self.assertEqual(rows[0].stale_flags, ("pid-dead",))
+
+    def test_contamination_marker_binds_to_session(self):
+        inventory = _inventory(sessions=(_clean_session(),))
+        marker = ContaminationMarker(
+            kind="runtime_recovery_contamination",
+            on_disk=True,
+            has_payload=True,
+            summary="manual daemon kill",
+            reason_class="manual_daemon_kill",
+            session_id="prgs-author-111-clean",
+            role="author",
+            command_summary="pkill -f mcp_server.py",
+            cleared=False,
+        )
+        rows = _build_session_rows(inventory, contamination=(marker,))
+        self.assertIn("runtime_recovery_contamination", rows[0].contamination_flags)
+
+    def test_process_wide_contamination_surfaces_on_all_sessions(self):
+        inventory = _inventory(sessions=(_clean_session(), _stale_session()))
+        marker = ContaminationMarker(
+            kind="stable_branch_contamination",
+            on_disk=True,
+            has_payload=True,
+            summary="direct master push attempt",
+            reason_class="stable_branch_push",
+            session_id=None,
+            cleared=False,
+        )
+        rows = _build_session_rows(inventory, contamination=(marker,))
+        self.assertEqual(len(rows), 2)
+        for row in rows:
+            self.assertTrue(
+                any("stable_branch_contamination" in f for f in row.contamination_flags)
+            )
+
+
+class TestRenderSessionsPage(unittest.TestCase):
+    def _snapshot(
+        self,
+        *,
+        sessions: tuple[dict, ...],
+        contamination: tuple[ContaminationMarker, ...] = (),
+        stale_runtime: str | None = None,
+    ) -> SessionViewSnapshot:
+        inventory = _inventory(
+            sessions=sessions,
+            leases=(
+                {
+                    "lease_id": "lease-1",
+                    "session_id": sessions[0]["session_id"] if sessions else "",
+                    "status": "active",
+                    "expired": False,
+                    "work_kind": "issue",
+                    "work_number": 641,
+                },
+            )
+            if sessions
+            else (),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "worktree_path": "branches/feat-issue-641",
+                },
+            )
+            if sessions
+            else (),
+            worktrees=(
+                {
+                    "rel_path": "branches/feat-issue-641",
+                    "branch": "feat/issue-641-runtime-session-view",
+                    "classification": "active_issue_work",
+                    "registered_worktree": True,
+                    "dirty": False,
+                },
+            ),
+        )
+        rows = _build_session_rows(inventory, contamination)
+        return SessionViewSnapshot(
+            runtime=_runtime(stale=stale_runtime),
+            inventory=inventory,
+            sessions=rows,
+            contamination_markers=contamination,
+        )
+
+    def test_clean_session_render(self):
+        html = render_sessions_page(self._snapshot(sessions=(_clean_session(),)))
+        self.assertIn("Runtime and sessions", html)
+        self.assertIn("prgs-author-111-clean", html)
+        self.assertIn("gitea-author", html)
+        self.assertIn("branches/feat-issue-641", html)
+        self.assertIn("Sanctioned recovery", html)
+        self.assertIn("docs/mcp-namespace-eof-recovery.md", html)
+        # Recovery section must name reconnect and forbid manual kill.
+        recovery_idx = html.lower().find("sanctioned recovery")
+        self.assertGreaterEqual(recovery_idx, 0)
+        recovery = html[recovery_idx:].lower()
+        self.assertIn("reconnect", recovery)
+        self.assertIn("contamination", recovery)
+        self.assertIn("not recovery", recovery)
+        self.assertNotIn("run pkill", recovery)
+        self.assertNotIn("killall", recovery)
+
+    def test_stale_session_render(self):
+        html = render_sessions_page(self._snapshot(sessions=(_stale_session(),)))
+        self.assertIn("prgs-author-222-stale", html)
+        self.assertIn("pid-dead", html)
+        self.assertIn("badge-stale", html)
+
+    def test_contamination_render_is_not_silent(self):
+        marker = ContaminationMarker(
+            kind="runtime_recovery_contamination",
+            on_disk=True,
+            has_payload=True,
+            summary="manual kill",
+            reason_class="manual_daemon_kill",
+            session_id="prgs-author-111-clean",
+            command_summary="pkill -f mcp_server.py",
+            cleared=False,
+        )
+        html = render_sessions_page(
+            self._snapshot(sessions=(_clean_session(),), contamination=(marker,))
+        )
+        self.assertIn("Contamination markers", html)
+        self.assertIn("runtime_recovery_contamination", html)
+        self.assertIn("ACTIVE", html)
+        self.assertIn("badge-blocked", html)
+
+    def test_stale_runtime_banner(self):
+        html = render_sessions_page(
+            self._snapshot(
+                sessions=(_clean_session(),),
+                stale_runtime="server behind master by 3 commits",
+            )
+        )
+        self.assertIn("Stale runtime", html)
+        self.assertIn("server behind master", html)
+
+
+class TestSessionLoaderComposition(unittest.TestCase):
+    def test_load_with_injected_sources(self):
+        inventory = _inventory(sessions=(_clean_session(), _stale_session()))
+        snap = load_session_view_snapshot(
+            load_runtime=lambda: _runtime(),
+            load_inventory=lambda: inventory,
+            inspect_contamination=lambda **_k: {
+                "on_disk": False,
+                "has_payload": False,
+                "summary": "absent",
+            },
+            load_contamination_payload=lambda **_k: None,
+        )
+        self.assertEqual(len(snap.sessions), 2)
+        self.assertEqual(snap.stale_session_count, 1)
+        self.assertEqual(snap.contaminated_session_count, 0)
+        data = snapshot_to_dict(snap)
+        self.assertEqual(data["view"], "runtime-sessions")
+        self.assertEqual(data["issue"], 641)
+        self.assertTrue(data["read_only"])
+        self.assertEqual(data["session_counts"]["total"], 2)
+        self.assertEqual(data["session_counts"]["stale"], 1)
+        self.assertIn("recovery_docs", data)
+
+
+class TestSessionsRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+        inventory = _inventory(
+            sessions=(_clean_session(), _stale_session()),
+            leases=(
+                {
+                    "lease_id": "lease-x",
+                    "session_id": "prgs-author-111-clean",
+                    "status": "active",
+                    "expired": False,
+                    "work_kind": "issue",
+                    "work_number": 641,
+                },
+            ),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "worktree_path": "branches/feat-issue-641",
+                },
+            ),
+            worktrees=(
+                {
+                    "rel_path": "branches/feat-issue-641",
+                    "branch": "feat/issue-641-runtime-session-view",
+                    "classification": "active_issue_work",
+                    "registered_worktree": True,
+                    "dirty": False,
+                },
+            ),
+        )
+        rows = _build_session_rows(inventory, contamination=())
+        self.snapshot = SessionViewSnapshot(
+            runtime=_runtime(stale="stale for test"),
+            inventory=inventory,
+            sessions=rows,
+            contamination_markers=(),
+        )
+        self._patch = mock.patch(
+            "webui.app.load_session_view_snapshot",
+            return_value=self.snapshot,
+        )
+        self._patch.start()
+
+    def tearDown(self):
+        self._patch.stop()
+
+    def test_sessions_page_live(self):
+        response = self.client.get("/sessions")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Runtime and sessions", response.text)
+        self.assertIn("prgs-author-111-clean", response.text)
+        self.assertIn("prgs-author-222-stale", response.text)
+        self.assertIn("pid-dead", response.text)
+        self.assertIn("Sanctioned recovery", response.text)
+        self.assertNotIn("Phase 1 shell placeholder", response.text)
+        self.assertNotIn("child issue of #425", response.text.lower())
+
+    def test_api_sessions_json(self):
+        for path in ("/api/sessions", "/api/v1/sessions"):
+            response = self.client.get(path)
+            self.assertEqual(response.status_code, 200, path)
+            data = response.json()
+            self.assertEqual(data["view"], "runtime-sessions")
+            self.assertEqual(data["session_counts"]["total"], 2)
+            self.assertEqual(data["session_counts"]["stale"], 1)
+            self.assertTrue(data["read_only"])
+            self.assertEqual(data["mutations"], [])
+
+    def test_nav_marks_sessions_live(self):
+        sessions_items = [
+            item for item in iter_nav_items() if item.href == "/sessions"
+        ]
+        self.assertEqual(len(sessions_items), 1)
+        self.assertEqual(sessions_items[0].status, "live")
+        self.assertNotIn("/sessions", STUB_PAGES)
+        # Home page should not mark Sessions as stub.
+        home = self.client.get("/")
+        self.assertEqual(home.status_code, 200)
+        self.assertIn('href="/sessions"', home.text)
+        # Stub marker only appears next to remaining stub destinations.
+        self.assertNotIn(
+            'href="/sessions">Sessions</a> <span class="muted">(stub)</span>',
+            home.text,
+        )
+
+
+class TestDegradedOwnershipAuthority(unittest.TestCase):
+    """B1: a section that could not be read must never render as absence."""
+
+    def _snapshot(self, inventory: InventorySnapshot) -> SessionViewSnapshot:
+        return SessionViewSnapshot(
+            runtime=_runtime(),
+            inventory=inventory,
+            sessions=_build_session_rows(inventory, contamination=()),
+            contamination_markers=(),
+        )
+
+    def test_unavailable_leases_mark_row_authority_unproven(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            statuses={"leases": STATUS_UNAVAILABLE},
+        )
+        row = _build_session_rows(inventory, contamination=())[0]
+        self.assertEqual(row.lease_ids, ())
+        self.assertEqual(row.lease_authority, STATUS_UNAVAILABLE)
+        # Worktree binding is correlated through lease work numbers, so it
+        # inherits the unreadable lease section.
+        self.assertEqual(row.worktree_authority, STATUS_UNAVAILABLE)
+        self.assertFalse(row.ownership_authority_complete)
+
+    def test_readable_locks_are_not_reported_unbound_when_leases_degrade(self):
+        # The narrow variant: locks hold a real worktree_path and read cleanly,
+        # but the lease section that supplies the correlating work number does
+        # not. The row must say unknown, not "unbound".
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "worktree_path": "branches/feat-issue-641",
+                },
+            ),
+            statuses={"leases": STATUS_DEGRADED},
+        )
+        row = _build_session_rows(inventory, contamination=())[0]
+        self.assertEqual(row.worktree_paths, ())
+        self.assertEqual(row.worktree_authority, STATUS_DEGRADED)
+        self.assertFalse(row.ownership_authority_complete)
+
+    def test_degraded_render_says_unknown_not_none_or_unbound(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            statuses={"leases": STATUS_UNAVAILABLE, "locks": STATUS_UNAVAILABLE},
+        )
+        html = render_sessions_page(self._snapshot(inventory))
+        self.assertIn("unknown (inventory unavailable)", html)
+        self.assertIn("authority unproven", html)
+        self.assertIn("Ownership authority incomplete", html)
+        # The affirmative-absence strings must be gone from the row entirely.
+        self.assertNotIn(">none<", html)
+        self.assertNotIn(">unbound<", html)
+
+    def test_clean_inventory_still_renders_affirmative_absence(self):
+        # Guards against over-correcting B1 into "everything is unknown".
+        inventory = _inventory(sessions=(_clean_session(),))
+        html = render_sessions_page(self._snapshot(inventory))
+        self.assertIn(">none<", html)
+        self.assertIn(">unbound<", html)
+        # The column legend mentions "unknown (inventory …)" as static copy, so
+        # assert on the per-row marker and the concrete statuses instead.
+        self.assertNotIn("authority unproven", html)
+        self.assertNotIn("unknown (inventory unavailable)", html)
+        self.assertNotIn("unknown (inventory degraded)", html)
+        self.assertNotIn("Ownership authority incomplete", html)
+
+    def test_json_export_carries_snapshot_and_per_row_authority(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            statuses={"locks": STATUS_UNAVAILABLE},
+        )
+        data = snapshot_to_dict(self._snapshot(inventory))
+        self.assertFalse(data["ownership_authority_complete"])
+        self.assertEqual(
+            data["ownership_section_status"]["locks"], STATUS_UNAVAILABLE
+        )
+        self.assertEqual(data["ownership_section_status"]["leases"], STATUS_OK)
+        self.assertIn("unknown, not unowned", data["ownership_note"])
+
+        row = data["sessions"][0]
+        self.assertTrue(row["lease_authority_complete"])
+        self.assertFalse(row["worktree_authority_complete"])
+        self.assertEqual(row["worktree_authority"], STATUS_UNAVAILABLE)
+        self.assertFalse(row["ownership_authority_complete"])
+        self.assertIn("unknown, not unowned", row["ownership_note"])
+
+    def test_json_export_is_affirmative_when_every_source_reads(self):
+        inventory = _inventory(sessions=(_clean_session(),))
+        data = snapshot_to_dict(self._snapshot(inventory))
+        self.assertTrue(data["ownership_authority_complete"])
+        self.assertTrue(data["sessions"][0]["ownership_authority_complete"])
+
+    def test_missing_session_list_is_not_reported_as_no_sessions(self):
+        inventory = _inventory(statuses={"sessions": STATUS_UNAVAILABLE})
+        html = render_sessions_page(self._snapshot(inventory))
+        self.assertIn("could not be read", html)
+        self.assertIn("not evidence that no sessions exist", html)
+
+    def test_expired_lease_flags_row_as_stale(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            leases=(
+                {
+                    "lease_id": "lease-expired-1",
+                    "session_id": "prgs-author-111-clean",
+                    "status": "active",
+                    "expired": True,
+                    "work_kind": "issue",
+                    "work_number": 641,
+                },
+            ),
+        )
+        row = _build_session_rows(inventory, contamination=())[0]
+        self.assertIn("lease-expired", row.stale_flags)
+        self.assertIn("active-lease-past-expiry", row.stale_flags)
+
+
+class TestContaminationRedaction(unittest.TestCase):
+    """B2: marker payload text is redacted at the display boundary."""
+
+    def _marker(self, payload: dict) -> ContaminationMarker:
+        return _inspect_contamination(
+            "runtime_recovery_contamination",
+            remote="prgs",
+            inspect=lambda **_k: {
+                "on_disk": True,
+                "has_payload": True,
+                "summary": "",
+            },
+            load=lambda **_k: payload,
+        )
+
+    def test_home_paths_are_collapsed(self):
+        home = os.path.expanduser("~")
+        marker = self._marker(
+            {"command_summary": f"pkill -f {home}/Development/Gitea-Tools/x.py"}
+        )
+        self.assertNotIn(home, marker.command_summary)
+        self.assertIn("~/Development/Gitea-Tools/x.py", marker.command_summary)
+
+    def test_secrets_missed_by_the_write_time_denylist_are_redacted(self):
+        # Each of these was verified in review to survive
+        # stable_branch_push_guard.redact_command untouched.
+        cases = (
+            ("curl -H 'X-Api-Key: SUPERSECRET123' https://example.invalid", "SUPERSECRET123"),
+            ("cmd --password hunter2 origin master", "hunter2"),
+            ("PRIVATE_KEY=abc123 python deploy.py", "abc123"),
+            ("fetch https://user:[email protected]/x.git", "user:pw"),
+        )
+        for raw, secret in cases:
+            with self.subTest(raw=raw):
+                marker = self._marker({"command_summary": raw})
+                self.assertNotIn(secret, marker.command_summary)
+                self.assertIn("[redacted]", marker.command_summary)
+
+    def test_command_summary_is_redacted_not_removed(self):
+        # It is legitimate #630 evidence: the operator must still see which
+        # daemon was killed.
+        marker = self._marker(
+            {
+                "command_summary": "pkill -f gitea_mcp_server.py",
+                "reason_class": "manual_daemon_kill",
+                "session_id": "prgs-author-111-clean",
+                "role": "author",
+            }
+        )
+        self.assertIn("pkill -f gitea_mcp_server.py", marker.command_summary)
+        self.assertEqual(marker.reason_class, "manual_daemon_kill")
+        self.assertEqual(marker.session_id, "prgs-author-111-clean")
+        self.assertEqual(marker.role, "author")
+
+    def test_rendered_page_exposes_no_home_path_from_a_marker(self):
+        home = os.path.expanduser("~")
+        marker = self._marker(
+            {
+                "command_summary": f"pkill -f {home}/Development/Gitea-Tools/x.py",
+                "reason_class": "manual_daemon_kill",
+            }
+        )
+        inventory = _inventory(sessions=(_clean_session(),))
+        html = render_sessions_page(
+            SessionViewSnapshot(
+                runtime=_runtime(),
+                inventory=inventory,
+                sessions=_build_session_rows(inventory, (marker,)),
+                contamination_markers=(marker,),
+            )
+        )
+        self.assertIn("Contamination markers", html)
+        self.assertNotIn(home, html)
+
+
+class TestSessionsPageEscaping(unittest.TestCase):
+    """Hostile values from every rendered source stay inert (N2)."""
+
+    HOSTILE = '<script>alert("xss")</script>'
+
+    def test_hostile_session_and_marker_values_are_escaped(self):
+        session = dict(_clean_session())
+        session["session_id"] = f"sid-{self.HOSTILE}"
+        session["role"] = self.HOSTILE
+        session["profile"] = self.HOSTILE
+        session["namespace"] = self.HOSTILE
+        session["status"] = self.HOSTILE
+        inventory = _inventory(
+            sessions=(session,),
+            leases=(
+                {
+                    "lease_id": self.HOSTILE,
+                    "session_id": session["session_id"],
+                    "status": "active",
+                    "expired": False,
+                    "work_kind": self.HOSTILE,
+                    "work_number": 641,
+                },
+            ),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "worktree_path": self.HOSTILE,
+                },
+            ),
+        )
+        marker = ContaminationMarker(
+            kind="runtime_recovery_contamination",
+            on_disk=True,
+            has_payload=True,
+            summary=self.HOSTILE,
+            reason_class=self.HOSTILE,
+            session_id=session["session_id"],
+            role=self.HOSTILE,
+            command_summary=self.HOSTILE,
+            cleared=False,
+        )
+        html = render_sessions_page(
+            SessionViewSnapshot(
+                runtime=_runtime(),
+                inventory=inventory,
+                sessions=_build_session_rows(inventory, (marker,)),
+                contamination_markers=(marker,),
+            )
+        )
+        self.assertNotIn("<script>", html)
+        self.assertNotIn('alert("xss")', html)
+        self.assertIn("&lt;script&gt;", html)
+
+    def test_hostile_values_in_a_degraded_render_are_escaped(self):
+        inventory = _inventory(
+            sessions=(dict(_clean_session(), session_id=f"sid-{self.HOSTILE}"),),
+            statuses={"leases": STATUS_UNAVAILABLE, "locks": STATUS_DEGRADED},
+        )
+        html = render_sessions_page(
+            SessionViewSnapshot(
+                runtime=_runtime(),
+                inventory=inventory,
+                sessions=_build_session_rows(inventory, contamination=()),
+                contamination_markers=(),
+            )
+        )
+        self.assertNotIn("<script>", html)
+        self.assertIn("&lt;script&gt;", html)
+        self.assertIn("unknown (inventory", html)
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/app.py

@@ -48,6 +48,11 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
 from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 from webui.runtime_views import render_runtime_page
+from webui.session_loader import (
+    load_session_view_snapshot,
+    snapshot_to_dict as session_view_snapshot_to_dict,
+)
+from webui.session_views import render_sessions_page
 from webui.inventory import (
     SECTION_NAMES as _INVENTORY_SECTIONS,
     load_inventory_snapshot,
@@ -67,6 +72,8 @@ 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"})
@@ -87,6 +94,7 @@ _LEGACY_PAGES = (
     ("/projects", "Projects", "registry and onboarding (#427)"),
     ("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
     ("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
+    ("/sessions", "Sessions", "runtime and session view (#641)"),
     ("/audit", "Audit", "final-report paste and validator preview (#431)"),
     ("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
     ("/leases", "Leases", "collision and lease visibility (#433)"),
@@ -325,6 +333,17 @@ async def api_runtime(_request: Request) -> JSONResponse:
     return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
 
 
+async def sessions(_request: Request) -> HTMLResponse:
+    """Runtime and session view (#641) — read-only composition of health + inventory."""
+    snapshot = load_session_view_snapshot()
+    return HTMLResponse(render_sessions_page(snapshot))
+
+
+async def api_sessions(_request: Request) -> JSONResponse:
+    """JSON export for the runtime/session view (#641)."""
+    return JSONResponse(session_view_snapshot_to_dict(load_session_view_snapshot()))
+
+
 async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
     if request.method == "GET":
         return "", None
@@ -722,6 +741,109 @@ 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":
@@ -764,6 +886,9 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
             Route("/api/prompts", api_prompts, methods=["GET"]),
             Route("/runtime", runtime, methods=["GET"]),
             Route("/api/runtime", api_runtime, methods=["GET"]),
+            Route("/sessions", sessions, methods=["GET"]),
+            Route("/api/sessions", api_sessions, methods=["GET"]),
+            Route("/api/v1/sessions", api_sessions, methods=["GET"]),
             Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
             Route("/analytics", analytics, methods=["GET"]),
             Route("/api/analytics", api_v1_analytics, methods=["GET"]),
@@ -786,6 +911,17 @@ 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,6 +115,12 @@ 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:
@@ -277,6 +283,27 @@ _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}
@@ -430,6 +457,33 @@ 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:
@@ -469,16 +523,19 @@ 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.
-    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.
+    ``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.
     """
     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(
@@ -497,7 +554,7 @@ def authorize(
         "requires_confirmation": action.requires_confirmation,
         "dual_control": action.dual_control,
         "break_glass": action.break_glass,
-        "execution_enabled": False,
+        "execution_enabled": wired,
     }
 
     if not who.authenticated:
@@ -530,13 +587,19 @@ def authorize(
             **base,
         )
 
-    if for_execution and action.phase > ACTIVE_PHASE:
+    if for_execution and not wired:
         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}. Execution is not wired."
+                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."
             ),
             **base,
         )
@@ -545,8 +608,8 @@ def authorize(
         allowed=True,
         reason_code=ALLOW_PREVIEW,
         detail=(
-            "Principal holds the required role. Preview only — execution "
-            "remains disabled until the Phase 2 action framework ships."
+            "Principal holds the required role. Execution proceeds only for an "
+            "action with a wired execution path; everything else is preview."
         ),
         **base,
     )

webui/inventory.py

@@ -76,6 +76,35 @@ _CREDENTIAL_KEY_RE = re.compile(
 )
 _REDACTED = "[redacted]"
 
+#: Credential-shaped *name* as it appears inside a free-form command line. This
+#: is deliberately broader than :data:`_CREDENTIAL_KEY_RE` — it also matches a
+#: bare ``key`` component, so ``PRIVATE_KEY=`` is caught. Over-redacting a
+#: displayed string is safe; under-redacting one is not.
+_TEXT_CREDENTIAL_NAME = (
+    r"[A-Za-z0-9_.\-]*"
+    r"(?:token|secret|password|passwd|key|authorization|bearer|credential)"
+    r"[A-Za-z0-9_.\-]*"
+)
+#: A value following such a name: single-quoted, double-quoted, or bare. The
+#: bare form stops at a quote so an enclosing quote survives the redaction.
+_TEXT_CREDENTIAL_VALUE = r"'[^']*'|\"[^\"]*\"|[^\s'\"]+"
+
+_TEXT_CREDENTIAL_FLAG_RE = re.compile(
+    rf"(?P<key>(?<![\w\-])--?{_TEXT_CREDENTIAL_NAME})"
+    rf"(?P<sep>[=\s]+)"
+    rf"(?P<value>{_TEXT_CREDENTIAL_VALUE})",
+    re.IGNORECASE,
+)
+_TEXT_CREDENTIAL_ASSIGN_RE = re.compile(
+    rf"(?P<key>(?<![\w\-]){_TEXT_CREDENTIAL_NAME})"
+    rf"(?P<sep>\s*[:=]\s*)"
+    rf"(?P<value>{_TEXT_CREDENTIAL_VALUE})",
+    re.IGNORECASE,
+)
+_TEXT_URL_USERINFO_RE = re.compile(
+    r"(?P<scheme>\b[A-Za-z][A-Za-z0-9+.\-]*://)[^\s/@]+@"
+)
+
 
 @dataclass(frozen=True)
 class InventorySection:
@@ -225,6 +254,42 @@ def scrub(value: Any, *, key: str | None = None) -> Any:
     return repr(value)
 
 
+def collapse_home(text: str) -> str:
+    """Collapse every ``$HOME`` occurrence *inside* a string, not just a prefix."""
+    home = os.path.expanduser("~")
+    if not home or home == "/":
+        return text
+    return text.replace(home, "~")
+
+
+def scrub_text(value: Any) -> Any:
+    """Redact a free-form text blob such as a recorded command line.
+
+    :func:`scrub` keys off structured field *names* and whole-value prefixes,
+    which is right for inventory records but blind to a secret embedded in the
+    middle of a sentence. This collapses ``$HOME`` and redacts credential-shaped
+    tokens and URL userinfo *anywhere* in the string, so operator-supplied text
+    rendered verbatim — contamination ``command_summary`` (#630) above all — is
+    held to the same standard as every other field on the page.
+
+    Returns ``None`` unchanged so callers can keep "absent" distinct from "".
+    """
+    if value is None:
+        return None
+    text = value if isinstance(value, str) else str(value)
+    text = collapse_home(text)
+    text = _TEXT_URL_USERINFO_RE.sub(
+        lambda m: f"{m.group('scheme')}{_REDACTED}@", text
+    )
+    text = _TEXT_CREDENTIAL_FLAG_RE.sub(
+        lambda m: f"{m.group('key')}{m.group('sep')}{_REDACTED}", text
+    )
+    text = _TEXT_CREDENTIAL_ASSIGN_RE.sub(
+        lambda m: f"{m.group('key')}{m.group('sep')}{_REDACTED}", text
+    )
+    return text
+
+
 # ── control-plane database (read-only) ───────────────────────────────────────
 
 

webui/nav.py

@@ -45,10 +45,11 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
         NavItem("/queue", "Queue"),
         NavItem("/leases", "Leases"),
         NavItem("/actions", "Actions"),
+        NavItem("/requests", "Requests"),
     )),
     NavGroup("Runtime/Sessions", (
         NavItem("/runtime", "Runtime health"),
-        NavItem("/sessions", "Sessions", "stub"),
+        NavItem("/sessions", "Sessions"),
     )),
     NavGroup("Projects", (
         NavItem("/projects", "Projects"),
@@ -76,11 +77,6 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
 # issues of epic #631. Each maps a path to (title, description). Routes are
 # registered so nav links resolve to a graceful, read-only stub page.
 STUB_PAGES: dict[str, tuple[str, str]] = {
-    "/sessions": (
-        "Sessions",
-        "Active session, capability, and role inventory. Backed by the unified "
-        "inventory API (#636) once it lands.",
-    ),
     "/inventory": (
         "Inventory",
         "Unified sessions, leases, locks, namespaces, and worktree inventory. "

webui/request_views.py

@@ -0,0 +1,164 @@
+"""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/runtime_views.py

@@ -88,5 +88,7 @@ def render_runtime_page(snapshot: RuntimeSnapshot) -> str:
         "<p class='muted'>MVP is read-only — restart MCP servers from your IDE/operator "
         "workflow. Related issue: <code>#420</code>. Guidance: "
         f"<code>{html.escape(snapshot.restart_guidance)}</code></p>"
-        "<p class='muted'>This page does not expose tokens or perform MCP restarts.</p>"
+        "<p class='muted'>This page does not expose tokens or perform MCP restarts. "
+        "Correlated sessions, worktree bindings, and contamination markers: "
+        "<a href='/sessions'>/sessions</a> (#641).</p>"
     )

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

@@ -0,0 +1,517 @@
+"""Compose runtime health + inventory into a sessions/runtime view (#641).
+
+Phase 1 is read-only. It correlates namespaces, sessions, capabilities,
+worktree bindings, lease ownership, stale flags, and contamination markers
+when they are detectable on disk (#630 / #671). It never restarts, kills, or
+takes over a session.
+
+Sources:
+
+* :mod:`webui.runtime_health` — profile, role, stale runtime, shell health.
+* :mod:`webui.inventory` — sessions, leases, locks, worktrees, namespaces
+  and collision signals from the control-plane DB + filesystem.
+* :mod:`mcp_session_state` — durable contamination markers (inspect only).
+
+Secrets are never read. Inventory-sourced values arrive already redacted by
+:func:`webui.inventory.scrub`; free-form marker text this module loads itself is
+put through :func:`webui.inventory.scrub_text`, which collapses ``$HOME`` and
+redacts credential-shaped tokens *inside* a string rather than only at its start.
+
+Ownership columns are authority-aware. A lease or lock section that could not be
+read renders as ``unknown``, never as ``none`` or ``unbound``: a lease the reader
+could not load is not an absent lease (see
+:attr:`webui.inventory.InventorySnapshot.ownership_authority_complete`).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable
+
+import mcp_session_state
+from webui.inventory import (
+    OWNERSHIP_SECTIONS,
+    STATUS_OK,
+    InventorySection,
+    InventorySnapshot,
+    load_inventory_snapshot,
+    scrub_text,
+    snapshot_to_dict as inventory_snapshot_to_dict,
+)
+from webui.runtime_health import (
+    RuntimeSnapshot,
+    load_runtime_snapshot,
+    snapshot_to_dict as runtime_snapshot_to_dict,
+)
+
+# Sanctioned recovery pointers only — never pkill / killall (#630).
+SANCTIONED_RECOVERY_DOCS: tuple[dict[str, str], ...] = (
+    {
+        "label": "MCP namespace EOF recovery (reconnect only)",
+        "path": "docs/mcp-namespace-eof-recovery.md",
+        "note": "IDE/client reconnect or operator-owned restart; never kill daemons.",
+    },
+    {
+        "label": "MCP namespace health",
+        "path": "docs/mcp-namespace-health.md",
+        "note": "client_namespace probe proves namespace health.",
+    },
+    {
+        "label": "Restart path inventory",
+        "path": "docs/mcp-restart-path-inventory.md",
+        "note": "Catalog of sanctioned reconnect/restart paths.",
+    },
+    {
+        "label": "Local web UI recovery",
+        "path": "docs/webui-local-dev.md",
+        "note": "Operator console start and documented recovery sequence.",
+    },
+)
+
+_CONTAMINATION_KINDS: tuple[str, ...] = (
+    mcp_session_state.KIND_RUNTIME_RECOVERY_CONTAMINATION,
+    mcp_session_state.KIND_STABLE_BRANCH_CONTAMINATION,
+)
+
+#: Status recorded on a row when the backing inventory section is absent
+#: entirely — distinct from a section that reported itself degraded.
+AUTHORITY_MISSING = "missing"
+
+
+def _section_status(section: InventorySection | None) -> str:
+    """Status of an ownership section, treating an absent section as missing."""
+    if section is None:
+        return AUTHORITY_MISSING
+    return section.status
+
+
+def _combined_authority(*statuses: str) -> str:
+    """Worst status of the sections a derived column depends on.
+
+    A column proved from two sections is only trustworthy when *both* read
+    cleanly, so the first non-``ok`` status wins.
+    """
+    for status in statuses:
+        if status != STATUS_OK:
+            return status
+    return STATUS_OK
+
+
+@dataclass(frozen=True)
+class ContaminationMarker:
+    """A detectable durable contamination marker (audit-safe summary)."""
+
+    kind: str
+    on_disk: bool
+    has_payload: bool
+    summary: str
+    reason_class: str | None = None
+    session_id: str | None = None
+    role: str | None = None
+    command_summary: str | None = None
+    cleared: bool = False
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "kind": self.kind,
+            "on_disk": self.on_disk,
+            "has_payload": self.has_payload,
+            "summary": self.summary,
+            "reason_class": self.reason_class,
+            "session_id": self.session_id,
+            "role": self.role,
+            "command_summary": self.command_summary,
+            "cleared": self.cleared,
+            "active": self.on_disk and self.has_payload and not self.cleared,
+        }
+
+
+@dataclass(frozen=True)
+class SessionRow:
+    """One correlated session row for the sessions table."""
+
+    session_id: str
+    role: str | None
+    profile: str | None
+    namespace: str | None
+    pid: int | None
+    pid_alive: bool | None
+    status: str | None
+    started_at: str | None
+    last_heartbeat_at: str | None
+    lease_ids: tuple[str, ...] = ()
+    work_refs: tuple[str, ...] = ()
+    worktree_paths: tuple[str, ...] = ()
+    stale_flags: tuple[str, ...] = ()
+    contamination_flags: tuple[str, ...] = ()
+    #: Status of the section backing ``lease_ids``/``work_refs``. While this is
+    #: not ``ok`` those tuples mean "could not be read", never "none held".
+    lease_authority: str = STATUS_OK
+    #: Worst status across the sections backing ``worktree_paths`` (locks are
+    #: correlated through lease work numbers, so both must read cleanly).
+    worktree_authority: str = STATUS_OK
+
+    @property
+    def ownership_authority_complete(self) -> bool:
+        """True only when this row's ownership columns are provable."""
+        return self.lease_authority == STATUS_OK and self.worktree_authority == STATUS_OK
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "session_id": self.session_id,
+            "role": self.role,
+            "profile": self.profile,
+            "namespace": self.namespace,
+            "pid": self.pid,
+            "pid_alive": self.pid_alive,
+            "status": self.status,
+            "started_at": self.started_at,
+            "last_heartbeat_at": self.last_heartbeat_at,
+            "lease_ids": list(self.lease_ids),
+            "work_refs": list(self.work_refs),
+            "worktree_paths": list(self.worktree_paths),
+            "stale_flags": list(self.stale_flags),
+            "contamination_flags": list(self.contamination_flags),
+            "is_stale": bool(self.stale_flags),
+            "is_contaminated": bool(self.contamination_flags),
+            "lease_authority": self.lease_authority,
+            "worktree_authority": self.worktree_authority,
+            "lease_authority_complete": self.lease_authority == STATUS_OK,
+            "worktree_authority_complete": self.worktree_authority == STATUS_OK,
+            "ownership_authority_complete": self.ownership_authority_complete,
+            "ownership_note": (
+                "Lease and worktree columns are proved from sections that read "
+                "cleanly."
+                if self.ownership_authority_complete
+                else "An ownership source could not be read; empty lease_ids or "
+                "worktree_paths on this row mean unknown, not unowned."
+            ),
+        }
+
+
+@dataclass(frozen=True)
+class SessionViewSnapshot:
+    """Composed runtime + session inventory view (#641)."""
+
+    runtime: RuntimeSnapshot
+    inventory: InventorySnapshot
+    sessions: tuple[SessionRow, ...]
+    contamination_markers: tuple[ContaminationMarker, ...]
+    recovery_docs: tuple[dict[str, str], ...] = SANCTIONED_RECOVERY_DOCS
+    fetch_error: str | None = None
+
+    @property
+    def stale_session_count(self) -> int:
+        return sum(1 for row in self.sessions if row.stale_flags)
+
+    @property
+    def contaminated_session_count(self) -> int:
+        return sum(1 for row in self.sessions if row.contamination_flags)
+
+    @property
+    def active_contamination(self) -> tuple[ContaminationMarker, ...]:
+        return tuple(m for m in self.contamination_markers if m.to_dict()["active"])
+
+    @property
+    def ownership_authority_complete(self) -> bool:
+        """False while any ownership section is degraded, absent, or unavailable."""
+        return self.inventory.ownership_authority_complete
+
+    @property
+    def ownership_section_status(self) -> dict[str, str]:
+        """Per-section status for the three ownership-bearing sections."""
+        return {
+            name: _section_status(self.inventory.section(name))
+            for name in OWNERSHIP_SECTIONS
+        }
+
+
+def _inspect_contamination(
+    kind: str,
+    *,
+    remote: str | None,
+    inspect: Callable[..., dict[str, Any]] | None = None,
+    load: Callable[..., dict[str, Any] | None] | None = None,
+) -> ContaminationMarker:
+    """Inspect one contamination kind; never raises into the page render path."""
+    inspect_fn = inspect or mcp_session_state.inspect_state_envelope
+    load_fn = load or mcp_session_state.load_state
+    try:
+        envelope = inspect_fn(kind=kind, remote=remote)
+    except Exception as exc:  # noqa: BLE001 — fail soft for the dashboard
+        return ContaminationMarker(
+            kind=kind,
+            on_disk=False,
+            has_payload=False,
+            summary=f"contamination inspect failed: {type(exc).__name__}",
+        )
+
+    reason_class = None
+    session_id = None
+    role = None
+    command_summary = None
+    cleared = False
+    summary = scrub_text(str(envelope.get("summary") or ""))
+
+    if envelope.get("on_disk") and envelope.get("has_payload"):
+        try:
+            payload = load_fn(kind=kind, remote=remote) or {}
+        except Exception:  # noqa: BLE001
+            payload = {}
+        if isinstance(payload, dict):
+            reason_class = payload.get("reason_class")
+            session_id = payload.get("session_id")
+            role = payload.get("role")
+            command_summary = payload.get("command_summary") or payload.get("detail")
+            cleared = bool(payload.get("cleared_by_reconciler"))
+            if not summary:
+                summary = scrub_text(
+                    f"{kind}: {reason_class or 'present'}"
+                    + (" (cleared)" if cleared else "")
+                )
+
+    # Marker payloads are operator-supplied free text and are the one thing on
+    # this page that does not arrive through inventory scrubbing. The write-time
+    # redactor is a narrow denylist, so redact again at the display boundary:
+    # it leaves $HOME paths, `-H 'X-Api-Key: …'`, `--password …`, and
+    # `PRIVATE_KEY=…` intact. The field itself stays — it is #630 evidence.
+    return ContaminationMarker(
+        kind=kind,
+        on_disk=bool(envelope.get("on_disk")),
+        has_payload=bool(envelope.get("has_payload")),
+        summary=summary or f"{kind}: not present",
+        reason_class=scrub_text(str(reason_class)) if reason_class else None,
+        session_id=scrub_text(str(session_id)) if session_id else None,
+        role=scrub_text(str(role)) if role else None,
+        command_summary=scrub_text(str(command_summary)) if command_summary else None,
+        cleared=cleared,
+    )
+
+
+def _load_contamination_markers(
+    *,
+    remote: str | None,
+    inspect: Callable[..., dict[str, Any]] | None = None,
+    load: Callable[..., dict[str, Any] | None] | None = None,
+) -> tuple[ContaminationMarker, ...]:
+    return tuple(
+        _inspect_contamination(kind, remote=remote, inspect=inspect, load=load)
+        for kind in _CONTAMINATION_KINDS
+    )
+
+
+def _build_session_rows(
+    inventory: InventorySnapshot,
+    contamination: tuple[ContaminationMarker, ...],
+) -> tuple[SessionRow, ...]:
+    sessions_section = inventory.section("sessions")
+    leases_section = inventory.section("leases")
+    locks_section = inventory.section("locks")
+
+    # Ownership columns may only assert absence when their source read cleanly.
+    lease_authority = _section_status(leases_section)
+    # Locks carry claimant profile/username, not control-plane session ids, so a
+    # worktree binding is correlated through lease work numbers: it depends on
+    # the locks *and* the leases section.
+    worktree_authority = _combined_authority(
+        lease_authority, _section_status(locks_section)
+    )
+
+    leases_by_session: dict[str, list[dict[str, Any]]] = {}
+    for lease in (leases_section.items if leases_section else ()):
+        sid = str(lease.get("session_id") or "")
+        if sid:
+            leases_by_session.setdefault(sid, []).append(lease)
+
+    active_markers = [m for m in contamination if m.to_dict()["active"]]
+    marker_session_ids = {
+        m.session_id for m in active_markers if m.session_id
+    }
+
+    rows: list[SessionRow] = []
+    for raw in sessions_section.items if sessions_section else ():
+        sid = str(raw.get("session_id") or "")
+        if not sid:
+            continue
+        session_leases = leases_by_session.get(sid, [])
+        lease_ids = tuple(
+            str(lease["lease_id"])
+            for lease in session_leases
+            if lease.get("lease_id")
+        )
+        work_refs: list[str] = []
+        work_numbers: list[int] = []
+        for lease in session_leases:
+            kind = lease.get("work_kind")
+            number = lease.get("work_number")
+            if kind and number is not None:
+                work_refs.append(f"{kind}#{number}")
+                try:
+                    work_numbers.append(int(number))
+                except (TypeError, ValueError):
+                    pass
+
+        worktree_paths: list[str] = []
+        for lock in (locks_section.items if locks_section else ()):
+            try:
+                issue_no = int(lock.get("issue_number"))
+            except (TypeError, ValueError):
+                continue
+            if issue_no in work_numbers and lock.get("worktree_path"):
+                worktree_paths.append(str(lock["worktree_path"]))
+
+        stale_flags: list[str] = []
+        if raw.get("pid_alive") is False:
+            stale_flags.append("pid-dead")
+        status = str(raw.get("status") or "").lower()
+        if status and status not in {"active", "alive", "running", "ok"}:
+            stale_flags.append(f"status:{status}")
+        for lease in session_leases:
+            if lease.get("expired") is True:
+                stale_flags.append("lease-expired")
+            if str(lease.get("status") or "").lower() == "active" and lease.get(
+                "expired"
+            ) is True:
+                stale_flags.append("active-lease-past-expiry")
+
+        contamination_flags: list[str] = []
+        if sid in marker_session_ids:
+            for marker in active_markers:
+                if marker.session_id == sid:
+                    contamination_flags.append(marker.kind)
+        # Process-wide contamination with no session binding still surfaces
+        # against every live session so it cannot be silent (#630).
+        for marker in active_markers:
+            if not marker.session_id and marker.kind not in contamination_flags:
+                contamination_flags.append(f"{marker.kind}:process-wide")
+
+        rows.append(
+            SessionRow(
+                session_id=sid,
+                role=raw.get("role"),
+                profile=raw.get("profile"),
+                namespace=raw.get("namespace"),
+                pid=raw.get("pid") if isinstance(raw.get("pid"), int) else None,
+                pid_alive=raw.get("pid_alive")
+                if isinstance(raw.get("pid_alive"), bool)
+                else None,
+                status=raw.get("status"),
+                started_at=raw.get("started_at"),
+                last_heartbeat_at=raw.get("last_heartbeat_at"),
+                lease_ids=lease_ids,
+                work_refs=tuple(work_refs),
+                worktree_paths=tuple(worktree_paths),
+                stale_flags=tuple(dict.fromkeys(stale_flags)),
+                contamination_flags=tuple(dict.fromkeys(contamination_flags)),
+                lease_authority=lease_authority,
+                worktree_authority=worktree_authority,
+            )
+        )
+
+    return tuple(rows)
+
+
+def load_session_view_snapshot(
+    *,
+    load_runtime: Callable[..., RuntimeSnapshot] | None = None,
+    load_inventory: Callable[..., InventorySnapshot] | None = None,
+    inspect_contamination: Callable[..., dict[str, Any]] | None = None,
+    load_contamination_payload: Callable[..., dict[str, Any] | None] | None = None,
+) -> SessionViewSnapshot:
+    """Build the composed sessions/runtime view. Fail-soft on partial sources."""
+    runtime_loader = load_runtime or load_runtime_snapshot
+    inventory_loader = load_inventory or load_inventory_snapshot
+    fetch_error: str | None = None
+
+    try:
+        runtime = runtime_loader()
+    except Exception as exc:  # noqa: BLE001
+        fetch_error = f"runtime snapshot failed: {type(exc).__name__}: {exc}"
+        # Minimal placeholder so the page still renders inventory + recovery.
+        from webui.runtime_health import RuntimeSnapshot as _RS
+
+        runtime = _RS(
+            project_id="unknown",
+            repo_root="",
+            remote="",
+            host="",
+            profile_name="unknown",
+            role_kind="unknown",
+            config_model="unknown",
+            profile_mode="unknown",
+            profile_source="unknown",
+            authenticated_username=None,
+            identity_error=str(exc),
+            repo_sha=None,
+            remote_master_sha=None,
+            commits_behind_master=None,
+            stale_runtime_warning=None,
+            shell_health={},
+            workflow_hashes=(),
+            schema_hashes=(),
+            restart_guidance="docs/mcp-namespace-eof-recovery.md",
+            fetch_error=str(exc),
+        )
+
+    try:
+        inventory = inventory_loader()
+    except Exception as exc:  # noqa: BLE001
+        msg = f"inventory snapshot failed: {type(exc).__name__}: {exc}"
+        fetch_error = f"{fetch_error}; {msg}" if fetch_error else msg
+        inventory = load_inventory_snapshot(
+            db_path="/nonexistent-for-fail-soft",
+            lock_dir="/nonexistent-for-fail-soft",
+        )
+
+    remote = getattr(runtime, "remote", None)
+    contamination = _load_contamination_markers(
+        remote=remote,
+        inspect=inspect_contamination,
+        load=load_contamination_payload,
+    )
+    sessions = _build_session_rows(inventory, contamination)
+    return SessionViewSnapshot(
+        runtime=runtime,
+        inventory=inventory,
+        sessions=sessions,
+        contamination_markers=contamination,
+        fetch_error=fetch_error,
+    )
+
+
+def snapshot_to_dict(snapshot: SessionViewSnapshot) -> dict[str, Any]:
+    """JSON export for ``/api/sessions`` (read-only)."""
+    return {
+        "api_version": "v1",
+        "view": "runtime-sessions",
+        "issue": 641,
+        "fetch_error": snapshot.fetch_error,
+        "runtime": runtime_snapshot_to_dict(snapshot.runtime),
+        "inventory": inventory_snapshot_to_dict(snapshot.inventory),
+        "sessions": [row.to_dict() for row in snapshot.sessions],
+        "session_counts": {
+            "total": len(snapshot.sessions),
+            "stale": snapshot.stale_session_count,
+            "contaminated": snapshot.contaminated_session_count,
+        },
+        "ownership_authority_complete": snapshot.ownership_authority_complete,
+        "ownership_section_status": snapshot.ownership_section_status,
+        "ownership_note": (
+            "Every ownership source read cleanly; a session with no lease and no "
+            "worktree path genuinely holds neither."
+            if snapshot.ownership_authority_complete
+            else "An ownership source is degraded or unavailable. Empty lease_ids "
+            "and worktree_paths mean unknown, not unowned; consult each row's "
+            "lease_authority and worktree_authority."
+        ),
+        "contamination_markers": [
+            marker.to_dict() for marker in snapshot.contamination_markers
+        ],
+        "active_contamination": [
+            marker.to_dict() for marker in snapshot.active_contamination
+        ],
+        "recovery_docs": [dict(doc) for doc in snapshot.recovery_docs],
+        "read_only": True,
+        "phase": 1,
+        "mutations": [],
+    }

webui/session_views.py

@@ -0,0 +1,403 @@
+"""HTML views for the Runtime and session view (Phase 1, #641).
+
+Read-only composition of runtime health (#430) and inventory sessions /
+namespaces / worktrees (#636). Surfaces stale and contamination indicators
+when detectable. Recovery links point only at sanctioned reconnect/restart
+docs — never at manual process kill (#630).
+"""
+
+from __future__ import annotations
+
+from html import escape
+from typing import Sequence
+
+from webui.inventory import STATUS_OK
+from webui.layout import render_page
+from webui.session_loader import (
+    ContaminationMarker,
+    SessionRow,
+    SessionViewSnapshot,
+)
+
+
+def _badge(text: str, css: str) -> str:
+    return f'<span class="badge {css}">{escape(text)}</span>'
+
+
+def _flags(flags: Sequence[str], *, css: str) -> str:
+    if not flags:
+        return '<span class="muted">—</span>'
+    return " ".join(_badge(flag, css) for flag in flags)
+
+
+def _unproven_cell(status: str) -> str:
+    """Render an ownership column whose backing inventory section failed to read.
+
+    Never "none" and never "unbound": an unreadable source proves nothing about
+    ownership, and claiming otherwise is the exact failure the
+    ``ownership_authority_complete`` invariant exists to prevent.
+    """
+    return (
+        f'<span class="muted">unknown (inventory {escape(status)})</span><br>'
+        f'{_badge("authority unproven", "badge-health-degraded")}'
+    )
+
+
+def _runtime_banner(snapshot: SessionViewSnapshot) -> str:
+    runtime = snapshot.runtime
+    stale = runtime.stale_runtime_warning
+    stale_html = ""
+    if stale:
+        stale_html = (
+            f'<div class="health-card health-stale" style="margin-top:0.75rem;">'
+            f"<strong>Stale runtime:</strong> {escape(stale)}</div>"
+        )
+    identity = runtime.authenticated_username or "unresolved"
+    if runtime.identity_error:
+        identity = f"unresolved ({runtime.identity_error})"
+
+    return f"""<div class="health-card">
+  <h3>Runtime context</h3>
+  <table class="detail">
+    <tr><th>Profile</th><td><code>{escape(runtime.profile_name)}</code></td></tr>
+    <tr><th>Role kind</th><td>{escape(runtime.role_kind)}</td></tr>
+    <tr><th>Identity</th><td>{escape(str(identity))}</td></tr>
+    <tr><th>Remote / host</th>
+      <td><code>{escape(runtime.remote)}</code> · <code>{escape(runtime.host)}</code></td>
+    </tr>
+    <tr><th>Local HEAD</th>
+      <td><code>{escape(runtime.repo_sha or "unknown")}</code></td>
+    </tr>
+    <tr><th>Remote master</th>
+      <td><code>{escape(runtime.remote_master_sha or "unknown")}</code></td>
+    </tr>
+    <tr><th>Commits behind</th>
+      <td>{escape(str(runtime.commits_behind_master if runtime.commits_behind_master is not None else "unknown"))}</td>
+    </tr>
+  </table>
+  <p class="muted">Full runtime detail: <a href="/runtime">/runtime</a> ·
+  Inventory API: <a href="/api/v1/inventory"><code>/api/v1/inventory</code></a></p>
+  {stale_html}
+</div>"""
+
+
+def _summary_bar(snapshot: SessionViewSnapshot) -> str:
+    total = len(snapshot.sessions)
+    stale = snapshot.stale_session_count
+    contaminated = snapshot.contaminated_session_count
+    active_markers = len(snapshot.active_contamination)
+    inv_status = snapshot.inventory.status
+    authority_complete = snapshot.ownership_authority_complete
+    authority_text = "complete" if authority_complete else "incomplete"
+    authority_css = "badge-health-ok" if authority_complete else "badge-health-degraded"
+    return f"""<div class="health-card" style="display:flex; flex-wrap:wrap; gap:1rem; align-items:center;">
+  <div><strong>Sessions:</strong> <span class="badge badge-health-ok">{total}</span></div>
+  <div><strong>Stale:</strong> <span class="badge badge-stale">{stale}</span></div>
+  <div><strong>Contaminated:</strong> <span class="badge badge-blocked">{contaminated}</span></div>
+  <div><strong>Active markers:</strong> <span class="badge badge-health-unproven">{active_markers}</span></div>
+  <div><strong>Inventory:</strong> <span class="badge badge-health-skipped">{escape(inv_status)}</span></div>
+  <div><strong>Ownership authority:</strong> <span class="badge {authority_css}">{authority_text}</span></div>
+</div>"""
+
+
+def _ownership_caveat(snapshot: SessionViewSnapshot) -> str:
+    """Name the unreadable ownership sections, or render nothing when all read."""
+    if snapshot.ownership_authority_complete:
+        return ""
+    degraded = ", ".join(
+        f"{name}: {status}"
+        for name, status in snapshot.ownership_section_status.items()
+        if status != STATUS_OK
+    )
+    return (
+        '<div class="health-card health-stale">'
+        "<strong>Ownership authority incomplete:</strong> "
+        f"{escape(degraded)}. Columns marked <em>unknown</em> could not be read. "
+        "No session below may be treated as holding no lease or no worktree "
+        "binding — absence of evidence is not evidence of absence.</div>"
+    )
+
+
+def _render_session_row(row: SessionRow) -> str:
+    pid = "—" if row.pid is None else str(row.pid)
+    pid_alive = "—" if row.pid_alive is None else ("alive" if row.pid_alive else "dead")
+    pid_css = (
+        "badge-health-ok"
+        if row.pid_alive is True
+        else ("badge-blocked" if row.pid_alive is False else "badge-health-skipped")
+    )
+    # An empty tuple only means "holds none" when its source read cleanly.
+    if row.lease_authority != STATUS_OK:
+        lease_cell = _unproven_cell(row.lease_authority)
+    else:
+        leases = (
+            ", ".join(f"<code>{escape(lid)}</code>" for lid in row.lease_ids)
+            if row.lease_ids
+            else '<span class="muted">none</span>'
+        )
+        work = (
+            ", ".join(escape(ref) for ref in row.work_refs)
+            if row.work_refs
+            else '<span class="muted">—</span>'
+        )
+        lease_cell = (
+            f'{leases}<div class="muted" style="font-size:0.82rem; '
+            f'margin-top:0.2rem;">{work}</div>'
+        )
+
+    if row.worktree_authority != STATUS_OK:
+        worktrees = _unproven_cell(row.worktree_authority)
+    else:
+        worktrees = (
+            "<br>".join(f"<code>{escape(path)}</code>" for path in row.worktree_paths)
+            if row.worktree_paths
+            else '<span class="muted">unbound</span>'
+        )
+    return f"""<tr>
+  <td><code>{escape(row.session_id)}</code></td>
+  <td>
+    <div><code>{escape(str(row.role or "—"))}</code> / <code>{escape(str(row.profile or "—"))}</code></div>
+    <div class="muted" style="font-size:0.82rem;">ns: <code>{escape(str(row.namespace or "—"))}</code></div>
+  </td>
+  <td>
+    <code>{escape(pid)}</code>
+    {_badge(pid_alive, pid_css)}
+  </td>
+  <td>{escape(str(row.status or "—"))}<div class="muted" style="font-size:0.82rem;">{escape(str(row.last_heartbeat_at or ""))}</div></td>
+  <td>{lease_cell}</td>
+  <td>{worktrees}</td>
+  <td>{_flags(row.stale_flags, css="badge-stale")}</td>
+  <td>{_flags(row.contamination_flags, css="badge-blocked")}</td>
+</tr>"""
+
+
+def _sessions_table(snapshot: SessionViewSnapshot) -> str:
+    rows: Sequence[SessionRow] = snapshot.sessions
+    if not rows:
+        sessions_status = snapshot.ownership_section_status.get("sessions", STATUS_OK)
+        if sessions_status != STATUS_OK:
+            return (
+                '<p class="muted">Session inventory is '
+                f"<strong>{escape(sessions_status)}</strong> — the session list "
+                "could not be read. This is not evidence that no sessions "
+                "exist.</p>"
+            )
+        return (
+            '<p class="muted">No control-plane sessions recorded. Inventory may '
+            "be unavailable, or no MCP workers have registered yet.</p>"
+        )
+    body = "".join(_render_session_row(row) for row in rows)
+    return f"""<table class="registry">
+  <thead>
+    <tr>
+      <th>Session</th>
+      <th>Role / profile / namespace</th>
+      <th>PID</th>
+      <th>Status</th>
+      <th>Leases / work</th>
+      <th>Worktree binding</th>
+      <th>Stale</th>
+      <th>Contamination</th>
+    </tr>
+  </thead>
+  <tbody>
+    {body}
+  </tbody>
+</table>"""
+
+
+def _namespaces_section(snapshot: SessionViewSnapshot) -> str:
+    section = snapshot.inventory.section("namespaces")
+    if section is None:
+        return (
+            '<div class="prompt-card"><h3>Namespaces</h3>'
+            '<p class="muted">Namespaces section not loaded.</p></div>'
+        )
+    if not section.ok:
+        return f"""<div class="prompt-card">
+  <h3>Namespaces {_badge(section.status, "badge-health-degraded")}</h3>
+  <p class="muted">{escape(section.reason or "unavailable")}</p>
+</div>"""
+
+    rows = []
+    for item in section.items:
+        caps = item.get("capability_summary") or {}
+        cap_bits = ", ".join(
+            name for name, ok in sorted(caps.items()) if ok
+        ) or "none"
+        rows.append(
+            "<tr>"
+            f"<td><code>{escape(str(item.get('mcp_namespace') or '—'))}</code></td>"
+            f"<td><code>{escape(str(item.get('profile_name') or '—'))}</code></td>"
+            f"<td>{escape(str(item.get('role') or '—'))}</td>"
+            f"<td>{escape(cap_bits)}</td>"
+            f"<td>{'yes' if item.get('active') else 'no'}</td>"
+            "</tr>"
+        )
+    reason = (
+        f'<p class="muted">{escape(section.reason)}</p>'
+        if section.reason
+        else ""
+    )
+    return f"""<div class="prompt-card">
+  <h3>Namespaces / capabilities</h3>
+  {reason}
+  <table class="registry">
+    <thead>
+      <tr>
+        <th>Namespace</th>
+        <th>Profile</th>
+        <th>Role</th>
+        <th>Capabilities</th>
+        <th>Active in process</th>
+      </tr>
+    </thead>
+    <tbody>
+      {"".join(rows) if rows else '<tr><td colspan="5" class="muted">No namespace rows.</td></tr>'}
+    </tbody>
+  </table>
+</div>"""
+
+
+def _worktrees_section(snapshot: SessionViewSnapshot) -> str:
+    section = snapshot.inventory.section("worktrees")
+    if section is None:
+        return ""
+    if not section.ok and not section.items:
+        return f"""<div class="prompt-card">
+  <h3>Worktrees {_badge(section.status, "badge-health-degraded")}</h3>
+  <p class="muted">{escape(section.reason or "unavailable")}</p>
+</div>"""
+
+    rows = []
+    for item in section.items[:50]:
+        rows.append(
+            "<tr>"
+            f"<td><code>{escape(str(item.get('rel_path') or item.get('path') or '—'))}</code></td>"
+            f"<td><code>{escape(str(item.get('branch') or '—'))}</code></td>"
+            f"<td>{escape(str(item.get('classification') or '—'))}</td>"
+            f"<td>{'yes' if item.get('registered_worktree') else 'no'}</td>"
+            f"<td>{'dirty' if item.get('dirty') else 'clean'}</td>"
+            "</tr>"
+        )
+    more = ""
+    if len(section.items) > 50:
+        more = f'<p class="muted">Showing 50 of {len(section.items)}. Full list: <a href="/worktrees">/worktrees</a>.</p>'
+    return f"""<div class="prompt-card">
+  <h3>Worktree bindings</h3>
+  <p class="muted">Registered issue worktrees under <code>branches/</code>. Hygiene detail: <a href="/worktrees">/worktrees</a>.</p>
+  <table class="registry">
+    <thead>
+      <tr>
+        <th>Path</th>
+        <th>Branch</th>
+        <th>Classification</th>
+        <th>Registered</th>
+        <th>State</th>
+      </tr>
+    </thead>
+    <tbody>
+      {"".join(rows) if rows else '<tr><td colspan="5" class="muted">No worktrees recorded.</td></tr>'}
+    </tbody>
+  </table>
+  {more}
+</div>"""
+
+
+def _contamination_section(markers: Sequence[ContaminationMarker]) -> str:
+    if not markers:
+        return (
+            '<div class="prompt-card"><h3>Contamination markers</h3>'
+            '<p class="muted">No contamination kinds inspected.</p></div>'
+        )
+    rows = []
+    for marker in markers:
+        active = marker.to_dict()["active"]
+        status = "ACTIVE" if active else ("cleared" if marker.cleared else "absent")
+        css = "badge-blocked" if active else "badge-health-ok"
+        rows.append(
+            "<tr>"
+            f"<td><code>{escape(marker.kind)}</code></td>"
+            f"<td>{_badge(status, css)}</td>"
+            f"<td>{escape(marker.reason_class or '—')}</td>"
+            f"<td><code>{escape(marker.session_id or '—')}</code></td>"
+            f"<td>{escape(marker.command_summary or marker.summary)}</td>"
+            "</tr>"
+        )
+    return f"""<div class="prompt-card">
+  <h3>Contamination markers (#630 / #671)</h3>
+  <p class="muted">Durable markers only — never silent when present. Clearance is reconciler-only.</p>
+  <table class="registry">
+    <thead>
+      <tr>
+        <th>Kind</th>
+        <th>State</th>
+        <th>Reason class</th>
+        <th>Session</th>
+        <th>Summary</th>
+      </tr>
+    </thead>
+    <tbody>
+      {"".join(rows)}
+    </tbody>
+  </table>
+</div>"""
+
+
+def _recovery_section(snapshot: SessionViewSnapshot) -> str:
+    items = []
+    for doc in snapshot.recovery_docs:
+        items.append(
+            "<li>"
+            f"<code>{escape(doc['path'])}</code> — "
+            f"<strong>{escape(doc['label'])}</strong>: {escape(doc['note'])}"
+            "</li>"
+        )
+    return f"""<div class="prompt-card">
+  <h3>Sanctioned recovery (read-only)</h3>
+  <p class="muted">This view does <strong>not</strong> restart, kill, or take over sessions.
+  Manual <code>pkill</code> / <code>kill</code> of MCP daemons is contamination (#630), not recovery.</p>
+  <ul class="reasons">
+    {"".join(items)}
+    <li>Prefer IDE/client reconnect (<code>/mcp reconnect</code>) or an operator-owned restart recorded in the restart inventory.</li>
+  </ul>
+</div>"""
+
+
+def render_sessions_page(snapshot: SessionViewSnapshot) -> str:
+    """Render the full HTML body for the runtime/session view."""
+    error_block = ""
+    if snapshot.fetch_error:
+        error_block = (
+            f'<div class="health-card health-stale"><strong>Partial load:</strong> '
+            f"{escape(snapshot.fetch_error)}</div>"
+        )
+    if snapshot.runtime.fetch_error:
+        error_block += (
+            f'<div class="health-card health-stale"><strong>Runtime note:</strong> '
+            f"{escape(snapshot.runtime.fetch_error)}</div>"
+        )
+
+    body = f"""
+{error_block}
+{_runtime_banner(snapshot)}
+{_summary_bar(snapshot)}
+
+<div class="prompt-card">
+  <h3>Sessions</h3>
+  <p class="muted">Control-plane sessions correlated with leases and worktree bindings.
+  Stale and contamination flags are fail-soft: absence of a marker is not proof of cleanliness when inventory is degraded.
+  Lease and worktree columns read <em>unknown (inventory …)</em> when their source could not be loaded.</p>
+  {_ownership_caveat(snapshot)}
+  {_sessions_table(snapshot)}
+</div>
+
+{_namespaces_section(snapshot)}
+{_worktrees_section(snapshot)}
+{_contamination_section(snapshot.contamination_markers)}
+{_recovery_section(snapshot)}
+"""
+    return render_page(title="Sessions", body_html=f"""<h2>Runtime and sessions</h2>
+<p class="meta">Phase 1 read-only view (#641). Combines runtime health (#430) with
+unified inventory sessions/namespaces/worktrees (#636). No restart or session-takeover controls.</p>
+{body}""")

webui/system_health_views.py

@@ -278,8 +278,10 @@ def _recovery_card() -> str:
         "controls arrive in Phase 2 (#642); until then recovery runs through "
         "the sanctioned client reconnect / operator restart path.</p>"
         "<ul class='reasons'>"
-        "<li><a href='/runtime'>Runtime and session view</a> — active profile, "
-        "workflow hashes, and shell health.</li>"
+        "<li><a href='/runtime'>Runtime health</a> — active profile, workflow "
+        "hashes, and shell health.</li>"
+        "<li><a href='/sessions'>Runtime and sessions</a> — namespaces, session "
+        "rows, worktree bindings, and contamination markers (#641).</li>"
         "<li>Reconnect the MCP client from the IDE, then re-run the blocked "
         "cycle. Never kill the daemon process manually: unmanaged kills are "
         "recorded as runtime contamination (#630).</li>"

webui/traffic_loader.py

@@ -201,6 +201,16 @@ 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.