Merge remote-tracking branch 'prgs/master' into feat/issue-666-concurrent-mcp-restart-tests

tests/test_allocator_service.py

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

tests/test_issue_886_apply_authorization_conjunction.py

@@ -35,6 +35,17 @@ BREAK_GLASS_ENV = "GITEA_BREAKGLASS_RESTART_AUTHORIZATION"
 QUIET_SESSIONS: list[dict] = []
 QUIET_LEASES: list[dict] = []
 
+# #669: broad restarts need a prior narrow-attempt log (unless break-glass).
+PRIOR_NARROW_ATTEMPTS_JSON = json.dumps(
+    [
+        {
+            "action": "client_reconnect",
+            "outcome": "insufficient",
+            "reason": "still flapping after reconnect",
+        }
+    ]
+)
+
 
 class _FakeDB:
     """Minimal control-plane DB stand-in for the restart inventory."""
@@ -128,6 +139,7 @@ class TestConjunction(_RestartToolHarness):
         preview = self._call(
             role="operator",
             restart_class="full_mcp_restart",
+            prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
             env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
         )
         self.assertTrue(preview["allow_restart"],
@@ -136,6 +148,7 @@ class TestConjunction(_RestartToolHarness):
         result = self._call(
             role="operator",
             restart_class="full_mcp_restart",
+            prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
             dry_run=False,
             drain_proof_json=self._clean_proof_for(preview),
             env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
@@ -342,6 +355,7 @@ class TestExistingPathsStillWork(_RestartToolHarness):
         result = self._call(
             role="operator",
             restart_class="full_mcp_restart",
+            prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
             dry_run=False,
             env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
         )

tests/test_recovery_playbook.py

@@ -0,0 +1,217 @@
+"""Unit tests for the scoped recovery playbook (#669)."""
+
+from __future__ import annotations
+
+import recovery_playbook as rp
+import restart_coordinator as rc
+
+
+def test_ladder_covers_eleven_ordered_rungs():
+    ranks = [r.rank for r in rp.RECOVERY_LADDER]
+    assert ranks == list(range(len(rp.RECOVERY_LADDER)))
+    assert len(rp.RECOVERY_LADDER) == 11
+    assert rp.RECOVERY_LADDER[0].action is rp.RecoveryAction.CLIENT_RECONNECT
+    assert rp.RECOVERY_LADDER[-1].action is rp.RecoveryAction.HOST_RESTART
+
+
+def test_ladder_document_links_parent_issues():
+    doc = rp.ladder_document()
+    assert "#655" in doc["parent_issues"]
+    assert "#652" in doc["parent_issues"]
+    assert "#653" in doc["parent_issues"]
+    assert doc["enforcement_issue"] == "#669"
+    assert "full_mcp_restart" in doc["broad_restart_actions"]
+
+
+def test_recommend_transport_eof_starts_at_client_reconnect():
+    plan = rp.recommend_actions(symptoms=["transport_eof"])
+    assert plan["recommended_actions"][0]["action"] == "client_reconnect"
+    assert plan["recommended_actions"][0]["issue_links"]
+
+
+def test_recommend_skips_successful_prior_attempts():
+    attempts = [
+        rp.build_attempt_record(
+            "client_reconnect", outcome="success", reason="reconnected"
+        )
+    ]
+    plan = rp.recommend_actions(
+        symptoms=["transport_eof"], prior_recovery_attempts=attempts
+    )
+    actions = [a["action"] for a in plan["recommended_actions"]]
+    assert "client_reconnect" not in actions
+    assert actions[0] == "capability_refresh"
+
+
+def test_escalation_denied_without_attempt_log():
+    result = rp.assess_escalation("full_mcp_restart", prior_recovery_attempts=[])
+    assert result.allowed is False
+    assert result.require_attempt_log is True
+    assert any("#669" in r for r in result.reasons)
+    assert result.recommended_next  # soft recommendations still provided
+
+
+def test_escalation_allowed_after_insufficient_narrower():
+    attempts = [
+        rp.build_attempt_record(
+            "client_reconnect",
+            outcome="insufficient",
+            reason="still flapping",
+        ),
+        rp.build_attempt_record(
+            "session_reconnect",
+            outcome="failed",
+            reason="namespace still dead",
+        ),
+    ]
+    result = rp.assess_escalation(
+        "full_mcp_restart", prior_recovery_attempts=attempts
+    )
+    assert result.allowed is True
+    assert len(result.qualifying_attempts) == 2
+
+
+def test_escalation_break_glass_bypasses_attempt_log():
+    result = rp.assess_escalation(
+        "host_restart", prior_recovery_attempts=[], break_glass=True
+    )
+    assert result.allowed is True
+    assert result.break_glass is True
+
+
+def test_narrow_action_does_not_require_attempt_log():
+    result = rp.assess_escalation(
+        "client_reconnect", prior_recovery_attempts=[]
+    )
+    assert result.allowed is True
+    assert result.require_attempt_log is False
+
+
+def test_same_rank_attempt_does_not_qualify_for_escalation():
+    attempts = [
+        rp.build_attempt_record(
+            "full_mcp_restart", outcome="failed", reason="already failed full"
+        )
+    ]
+    result = rp.assess_escalation(
+        "full_mcp_restart", prior_recovery_attempts=attempts
+    )
+    assert result.allowed is False
+
+
+def test_success_outcome_does_not_qualify_for_escalation():
+    attempts = [
+        rp.build_attempt_record(
+            "client_reconnect", outcome="success", reason="fixed"
+        )
+    ]
+    result = rp.assess_escalation(
+        "full_mcp_restart", prior_recovery_attempts=attempts
+    )
+    assert result.allowed is False
+
+
+def test_recovery_metrics_fraction_avoided():
+    attempts = [
+        rp.build_attempt_record("client_reconnect", outcome="success"),
+        rp.build_attempt_record("session_reconnect", outcome="success"),
+        rp.build_attempt_record("full_mcp_restart", outcome="success"),
+    ]
+    metrics = rp.recovery_metrics(attempts)
+    assert metrics["successes_total"] == 3
+    assert metrics["successes_avoided_full_restart"] == 2
+    assert metrics["successes_full_or_host_restart"] == 1
+    assert abs(metrics["fraction_avoided_full_restart"] - (2 / 3)) < 1e-9
+
+
+def test_coordinator_denies_full_restart_without_attempt_log():
+    inv = {
+        "inventory_complete": True,
+        "sessions": [],
+        "leases": [],
+        "prior_recovery_attempts": [],
+    }
+    report = rc.evaluate_restart_impact(
+        inv,
+        restart_class=rc.RestartClass.FULL_MCP_RESTART,
+        requester_role="controller",
+        requester_permissions=rc.permissions_for_role("controller"),
+        controller_approved=True,
+        operator_authorized=True,
+    )
+    assert report.allow_restart is False
+    assert report.attempt_log_satisfied is False
+    assert report.verdict == rc.VERDICT_UNSAFE
+    blob = " ".join(report.reasons + report.authorization_reasons)
+    assert "#669" in blob or "attempt log" in blob
+
+
+def test_coordinator_allows_full_restart_with_attempt_log():
+    inv = {
+        "inventory_complete": True,
+        "sessions": [],
+        "leases": [],
+        "prior_recovery_attempts": [
+            {
+                "action": "client_reconnect",
+                "outcome": "insufficient",
+                "reason": "still broken",
+            }
+        ],
+    }
+    report = rc.evaluate_restart_impact(
+        inv,
+        restart_class=rc.RestartClass.FULL_MCP_RESTART,
+        requester_role="controller",
+        requester_permissions=rc.permissions_for_role("controller"),
+        controller_approved=True,
+        operator_authorized=True,
+    )
+    assert report.attempt_log_satisfied is True
+    assert report.allow_restart is True
+    assert report.verdict == rc.VERDICT_SAFE
+
+
+def test_coordinator_break_glass_allows_without_log():
+    inv = {
+        "inventory_complete": True,
+        "sessions": [],
+        "leases": [],
+        "prior_recovery_attempts": [],
+    }
+    report = rc.evaluate_restart_impact(
+        inv,
+        restart_class=rc.RestartClass.FULL_MCP_RESTART,
+        requester_role="controller",
+        requester_permissions=rc.permissions_for_role("controller"),
+        controller_approved=True,
+        operator_authorized=True,
+        break_glass=True,
+    )
+    assert report.break_glass is True
+    assert report.attempt_log_satisfied is True
+    assert report.allow_restart is True
+
+
+def test_coordinator_client_reconnect_unaffected():
+    inv = {
+        "inventory_complete": True,
+        "sessions": [],
+        "leases": [],
+        "prior_recovery_attempts": [],
+    }
+    report = rc.evaluate_restart_impact(
+        inv,
+        restart_class=rc.RestartClass.CLIENT_RECONNECT,
+        requester_role="author",
+        requester_permissions=rc.permissions_for_role("author"),
+    )
+    assert report.attempt_log_satisfied is True
+    assert report.allow_restart is True
+
+
+def test_restart_class_alias_accepted():
+    assert (
+        rp.resolve_action("full_mcp_restart")
+        is rp.RecoveryAction.FULL_MCP_RESTART
+    )

tests/test_webui_console_recovery.py

@@ -0,0 +1,506 @@
+"""Unit and integration tests for Phase 2 Web Console recovery controls (#644)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import types
+import unittest
+from unittest.mock import patch
+
+from starlette.testclient import TestClient
+
+import merged_cleanup_reconcile
+import runtime_recovery_guard
+import stable_branch_push_guard
+import stale_binding_recovery
+from webui import console_authz, console_recovery, system_health
+from webui.app import create_app
+
+
+class TestConsoleRecovery(unittest.TestCase):
+
+    def test_diagnose_recovery_healthy(self) -> None:
+        diag = console_recovery.diagnose_recovery()
+        self.assertIn(diag.status, {console_recovery.STATUS_HEALTHY, console_recovery.STATUS_ACTION_REQUIRED})
+        self.assertIsInstance(diag.playbooks, tuple)
+        self.assertGreaterEqual(len(diag.playbooks), 4)
+
+        playbook_ids = {pb.playbook_id for pb in diag.playbooks}
+        self.assertIn(console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, playbook_ids)
+        self.assertIn(console_recovery.PLAYBOOK_REBIND_SESSION, playbook_ids)
+        self.assertIn(console_recovery.PLAYBOOK_RECONCILE_CLEANUPS, playbook_ids)
+        self.assertIn(console_recovery.PLAYBOOK_SANCTIONED_RESTART, playbook_ids)
+
+    def test_confirmation_phrase_generation_and_matching(self) -> None:
+        phrase = console_recovery.confirmation_phrase("clear_stale_binding")
+        self.assertEqual(phrase, "confirm clear_stale_binding")
+        self.assertTrue(console_recovery.confirmation_matches("clear_stale_binding", "confirm clear_stale_binding"))
+        self.assertFalse(console_recovery.confirmation_matches("clear_stale_binding", "wrong phrase"))
+
+        phrase_target = console_recovery.confirmation_phrase("sanctioned_restart", "gitea-author")
+        self.assertEqual(phrase_target, "confirm sanctioned_restart gitea-author")
+        self.assertTrue(console_recovery.confirmation_matches("sanctioned_restart", "confirm sanctioned_restart gitea-author", "gitea-author"))
+
+    def test_build_recovery_preview(self) -> None:
+        principal = console_authz.Principal("[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True)
+        preview = console_recovery.build_recovery_preview(
+            playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
+            target="test-worktree",
+            principal=principal,
+        )
+        self.assertEqual(preview["playbook_id"], console_recovery.PLAYBOOK_CLEAR_STALE_BINDING)
+        self.assertEqual(preview["action_id"], console_recovery.ACTION_CLEAR_STALE_BINDING)
+        self.assertEqual(preview["confirmation_phrase"], "confirm clear_stale_binding test-worktree")
+        self.assertTrue(len(preview["mutation_ledger"]) >= 3)
+        self.assertTrue(preview["authorization"]["allowed"])
+
+    def test_build_recovery_preview_unknown_playbook(self) -> None:
+        preview = console_recovery.build_recovery_preview("unknown_playbook")
+        self.assertFalse(preview.get("allowed"))
+        self.assertEqual(preview.get("error"), "unknown_playbook")
+
+    def test_execute_recovery_playbook_confirmation_mismatch(self) -> None:
+        # Authorization is checked before confirmation, so the phase gate has to
+        # pass for this test to reach the branch it is about.
+        principal = console_authz.Principal("[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True)
+        with self._phase_two_enabled():
+            result = console_recovery.execute_recovery_playbook(
+                playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
+                confirmation="invalid confirmation",
+                principal=principal,
+            )
+        self.assertFalse(result["success"])
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["error"], "confirmation_mismatch")
+
+    def test_execute_recovery_playbook_unauthorized(self) -> None:
+        # Anonymous principal has viewer role -> should be denied
+        result = console_recovery.execute_recovery_playbook(
+            playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
+            confirmation="confirm clear_stale_binding",
+            principal=console_authz.ANONYMOUS,
+        )
+        self.assertFalse(result["success"])
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["error"], console_authz.DENY_UNAUTHENTICATED)
+
+    def test_execute_refuses_phase_two_write_while_console_is_phase_one(self) -> None:
+        """B1: the apply path must arm the phase gate, not skip it.
+
+        ``authorize`` only applies the phase branch when ``for_execution=True``.
+        The apply path used the default, so an operator executed a phase-2 write
+        while ``ACTIVE_PHASE`` was 1.
+        """
+        self.assertGreater(
+            console_authz.get_action(console_recovery.ACTION_CLEAR_STALE_BINDING).phase,
+            console_authz.ACTIVE_PHASE,
+            "fixture assumes the recovery actions are ahead of the active phase",
+        )
+        principal = console_authz.Principal(
+            "[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True
+        )
+        phrase = console_recovery.confirmation_phrase(
+            console_recovery.PLAYBOOK_CLEAR_STALE_BINDING
+        )
+        result = console_recovery.execute_recovery_playbook(
+            playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
+            confirmation=phrase,
+            principal=principal,
+        )
+        self.assertFalse(result["success"])
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["error"], console_authz.DENY_PHASE_NOT_ACTIVE)
+
+    def test_preview_execution_enabled_matches_the_execution_decision(self) -> None:
+        """B1: preview must not report a bare False it cannot explain."""
+        principal = console_authz.Principal(
+            "[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True
+        )
+        preview = console_recovery.build_recovery_preview(
+            playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION,
+            target="branches/feat-issue-644",
+            principal=principal,
+        )
+        self.assertFalse(preview["execution_enabled"])
+        self.assertEqual(
+            preview["execution_blocked_reason"], console_authz.DENY_PHASE_NOT_ACTIVE
+        )
+        self.assertFalse(preview["execution_authorization"]["allowed"])
+        # The preview (non-execution) decision still allows, by role.
+        self.assertTrue(preview["authorization"]["allowed"])
+
+    def _phase_two_enabled(self):
+        """Raise ACTIVE_PHASE so the execution branches are reachable in tests."""
+        return patch.object(console_authz, "ACTIVE_PHASE", 2)
+
+    def _operator(self) -> console_authz.Principal:
+        return console_authz.Principal(
+            "[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True
+        )
+
+    def test_rebind_mutates_the_live_environment_not_a_copy(self) -> None:
+        """B2: the playbook must change the mapping it claims to have changed."""
+        live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: "branches/stale-old"}
+        phrase = console_recovery.confirmation_phrase(
+            console_recovery.PLAYBOOK_REBIND_SESSION, "branches/feat-issue-644"
+        )
+        with self._phase_two_enabled():
+            result = console_recovery.execute_recovery_playbook(
+                playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION,
+                confirmation=phrase,
+                target="branches/feat-issue-644",
+                principal=self._operator(),
+                env=live_env,
+            )
+        self.assertTrue(result["success"])
+        self.assertEqual(
+            live_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV],
+            "branches/feat-issue-644",
+            "rebind reported success without changing the caller's environment",
+        )
+        self.assertTrue(result["applied_result"]["binding_changed"])
+        self.assertEqual(result["applied_result"]["binding_before"], "branches/stale-old")
+        self.assertEqual(
+            result["applied_result"]["binding_after"], "branches/feat-issue-644"
+        )
+
+    def test_clear_stale_binding_reports_failure_when_nothing_changed(self) -> None:
+        """B2: a no-op recovery must never be reported as success."""
+        live_env: dict[str, str] = {}
+        phrase = console_recovery.confirmation_phrase(
+            console_recovery.PLAYBOOK_CLEAR_STALE_BINDING
+        )
+        with self._phase_two_enabled():
+            result = console_recovery.execute_recovery_playbook(
+                playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
+                confirmation=phrase,
+                principal=self._operator(),
+                env=live_env,
+            )
+        self.assertFalse(
+            result["success"],
+            "a clear that changed no binding must not report success",
+        )
+        self.assertFalse(result["applied_result"]["binding_changed"])
+
+    def test_clear_stale_binding_clears_the_live_binding(self) -> None:
+        """B2: the sanctioned clear must reach the caller's environment."""
+        missing = "/nonexistent/branches/deleted-worktree"
+        live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: missing}
+        phrase = console_recovery.confirmation_phrase(
+            console_recovery.PLAYBOOK_CLEAR_STALE_BINDING
+        )
+        with self._phase_two_enabled():
+            result = console_recovery.execute_recovery_playbook(
+                playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
+                confirmation=phrase,
+                principal=self._operator(),
+                env=live_env,
+            )
+        if result["success"]:
+            self.assertNotIn(stale_binding_recovery.ACTIVE_WORKTREE_ENV, live_env)
+            self.assertEqual(result["applied_result"]["binding_before"], missing)
+            self.assertIsNone(result["applied_result"]["binding_after"])
+        else:
+            # Fail closed is acceptable; reporting a clear that did not happen
+            # is not. This is the invariant the blocker was about.
+            self.assertFalse(result["applied_result"]["binding_changed"])
+            self.assertEqual(
+                live_env.get(stale_binding_recovery.ACTIVE_WORKTREE_ENV), missing
+            )
+
+    def test_reconcile_playbook_calls_an_entry_point_that_exists(self) -> None:
+        """B3: the previous call named a function absent from the module."""
+        phrase = console_recovery.confirmation_phrase(
+            console_recovery.PLAYBOOK_RECONCILE_CLEANUPS
+        )
+        fake_server = types.SimpleNamespace(
+            gitea_reconcile_merged_cleanups=lambda **kwargs: {
+                "success": True,
+                "entries": [{"issue_number": 100}],
+            }
+        )
+        with self._phase_two_enabled(), patch.dict(
+            sys.modules, {"gitea_mcp_server": fake_server}
+        ):
+            result = console_recovery.execute_recovery_playbook(
+                playbook_id=console_recovery.PLAYBOOK_RECONCILE_CLEANUPS,
+                confirmation=phrase,
+                principal=console_authz.Principal(
+                    "[email protected]",
+                    console_authz.ADMIN,
+                    console_authz.IDENTITY_LOCAL_DEV,
+                    True,
+                ),
+            )
+        self.assertTrue(result["success"], result.get("applied_result"))
+        self.assertNotIn("error_type", result["applied_result"])
+        self.assertEqual(result["applied_result"]["reconciled_count"], 1)
+
+    def test_reconcile_entry_point_exists_on_the_real_module(self) -> None:
+        """B3 regression: guard the symbol itself, not just the call shape."""
+        import gitea_mcp_server
+
+        self.assertTrue(
+            hasattr(gitea_mcp_server, "gitea_reconcile_merged_cleanups"),
+            "console recovery depends on this reconciler entry point",
+        )
+        self.assertFalse(
+            hasattr(merged_cleanup_reconcile, "reconcile_merged_cleanups"),
+            "if this module grows the orchestrator, point the playbook back at it",
+        )
+
+    def test_contamination_gate_blocks_a_writing_playbook(self) -> None:
+        """B4: a live marker plus a gated task key must actually block."""
+        marker = {
+            "kind": "manual_daemon_kill",
+            "reason_class": "manual_daemon_kill",
+            "command_summary": "pkill -f gitea_mcp_server",
+            "active": True,
+        }
+        phrase = console_recovery.confirmation_phrase(
+            console_recovery.PLAYBOOK_REBIND_SESSION, "branches/feat-issue-644"
+        )
+        live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: "branches/stale-old"}
+        with self._phase_two_enabled(), patch.object(
+            console_recovery, "load_active_contamination_marker", return_value=marker
+        ):
+            result = console_recovery.execute_recovery_playbook(
+                playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION,
+                confirmation=phrase,
+                target="branches/feat-issue-644",
+                principal=self._operator(),
+                env=live_env,
+            )
+        self.assertFalse(result["success"])
+        self.assertEqual(result["error"], "contaminated_runtime")
+        self.assertEqual(
+            live_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV],
+            "branches/stale-old",
+            "a blocked playbook must not have mutated anything",
+        )
+
+    def test_contamination_gate_exempts_the_reconciler_remedy(self) -> None:
+        """B4: the designated remedy must stay reachable while contaminated."""
+        marker = {
+            "kind": "manual_daemon_kill",
+            "reason_class": "manual_daemon_kill",
+            "command_summary": "pkill -f gitea_mcp_server",
+            "active": True,
+        }
+        phrase = console_recovery.confirmation_phrase(
+            console_recovery.PLAYBOOK_RECONCILE_CLEANUPS
+        )
+        fake_server = types.SimpleNamespace(
+            gitea_reconcile_merged_cleanups=lambda **kwargs: {
+                "success": True,
+                "entries": [],
+            }
+        )
+        with self._phase_two_enabled(), patch.object(
+            console_recovery, "load_active_contamination_marker", return_value=marker
+        ), patch.dict(sys.modules, {"gitea_mcp_server": fake_server}):
+            result = console_recovery.execute_recovery_playbook(
+                playbook_id=console_recovery.PLAYBOOK_RECONCILE_CLEANUPS,
+                confirmation=phrase,
+                principal=console_authz.Principal(
+                    "[email protected]",
+                    console_authz.ADMIN,
+                    console_authz.IDENTITY_LOCAL_DEV,
+                    True,
+                ),
+            )
+        self.assertNotEqual(result.get("error"), "contaminated_runtime")
+
+    def test_gated_task_key_is_actually_gated(self) -> None:
+        """B4: the console action id was never a member of the gated set."""
+        self.assertIn(
+            console_recovery.CONTAMINATION_GATED_TASK,
+            stable_branch_push_guard.CONTAMINATION_GATED_TASKS,
+        )
+        self.assertNotIn(
+            console_recovery.ACTION_CLEAR_STALE_BINDING,
+            stable_branch_push_guard.CONTAMINATION_GATED_TASKS,
+        )
+
+    def test_diagnosis_reads_the_key_the_gate_returns(self) -> None:
+        """B4: ``contaminated`` is a key assess_contamination_gate never returns."""
+        gate = runtime_recovery_guard.assess_contamination_gate(
+            None, task=console_recovery.CONTAMINATION_GATED_TASK, actual_role="operator"
+        )
+        self.assertNotIn("contaminated", gate)
+        self.assertIn("block", gate)
+
+    def test_contaminated_runtime_is_reported_unclean(self) -> None:
+        """B4: verify_post_recovery reported contamination_clean unconditionally."""
+        marker = {
+            "kind": "manual_daemon_kill",
+            "reason_class": "manual_daemon_kill",
+            "command_summary": "pkill -f gitea_mcp_server",
+            "active": True,
+        }
+        with patch.object(
+            console_recovery, "load_active_contamination_marker", return_value=marker
+        ):
+            verification = console_recovery.verify_post_recovery()
+            diag = console_recovery.diagnose_recovery()
+        self.assertFalse(verification["contamination_clean"])
+        self.assertFalse(verification["clean"])
+        self.assertEqual(diag.status, console_recovery.STATUS_BLOCKED_CONTAMINATION)
+
+    def test_master_parity_baseline_is_not_the_head_it_is_compared_against(self) -> None:
+        """B5: capture_startup_parity was fed the head it was then compared to."""
+        stale = system_health.StaleRuntime(
+            daemon_head="a" * 40,
+            checkout_head="b" * 40,
+            remote_head="b" * 40,
+            stale=True,
+            determinable=True,
+            mutation_safe=False,
+            reasons=("daemon is behind the checkout",),
+        )
+        with patch.object(system_health, "assess_stale_runtime", return_value=stale):
+            diag = console_recovery.diagnose_recovery()
+        parity = diag.master_parity
+        self.assertEqual(parity["startup_head"], "a" * 40)
+        self.assertEqual(parity["current_head"], "b" * 40)
+        self.assertNotEqual(parity["startup_head"], parity["current_head"])
+        self.assertFalse(parity["in_parity"])
+
+    def test_master_parity_carries_the_live_remote_dimension(self) -> None:
+        """B5: live_remote_head was never passed, dropping the #610 dimension."""
+        stale = system_health.StaleRuntime(
+            daemon_head="c" * 40,
+            checkout_head="c" * 40,
+            remote_head="d" * 40,
+            stale=False,
+            determinable=True,
+            mutation_safe=False,
+            reasons=(),
+        )
+        with patch.object(system_health, "assess_stale_runtime", return_value=stale):
+            diag = console_recovery.diagnose_recovery()
+        self.assertEqual(diag.master_parity.get("live_remote_head"), "d" * 40)
+
+    def test_verify_post_recovery(self) -> None:
+        verification = console_recovery.verify_post_recovery()
+        self.assertIn("clean", verification)
+        self.assertIn("status", verification)
+        self.assertIn("reasons", verification)
+
+    def test_unverified_inherited_binding_is_not_reported_clean(self) -> None:
+        """B2: ``not clear_eligible`` also read clean for unproven bindings."""
+        binding = {
+            "classification": stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED,
+            "clear_eligible": False,
+        }
+        diag = console_recovery.diagnose_recovery()
+        patched = console_recovery.RecoveryDiagnosis(
+            status=diag.status,
+            clean=diag.clean,
+            stale_runtime=diag.stale_runtime,
+            master_parity=diag.master_parity,
+            stale_binding=binding,
+            contamination=diag.contamination,
+            worktree_anomalies=diag.worktree_anomalies,
+            playbooks=diag.playbooks,
+            reasons=diag.reasons,
+        )
+        with patch.object(console_recovery, "diagnose_recovery", return_value=patched):
+            verification = console_recovery.verify_post_recovery()
+        self.assertFalse(verification["binding_clean"])
+        self.assertEqual(
+            verification["binding_classification"],
+            stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED,
+        )
+
+
+class TestConsoleRecoveryApi(unittest.TestCase):
+    def setUp(self) -> None:
+        self.app = create_app()
+        self.client = TestClient(self.app)
+
+    def test_api_recovery_diagnose(self) -> None:
+        res = self.client.get("/api/v1/system/recovery/diagnose")
+        self.assertEqual(res.status_code, 200)
+        data = res.json()
+        self.assertIn("status", data)
+        self.assertIn("clean", data)
+        self.assertIn("playbooks", data)
+        self.assertTrue(len(data["playbooks"]) >= 4)
+
+    def test_api_recovery_preview(self) -> None:
+        res = self.client.post(
+            "/api/v1/system/recovery/preview",
+            json={"playbook_id": "clear_stale_binding", "target": "active"},
+        )
+        self.assertEqual(res.status_code, 200)
+        data = res.json()
+        self.assertEqual(data["playbook_id"], "clear_stale_binding")
+        self.assertEqual(data["confirmation_phrase"], "confirm clear_stale_binding active")
+        self.assertIn("mutation_ledger", data)
+
+    def test_api_recovery_apply_denied_without_auth(self) -> None:
+        res = self.client.post(
+            "/api/v1/system/recovery/apply",
+            json={"playbook_id": "clear_stale_binding", "confirmation": "confirm clear_stale_binding"},
+        )
+        self.assertEqual(res.status_code, 400)
+        data = res.json()
+        self.assertFalse(data["success"])
+        self.assertFalse(data["allowed"])
+
+    def test_api_recovery_apply_refuses_phase_two_write_with_dev_auth(self) -> None:
+        """B1: this previously asserted the phase-gate bypass as intended.
+
+        An authenticated operator posting a valid confirmation still must not
+        execute a phase-2 write while the console is in phase 1. The refusal is
+        the contract; a 200 here means the gate is not armed.
+        """
+        env = {
+            "WEBUI_AUTH_MODE": "local_dev",
+            "WEBUI_DEV_SUBJECT": "[email protected]",
+            "WEBUI_DEV_ROLE": "operator",
+        }
+        before = os.environ.get("GITEA_ACTIVE_WORKTREE")
+        with patch.dict(os.environ, env):
+            res = self.client.post(
+                "/api/v1/system/recovery/apply",
+                json={
+                    "playbook_id": "rebind_session_worktree",
+                    "target": "branches/feat-issue-644",
+                    "confirmation": "confirm rebind_session_worktree branches/feat-issue-644",
+                },
+            )
+            self.assertEqual(res.status_code, 400)
+            data = res.json()
+            self.assertFalse(data["success"])
+            self.assertFalse(data["allowed"])
+            self.assertEqual(data["error"], console_authz.DENY_PHASE_NOT_ACTIVE)
+        self.assertEqual(
+            os.environ.get("GITEA_ACTIVE_WORKTREE"),
+            before,
+            "a refused apply must not have rebound the live process environment",
+        )
+
+    def test_api_recovery_preview_reports_why_execution_is_disabled(self) -> None:
+        res = self.client.post(
+            "/api/v1/system/recovery/preview",
+            json={"playbook_id": "rebind_session_worktree", "target": "active"},
+        )
+        self.assertEqual(res.status_code, 200)
+        data = res.json()
+        self.assertFalse(data["execution_enabled"])
+        self.assertIn("execution_authorization", data)
+
+    def test_api_recovery_verify(self) -> None:
+        res = self.client.get("/api/v1/system/recovery/verify")
+        self.assertEqual(res.status_code, 200)
+        data = res.json()
+        self.assertIn("clean", data)
+        self.assertIn("status", data)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_gitea_linkage.py

