docs(webui): document the read-only restart console surface (#667 )

Records the two GET routes, what each panel consumes, and the three properties the surface is held to: an unreadable source reports unavailable rather than green, authorization is probed with for_execution=True so a Phase 1 refusal is never shown as an allow, and the control-plane database is opened mode=ro so reading status never creates it. Refs #655 #642 #658 #661 #662 #663 #633 #652 #653 #664 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
feat(webui): add read-only console restart status and impact controls (Closes #667 )
2026-07-25 17:26:30 -04:00 · 2026-07-25 17:23:52 -04:00
15 changed files with 1511 additions and 3260 deletions
@@ -23,7 +23,6 @@ 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 (
@@ -739,46 +738,6 @@ 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],
    *,
@@ -867,22 +826,12 @@ 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
@@ -936,24 +885,7 @@ 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,
@@ -1021,12 +953,6 @@ 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)
@@ -94,7 +94,6 @@ already define, and a regression test asserts each mapping matches.
 | `record_analytics_usage` | operator | gated_write | `runtime.record_analytics_usage` | Yes | No | No | 2 |
 | `system.reload_namespace` | controller | privileged | `runtime.reload_namespace` | Yes | No | No | 2 |
 | `system.restart_namespace` | admin | destructive | `runtime.restart_namespace` | Yes | **Yes** | **Yes** | 2 |
 | `initiate_workflow` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
 **Dual control** means the acting principal may not be the sole authority: a
 second distinct principal must confirm. **Break-glass** means the action is
@@ -113,12 +112,6 @@ by the console — both hand off to a host supervisor, and neither exposes a raw
 process kill. See
 [`sanctioned-restart-controls.md`](sanctioned-restart-controls.md) (#642).
 `initiate_workflow` (#643) is operator-class because its outcome is a *claim*,
 not a Gitea verdict. Requesting reviewer or merger work reserves that work
 through the allocator; it does not grant the right to approve or merge, which
 stays with the MCP role profile and its own capability gates. See
 [`webui-requests.md`](webui-requests.md).
 ### Authorization decision
 `authorize(action_id, principal, for_execution=False)` returns a decision
@@ -133,24 +126,9 @@ record and **denies by default**. The deny reasons are closed and enumerated:
 | `phase_not_active` | Execution requested for an action whose phase is not open. |
 | `allowed_preview_only` | Authorized — preview only, execution still disabled. |
-There is no implicit allow branch.
+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
-`execution_enabled` on the decision reports whether the action has a live
+read an allow as permission to mutate.
 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
@@ -257,22 +235,13 @@ second one. The integration points are already wired and observable:
  instead of adding a parallel check.
 - **`GET /api/console/security-model`** publishes the RBAC matrix, redaction
  policy, and audit policy as JSON for operators and tests.
 - **`POST /api/v1/requests/preview` and `.../apply`** (#643) are the first
  actions to use this model for a real execution path. Preview always returns a
  decision and an audited `previewed` record; apply requires `confirm=true`,
  emits `succeeded` or `denied`, and reserves work only through the allocator.
  See [`webui-requests.md`](webui-requests.md).
-A Phase 2 action must: use `execution_wired` rather than a private enable flag,
+To open Phase 2, a child issue must: raise `ACTIVE_PHASE`, implement the
-implement the confirmation and dual-control flow the matrix already declares,
+confirmation and dual-control flow the matrix already declares, emit a
-emit a `succeeded` or `failed` record alongside the `gitea_audit` mutation
+`succeeded` or `failed` record alongside the `gitea_audit` mutation record, and
-record, and keep `viewer` unable to reach any of it. Turning on execution
+keep `viewer` unable to reach any of it. Turning on execution without the
-without the confirmation flow contradicts a declared requirement and is a
+confirmation flow contradicts a declared requirement and is a review failure,
-review failure, not a shortcut.
+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
@@ -325,7 +294,6 @@ Until Phase 2 wires it, probe protection rests on network placement alone, as
 | `WEBUI_ROLE_MAP` | unset | JSON subject → role map |
 | `WEBUI_REQUIRE_PROBE_AUTH` | unset | Require auth for non-public probes |
 | `WEBUI_CONSOLE_AUDIT_LOG` | unset | Append-only audit sink path |
 | `WEBUI_REQUESTS_EXECUTION` | unset | Opt in to `initiate_workflow` execution (#643) |
 All are read server-side only. None is ever rendered into a page or returned by
 an API.
@@ -1,160 +0,0 @@
 # Web console requests: intent preview and workflow initiation (#643)
 **Phase 2. Preview is always live and always read-only. Initiation is wired but
 denied until an operator opts in.**
 Before this surface, starting role work meant pasting a prompt into a terminal
 and trusting the operator to have checked the allocator first. Nothing enforced
 that check, so two sessions could reach for the same issue and each believe it
 was theirs. This page replaces the paste with a *request*: a desired role, an
 issue or PR, and a stated intent, answered by an authorization decision and —
 on confirmation — an exclusive assignment from the allocator.
 | Concern | Module |
 |---------|--------|
 | Request model, preview, initiation | `webui/request_service.py` |
 | Form and preview rendering | `webui/request_views.py` |
 | Authorization | `webui/console_authz.py` (`initiate_workflow`) |
 | Audit | `webui/console_audit.py` |
 | Ownership substrate | `allocator_service.py` + `control_plane_db.py` |
 ## Surfaces
 | Path | Method | Purpose |
 |------|--------|---------|
 | `/requests` | GET | Request form |
 | `/requests` | POST | Render an intent preview. **Never assigns.** |
 | `/api/v1/requests/preview` | POST | Intent preview as JSON |
 | `/api/v1/requests/apply` | POST | Initiate — confirmed, audited, allocator-owned |
 The HTML form has no initiate button on purpose. Initiating requires a
 confirmed POST to `/api/v1/requests/apply`, so a stray form submission cannot
 reserve work as a side effect.
 ## The request
 ```json
 {
  "desired_role": "author",
  "work_kind": "issue",
  "work_number": 643,
  "intent_summary": "implement request preview and initiation",
  "remote": "prgs",
  "org": "Scaled-Tech-Consulting",
  "repo": "Gitea-Tools",
  "expected_head_sha": null
 }
 ```
 `desired_role` is one of `author`, `reviewer`, `merger`, `reconciler`,
 `controller`. `work_kind` is `issue` or `pr`. `remote`/`org`/`repo` default to
 the first project in the registry when omitted; when neither the request nor
 the registry resolves them, the request is rejected rather than pointed at some
 other repository. `intent_summary` is required — it is what the audit record
 states as the reason — and is truncated to 500 characters.
 Parsing rejects rather than corrects. An unknown role, an unknown work kind, a
 non-positive number, or a missing intent each return `400` with a `reason_code`
 and the offending `field`.
 ## Preview
 Five checks, each with its own verdict, reason code, and detail:
 | Check | Passes when |
 |-------|-------------|
 | `authorization` | The console principal holds `operator` or above |
 | `capability` | The desired role maps to a declared profile and MCP namespace |
 | `lease_availability` | No active claim holds the work unit |
 | `next_safe_action` | The allocator would independently select this exact work unit |
 | `head_pin` | PR work resolves to a head SHA, and a supplied SHA still matches |
 A preview also returns the role's `allowed_actions` and `prohibited_actions`
 (from `allocator_service.ROLE_ACTIONS`), the `required_profile` and
 `required_namespace` the work must run under, and a `correlation_id` that ties
 the preview to its audit record and to any assignment that follows.
 Preview is read-only in the strict sense: it calls the allocator with
 `apply=false` and writes nothing but an audit line. An unauthorized principal
 never reaches the allocator or the control-plane DB at all, so a denial cannot
 be used to enumerate the queue.
 ## Initiation
 `POST /api/v1/requests/apply` refuses in this order, and every refusal returns
 before any assignment is attempted:
 | Condition | Outcome | Status |
 |-----------|---------|--------|
 | Unparseable request | `invalid_request` | 400 |
 | Not authorized, or execution not wired | `denied` | 403 |
 | `confirm` not set | `denied` / `confirmation_required` | 409 |
 | Work unit already claimed | `blocked` / `duplicate_assignment` | 409 |
 | Allocator would select other work | `wait` / `not_next_safe_work` | 409 |
 | Allocator declines on apply | `blocked` or `wait` | 409 |
 | Evidence unavailable | `wait` / `evidence_unavailable` | 503 |
 | Assigned | `assigned_work` | 201 |
 A success returns the assignment plus a `handoff` block naming the profile, the
 namespace, and the actions that stay forbidden — enough for the operator to
 continue in the right MCP namespace without guessing.
 ### Why apply runs the allocator twice
 The allocator is the only source of exclusive ownership (#600 / #613), and it
 selects work; it does not take orders. So `apply` runs a dry-run first and
 proceeds only when the allocator would independently pick the requested work
 unit. If it would not, the request reports `wait` and mutates nothing.
 A request is therefore a *confirmation* of the allocator's decision, never an
 override of it. The apply call carries the dry-run's
 `candidate_set_fingerprint` as a CAS pin (#776), so a queue that changed
 between the two calls fails closed rather than assigning against a stale view.
 The result is checked again on the way out: an assignment naming a different
 work unit is not read as success.
 ### Fail-closed defaults
 - An unreadable control-plane DB denies. It is never treated as "nothing holds
  this work unit".
 - An incomplete queue inventory denies (#758). Ranking a partial candidate set
  can select the wrong work.
 - An allocator that raises denies.
 - PR work with no resolvable head SHA denies; a supplied SHA that no longer
  matches denies with `head_moved`.
 ## Enabling initiation
 Execution is wired off. Set `WEBUI_REQUESTS_EXECUTION=1` to enable it for the
 `initiate_workflow` action only — see
 [`webui-authz-audit.md`](webui-authz-audit.md) for why this is an
 action-scoped flag rather than a phase bump. With the variable unset, `apply`
 returns `403` with `reason_code: unauthorized` no matter who asks.
 Enabling execution does **not** enable approvals or merges. Those are phase 3
 console actions and remain forbidden in every path here; the console reserves
 work and hands off, and the MCP role profile enforces what that role may then
 do.
 ## Audit
 Every preview and every apply emits a console audit record (schema in
 [`webui-authz-audit.md`](webui-authz-audit.md)):
 | Event | `result` |
 |-------|----------|
 | Preview | `previewed` |
 | Refusal at any stage | `denied` |
 | Assignment created | `succeeded` |
 `correlation.request_id` carries the request's `correlation_id`, and a
 successful record's `metadata` carries `assignment_id` and `lease_id`, so an
 assignment can be traced back to the intent that produced it. The operator's
 `intent_summary` travels in `metadata` and passes through the standard
 redaction pass before persistence like every other field.
 ## Non-goals
 - No browser-initiated approve or merge, in this phase or any other.
 - No bypass of allocator exclusive ownership; no self-selection of work.
 - No auto-start from raw monitoring incidents (#612 stays downstream).
@@ -0,0 +1,102 @@
 # Web Console: restart status, impact preview, and approval state (#667)
 Phase 1 of the console restart surface. It consumes the #655 coordinator
 substrate and displays it. It performs no restart, reload, drain, approval, or
 process action, and it registers no write endpoint.
 Issue #667's rollout is explicit — *status views first, write approval after the
 backend gates are green* — and this change delivers only the status half.
 ## Surfaces
 | Path | Method | Purpose |
 |------|--------|---------|
 | `/runtime/restart` | GET | Restart status page |
 | `/api/v1/system/restart/status` | GET | Same snapshot as JSON |
 Both accept an optional `restart_class` query parameter (default
 `full_mcp_restart`). An unrecognised class is not an error: the coordinator
 resolves it as unknown and fails closed, and the page shows the resulting deny.
 Neither path accepts `POST`; a write attempt returns `405`, and a test asserts
 it.
 ## What it shows
 * **Impact preview (#658)** — verdict, blast radius, affected sessions, leases,
  critical sections, mutations, and the counts behind them, evaluated
  `dry_run=True` against live control-plane state.
 * **Drain proof (#661)** — verification of a supplied proof: valid, clean,
  expired, tampered, and the reasons behind a refusal.
 * **Post-restart reconcile (#662)** — the most recent completion proof, its
  overall status, and which dimensions still require follow-up.
 * **Restart classes (#663)** — the least-privilege matrix, with *you may
  request* and *you may execute* computed for the viewing role rather than for a
  generic operator.
 * **Approval controls (#633)** — the authorization state of
  `system.restart_namespace` and `system.reload_namespace`.
 * **Break-glass (#664)** — declared and marked unavailable; see below.
 ## Three rules this surface holds itself to
 A status page that is wrong is worse than one that is missing, because an
 operator acts on it. Three properties are enforced by tests, and each was
 verified by reverting the guard and watching a test fail.
 ### An unreadable source reports unavailable, never green
 Every source carries its own `SourceStatus`. Nothing substitutes a default,
 placeholder, or self-comparison for a reading that failed. An unreadable
 control-plane database yields `inventory_complete: false`, which the coordinator
 itself turns into a fail-closed verdict, and the page says the blast radius is
 unknown rather than showing an empty affected-sessions table.
 An absent drain proof is reported as absent — not as a pass. The #661 gate
 authorizes a restart only against a valid, unexpired, clean proof, so no proof
 is precisely the state that gate denies on.
 ### Authorization is asked the way execution would ask it
 Every probe passes `for_execution=True`.
 Asked without it, an admin is `allowed` for `system.restart_namespace`. On a
 control surface that reads as a live button. Asked the way an execution attempt
 would ask, the same principal is refused `phase_not_active`, because the console
 is in Phase 1 and the action is Phase 2. This surface reports the second answer.
 `execution_enabled` is therefore `false` for every action and every role today,
 and a test asserts that across the whole role matrix.
 ### The control-plane database is opened read-only
 `ControlPlaneDB()` creates directories and runs migrations on construction — a
 write. This surface never constructs one. It opens the sqlite file with
 `mode=ro`, exactly as `webui/inventory.py` does, and treats a missing file as
 missing authority rather than as an empty inventory.
 The test that protects this points at a path inside a directory that already
 exists, so a read-write `connect` would really create the file. A nested
 missing-directory path would have passed for the wrong reason.
 ## Break-glass is declared, not offered
 The break-glass workflow (#664) is not available on this branch's base. The
 panel is rendered to operator-class roles as **unavailable**, naming the issue
 that tracks it. It is not silently omitted, because an operator who has been
 told a governance path exists needs to see that it is not wired here; and it is
 not rendered as a control, because there is nothing behind it.
 Unprivileged viewers see only a note that the surface is operator-class.
 ## Redaction and escaping
 Every interpolated value passes through `_esc` (`html.escape(..., quote=True)`).
 Free-form text and anything that can carry a filesystem path additionally passes
 through `webui.inventory.scrub_text`, which redacts credential-shaped tokens
 inside a string rather than only at its start. The impact payload is passed
 through `webui.inventory.scrub` before rendering.
 ## Linkage
 Parent #655 · extends #642 · consumes #658, #661, #662, #663 · RBAC #633 ·
 console #631 · vision #652 · roadmap #653 · break-glass #664.
@@ -7,7 +7,6 @@ import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
 from allocator_service import (
    OUTCOME_ASSIGNED,
@@ -16,7 +15,6 @@ from allocator_service import (
    OUTCOME_PREVIEW,
    OUTCOME_WAIT,
    WorkCandidate,
    _drop_expired_claims,
    allocate_next_work,
    candidate_from_dict,
    classify_skip,
@@ -364,161 +362,5 @@ 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()
@@ -0,0 +1,452 @@
 """Read-only restart console: views, gates, and honesty rules (#667).
 The console consumes the #655 substrate. These tests hold it to the three
 properties that make a status surface trustworthy:
 * an unreadable source is reported unavailable, never rendered as green;
 * authorization is probed the way execution would probe it, so an allow is
  never shown for something that could not run;
 * the surface performs no mutation, including no write to the control-plane DB.
 """
 from __future__ import annotations
 import os
 import sqlite3
 import sys
 import tempfile
 import unittest
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from starlette.testclient import TestClient  # noqa: E402
 import restart_coordinator  # noqa: E402
 from webui import console_authz, restart_console, restart_views  # noqa: E402
 from webui.app import create_app  # noqa: E402
 NOW = datetime(2026, 7, 25, 21, 0, 0, tzinfo=timezone.utc)
 def _principal(role: str) -> console_authz.Principal:
    return console_authz.Principal(
        subject="[email protected]",
        role=role,
        identity_source=console_authz.IDENTITY_LOCAL_DEV,
        authenticated=True,
    )
 def _inventory(*, complete: bool = True, sessions=(), leases=()):
    def _read(**_kwargs):
        return {
            "sessions": list(sessions),
            "leases": list(leases),
            "terminal_lock": None,
            "prior_recovery_attempts": [],
            "inventory_complete": complete,
            "incomplete_reasons": (
                [] if complete else ["fixture: inventory withheld"]
            ),
        }
    return _read
 def _live_session(session_id: str = "prgs-author-1234-abcd") -> dict:
    return {
        "session_id": session_id,
        "role": "author",
        "profile": "prgs-author",
        "pid": os.getpid(),
        "status": "active",
        "last_heartbeat_at": (NOW - timedelta(seconds=30)).isoformat(),
    }
 def drain_proof_fixture() -> dict:
    """A structurally complete but unsigned drain proof."""
    return {
        "version": "drain-proof/v1",
        "proof_id": "deadbeef" * 8,
        "clean": True,
        "issued_at": (NOW - timedelta(minutes=1)).isoformat(),
        "expires_at": (NOW + timedelta(minutes=5)).isoformat(),
        "requesting_session_id": "s-live",
        "impact_fingerprint": "f" * 64,
        "checks": [],
        "failed_checks": [],
    }
 class RestartClassMatrixTest(unittest.TestCase):
    def test_every_policy_class_is_rendered(self) -> None:
        views = restart_console.build_restart_class_views("operator")
        self.assertEqual(len(views), len(restart_coordinator.RESTART_CLASS_POLICIES))
    def test_viewer_capability_is_role_scoped_not_generic(self) -> None:
        """A worker role must not be shown as able to request a full restart."""
        author = {
            v.restart_class: v
            for v in restart_console.build_restart_class_views("author")
        }
        operator = {
            v.restart_class: v
            for v in restart_console.build_restart_class_views("operator")
        }
        full = restart_coordinator.RestartClass.FULL_MCP_RESTART.value
        self.assertFalse(author[full].viewer_may_request)
        self.assertFalse(author[full].viewer_may_execute)
        self.assertTrue(operator[full].viewer_may_request)
        self.assertTrue(operator[full].viewer_may_execute)
    def test_unknown_role_may_do_nothing(self) -> None:
        views = restart_console.build_restart_class_views("not-a-role")
        self.assertTrue(all(not v.viewer_may_request for v in views))
        self.assertTrue(all(not v.viewer_may_execute for v in views))
 class AuthorizationProbeTest(unittest.TestCase):
    def test_probe_asks_for_execution_so_phase_gate_is_reported(self) -> None:
        """An admin clears the role bar and still cannot execute in Phase 1.
        This is the case that distinguishes the two probes. Asked without
        ``for_execution`` an admin is *allowed* for ``system.restart_namespace``,
        which on a control surface reads as a live button. Asked the way
        execution asks, the same principal is refused ``phase_not_active``. The
        console must report the second answer.
        """
        by_id = {
            a.action_id: a
            for a in restart_console.build_action_authorizations(
                _principal(console_authz.ADMIN)
            )
        }
        restart = by_id["system.restart_namespace"]
        self.assertFalse(restart.execution_enabled)
        self.assertEqual(restart.reason_code, console_authz.DENY_PHASE_NOT_ACTIVE)
        permissive = console_authz.authorize(
            "system.restart_namespace", _principal(console_authz.ADMIN)
        )
        self.assertTrue(
            permissive.allowed,
            "guard precondition: without for_execution an admin is allowed, "
            "which is exactly why the console must not probe that way",
        )
    def test_operator_is_refused_the_admin_only_restart_action(self) -> None:
        """Role refusal precedes the phase gate and is reported as such."""
        by_id = {
            a.action_id: a
            for a in restart_console.build_action_authorizations(
                _principal(console_authz.OPERATOR)
            )
        }
        self.assertEqual(
            by_id["system.restart_namespace"].reason_code,
            console_authz.DENY_INSUFFICIENT_ROLE,
        )
    def test_anonymous_is_denied_unauthenticated(self) -> None:
        by_id = {
            a.action_id: a for a in restart_console.build_action_authorizations(None)
        }
        self.assertEqual(
            by_id["system.restart_namespace"].reason_code,
            console_authz.DENY_UNAUTHENTICATED,
        )
    def test_no_authorization_ever_reports_execution_enabled(self) -> None:
        for role in (
            console_authz.VIEWER,
            console_authz.OPERATOR,
            console_authz.CONTROLLER,
            console_authz.ADMIN,
        ):
            for auth in restart_console.build_action_authorizations(_principal(role)):
                self.assertFalse(
                    auth.execution_enabled,
                    f"{role} reported execution_enabled for {auth.action_id}",
                )
 class ImpactPreviewTest(unittest.TestCase):
    def test_impact_renders_from_coordinator_dto(self) -> None:
        impact, source = restart_console.load_impact_report(
            principal=_principal(console_authz.OPERATOR),
            read_inventory=_inventory(sessions=[_live_session()]),
            now=NOW,
        )
        self.assertTrue(source.available)
        self.assertIsNotNone(impact)
        self.assertEqual(
            impact["restart_class"],
            restart_coordinator.RestartClass.FULL_MCP_RESTART.value,
        )
        self.assertIn("verdict", impact)
        self.assertFalse(impact["restart_performed"])
        self.assertTrue(impact["dry_run"])
    def test_incomplete_inventory_is_surfaced_and_denies(self) -> None:
        impact, source = restart_console.load_impact_report(
            principal=_principal(console_authz.OPERATOR),
            read_inventory=_inventory(complete=False),
            now=NOW,
        )
        self.assertFalse(impact["inventory_complete"])
        self.assertFalse(impact["allow_restart"])
        self.assertTrue(source.detail, "incomplete inventory must explain itself")
    def test_inventory_reader_failure_is_unavailable_not_empty(self) -> None:
        """A reader that raises must not be rendered as 'no sessions affected'."""
        def _boom(**_kwargs):
            raise RuntimeError("control-plane unreachable")
        impact, source = restart_console.load_impact_report(
            principal=_principal(console_authz.OPERATOR),
            read_inventory=_boom,
            now=NOW,
        )
        self.assertIsNone(impact)
        self.assertFalse(source.available)
        self.assertIn("control-plane unreachable", source.detail)
 class ControlPlaneReadTest(unittest.TestCase):
    def test_missing_database_is_incomplete_not_empty(self) -> None:
        inventory = restart_console.read_control_plane_inventory(
            db_path="/nonexistent/control-plane.sqlite3"
        )
        self.assertFalse(inventory["inventory_complete"])
        self.assertEqual(inventory["sessions"], [])
        self.assertTrue(inventory["incomplete_reasons"])
    def test_reader_never_creates_the_database(self) -> None:
        """Reading status must not bring a control-plane DB into existence.
        The path deliberately sits in a directory that already exists: a
        read-write ``sqlite3.connect`` would happily create the file there, so
        this fails if the reader ever stops opening the database ``mode=ro``.
        A nested-missing-directory path would pass for the wrong reason,
        because sqlite cannot create the parent directory either way.
        """
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "control_plane.sqlite3")
            self.assertTrue(os.path.isdir(os.path.dirname(path)))
            inventory = restart_console.read_control_plane_inventory(db_path=path)
            self.assertFalse(
                os.path.exists(path),
                "reading restart status created a control-plane database",
            )
            self.assertFalse(inventory["inventory_complete"])
    def test_reads_active_sessions_from_a_real_database(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "cp.sqlite3")
            conn = sqlite3.connect(path)
            conn.execute(
                "CREATE TABLE sessions (session_id TEXT, role TEXT, profile TEXT,"
                " pid INTEGER, status TEXT, last_heartbeat_at TEXT)"
            )
            conn.execute(
                "CREATE TABLE work_items (work_item_id INTEGER, kind TEXT,"
                " number INTEGER)"
            )
            conn.execute(
                "CREATE TABLE leases (lease_id TEXT, session_id TEXT, role TEXT,"
                " phase TEXT, status TEXT, worktree_path TEXT,"
                " work_item_id INTEGER, expires_at TEXT)"
            )
            conn.execute(
                "INSERT INTO sessions VALUES (?,?,?,?,?,?)",
                ("s-live", "author", "prgs-author", 4242, "active", NOW.isoformat()),
            )
            conn.execute(
                "INSERT INTO sessions VALUES (?,?,?,?,?,?)",
                ("s-done", "author", "prgs-author", 11, "closed", NOW.isoformat()),
            )
            conn.execute("INSERT INTO work_items VALUES (1, 'issue', 667)")
            conn.execute(
                "INSERT INTO leases VALUES (?,?,?,?,?,?,?,?)",
                (
                    "l-1",
                    "s-live",
                    "author",
                    "allocated",
                    "active",
                    None,
                    1,
                    NOW.isoformat(),
                ),
            )
            conn.commit()
            conn.close()
            inventory = restart_console.read_control_plane_inventory(db_path=path)
        self.assertTrue(inventory["inventory_complete"])
        self.assertEqual([s["session_id"] for s in inventory["sessions"]], ["s-live"])
        self.assertEqual(inventory["leases"][0]["work_number"], 667)
 class DrainAndReconcileTest(unittest.TestCase):
    def test_absent_drain_proof_is_not_a_pass(self) -> None:
        drain, source = restart_console.load_drain_status(proof=None, now=NOW)
        self.assertIsNone(drain)
        self.assertFalse(source.available)
        self.assertIn("denies", source.detail)
    def test_tampered_drain_proof_is_reported_invalid(self) -> None:
        proof = drain_proof_fixture()
        proof["clean"] = True
        proof["proof_id"] = "0" * 64
        drain, source = restart_console.load_drain_status(proof=proof, now=NOW)
        self.assertTrue(source.available)
        self.assertFalse(drain["valid"])
    def test_absent_reconcile_proof_is_unavailable(self) -> None:
        reconcile, source = restart_console.load_reconcile_status(load_proof=None)
        self.assertIsNone(reconcile)
        self.assertFalse(source.available)
    def test_reconcile_proof_is_rendered_when_supplied(self) -> None:
        payload = {
            "overall_status": "degraded",
            "mode": "log_only",
            "resolved_count": 3,
            "unresolved_count": 2,
            "items": [
                {
                    "dimension": "leases",
                    "status": "unresolved",
                    "summary": "2 orphaned leases",
                    "follow_up_required": True,
                }
            ],
        }
        reconcile, source = restart_console.load_reconcile_status(
            load_proof=lambda: payload
        )
        self.assertTrue(source.available)
        self.assertEqual(reconcile["unresolved_count"], 2)
 class RenderingTest(unittest.TestCase):
    def _snapshot(self, **kwargs):
        params = {
            "principal": _principal(console_authz.OPERATOR),
            "read_inventory": _inventory(sessions=[_live_session()]),
            "now": NOW,
        }
        params.update(kwargs)
        return restart_console.load_restart_console_snapshot(**params)
    def test_page_renders_every_section(self) -> None:
        html = restart_views.render_restart_console_page(self._snapshot())
        for heading in (
            "Impact preview",
            "Drain proof",
            "Post-restart reconcile",
            "Restart classes",
            "Approval controls",
            "Break-glass",
        ):
            self.assertIn(heading, html)
    def test_hostile_session_id_is_escaped(self) -> None:
        hostile = "<script>alert('x')</script>"
        html = restart_views.render_restart_console_page(
            self._snapshot(read_inventory=_inventory(sessions=[_live_session(hostile)]))
        )
        self.assertNotIn("<script>alert", html)
        self.assertIn("&lt;script&gt;", html)
    def test_unavailable_impact_says_unsafe_rather_than_clean(self) -> None:
        def _boom(**_kwargs):
            raise RuntimeError("nope")
        snapshot = self._snapshot(read_inventory=_boom)
        html = restart_views.render_restart_console_page(snapshot)
        self.assertIn("blast radius of a restart is unknown", html)
        self.assertIn("unavailable", html)
    def test_break_glass_is_hidden_from_unprivileged_viewers(self) -> None:
        viewer_html = restart_views.render_restart_console_page(
            self._snapshot(principal=_principal(console_authz.VIEWER))
        )
        self.assertIn("visible to operator-class", viewer_html)
        self.assertNotIn(
            f"#{restart_console.BREAK_GLASS_ISSUE}", viewer_html
        )
    def test_break_glass_shown_to_operator_is_marked_unavailable(self) -> None:
        html = restart_views.render_restart_console_page(self._snapshot())
        self.assertIn("unavailable", html)
        self.assertIn(f"#{restart_console.BREAK_GLASS_ISSUE}", html)
    def test_snapshot_always_declares_itself_read_only(self) -> None:
        self.assertTrue(self._snapshot().read_only)
 class RestartConsoleRouteTest(unittest.TestCase):
    def setUp(self) -> None:
        self.client = TestClient(create_app())
    def test_page_route_renders(self) -> None:
        res = self.client.get("/runtime/restart")
        self.assertEqual(res.status_code, 200)
        self.assertIn("Restart status and impact", res.text)
    def test_api_route_exports_snapshot(self) -> None:
        res = self.client.get("/api/v1/system/restart/status")
        self.assertEqual(res.status_code, 200)
        payload = res.json()
        self.assertTrue(payload["read_only"])
        self.assertEqual(payload["links"]["issue"], 667)
        self.assertEqual(
            len(payload["restart_classes"]),
            len(restart_coordinator.RESTART_CLASS_POLICIES),
        )
    def test_restart_class_is_selectable(self) -> None:
        res = self.client.get(
            "/api/v1/system/restart/status?restart_class=client_reconnect"
        )
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res.json()["impact"]["restart_class"], "client_reconnect")
    def test_unknown_restart_class_fails_closed(self) -> None:
        res = self.client.get(
            "/api/v1/system/restart/status?restart_class=obliterate-everything"
        )
        self.assertEqual(res.status_code, 200)
        impact = res.json()["impact"]
        self.assertFalse(impact["allow_restart"])
    def test_anonymous_api_reader_gets_no_execution_grant(self) -> None:
        payload = self.client.get("/api/v1/system/restart/status").json()
        self.assertFalse(payload["break_glass"]["available"])
        for auth in payload["authorizations"]:
            self.assertFalse(auth["execution_enabled"])
    def test_route_is_registered_in_nav(self) -> None:
        from webui.nav import nav_hrefs
        self.assertIn("/runtime/restart", nav_hrefs())
    def test_no_write_method_is_exposed(self) -> None:
        """The surface is read-only: nothing accepts a POST."""
        for path in ("/runtime/restart", "/api/v1/system/restart/status"):
            self.assertEqual(self.client.post(path).status_code, 405, path)
 if __name__ == "__main__":
    unittest.main()
@@ -47,6 +47,9 @@ from webui.traffic_views import render_traffic_page
 from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as worktree_snapshot_to_dict
 from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 import restart_coordinator
 from webui.restart_console import load_restart_console_snapshot
 from webui.restart_views import render_restart_console_page
 from webui.runtime_views import render_runtime_page
 from webui.session_loader import (
    load_session_view_snapshot,
@@ -72,8 +75,6 @@ from webui.system_health import (
    snapshot_to_dict as system_health_to_dict,
 )
 from webui.system_health_views import render_system_health_page
 from webui import request_service
 from webui.request_views import render_requests_page
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -333,6 +334,33 @@ async def api_runtime(_request: Request) -> JSONResponse:
    return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
 def _restart_console_snapshot(request: Request):
    """Build the read-only restart snapshot for the requesting principal (#667)."""
    principal = resolve_principal(request.headers)
    restart_class = (
        request.query_params.get("restart_class")
        or restart_coordinator.RestartClass.FULL_MCP_RESTART.value
    )
    return load_restart_console_snapshot(
        principal=principal, restart_class=restart_class
    )
 async def restart_console_page(request: Request) -> HTMLResponse:
    """Restart status, impact preview, and approval state (#667). Read-only."""
    snapshot = _restart_console_snapshot(request)
    return HTMLResponse(
        render_page(
            title="Restart", body_html=render_restart_console_page(snapshot)
        )
    )
 async def api_restart_status(request: Request) -> JSONResponse:
    """JSON export of the read-only restart console snapshot (#667)."""
    return JSONResponse(_restart_console_snapshot(request).as_dict())
 async def sessions(_request: Request) -> HTMLResponse:
    """Runtime and session view (#641) — read-only composition of health + inventory."""
    snapshot = load_session_view_snapshot()
@@ -741,109 +769,6 @@ async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
    )
 def _default_request_scope() -> dict[str, str]:
    """Resolve remote/org/repo from the project registry for request forms.
    Returns an empty mapping when the registry cannot be read, which makes
    ``parse_request`` reject a request that did not name its own scope rather
    than letting it default to some other repository.
    """
    from webui.queue_loader import _host_from_url  # host normalisation helper
    registry, error = _load_project_registry()
    if error is not None or not registry.projects:
        return {}
    project = registry.projects[0]
    host = _host_from_url(project.remote_host)
    return {
        "remote": _derive_remote(host),
        "org": project.gitea_owner or "",
        "repo": project.repo_name or "",
    }
 async def _request_payload(request: Request) -> dict[str, object]:
    """Read a request body as JSON or form-encoded. Never raises."""
    content_type = (request.headers.get("content-type") or "").lower()
    if "application/json" in content_type:
        try:
            body = await request.json()
        except Exception:
            return {}
        return dict(body) if isinstance(body, dict) else {}
    try:
        form = await request.form()
    except Exception:
        return {}
    return {key: form[key] for key in form}
 async def requests_page(request: Request) -> HTMLResponse:
    """Operator request form and intent preview (#643).
    POST here only ever *previews*. Initiation is a separate confirmed call to
    ``/api/v1/requests/apply`` so that submitting this form cannot reserve
    work as a side effect.
    """
    submitted: dict[str, object] = {}
    preview = None
    error = None
    if request.method == "POST":
        submitted = await _request_payload(request)
        work_request, error = request_service.parse_request(
            submitted, default_scope=_default_request_scope()
        )
        if work_request is not None:
            preview = request_service.preview_request(
                work_request,
                principal=resolve_principal(headers=dict(request.headers)),
            )
    return HTMLResponse(
        render_requests_page(
            preview=preview, error=error, submitted=submitted
        )
    )
 async def api_v1_request_preview(request: Request) -> JSONResponse:
    """Dry-run authorization and intent preview for a work request (#643)."""
    payload = await _request_payload(request)
    work_request, error = request_service.parse_request(
        payload, default_scope=_default_request_scope()
    )
    if work_request is None:
        return JSONResponse(error.to_dict(), status_code=400)
    preview = request_service.preview_request(
        work_request,
        principal=resolve_principal(headers=dict(request.headers)),
    )
    return JSONResponse(
        preview.to_dict(), status_code=200 if preview.authorized else 403
    )
 async def api_v1_request_apply(request: Request) -> JSONResponse:
    """Initiate a previewed work request through the allocator (#643).
    Fail-closed at every step: unauthorized, unconfirmed, not-next-safe, and
    already-claimed all return without attempting an assignment.
    """
    payload = await _request_payload(request)
    work_request, error = request_service.parse_request(
        payload, default_scope=_default_request_scope()
    )
    if work_request is None:
        return JSONResponse(error.to_dict(), status_code=400)
    confirm = _truthy_flag(str(payload.get("confirm") or ""))
    result = request_service.apply_request(
        work_request,
        principal=resolve_principal(headers=dict(request.headers)),
        confirm=confirm,
    )
    status = int(result.pop("status_code", 403))
    return JSONResponse(result, status_code=status)
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
    path = request.url.path
    if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -886,6 +811,13 @@ 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"]),
            # #667 read-only restart status / impact preview / approval state.
            Route("/runtime/restart", restart_console_page, methods=["GET"]),
            Route(
                "/api/v1/system/restart/status",
                api_restart_status,
                methods=["GET"],
            ),
            Route("/sessions", sessions, methods=["GET"]),
            Route("/api/sessions", api_sessions, methods=["GET"]),
            Route("/api/v1/sessions", api_sessions, methods=["GET"]),
@@ -911,17 +843,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                api_action_attempt,
                methods=["POST"],
            ),
            Route("/requests", requests_page, methods=["GET", "POST"]),
            Route(
                "/api/v1/requests/preview",
                api_v1_request_preview,
                methods=["POST"],
            ),
            Route(
                "/api/v1/requests/apply",
                api_v1_request_apply,
                methods=["POST"],
            ),
            Route("/api/leases", api_leases, methods=["GET"]),
            Route("/api/v1/inventory", api_inventory, methods=["GET"]),
            Route(
@@ -115,12 +115,6 @@ class ConsoleAction:
    break_glass: bool
    phase: int
    summary: str
    # Opt-in switch for an action whose execution path is genuinely wired
    # ahead of its phase becoming globally active (#643). Naming a variable
    # here enables nothing on its own: the variable must also be set in the
    # environment. An action that leaves this ``None`` can only execute once
    # ACTIVE_PHASE reaches its phase, exactly as before.
    execution_env_flag: str | None = None
    @property
    def mcp_permission(self) -> str:
@@ -283,27 +277,6 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
        phase=2,
        summary="Restart one MCP namespace via the host supervisor.",
    ),
    # #643: submit a work request — desired role, issue/PR, intent — and let
    # the allocator reserve it. This is the one Phase 2 action whose execution
    # path is actually implemented (``webui.request_service``), so it carries
    # the opt-in flag; it stays denied until an operator sets that variable.
    # Authority is operator-class because the outcome is a claim, not a Gitea
    # verdict: initiating reviewer or merger *work* does not grant the right
    # to approve or merge, which stays with the MCP role profile.
    ConsoleAction(
        action_id="initiate_workflow",
        task_key="allocate_next_work",
        action_class=CLASS_WRITE,
        minimum_role=OPERATOR,
        requires_confirmation=True,
        dual_control=False,
        break_glass=False,
        phase=2,
        summary=(
            "Preview and initiate allocator-owned workflow work for a role."
        ),
        execution_env_flag="WEBUI_REQUESTS_EXECUTION",
    ),
 )
 ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
@@ -457,33 +430,6 @@ ALLOW_PREVIEW = "allowed_preview_only"
 # gated on this model landing; nothing here enables it.
 ACTIVE_PHASE = 1
 _TRUTHY = frozenset({"1", "true", "yes", "on"})
 def execution_wired(
    action: ConsoleAction | None, env: dict[str, str] | None = None
 ) -> bool:
    """Whether *action* has a live execution path right now.
    Two ways to be wired, and only two. The action's phase is active, or the
    action declares an opt-in environment variable *and* that variable is set.
    Everything else — including every action that never declares a flag — is
    unwired, so the default across the registry stays deny.
    Bumping ``ACTIVE_PHASE`` would enable execution for every action of that
    phase at once. The per-action flag exists so a single implemented action
    can go live without dragging its unimplemented phase-mates with it.
    """
    if action is None:
        return False
    if action.phase <= ACTIVE_PHASE:
        return True
    flag = (action.execution_env_flag or "").strip()
    if not flag:
        return False
    source = env if env is not None else os.environ
    return (source.get(flag) or "").strip().lower() in _TRUTHY
@dataclass(frozen=True)
 class AuthorizationDecision:
@@ -523,19 +469,16 @@ def authorize(
    principal: Principal | None = None,
    *,
    for_execution: bool = False,
    env: dict[str, str] | None = None,
 ) -> AuthorizationDecision:
    """Decide whether *principal* may invoke *action_id*. Deny by default.
    ``for_execution`` distinguishes a read-only preview from a real invocation.
-    ``execution_enabled`` reports whether the action has a live execution path
+    Even an allowed decision reports ``execution_enabled=False`` while the
-    at all (:func:`execution_wired`) — for every action without an explicit
+    console is in Phase 1, so no caller can read an allow as permission to
-    opt-in flag that stays ``False`` while the console is in Phase 1, so no
+    mutate.
    caller can read an allow as permission to mutate.
    """
    who = principal if principal is not None else ANONYMOUS
    action = get_action(action_id)
    wired = execution_wired(action, env)
    if action is None:
        return AuthorizationDecision(
@@ -554,7 +497,7 @@ def authorize(
        "requires_confirmation": action.requires_confirmation,
        "dual_control": action.dual_control,
        "break_glass": action.break_glass,
-        "execution_enabled": wired,
+        "execution_enabled": False,
    }
    if not who.authenticated:
@@ -587,19 +530,13 @@ def authorize(
            **base,
        )
-    if for_execution and not wired:
+    if for_execution and action.phase > ACTIVE_PHASE:
        return AuthorizationDecision(
            allowed=False,
            reason_code=DENY_PHASE_NOT_ACTIVE,
            detail=(
                f"Action {action_id!r} belongs to phase {action.phase}; the "
-                f"console is in phase {ACTIVE_PHASE}"
+                f"console is in phase {ACTIVE_PHASE}. Execution is not wired."
                + (
                    f" and {action.execution_env_flag} is not set"
                    if action.execution_env_flag
                    else ""
                )
                + ". Execution is not wired."
            ),
            **base,
        )
@@ -608,8 +545,8 @@ def authorize(
        allowed=True,
        reason_code=ALLOW_PREVIEW,
        detail=(
-            "Principal holds the required role. Execution proceeds only for an "
+            "Principal holds the required role. Preview only — execution "
-            "action with a wired execution path; everything else is preview."
+            "remains disabled until the Phase 2 action framework ships."
        ),
        **base,
    )
@@ -45,10 +45,10 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
        NavItem("/queue", "Queue"),
        NavItem("/leases", "Leases"),
        NavItem("/actions", "Actions"),
        NavItem("/requests", "Requests"),
    )),
    NavGroup("Runtime/Sessions", (
        NavItem("/runtime", "Runtime health"),
        NavItem("/runtime/restart", "Restart status"),
        NavItem("/sessions", "Sessions"),
    )),
    NavGroup("Projects", (
@@ -1,164 +0,0 @@
 """HTML views for the operator request surface (#643).
 The form is deliberately a *preview* form. It has no initiate button, because
 initiating requires a confirmed POST to ``/api/v1/requests/apply`` and a stray
 form submission must not be able to produce one by accident.
 Nothing rendered here is trusted input: every interpolated value is escaped,
 and the page renders only values the service already produced rather than
 echoing a raw request body back.
 """
 from __future__ import annotations
 import html
 import json
 from typing import Any
 from webui.layout import render_page
 from webui.request_service import (
    REQUESTABLE_ROLES,
    WORK_KINDS,
    RequestError,
    RequestPreview,
 )
 REQUESTS_PATH = "/requests"
 PREVIEW_API_PATH = "/api/v1/requests/preview"
 APPLY_API_PATH = "/api/v1/requests/apply"
 def _escape(text: Any) -> str:
    return html.escape(str(text if text is not None else ""), quote=True)
 REQUEST_PAGE_STYLES = """
 <style>
  .request-form { display: grid; gap: 0.75rem; max-width: 44rem; }
  .request-form label { display: grid; gap: 0.25rem; font-size: 0.9rem; }
  .request-check { margin: 0.35rem 0; }
  .request-check .verdict-ok { color: var(--accent); }
  .request-check .verdict-fail { color: #d14; }
  .request-prohibited code { margin-right: 0.4rem; }
 </style>
 """
 def _options(values: tuple[str, ...], selected: Any) -> str:
    return "".join(
        f"<option value='{_escape(value)}'"
        + (" selected" if selected == value else "")
        + f">{_escape(value)}</option>"
        for value in values
    )
 def _form(values: dict[str, Any] | None = None) -> str:
    current = dict(values or {})
    number = current.get("work_number")
    return (
        f"<form class='request-form' method='post' action='{REQUESTS_PATH}'>"
        "<label>Desired role<select name='desired_role'>"
        f"{_options(REQUESTABLE_ROLES, current.get('desired_role'))}"
        "</select></label>"
        "<label>Work kind<select name='work_kind'>"
        f"{_options(WORK_KINDS, current.get('work_kind'))}"
        "</select></label>"
        "<label>Issue or PR number"
        "<input type='number' name='work_number' min='1' required "
        f"value='{_escape(number) if number else ''}'></label>"
        "<label>Intent summary"
        "<input type='text' name='intent_summary' maxlength='500' required "
        f"value='{_escape(current.get('intent_summary'))}'></label>"
        "<label>Expected head SHA <span class='muted'>(PR work only)</span>"
        "<input type='text' name='expected_head_sha' "
        f"value='{_escape(current.get('expected_head_sha'))}'></label>"
        "<button type='submit' class='copy-btn'>Preview request</button>"
        "<p class='muted meta'>Preview is read-only and creates no assignment. "
        f"Initiating requires a confirmed POST to <code>{APPLY_API_PATH}</code>."
        "</p>"
        "</form>"
    )
 def _checks_block(preview: RequestPreview) -> str:
    rows = []
    for check in preview.checks:
        verdict = "PASS" if check.ok else "FAIL"
        css = "verdict-ok" if check.ok else "verdict-fail"
        rows.append(
            "<li class='request-check'>"
            f"<span class='{css}'><strong>{verdict}</strong></span> "
            f"<code>{_escape(check.name)}</code> — {_escape(check.detail)} "
            f"<span class='muted meta'>({_escape(check.reason_code)})</span>"
            "</li>"
        )
    return "<ul>" + "".join(rows) + "</ul>"
 def _preview_block(preview: RequestPreview) -> str:
    verdict = "AUTHORIZED" if preview.authorized else "DENIED"
    prohibited = "".join(
        f"<code>{_escape(action)}</code>" for action in preview.prohibited_actions
    )
    request = preview.request
    evidence = json.dumps(preview.allocator_evidence, indent=2, default=str)
    return (
        "<h3>Intent preview</h3>"
        f"<p><strong>{verdict}</strong> — {_escape(preview.detail)}</p>"
        "<p class='meta'>"
        f"Role <code>{_escape(request.desired_role)}</code> · "
        f"{_escape(request.work_kind)} <code>{_escape(request.display_ref)}</code>"
        f" · profile <code>{_escape(preview.required_profile)}</code> · "
        f"namespace <code>{_escape(preview.required_namespace)}</code> · "
        f"permission <code>{_escape(preview.required_permission)}</code>"
        "</p>"
        f"<p>Intent: {_escape(request.intent_summary)}</p>"
        f"{_checks_block(preview)}"
        f"<p><strong>Next safe action:</strong> "
        f"{_escape(preview.next_safe_action)}</p>"
        "<p class='request-prohibited'><strong>Prohibited for this role:</strong> "
        + (prohibited or "<span class='muted'>none declared</span>")
        + "</p>"
        "<p class='muted meta'>Correlation id "
        f"<code>{_escape(preview.correlation_id)}</code></p>"
        "<details><summary>Allocator evidence</summary>"
        f"<pre class='prompt-text'>{_escape(evidence)}</pre>"
        "</details>"
    )
 def _error_block(error: RequestError) -> str:
    field = (
        f"<p class='meta'>Field: <code>{_escape(error.field_name)}</code></p>"
        if error.field_name
        else ""
    )
    return (
        "<h3>Request rejected</h3>"
        f"<p><strong>{_escape(error.reason_code)}</strong> — "
        f"{_escape(error.detail)}</p>{field}"
    )
 def render_requests_page(
    *,
    preview: RequestPreview | None = None,
    error: RequestError | None = None,
    submitted: dict[str, Any] | None = None,
 ) -> str:
    """Render the request form, plus a preview or rejection when one exists."""
    body = (
        "<h2>Requests</h2>"
        "<p>Submit a work request — desired role, issue or PR, and intent — "
        "and see whether it would be authorized before anything is reserved. "
        "Initiation goes through the allocator (#600/#613); this console never "
        "self-selects work, never approves, and never merges.</p>"
        + _form(submitted)
        + (_error_block(error) if error is not None else "")
        + (_preview_block(preview) if preview is not None else "")
        + f"<p class='meta'><a href='{PREVIEW_API_PATH}'>Preview API</a> · "
        "<a href='/api/console/security-model'>RBAC model</a></p>"
        + REQUEST_PAGE_STYLES
    )
    return render_page(title="Requests", body_html=body)
@@ -0,0 +1,579 @@
 """Read-only restart status, impact preview, and approval state (#667).
 Phase 1 of the console restart surface. It *consumes* the #655 coordinator
 substrate and renders it; it never restarts, reloads, drains, approves, or kills
 anything. There is no apply path in this module, so there is no execution gate
 here to arm incorrectly — the only writes the console could perform are the ones
 it does not implement.
 Sources, each independently fail-soft and each reported with its own
 :class:`SourceStatus`:
 * :mod:`restart_coordinator` — restart-class policy matrix (#663) and the
  blast-radius impact report (#658).
 * :mod:`drain_proof` — drain checklist and gate verdict (#661), verified
  read-only against a caller-supplied proof.
 * :mod:`post_restart_reconcile` — post-restart completion proof (#662).
 * :mod:`webui.console_authz` — role authorization for the approval controls
  (#633).
 Three rules this module holds itself to, because a status surface that lies is
 worse than one that is absent:
 **A source that could not be read is reported unavailable, never green.** No
 default, placeholder, or self-comparison is substituted for a reading that
 failed. An unreadable control-plane DB yields ``inventory_complete=False``,
 which the coordinator itself turns into a fail-closed verdict.
 **Authorization is asked the way execution would ask it.** Every authorization
 probe passes ``for_execution=True``, so the console reports whether the action
 could actually run rather than the weaker "this principal is the right role".
 While the console is in Phase 1 that answer is ``phase_not_active`` for every
 phase-2 action, and the surface says so plainly instead of showing an allow.
 **The database is opened read-only.** ``ControlPlaneDB()`` creates directories
 and runs migrations on construction, which is a write; this module opens the
 sqlite file with ``mode=ro`` exactly as :mod:`webui.inventory` does, and treats
 a missing file as missing authority rather than an empty inventory.
 """
 from __future__ import annotations
 import os
 import sqlite3
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any, Callable, Mapping
 import control_plane_db
 import drain_proof
 import restart_coordinator
 from webui import console_authz
 from webui.inventory import redact_path, scrub
 # --- Source status ----------------------------------------------------------
 STATUS_OK = "ok"
 STATUS_UNAVAILABLE = "unavailable"
 #: Console actions whose authorization state this surface reports. Both are
 #: pre-existing #642 actions; this module adds no new console action because it
 #: performs no console action.
 REPORTED_ACTIONS: tuple[str, ...] = (
    "system.restart_namespace",
    "system.reload_namespace",
 )
 #: The break-glass workflow (#664) is not consumed here. It is declared so the
 #: surface is honest about the gap rather than silently omitting a governance
 #: path the operator has been told exists.
 BREAK_GLASS_ISSUE = 664
 BREAK_GLASS_PENDING_REASON = (
    "The break-glass workflow (#664) is not yet available on this branch's "
    "base; no break-glass control is offered and none is implied."
 )
@dataclass(frozen=True)
 class SourceStatus:
    """Whether one backing source could be read, and why not when it could not."""
    name: str
    status: str
    detail: str = ""
    @property
    def available(self) -> bool:
        return self.status == STATUS_OK
    def as_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "status": self.status,
            "available": self.available,
            "detail": self.detail,
        }
@dataclass(frozen=True)
 class RestartClassView:
    """One row of the #663 restart-class matrix, scoped to the viewer's role."""
    restart_class: str
    required_permission: str
    expected_blast_radius: str
    drain_requirement: str
    full_drain_required: bool
    approval_requirement: str
    request_roles: tuple[str, ...]
    execution_roles: tuple[str, ...]
    viewer_may_request: bool
    viewer_may_execute: bool
    def as_dict(self) -> dict[str, Any]:
        return {
            "restart_class": self.restart_class,
            "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,
            "request_roles": list(self.request_roles),
            "execution_roles": list(self.execution_roles),
            "viewer_may_request": self.viewer_may_request,
            "viewer_may_execute": self.viewer_may_execute,
        }
@dataclass(frozen=True)
 class ActionAuthorization:
    """Authorization state for one console action, asked as execution would."""
    action_id: str
    summary: str
    required_role: str
    allowed: bool
    execution_enabled: bool
    reason_code: str
    detail: str
    def as_dict(self) -> dict[str, Any]:
        return {
            "action_id": self.action_id,
            "summary": self.summary,
            "required_role": self.required_role,
            "allowed": self.allowed,
            "execution_enabled": self.execution_enabled,
            "reason_code": self.reason_code,
            "detail": self.detail,
        }
@dataclass(frozen=True)
 class BreakGlassSurface:
    """Declared-but-unavailable break-glass panel (#664 is not on this base)."""
    available: bool
    issue: int
    reason: str
    viewer_is_privileged: bool
    def as_dict(self) -> dict[str, Any]:
        return {
            "available": self.available,
            "issue": self.issue,
            "reason": self.reason,
            "viewer_is_privileged": self.viewer_is_privileged,
        }
@dataclass(frozen=True)
 class RestartConsoleSnapshot:
    """Everything the read-only restart console renders."""
    generated_at: str
    viewer_role: str
    viewer_authenticated: bool
    read_only: bool
    impact: dict[str, Any] | None
    impact_source: SourceStatus
    drain: dict[str, Any] | None
    drain_source: SourceStatus
    reconcile: dict[str, Any] | None
    reconcile_source: SourceStatus
    restart_classes: tuple[RestartClassView, ...]
    authorizations: tuple[ActionAuthorization, ...]
    break_glass: BreakGlassSurface
    notes: tuple[str, ...] = field(default_factory=tuple)
    def as_dict(self) -> dict[str, Any]:
        return {
            "generated_at": self.generated_at,
            "viewer_role": self.viewer_role,
            "viewer_authenticated": self.viewer_authenticated,
            "read_only": self.read_only,
            "impact": self.impact,
            "impact_source": self.impact_source.as_dict(),
            "drain": self.drain,
            "drain_source": self.drain_source.as_dict(),
            "reconcile": self.reconcile,
            "reconcile_source": self.reconcile_source.as_dict(),
            "restart_classes": [c.as_dict() for c in self.restart_classes],
            "authorizations": [a.as_dict() for a in self.authorizations],
            "break_glass": self.break_glass.as_dict(),
            "notes": list(self.notes),
            "links": {
                "issue": 667,
                "extends": 642,
                "umbrella": 655,
                "coordinator": 658,
                "drain_proof": 661,
                "reconcile": 662,
                "restart_classes": 663,
                "break_glass": BREAK_GLASS_ISSUE,
                "vision": 652,
                "roadmap": 653,
            },
        }
 def _utc_now() -> datetime:
    return datetime.now(timezone.utc)
 # --- Control-plane inventory (read-only) ------------------------------------
 def read_control_plane_inventory(
    *,
    db_path: str | None = None,
    limit: int = 200,
 ) -> dict[str, Any]:
    """Read sessions and leases for an impact evaluation, read-only.
    Returns the inventory mapping
    :func:`restart_coordinator.evaluate_restart_impact` expects.
    ``inventory_complete`` is True only when every read succeeded, so a partial
    read denies rather than under-reporting the blast radius.
    The database is never created, migrated, or written: a missing file means
    the console has no session authority, which is not the same as there being
    no sessions.
    """
    path = (db_path or control_plane_db.default_db_path() or "").strip()
    incomplete: list[str] = []
    def _incomplete(reason: str) -> dict[str, Any]:
        return {
            "sessions": [],
            "leases": [],
            "terminal_lock": None,
            "prior_recovery_attempts": [],
            "inventory_complete": False,
            "incomplete_reasons": [reason],
        }
    if not path:
        return _incomplete("control-plane database path is not configured")
    if not os.path.exists(path):
        return _incomplete(
            f"control-plane database not present at {redact_path(path)}; "
            "no session or lease authority available"
        )
    try:
        conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5)
        conn.row_factory = sqlite3.Row
    except sqlite3.Error as exc:
        return _incomplete(f"control-plane database could not be opened: {exc}")
    sessions: list[dict[str, Any]] = []
    leases: list[dict[str, Any]] = []
    capped = max(1, int(limit))
    try:
        tables = {
            str(row[0])
            for row in conn.execute(
                "SELECT name FROM sqlite_master WHERE type = 'table'"
            ).fetchall()
        }
        if "sessions" not in tables:
            incomplete.append("control-plane database has no sessions table")
        else:
            sessions = [
                dict(row)
                for row in conn.execute(
                    "SELECT session_id, role, profile, pid, status,"
                    " last_heartbeat_at FROM sessions"
                    " WHERE status = 'active'"
                    " ORDER BY last_heartbeat_at DESC LIMIT ?",
                    (capped,),
                ).fetchall()
            ]
        if "leases" not in tables:
            incomplete.append("control-plane database has no leases table")
        elif "work_items" not in tables:
            incomplete.append(
                "control-plane database has no work_items table; lease work "
                "identity cannot be resolved"
            )
        else:
            leases = [
                dict(row)
                for row in conn.execute(
                    "SELECT l.lease_id, l.session_id, l.role, l.phase,"
                    " l.status AS freshness, l.worktree_path,"
                    " w.kind AS work_kind, w.number AS work_number"
                    " FROM leases l"
                    " JOIN work_items w ON w.work_item_id = l.work_item_id"
                    " WHERE l.status = 'active'"
                    " ORDER BY l.expires_at DESC LIMIT ?",
                    (capped,),
                ).fetchall()
            ]
    except sqlite3.Error as exc:
        return _incomplete(f"control-plane database read failed: {exc}")
    finally:
        conn.close()
    return {
        "sessions": sessions,
        "leases": leases,
        "terminal_lock": None,
        "prior_recovery_attempts": [],
        "inventory_complete": not incomplete,
        "incomplete_reasons": incomplete,
    }
 # --- Composition ------------------------------------------------------------
 def build_restart_class_views(viewer_role: str | None) -> tuple[RestartClassView, ...]:
    """Render the #663 class matrix, marking what this viewer may request."""
    normalized = str(viewer_role or "").strip().lower()
    views: list[RestartClassView] = []
    for policy in restart_coordinator.RESTART_CLASS_POLICIES.values():
        views.append(
            RestartClassView(
                restart_class=policy.restart_class.value,
                required_permission=policy.required_permission,
                expected_blast_radius=policy.expected_blast_radius,
                drain_requirement=policy.drain_requirement,
                full_drain_required=policy.full_drain_required,
                approval_requirement=policy.approval_requirement,
                request_roles=tuple(policy.request_roles),
                execution_roles=tuple(policy.execution_roles),
                viewer_may_request=normalized in policy.request_roles,
                viewer_may_execute=normalized in policy.execution_roles,
            )
        )
    return tuple(views)
 def build_action_authorizations(
    principal: console_authz.Principal | None,
 ) -> tuple[ActionAuthorization, ...]:
    """Authorization state for the approval controls, asked as execution.
    ``for_execution=True`` is deliberate. Asking without it answers "is this
    principal senior enough", which is not the question an operator looking at a
    control needs answered; asking with it answers "would this run", and while
    the console is in Phase 1 the honest answer is no.
    """
    results: list[ActionAuthorization] = []
    for action_id in REPORTED_ACTIONS:
        action = console_authz.get_action(action_id)
        decision = console_authz.authorize(action_id, principal, for_execution=True)
        results.append(
            ActionAuthorization(
                action_id=action_id,
                summary=action.summary if action else "",
                required_role=(
                    action.minimum_role if action else console_authz.OPERATOR
                ),
                allowed=bool(decision.allowed),
                execution_enabled=bool(decision.execution_enabled),
                reason_code=str(decision.reason_code or ""),
                detail=str(decision.detail or ""),
            )
        )
    return tuple(results)
 def viewer_is_privileged(principal: console_authz.Principal | None) -> bool:
    """True when the viewer holds at least the operator role."""
    who = principal if principal is not None else console_authz.ANONYMOUS
    if not who.authenticated:
        return False
    return who.rank >= console_authz.ROLE_ORDER.index(console_authz.OPERATOR)
 def load_impact_report(
    *,
    principal: console_authz.Principal | None = None,
    restart_class: str = restart_coordinator.RestartClass.FULL_MCP_RESTART.value,
    db_path: str | None = None,
    limit: int = 200,
    read_inventory: Callable[..., Mapping[str, Any]] | None = None,
    now: datetime | None = None,
 ) -> tuple[dict[str, Any] | None, SourceStatus]:
    """Evaluate the blast radius for *restart_class*, always dry-run."""
    reader = read_inventory or read_control_plane_inventory
    try:
        inventory = dict(reader(db_path=db_path, limit=limit))
    except Exception as exc:  # noqa: BLE001
        return None, SourceStatus(
            "impact",
            STATUS_UNAVAILABLE,
            f"control-plane inventory failed: {type(exc).__name__}: {exc}",
        )
    who = principal if principal is not None else console_authz.ANONYMOUS
    viewer_role = str(who.role or "").strip().lower()
    try:
        report = restart_coordinator.evaluate_restart_impact(
            inventory,
            now=now,
            dry_run=True,
            restart_class=restart_class,
            requester_role=viewer_role,
            requester_permissions=restart_coordinator.permissions_for_role(
                viewer_role
            ),
        )
    except Exception as exc:  # noqa: BLE001
        return None, SourceStatus(
            "impact",
            STATUS_UNAVAILABLE,
            f"impact evaluation failed: {type(exc).__name__}: {exc}",
        )
    payload = scrub(report.as_dict())
    detail = ""
    if not report.inventory_complete:
        detail = "; ".join(report.incomplete_reasons) or "inventory incomplete"
    return payload, SourceStatus("impact", STATUS_OK, detail)
 def load_drain_status(
    *,
    proof: Mapping[str, Any] | None = None,
    now: datetime | None = None,
    expected_impact_fingerprint: str | None = None,
 ) -> tuple[dict[str, Any] | None, SourceStatus]:
    """Verify a supplied drain proof read-only and report the verdict.
    No proof supplied is not a failure and not a pass: it is reported as the
    absence of a proof, which is exactly what the #661 gate would deny on.
    """
    if proof is None:
        return None, SourceStatus(
            "drain",
            STATUS_UNAVAILABLE,
            "no drain proof supplied; the #661 gate denies a restart without a "
            "valid unexpired clean proof",
        )
    try:
        verified = drain_proof.verify_drain_proof(
            proof,
            now=now,
            expected_impact_fingerprint=expected_impact_fingerprint,
        )
    except Exception as exc:  # noqa: BLE001
        return None, SourceStatus(
            "drain",
            STATUS_UNAVAILABLE,
            f"drain proof verification failed: {type(exc).__name__}: {exc}",
        )
    return scrub(verified.as_dict()), SourceStatus("drain", STATUS_OK)
 def load_reconcile_status(
    *,
    load_proof: Callable[[], Any] | None = None,
 ) -> tuple[dict[str, Any] | None, SourceStatus]:
    """Report the most recent post-restart completion proof (#662)."""
    if load_proof is None:
        return None, SourceStatus(
            "reconcile",
            STATUS_UNAVAILABLE,
            "no post-restart completion proof source is wired into this view",
        )
    try:
        proof = load_proof()
    except Exception as exc:  # noqa: BLE001
        return None, SourceStatus(
            "reconcile",
            STATUS_UNAVAILABLE,
            f"reconcile proof unavailable: {type(exc).__name__}: {exc}",
        )
    if proof is None:
        return None, SourceStatus(
            "reconcile",
            STATUS_UNAVAILABLE,
            "no post-restart reconcile has been recorded",
        )
    payload = proof.as_dict() if hasattr(proof, "as_dict") else dict(proof)
    return scrub(payload), SourceStatus("reconcile", STATUS_OK)
 def load_restart_console_snapshot(
    *,
    principal: console_authz.Principal | None = None,
    restart_class: str = restart_coordinator.RestartClass.FULL_MCP_RESTART.value,
    db_path: str | None = None,
    limit: int = 200,
    drain_proof_payload: Mapping[str, Any] | None = None,
    read_inventory: Callable[..., Mapping[str, Any]] | None = None,
    load_reconcile_proof: Callable[[], Any] | None = None,
    now: datetime | None = None,
 ) -> RestartConsoleSnapshot:
    """Compose the read-only restart console snapshot."""
    who = principal if principal is not None else console_authz.ANONYMOUS
    moment = now or _utc_now()
    impact, impact_source = load_impact_report(
        principal=who,
        restart_class=restart_class,
        db_path=db_path,
        limit=limit,
        read_inventory=read_inventory,
        now=moment,
    )
    fingerprint = None
    if impact is not None:
        try:
            fingerprint = drain_proof.impact_fingerprint(impact)
        except Exception:  # noqa: BLE001
            fingerprint = None
    drain, drain_source = load_drain_status(
        proof=drain_proof_payload,
        now=moment,
        expected_impact_fingerprint=fingerprint,
    )
    reconcile, reconcile_source = load_reconcile_status(
        load_proof=load_reconcile_proof
    )
    notes: list[str] = [
        "This surface is read-only: it evaluates and displays, and performs no "
        "restart, reload, drain, approval, or process action.",
    ]
    if not impact_source.available:
        notes.append(
            "Impact preview unavailable — a restart decision must not be made "
            "from this page while the blast radius is unknown."
        )
    return RestartConsoleSnapshot(
        generated_at=moment.isoformat(),
        viewer_role=str(who.role or "anonymous"),
        viewer_authenticated=bool(who.authenticated),
        read_only=True,
        impact=impact,
        impact_source=impact_source,
        drain=drain,
        drain_source=drain_source,
        reconcile=reconcile,
        reconcile_source=reconcile_source,
        restart_classes=build_restart_class_views(who.role),
        authorizations=build_action_authorizations(who),
        break_glass=BreakGlassSurface(
            available=False,
            issue=BREAK_GLASS_ISSUE,
            reason=BREAK_GLASS_PENDING_REASON,
            viewer_is_privileged=viewer_is_privileged(who),
        ),
        notes=tuple(notes),
    )
@@ -0,0 +1,299 @@
 """HTML views for the read-only restart console (#667).
 Every interpolated value passes through :func:`_esc`. Values that can carry a
 filesystem path or free-form operator text additionally pass through
 :func:`webui.inventory.scrub_text`, which redacts credential-shaped tokens
 *inside* a string rather than only at its start.
 The page renders state and never offers a control that would mutate anything:
 the approval and break-glass panels report authorization and availability, and
 there is no form, button, or endpoint behind them.
 """
 from __future__ import annotations
 import html
 from webui.inventory import scrub_text
 from webui.restart_console import RestartConsoleSnapshot, SourceStatus
 def _esc(value: object) -> str:
    """Escape any value for HTML text or a quoted attribute."""
    if value is None:
        return ""
    return html.escape(str(value), quote=True)
 def _esc_text(value: object) -> str:
    """Escape free-form text after redacting secrets embedded inside it."""
    if value is None:
        return ""
    return _esc(scrub_text(str(value)))
 def _bool_badge(
    value: bool, *, true_label: str = "yes", false_label: str = "no"
 ) -> str:
    css = "badge-ok" if value else "badge-blocked"
    label = true_label if value else false_label
    return f'<span class="badge {css}">{_esc(label)}</span>'
 def _source_badge(source: SourceStatus) -> str:
    css = "badge-ok" if source.available else "badge-blocked"
    badge = f'<span class="badge {css}">{_esc(source.status)}</span>'
    if source.detail:
        badge += f' <span class="muted">{_esc_text(source.detail)}</span>'
    return badge
 def _notes_block(snapshot: RestartConsoleSnapshot) -> str:
    if not snapshot.notes:
        return ""
    items = "".join(f"<li>{_esc_text(note)}</li>" for note in snapshot.notes)
    return f"<ul class='reasons'>{items}</ul>"
 def _impact_section(snapshot: RestartConsoleSnapshot) -> str:
    head = (
        "<section class='health-card'>"
        f"<h3>Impact preview {_source_badge(snapshot.impact_source)}</h3>"
    )
    impact = snapshot.impact
    if impact is None:
        return (
            head
            + "<p class='muted'>No impact preview is available, so the blast "
            "radius of a restart is unknown. Treat this as unsafe.</p></section>"
        )
    counts = impact.get("counts") or {}
    verdict = str(impact.get("verdict") or "unknown")
    verdict_css = "badge-ok" if verdict == "safe" else "badge-blocked"
    rows = "".join(
        f"<tr><th>{_esc(key.replace('_', ' '))}</th><td>{_esc(value)}</td></tr>"
        for key, value in sorted(counts.items())
    )
    reasons = "".join(
        f"<li>{_esc_text(reason)}</li>" for reason in (impact.get("reasons") or [])
    )
    incomplete = ""
    if not impact.get("inventory_complete", False):
        detail = "; ".join(str(r) for r in (impact.get("incomplete_reasons") or []))
        incomplete = (
            "<p class='error'><strong>Inventory incomplete:</strong> "
            f"{_esc_text(detail or 'unspecified')}. The coordinator fails "
            "closed on an incomplete inventory.</p>"
        )
    sessions = impact.get("affected_sessions") or []
    session_rows = "".join(
        "<tr>"
        f"<td><code>{_esc(s.get('session_id'))}</code></td>"
        f"<td>{_esc(s.get('role'))}</td>"
        f"<td>{_esc(s.get('pid'))}</td>"
        f"<td>{_bool_badge(bool(s.get('live')), true_label='live', false_label='idle')}</td>"
        f"<td>{_bool_badge(not s.get('heartbeat_stale'), true_label='fresh', false_label='stale')}</td>"
        "</tr>"
        for s in sessions[:50]
    )
    session_table = (
        "<h4>Sessions a restart would terminate</h4>"
        "<div class='table-scroll'><table class='registry'><thead><tr>"
        "<th>Session</th><th>Role</th><th>PID</th><th>State</th>"
        "<th>Heartbeat</th></tr></thead><tbody>"
        f"{session_rows}</tbody></table></div>"
        if session_rows
        else "<p class='muted'>No affected sessions reported.</p>"
    )
    truncated = (
        f"<p class='muted'>Showing the first 50 of {_esc(len(sessions))} "
        "affected sessions.</p>"
        if len(sessions) > 50
        else ""
    )
    return (
        head
        + "<p class='health-headline'>Verdict "
        f"<span class='badge {verdict_css}'>{_esc(verdict)}</span> · "
        f"blast radius <code>{_esc(impact.get('blast_radius'))}</code> · "
        f"class <code>{_esc(impact.get('restart_class'))}</code></p>"
        + incomplete
        + (f"<ul class='reasons'>{reasons}</ul>" if reasons else "")
        + (f"<table class='registry'><tbody>{rows}</tbody></table>" if rows else "")
        + session_table
        + truncated
        + "</section>"
    )
 def _drain_section(snapshot: RestartConsoleSnapshot) -> str:
    head = (
        "<section class='health-card'>"
        f"<h3>Drain proof {_source_badge(snapshot.drain_source)}</h3>"
    )
    drain = snapshot.drain
    if drain is None:
        return (
            head
            + "<p class='muted'>No drain proof has been presented to this view. "
            "The #661 gate authorizes a restart only against a valid, unexpired, "
            "clean proof, so the absence of one is a denial, not a pass.</p>"
            "</section>"
        )
    reasons = "".join(
        f"<li>{_esc_text(reason)}</li>" for reason in (drain.get("reasons") or [])
    )
    return (
        head
        + "<table class='registry'><tbody>"
        f"<tr><th>Valid</th><td>{_bool_badge(bool(drain.get('valid')))}</td></tr>"
        f"<tr><th>Clean</th><td>{_bool_badge(bool(drain.get('clean')))}</td></tr>"
        f"<tr><th>Expired</th><td>{_bool_badge(not drain.get('expired'), true_label='no', false_label='yes')}</td></tr>"
        f"<tr><th>Tampered</th><td>{_bool_badge(not drain.get('tampered'), true_label='no', false_label='yes')}</td></tr>"
        f"<tr><th>Proof id</th><td><code>{_esc(drain.get('proof_id'))}</code></td></tr>"
        "</tbody></table>"
        + (f"<ul class='reasons'>{reasons}</ul>" if reasons else "")
        + "</section>"
    )
 def _reconcile_section(snapshot: RestartConsoleSnapshot) -> str:
    head = (
        "<section class='health-card'>"
        f"<h3>Post-restart reconcile {_source_badge(snapshot.reconcile_source)}</h3>"
    )
    proof = snapshot.reconcile
    if proof is None:
        return (
            head
            + "<p class='muted'>No post-restart completion proof is recorded. "
            "Until one is, the last restart's recovery state is unproven.</p>"
            "</section>"
        )
    items = "".join(
        "<tr>"
        f"<td>{_esc(item.get('dimension'))}</td>"
        f"<td>{_esc(item.get('status'))}</td>"
        f"<td>{_esc_text(item.get('summary'))}</td>"
        f"<td>{_bool_badge(not item.get('follow_up_required'), true_label='no', false_label='yes')}</td>"
        "</tr>"
        for item in (proof.get("items") or [])
    )
    return (
        head
        + "<p class='health-headline'>Status "
        f"<code>{_esc(proof.get('overall_status'))}</code> · mode "
        f"<code>{_esc(proof.get('mode'))}</code> · resolved "
        f"{_esc(proof.get('resolved_count'))} · unresolved "
        f"{_esc(proof.get('unresolved_count'))}</p>"
        + (
            "<div class='table-scroll'><table class='registry'><thead><tr>"
            "<th>Dimension</th><th>Status</th><th>Summary</th>"
            "<th>Follow-up required</th></tr></thead><tbody>"
            f"{items}</tbody></table></div>"
            if items
            else "<p class='muted'>No reconcile dimensions reported.</p>"
        )
        + "</section>"
    )
 def _class_matrix_section(snapshot: RestartConsoleSnapshot) -> str:
    rows = "".join(
        "<tr>"
        f"<td><code>{_esc(view.restart_class)}</code></td>"
        f"<td><code>{_esc(view.required_permission)}</code></td>"
        f"<td>{_esc(view.expected_blast_radius)}</td>"
        f"<td>{_esc(view.drain_requirement)}</td>"
        f"<td>{_esc(view.approval_requirement)}</td>"
        f"<td>{_bool_badge(view.viewer_may_request)}</td>"
        f"<td>{_bool_badge(view.viewer_may_execute)}</td>"
        "</tr>"
        for view in snapshot.restart_classes
    )
    return (
        "<section class='health-card'>"
        "<h3>Restart classes</h3>"
        "<p class='muted'>The least-privilege matrix each restart request is "
        "resolved against. &ldquo;You may request&rdquo; and &ldquo;you may "
        "execute&rdquo; are computed for the current viewer role, not for a "
        "generic operator.</p>"
        "<div class='table-scroll'><table class='registry'><thead><tr>"
        "<th>Class</th><th>Permission</th><th>Blast radius</th>"
        "<th>Drain</th><th>Approval</th><th>You may request</th>"
        "<th>You may execute</th></tr></thead><tbody>"
        f"{rows}</tbody></table></div>"
        "</section>"
    )
 def _approval_section(snapshot: RestartConsoleSnapshot) -> str:
    rows = "".join(
        "<tr>"
        f"<td><code>{_esc(a.action_id)}</code></td>"
        f"<td>{_esc(a.required_role)}</td>"
        f"<td>{_bool_badge(a.allowed)}</td>"
        f"<td>{_bool_badge(a.execution_enabled)}</td>"
        f"<td><code>{_esc(a.reason_code)}</code></td>"
        f"<td>{_esc_text(a.detail)}</td>"
        "</tr>"
        for a in snapshot.authorizations
    )
    return (
        "<section class='health-card'>"
        "<h3>Approval controls</h3>"
        "<p class='muted'>Authorization is probed the way execution would probe "
        "it, so &ldquo;execution enabled&rdquo; answers whether the action would "
        "actually run — not merely whether this role outranks the requirement. "
        "No control on this page performs the action.</p>"
        "<div class='table-scroll'><table class='registry'><thead><tr>"
        "<th>Action</th><th>Required role</th><th>Authorized</th>"
        "<th>Execution enabled</th><th>Reason</th><th>Detail</th>"
        "</tr></thead><tbody>"
        f"{rows}</tbody></table></div>"
        "</section>"
    )
 def _break_glass_section(snapshot: RestartConsoleSnapshot) -> str:
    bg = snapshot.break_glass
    if not bg.viewer_is_privileged:
        return (
            "<section class='health-card'>"
            "<h3>Break-glass</h3>"
            "<p class='muted'>Break-glass status is visible to operator-class "
            "roles only. Your role does not carry that authority, so no "
            "emergency surface is shown.</p>"
            "</section>"
        )
    return (
        "<section class='health-card'>"
        "<h3>Break-glass "
        f"{_bool_badge(bg.available, true_label='available', false_label='unavailable')}"
        "</h3>"
        f"<p class='muted'>{_esc_text(bg.reason)}</p>"
        f"<p class='meta'>Tracked by issue #{_esc(bg.issue)}.</p>"
        "</section>"
    )
 def render_restart_console_page(snapshot: RestartConsoleSnapshot) -> str:
    """Render the whole read-only restart console body."""
    return (
        "<h2>Restart status and impact</h2>"
        f"<p class='meta'>Generated <code>{_esc(snapshot.generated_at)}</code> · "
        f"viewer role <code>{_esc(snapshot.viewer_role)}</code> · "
        f"authenticated {_bool_badge(snapshot.viewer_authenticated)} · "
        f"read-only {_bool_badge(snapshot.read_only)}</p>"
        + _notes_block(snapshot)
        + _impact_section(snapshot)
        + _drain_section(snapshot)
        + _reconcile_section(snapshot)
        + _class_matrix_section(snapshot)
        + _approval_section(snapshot)
        + _break_glass_section(snapshot)
    )
@@ -201,16 +201,6 @@ def _candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate
    return candidates
 def candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate]:
    """Public alias for :func:`_candidates_from_queue_snapshot` (#643).
    The request-initiation service ranks the same candidate set this view
    renders, so both must agree on how a queue row becomes a candidate. One
    construction, two callers — not two that can drift apart.
    """
    return _candidates_from_queue_snapshot(q_snap)
 def _claim_lease_records(inventory: dict[str, Any] | None) -> list[dict[str, Any]]:
    """Normalize ``build_claim_inventory`` entries into lease records.