diff --git a/tests/test_webui_restart_console.py b/tests/test_webui_restart_console.py
new file mode 100644
index 0000000..439bba5
--- /dev/null
+++ b/tests/test_webui_restart_console.py
@@ -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="operator@example.com",
+ 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 = ""
+ html = restart_views.render_restart_console_page(
+ self._snapshot(read_inventory=_inventory(sessions=[_live_session(hostile)]))
+ )
+ self.assertNotIn("