@@ -0,0 +1,511 @@
+"""Tests for the Gitea issue↔PR linkage console (#645, Phase 3).
+
+Covers the acceptance criteria of the issue:
+
+* AC1 — issue↔PR linkage is visible for the selected project/repo, in both
+  directions, with the evidence that produced each edge.
+* AC2 — the latest canonical handoff (CTH) is summarized for a focused thread.
+* AC3 — an external Gitea link appears only under the admin reveal opt-in.
+* AC4 — every case is driven by mocked Gitea payloads; no network.
+
+Plus the invariants this console must not violate: a partial or failed read is
+never rendered as "no link exists", an unfetched thread is never rendered as
+"no handoff", redaction happens before display, and the surface stays read-only.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from tests.webui_testclient import TestClient
+
+from canonical_thread_handoff import format_cth_body
+from webui.app import create_app
+from webui.linkage_loader import (
+    EVIDENCE_BRANCH,
+    EVIDENCE_CLOSES,
+    EVIDENCE_REFERENCE,
+    HANDOFF_LOADED,
+    HANDOFF_NOT_LOADED,
+    HANDOFF_UNAVAILABLE,
+    LinkageSnapshot,
+    load_linkage_snapshot,
+    resolve_linkage,
+    resolve_pr_links,
+    snapshot_to_dict,
+    summarize_handoff,
+)
+from webui.linkage_views import render_linkage_page
+from webui.nav import nav_hrefs
+from webui.queue_loader import PaginationMeta
+
+
+def _pagination(*, complete: bool = True, count: int = 0) -> PaginationMeta:
+    return PaginationMeta(
+        page=1,
+        per_page=50,
+        returned_count=count,
+        has_more=not complete,
+        is_final_page=complete,
+        inventory_complete=complete,
+        pages_fetched=1,
+    )
+
+
+def _pr(
+    number: int,
+    *,
+    title: str = "",
+    body: str = "",
+    head: str = "",
+    state: str = "open",
+    labels: tuple[str, ...] = (),
+) -> dict:
+    return {
+        "number": number,
+        "title": title or f"pr {number}",
+        "body": body,
+        "state": state,
+        "head": {"ref": head},
+        "labels": [{"name": name} for name in labels],
+    }
+
+
+def _issue(
+    number: int,
+    *,
+    title: str = "",
+    state: str = "open",
+    labels: tuple[str, ...] = (),
+) -> dict:
+    return {
+        "number": number,
+        "title": title or f"issue {number}",
+        "state": state,
+        "labels": [{"name": name} for name in labels],
+    }
+
+
+def _fetcher(items: list[dict], *, complete: bool = True):
+    def _fetch(*_args, **_kwargs):
+        return items, _pagination(complete=complete, count=len(items))
+
+    return _fetch
+
+
+def _load(
+    issues: list[dict],
+    prs: list[dict],
+    *,
+    complete: bool = True,
+    **kwargs,
+) -> LinkageSnapshot:
+    return load_linkage_snapshot(
+        fetch_prs=_fetcher(prs, complete=complete),
+        fetch_issues=_fetcher(issues, complete=complete),
+        **kwargs,
+    )
+
+
+def _cth(comment_id: int, *, created_at: str, status: str, next_owner: str) -> dict:
+    return {
+        "id": comment_id,
+        "created_at": created_at,
+        "user": {"login": "jcwalker3"},
+        "body": format_cth_body(
+            cth_type="Author Handoff",
+            status=status,
+            next_owner=next_owner,
+            current_blocker="none",
+            decision="implemented",
+            proof="full suite green",
+            next_action="review PR",
+            ready_to_paste_prompt="Review PR #902 now.",
+        ),
+    }
+
+
+class TestLinkageEvidence(unittest.TestCase):
+    """AC1 — every edge records how it was found, and keeps all candidates."""
+
+    def test_closes_keyword_in_body_is_strongest_evidence(self):
+        links = resolve_pr_links(_pr(902, body="Closes #643"))
+        self.assertEqual([link.issue_number for link in links], [643])
+        self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
+        self.assertTrue(links[0].closes)
+
+    def test_closes_keyword_in_title_counts(self):
+        links = resolve_pr_links(_pr(902, title="feat(webui): preview (Closes #643)"))
+        self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
+
+    def test_canonical_branch_marker_links_without_a_keyword(self):
+        links = resolve_pr_links(_pr(902, head="feat/issue-643-request-preview"))
+        self.assertEqual([link.issue_number for link in links], [643])
+        self.assertEqual(links[0].evidence, (EVIDENCE_BRANCH,))
+        self.assertFalse(links[0].closes)
+
+    def test_non_canonical_branch_is_not_treated_as_a_marker(self):
+        self.assertEqual(resolve_pr_links(_pr(902, head="issue-643-preview")), ())
+
+    def test_bare_mention_is_recorded_as_the_weakest_evidence(self):
+        links = resolve_pr_links(_pr(902, body="context in #643"))
+        self.assertEqual(links[0].evidence, (EVIDENCE_REFERENCE,))
+        self.assertFalse(links[0].closes)
+
+    def test_several_evidence_kinds_merge_onto_one_edge(self):
+        links = resolve_pr_links(
+            _pr(902, body="Closes #643 — see #643", head="feat/issue-643-preview")
+        )
+        self.assertEqual(len(links), 1)
+        self.assertEqual(
+            links[0].evidence,
+            (EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE),
+        )
+
+    def test_stronger_evidence_sorts_first(self):
+        links = resolve_pr_links(_pr(902, body="Closes #643, related #700"))
+        self.assertEqual([link.issue_number for link in links], [643, 700])
+
+    def test_self_reference_is_not_linkage(self):
+        links = resolve_pr_links(_pr(902, body="supersedes #902"))
+        self.assertEqual(links, ())
+
+    def test_every_candidate_is_kept_never_collapsed_to_a_guess(self):
+        links = resolve_pr_links(_pr(902, body="Closes #643\nCloses #644"))
+        self.assertEqual([link.issue_number for link in links], [643, 644])
+
+
+class TestLinkageIndex(unittest.TestCase):
+    def test_issue_direction_ignores_mention_only_edges(self):
+        index = resolve_linkage([_pr(902, body="context in #643")])
+        self.assertIsNone(index.issue_prs.get(643))
+        self.assertEqual(index.pr_links[902][0].evidence, (EVIDENCE_REFERENCE,))
+
+    def test_contested_issue_is_reported_when_two_prs_claim_it(self):
+        index = resolve_linkage(
+            [_pr(902, body="Closes #643"), _pr(903, head="feat/issue-643-again")]
+        )
+        self.assertEqual(index.contested_issues(), (643,))
+        self.assertEqual(index.issue_prs[643], (902, 903))
+
+    def test_single_claim_is_not_contested(self):
+        index = resolve_linkage([_pr(902, body="Closes #643")])
+        self.assertEqual(index.contested_issues(), ())
+
+    def test_ambiguous_when_two_issues_tie_at_the_strongest_evidence(self):
+        index = resolve_linkage([_pr(902, body="Closes #643\nCloses #644")])
+        self.assertTrue(index.ambiguous(902))
+
+    def test_weaker_candidate_alongside_a_stronger_one_is_not_ambiguous(self):
+        index = resolve_linkage([_pr(902, body="Closes #643, see #700")])
+        self.assertFalse(index.ambiguous(902))
+        self.assertEqual(index.primary_issue(902).issue_number, 643)
+
+    def test_malformed_pr_row_is_skipped_not_raised_on(self):
+        index = resolve_linkage([{"title": "no number"}, _pr(902, body="Closes #643")])
+        self.assertEqual(sorted(index.pr_links), [902])
+
+
+class TestLinkageSnapshot(unittest.TestCase):
+    """AC1 — linkage is visible per project/repo, in both directions."""
+
+    def test_both_directions_are_populated(self):
+        snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
+        self.assertTrue(snapshot.ok)
+        self.assertEqual([node.number for node in snapshot.issues], [643])
+        self.assertEqual(snapshot.issues[0].linked_prs, (902,))
+        self.assertEqual(snapshot.prs[0].links[0].issue_number, 643)
+
+    def test_repo_scope_comes_from_the_registry_project(self):
+        snapshot = _load([], [])
+        self.assertIn("/", snapshot.repo_label)
+        self.assertTrue(snapshot.project_id)
+
+    def test_unknown_project_fails_closed_with_a_reason(self):
+        snapshot = _load([_issue(643)], [], project_id="no-such-project")
+        self.assertFalse(snapshot.ok)
+        self.assertIn("not found in registry", snapshot.fetch_error)
+        self.assertEqual(snapshot.issues, ())
+
+    def test_orphan_pr_is_identifiable(self):
+        snapshot = _load([], [_pr(902), _pr(903, body="Closes #643")])
+        self.assertEqual([node.number for node in snapshot.orphan_prs], [902])
+
+    def test_state_scope_defaults_to_open_and_is_reported(self):
+        self.assertEqual(_load([], []).state_scope, "open")
+        self.assertEqual(_load([], [], state="all").state_scope, "all")
+
+    def test_unsupported_state_falls_back_to_open(self):
+        self.assertEqual(_load([], [], state="../etc").state_scope, "open")
+
+    def test_state_is_passed_through_to_the_fetchers(self):
+        seen: list[str] = []
+
+        def _fetch(*_args, **kwargs):
+            seen.append(kwargs.get("state", ""))
+            return [], _pagination()
+
+        load_linkage_snapshot(state="all", fetch_prs=_fetch, fetch_issues=_fetch)
+        self.assertEqual(seen, ["all", "all"])
+
+
+class TestPartialInventoryIsNotAnAbsenceClaim(unittest.TestCase):
+    """An empty edge list from a partial read must never read as 'no link'."""
+
+    def test_incomplete_pagination_marks_links_non_authoritative(self):
+        snapshot = _load([_issue(643)], [], complete=False)
+        self.assertFalse(snapshot.inventory_complete)
+        self.assertFalse(snapshot.issues[0].links_authoritative)
+
+    def test_complete_pagination_marks_links_authoritative(self):
+        snapshot = _load([_issue(643)], [], complete=True)
+        self.assertTrue(snapshot.inventory_complete)
+        self.assertTrue(snapshot.issues[0].links_authoritative)
+
+    def test_partial_window_renders_a_qualified_empty_cell(self):
+        html = render_linkage_page(_load([_issue(643)], [], complete=False))
+        self.assertIn("none found (partial inventory)", html)
+
+    def test_complete_window_renders_a_plain_none(self):
+        html = render_linkage_page(_load([_issue(643)], [], complete=True))
+        self.assertNotIn("partial inventory", html)
+        self.assertIn(">none<", html)
+
+    def test_missing_credentials_fail_closed_without_a_table(self):
+        with mock.patch(
+            "webui.linkage_loader._offline_test_mode", return_value=False
+        ), mock.patch("webui.linkage_loader.get_auth_header", return_value=""):
+            snapshot = load_linkage_snapshot()
+        self.assertFalse(snapshot.ok)
+        self.assertIn("credentials unavailable", snapshot.fetch_error)
+        html = render_linkage_page(snapshot)
+        self.assertIn("Linkage unavailable", html)
+        self.assertNotIn("Issues → pull requests", html)
+
+    def test_fetch_failure_is_reported_not_raised(self):
+        def _boom(*_args, **_kwargs):
+            raise RuntimeError("gitea 502")
+
+        snapshot = load_linkage_snapshot(fetch_prs=_boom, fetch_issues=_boom)
+        self.assertFalse(snapshot.ok)
+        self.assertIn("Gitea fetch failed", snapshot.fetch_error)
+
+
+class TestHandoffSummary(unittest.TestCase):
+    """AC2 — the latest canonical handoff is summarized for a focused thread."""
+
+    def test_latest_cth_wins(self):
+        summary = summarize_handoff([
+            _cth(1, created_at="2026-07-24T10:00:00Z", status="in progress",
+                 next_owner="author"),
+            _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
+                 next_owner="reviewer"),
+        ])
+        self.assertEqual(summary.comment_id, 2)
+        self.assertEqual(summary.status, "PR-open")
+        self.assertEqual(summary.next_owner, "reviewer")
+        self.assertTrue(summary.cth_type_known)
+
+    def test_thread_without_a_cth_summarizes_to_none(self):
+        self.assertIsNone(summarize_handoff([{"id": 1, "body": "ordinary comment"}]))
+
+    def test_unknown_heading_is_reported_not_republished(self):
+        summary = summarize_handoff([
+            {
+                "id": 5,
+                "created_at": "2026-07-25T10:00:00Z",
+                "user": {"login": "someone"},
+                "body": "<!-- cth:v1 -->\n## CTH: Totally Made Up\n\nStatus: odd\n",
+            }
+        ])
+        self.assertFalse(summary.cth_type_known)
+        self.assertEqual(summary.cth_type, "unrecognized")
+        self.assertNotIn("Totally Made Up", json.dumps(summary.to_dict()))
+
+    def test_focused_pr_loads_its_handoff(self):
+        snapshot = _load(
+            [_issue(643)],
+            [_pr(902, body="Closes #643")],
+            pr=902,
+            comment_source=lambda kind, number: [
+                _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
+                     next_owner="reviewer")
+            ],
+        )
+        self.assertEqual(snapshot.handoff_status.state, HANDOFF_LOADED)
+        self.assertEqual(snapshot.focus, ("pr", 902))
+        self.assertEqual(snapshot.prs[0].handoff.status, "PR-open")
+
+    def test_unfocused_rows_report_not_loaded_never_none(self):
+        snapshot = _load(
+            [_issue(643)],
+            [_pr(902, body="Closes #643"), _pr(903)],
+            pr=902,
+            comment_source=lambda kind, number: [],
+        )
+        other = next(node for node in snapshot.prs if node.number == 903)
+        self.assertIsNone(other.handoff)
+        self.assertEqual(other.handoff_status.state, HANDOFF_NOT_LOADED)
+        self.assertIn("not loaded", render_linkage_page(snapshot))
+
+    def test_no_focus_means_no_thread_is_claimed_handoff_free(self):
+        snapshot = _load([_issue(643)], [])
+        self.assertEqual(snapshot.handoff_status.state, HANDOFF_NOT_LOADED)
+        self.assertIn("thread-scoped", snapshot.handoff_status.reason)
+
+    def test_comment_source_failure_degrades_only_the_handoff(self):
+        def _boom(_kind, _number):
+            raise RuntimeError("comments 500")
+
+        snapshot = _load(
+            [_issue(643)], [_pr(902, body="Closes #643")], pr=902, comment_source=_boom
+        )
+        self.assertTrue(snapshot.ok)
+        self.assertEqual(snapshot.handoff_status.state, HANDOFF_UNAVAILABLE)
+        self.assertEqual(snapshot.issues[0].linked_prs, (902,))
+        self.assertIn("unavailable", render_linkage_page(snapshot))
+
+    def test_loaded_thread_with_no_cth_says_so_explicitly(self):
+        snapshot = _load(
+            [_issue(643)],
+            [_pr(902, body="Closes #643")],
+            pr=902,
+            comment_source=lambda kind, number: [{"id": 1, "body": "hi"}],
+        )
+        self.assertIn(
+            "no Canonical Thread Handoff comment found", render_linkage_page(snapshot)
+        )
+
+
+class TestDeepLinks(unittest.TestCase):
+    """AC3 — an external Gitea link is emitted only when permitted."""
+
+    def test_deep_links_are_withheld_by_default(self):
+        with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}):
+            snapshot = _load([_issue(643)], [])
+            html = render_linkage_page(snapshot)
+        self.assertFalse(snapshot.deep_links_enabled)
+        self.assertIsNone(snapshot.issues[0].deep_link)
+        self.assertIn("Gitea deep links are withheld", html)
+
+    def test_reveal_opt_in_emits_the_link(self):
+        with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}):
+            snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
+            html = render_linkage_page(snapshot)
+        self.assertTrue(snapshot.deep_links_enabled)
+        self.assertIn("/issues/643", snapshot.issues[0].deep_link)
+        self.assertIn("/pulls/902", snapshot.prs[0].deep_link)
+        self.assertIn(f'href="{snapshot.issues[0].deep_link}"', html)
+
+
+class TestRedactionBoundary(unittest.TestCase):
+    def test_secret_shaped_title_is_redacted_before_display(self):
+        snapshot = _load(
+            [_issue(643, title="token=ghp_thisisnotarealsecretvalue0001")], []
+        )
+        payload = json.dumps(snapshot_to_dict(snapshot))
+        self.assertNotIn("ghp_thisisnotarealsecretvalue0001", payload)
+        self.assertNotIn(
+            "ghp_thisisnotarealsecretvalue0001", render_linkage_page(snapshot)
+        )
+
+    def test_handoff_fields_are_redacted(self):
+        comment = _cth(
+            2, created_at="2026-07-25T10:00:00Z", status="ok", next_owner="reviewer"
+        )
+        comment["body"] += "\nDecision: password=hunter2hunter2\n"
+        snapshot = _load(
+            [_issue(643)],
+            [_pr(902, body="Closes #643")],
+            pr=902,
+            comment_source=lambda kind, number: [comment],
+        )
+        self.assertNotIn("hunter2hunter2", json.dumps(snapshot_to_dict(snapshot)))
+        self.assertNotIn("hunter2hunter2", render_linkage_page(snapshot))
+
+    def test_html_escapes_markup_in_a_title(self):
+        snapshot = _load([_issue(643, title="<script>alert(1)</script>")], [])
+        html = render_linkage_page(snapshot)
+        self.assertNotIn("<script>alert(1)</script>", html)
+        self.assertIn("&lt;script&gt;", html)
+
+
+class TestLinkageRoutes(unittest.TestCase):
+    def setUp(self):
+        self.snapshot = _load(
+            [_issue(643, labels=("status:ready",))],
+            [_pr(902, body="Closes #643", labels=("status:pr-open",))],
+        )
+        self.client = TestClient(create_app())
+
+    def test_page_renders_both_tables(self):
+        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
+            response = self.client.get("/gitea")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Issues → pull requests", response.text)
+        self.assertIn("Pull requests → issues", response.text)
+        self.assertIn("#643", response.text)
+
+    def test_api_exports_the_same_model(self):
+        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
+            response = self.client.get("/api/v1/gitea/linkage")
+        self.assertEqual(response.status_code, 200)
+        payload = response.json()
+        self.assertTrue(payload["ok"])
+        self.assertEqual(payload["issues"][0]["linked_prs"], [902])
+        self.assertEqual(payload["prs"][0]["links"][0]["issue_number"], 643)
+        self.assertEqual(payload["schema_version"], 1)
+
+    def test_api_declares_the_evidence_vocabulary(self):
+        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
+            payload = self.client.get("/api/v1/gitea/linkage").json()
+        names = {entry["name"] for entry in payload["evidence_kinds"]}
+        self.assertEqual(names, {EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE})
+
+    def test_api_fails_closed_with_a_non_200_when_the_read_failed(self):
+        failed = _load([], [], project_id="no-such-project")
+        with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
+            response = self.client.get("/api/v1/gitea/linkage")
+        self.assertEqual(response.status_code, 502)
+        self.assertFalse(response.json()["ok"])
+
+    def test_page_still_renders_when_the_read_failed(self):
+        failed = _load([], [], project_id="no-such-project")
+        with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
+            response = self.client.get("/gitea")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Linkage unavailable", response.text)
+
+    def test_query_parameters_reach_the_loader(self):
+        with mock.patch(
+            "webui.app.load_linkage_snapshot", return_value=self.snapshot
+        ) as loader:
+            self.client.get("/gitea?project=gitea-tools&state=all&pr=902")
+        loader.assert_called_once()
+        args, kwargs = loader.call_args
+        self.assertEqual(args[0], "gitea-tools")
+        self.assertEqual(kwargs["state"], "all")
+        self.assertEqual(kwargs["pr"], 902)
+        self.assertIsNone(kwargs["issue"])
+
+    def test_surface_stays_read_only(self):
+        for path in ("/gitea", "/api/v1/gitea/linkage"):
+            with self.subTest(path=path):
+                self.assertEqual(self.client.post(path).status_code, 405)
+
+    def test_nav_exposes_the_linkage_page_as_live(self):
+        self.assertIn("/gitea", nav_hrefs())
+        home = self.client.get("/").text
+        self.assertIn('href="/gitea"', home)
+        self.assertIn(">Gitea<", home)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_notifications.py

@@ -0,0 +1,465 @@
+"""Unit tests for Phase 3 Notifications and Human-Attention Console (#648)."""
+
+from __future__ import annotations
+
+import pytest
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.notifications import (
+    ATTENTION_HUMAN_REQUIRED,
+    ATTENTION_OPERATOR,
+    ATTENTION_ROUTINE,
+    CATEGORY_AUTH,
+    CATEGORY_BLOCKER,
+    CATEGORY_LEASE,
+    CATEGORY_SYSTEM,
+    CATEGORY_VALIDATION,
+    CATEGORY_WORKFLOW,
+    NotificationItem,
+    NotificationSnapshot,
+    classify_attention_event,
+    load_notifications_snapshot,
+    snapshot_to_dict,
+)
+from webui.notification_views import render_notifications_page
+from webui.project_registry import load_registry
+from webui.queue_loader import QueueItem, QueueSnapshot
+from webui.lease_loader import CollisionWarning, LeaseSnapshot
+from webui.system_health import DependencyProbe, SystemHealthSnapshot, VersionInfo, StaleRuntime
+
+
+def test_classify_attention_event_rules():
+    # 1. Critical escalation boundaries -> human-required
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_AUTH, "Auth error", "Unauthorized access attempt", is_auth_failure=True
+    )
+    assert att_cls == ATTENTION_HUMAN_REQUIRED
+    assert req_human is True
+
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_SYSTEM, "Hard stop", "Hard stop triggered", is_hard_stop=True
+    )
+    assert att_cls == ATTENTION_HUMAN_REQUIRED
+    assert req_human is True
+
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_VALIDATION, "Validation Error", "Report validation failed", is_validation_failure=True
+    )
+    assert att_cls == ATTENTION_HUMAN_REQUIRED
+    assert req_human is True
+
+    # 2. Operational issues -> operator
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_BLOCKER, "PR Blocked", "Merge conflict detected", is_blocker=True
+    )
+    assert att_cls == ATTENTION_OPERATOR
+    assert req_human is False
+
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_LEASE, "Lease Expired", "Session lease expired", is_stale=True
+    )
+    assert att_cls == ATTENTION_OPERATOR
+    assert req_human is False
+
+    # 3. Routine workflow transitions -> routine
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_WORKFLOW, "PR Active", "PR in review"
+    )
+    assert att_cls == ATTENTION_ROUTINE
+    assert req_human is False
+
+
+def test_notification_snapshot_aggregation():
+    reg = load_registry()
+    proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
+
+    mock_queue = QueueSnapshot(
+        project_id=proj_id,
+        repo_label="org/repo",
+        prs=(
+            QueueItem(
+                number=101,
+                title="Blocked PR",
+                badges=("blocked",),
+                extra={},
+            ),
+            QueueItem(
+                number=102,
+                title="Normal PR",
+                badges=("in-review",),
+                extra={},
+            ),
+        ),
+        issues=(),
+        pr_pagination=None,
+        issue_pagination=None,
+    )
+
+    mock_leases = LeaseSnapshot(
+        project_id=proj_id,
+        repo_label="org/repo",
+        issue_lock=None,
+        claim_inventory={},
+        reviewer_leases=(
+            {
+                "pr_number": 101,
+                "status": "expired",
+                "is_expired": True,
+            },
+        ),
+        duplicate_prs=(
+            CollisionWarning(
+                kind="duplicate_pr",
+                message="Multiple open PRs for issue #101",
+                issue_number=101,
+                pr_numbers=(101, 103),
+            ),
+        ),
+        duplicate_branches=(),
+        collision_history=(),
+        fetch_error=None,
+    )
+
+    mock_version = VersionInfo(
+        git_sha="abc1234",
+        git_describe="v1.0.0",
+        control_plane_schema_version=1,
+        python_version="3.11",
+        known=True,
+    )
+
+    mock_stale = StaleRuntime(
+        daemon_head="abc1234",
+        checkout_head="abc1234",
+        remote_head="abc1234",
+        stale=False,
+        determinable=True,
+        mutation_safe=True,
+        reasons=(),
+    )
+
+    mock_health = SystemHealthSnapshot(
+        status="degraded",
+        ready=False,
+        readiness_complete=True,
+        readiness_reasons=("Auth failure",),
+        service="webui",
+        mode="test",
+        version=mock_version,
+        started_at="2026-07-25T00:00:00Z",
+        uptime_seconds=100.0,
+        timestamp="2026-07-25T00:00:00Z",
+        deep_probes_requested=True,
+        dependencies=(
+            DependencyProbe(
+                name="auth_service",
+                kind="auth",
+                status="unauthorized",
+                detail="Token expired",
+                required=True,
+            ),
+        ),
+        mcp_namespaces=(),
+        stale_runtime=mock_stale,
+        probe_errors=(),
+    )
+
+    snapshot = load_notifications_snapshot(
+        proj_id,
+        load_queue=lambda _id: mock_queue,
+        load_leases=lambda **_kwargs: mock_leases,
+        load_health=lambda **_kwargs: mock_health,
+    )
+
+    assert snapshot.project_id == proj_id
+    assert snapshot.total_count == 5
+    assert snapshot.human_required_count >= 1  # auth probe failure
+    assert snapshot.operator_count >= 3  # blocked PR + expired lease + duplicate PR collision
+    assert snapshot.routine_count >= 1  # normal PR
+
+    # Inbox items should include operator and human-required items only
+    inbox_classes = {item.attention_class for item in snapshot.inbox_items}
+    assert ATTENTION_ROUTINE not in inbox_classes
+    assert ATTENTION_OPERATOR in inbox_classes
+    assert ATTENTION_HUMAN_REQUIRED in inbox_classes
+
+
+def test_snapshot_to_dict_and_redaction():
+    item = NotificationItem(
+        id="notif-1",
+        attention_class=ATTENTION_HUMAN_REQUIRED,
+        category=CATEGORY_AUTH,
+        title="Auth Error",
+        summary="Failed auth header: Bearer secret_token_12345",
+        work_kind="system",
+        work_number=None,
+        project_id="test-proj",
+        repo_label="org/repo",
+        created_at="2026-07-25T16:00:00Z",
+        requires_human=True,
+    )
+    snap = NotificationSnapshot(
+        project_id="test-proj",
+        repo_label="org/repo",
+        items=(item,),
+        human_required_count=1,
+        operator_count=0,
+        routine_count=0,
+        total_count=1,
+    )
+
+    data = snapshot_to_dict(snap)
+    assert data["project_id"] == "test-proj"
+    assert data["human_required_count"] == 1
+    assert len(data["inbox_items"]) == 1
+
+    # Redaction test
+    summary = data["inbox_items"][0]["summary"]
+    assert "secret_token_12345" not in summary
+    assert "<redacted>" in summary or "Bearer" in summary
+
+
+def test_notifications_html_views():
+    item = NotificationItem(
+        id="notif-1",
+        attention_class=ATTENTION_HUMAN_REQUIRED,
+        category=CATEGORY_AUTH,
+        title="Critical Auth Failure",
+        summary="Auth failure details",
+        work_kind="issue",
+        work_number=42,
+        project_id="test-proj",
+        repo_label="org/repo",
+        created_at="2026-07-25T16:00:00Z",
+        requires_human=True,
+    )
+    snap = NotificationSnapshot(
+        project_id="test-proj",
+        repo_label="org/repo",
+        items=(item,),
+        human_required_count=1,
+        operator_count=0,
+        routine_count=0,
+        total_count=1,
+    )
+
+    html = render_notifications_page(snap, filter_class="inbox")
+    assert "Notifications &amp; Attention Inbox" in html or "Notifications & Attention Inbox" in html
+    assert "Critical Auth Failure" in html
+    assert "HUMAN REQUIRED" in html
+    assert "Human Required" in html
+
+
+def test_notifications_app_routes():
+    app = create_app()
+    client = TestClient(app)
+
+    # 1. HTML Route
+    res = client.get("/notifications")
+    assert res.status_code == 200
+    assert "Notifications" in res.text
+    assert "Attention Inbox" in res.text
+
+    # 2. API Route /api/v1/notifications
+    res_api = client.get("/api/v1/notifications")
+    assert res_api.status_code == 200
+    json_data = res_api.json()
+    assert "human_required_count" in json_data
+    assert "operator_count" in json_data
+    assert "routine_count" in json_data
+    assert "inbox_items" in json_data
+
+    # 3. Compatibility Alias /api/notifications
+    res_alias = client.get("/api/notifications")
+    assert res_alias.status_code == 200
+    assert res_alias.json()["project_id"] == json_data["project_id"]
+
+
+def test_classify_ignores_human_authored_title_and_summary_keywords():
+    """B1: keywords in human-authored titles must not escalate routine work (#905)."""
+    # Routine transition whose title/summary mention critical-boundary words
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_WORKFLOW,
+        "record irrecoverable decision lock provenance",
+        "PR #999 'record irrecoverable decision lock provenance' is in routine state in-review.",
+    )
+    assert att_cls == ATTENTION_ROUTINE
+    assert req_human is False
+
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_WORKFLOW,
+        "fix unauthorized token path",
+        "Issue #1 'fix unauthorized token path' state: claimed. hard stop docs only.",
+    )
+    assert att_cls == ATTENTION_ROUTINE
+    assert req_human is False
+
+    # Structured flags still escalate (machine-driven)
+    att_cls, req_human = classify_attention_event(
+        CATEGORY_SYSTEM,
+        "anything",
+        "anything with hard stop in text",
+        is_hard_stop=True,
+    )
+    assert att_cls == ATTENTION_HUMAN_REQUIRED
+    assert req_human is True
+
+
+def test_notification_ids_are_unique_across_probe_errors_and_collisions():
+    """B2: published notification ids must be unique within a snapshot (#905)."""
+    reg = load_registry()
+    proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
+
+    mock_queue = QueueSnapshot(
+        project_id=proj_id,
+        repo_label="org/repo",
+        prs=(),
+        issues=(),
+        pr_pagination=None,
+        issue_pagination=None,
+    )
+    mock_leases = LeaseSnapshot(
+        project_id=proj_id,
+        repo_label="org/repo",
+        issue_lock=None,
+        claim_inventory={},
+        reviewer_leases=(),
+        duplicate_prs=(
+            CollisionWarning(
+                kind="duplicate_pr",
+                message="Multiple open PRs for issue #10",
+                issue_number=10,
+                pr_numbers=(10, 11),
+            ),
+            CollisionWarning(
+                kind="duplicate_branch",
+                message="Another collision without issue",
+                issue_number=None,
+                pr_numbers=(12, 13),
+            ),
+            CollisionWarning(
+                kind="duplicate_pr",
+                message="Second issue collision",
+                issue_number=10,
+                pr_numbers=(14, 15),
+            ),
+        ),
+        duplicate_branches=(),
+        collision_history=(),
+        fetch_error=None,
+    )
+    mock_version = VersionInfo(
+        git_sha="abc1234",
+        git_describe="v1.0.0",
+        control_plane_schema_version=1,
+        python_version="3.11",
+        known=True,
+    )
+    mock_stale = StaleRuntime(
+        daemon_head="abc1234",
+        checkout_head="abc1234",
+        remote_head="abc1234",
+        stale=False,
+        determinable=True,
+        mutation_safe=True,
+        reasons=(),
+    )
+    mock_health = SystemHealthSnapshot(
+        status="degraded",
+        ready=False,
+        readiness_complete=True,
+        readiness_reasons=(),
+        service="webui",
+        mode="test",
+        version=mock_version,
+        started_at="2026-07-25T00:00:00Z",
+        uptime_seconds=100.0,
+        timestamp="2026-07-25T00:00:00Z",
+        deep_probes_requested=True,
+        dependencies=(),
+        mcp_namespaces=(),
+        stale_runtime=mock_stale,
+        probe_errors=("error alpha", "error beta"),
+    )
+
+    snapshot = load_notifications_snapshot(
+        proj_id,
+        load_queue=lambda _id: mock_queue,
+        load_leases=lambda **_kwargs: mock_leases,
+        load_health=lambda **_kwargs: mock_health,
+    )
+    ids = [item.id for item in snapshot.items]
+    assert len(ids) == len(set(ids)), f"duplicate notification ids: {ids}"
+    assert any(i.startswith(f"notif-sys-err-{proj_id}-") for i in ids)
+    assert any(i.startswith("notif-collision-") for i in ids)
+
+
+def test_probe_errors_do_not_set_fetch_error():
+    """B3: probe_errors must not be reported as fetch_error (#905)."""
+    reg = load_registry()
+    proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
+
+    mock_queue = QueueSnapshot(
+        project_id=proj_id,
+        repo_label="org/repo",
+        prs=(),
+        issues=(),
+        pr_pagination=None,
+        issue_pagination=None,
+        fetch_error=None,
+    )
+    mock_leases = LeaseSnapshot(
+        project_id=proj_id,
+        repo_label="org/repo",
+        issue_lock=None,
+        claim_inventory={},
+        reviewer_leases=(),
+        duplicate_prs=(),
+        duplicate_branches=(),
+        collision_history=(),
+        fetch_error=None,
+    )
+    mock_version = VersionInfo(
+        git_sha="abc1234",
+        git_describe="v1.0.0",
+        control_plane_schema_version=1,
+        python_version="3.11",
+        known=True,
+    )
+    mock_stale = StaleRuntime(
+        daemon_head="abc1234",
+        checkout_head="abc1234",
+        remote_head="abc1234",
+        stale=False,
+        determinable=True,
+        mutation_safe=True,
+        reasons=(),
+    )
+    mock_health = SystemHealthSnapshot(
+        status="degraded",
+        ready=False,
+        readiness_complete=True,
+        readiness_reasons=(),
+        service="webui",
+        mode="test",
+        version=mock_version,
+        started_at="2026-07-25T00:00:00Z",
+        uptime_seconds=100.0,
+        timestamp="2026-07-25T00:00:00Z",
+        deep_probes_requested=True,
+        dependencies=(),
+        mcp_namespaces=(),
+        stale_runtime=mock_stale,
+        probe_errors=("probe blew up",),
+    )
+
+    snapshot = load_notifications_snapshot(
+        proj_id,
+        load_queue=lambda _id: mock_queue,
+        load_leases=lambda **_kwargs: mock_leases,
+        load_health=lambda **_kwargs: mock_health,
+    )
+    assert snapshot.fetch_error is None
+    # probe errors still appear as items
+    assert any("probe blew up" in item.summary for item in snapshot.items)

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="[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()