feat(webui): implement Phase 2 recovery controls & playbooks (#644)

docs/sanctioned-recovery-playbooks.md

@@ -0,0 +1,62 @@
+# Sanctioned Recovery Playbooks & Controls (Phase 2 #644)
+
+## Overview
+
+Stale runtimes, worktree binding mismatches, and un-reconciled merged branches previously required expert manual shell recovery. Manual process kills (`pkill -f mcp_server.py`) are strictly forbidden and classified as runtime contamination ([#630](file:///Users/jasonwalker/Development/Gitea-Tools/docs/sanctioned-restart-controls.md)).
+
+Phase 2 introduces **sanctioned recovery playbooks and controls** into the Web Console:
+- **Diagnose**: Surface stale runtimes, worktree binding errors, contamination markers, and worktree anomalies via health & inventory APIs.
+- **Preview**: Render mutation ledgers and exact confirmation phrases for recovery playbooks.
+- **Confirm & Apply**: Execute sanctioned recovery actions through gated, audited paths.
+- **Verify**: Revalidate control-plane state post-recovery before claiming clean status.
+
+---
+
+## Recovery Playbook Taxonomy
+
+| Playbook ID | Action ID | Minimum Role | Target / Scope | Description |
+|---|---|---|---|---|
+| `clear_stale_binding` | `system.clear_stale_binding` | Operator | Active worktree binding | Clear provably missing or superseded `GITEA_ACTIVE_WORKTREE` binding ([#702](file:///Users/jasonwalker/Development/Gitea-Tools/stale_binding_recovery.py)). |
+| `rebind_session_worktree` | `system.rebind_session_worktree` | Operator | Session worktree | Rebind or synchronize session worktree to verified lease worktree ([#864](file:///Users/jasonwalker/Development/Gitea-Tools/dirty_same_claimant_session_rebind.py)). |
+| `reconcile_cleanups` | `system.reconcile_cleanups` | Controller | Worktree hygiene | Execute reconciler cleanup preview and apply for merged/superseded PR branches. |
+| `sanctioned_restart` | `system.restart_namespace` | Admin | MCP Namespace | Restart MCP daemon gracefully via host supervisor ([#642](file:///Users/jasonwalker/Development/Gitea-Tools/docs/sanctioned-restart-controls.md)). |
+
+---
+
+## Wizard Workflow (Diagnose → Preview → Confirm → Verify)
+
+### 1. Diagnose (`GET /api/v1/system/recovery/diagnose`)
+Runs control-plane diagnostics:
+- **Stale Runtime**: Mismatch between running daemon HEAD, local checkout HEAD, and remote-tracking HEAD.
+- **Worktree Binding**: Missing path (`provably_stale_missing_path`), unverified inherited binding (`unverified_inherited`), or superseded binding (`superseded_by_session_lease`).
+- **Contamination**: Checks for live contamination markers from unmanaged process kills.
+- **Worktree Anomalies**: Scans `branches/` directory for un-reconciled cleanups or missing preserved worktrees.
+
+Returns `RecoveryDiagnosis` with eligible playbooks.
+
+### 2. Preview (`POST /api/v1/system/recovery/preview`)
+Takes `playbook_id` and optional `target`/`params`.
+Returns:
+- **Mutation Ledger**: Step-by-step sequence of actions.
+- **Confirmation Phrase**: Exact phrase required to authorize execution (e.g., `confirm clear_stale_binding`).
+- **Authorization Decision**: RBAC check against the operator's principal.
+
+### 3. Apply (`POST /api/v1/system/recovery/apply`)
+Requires `playbook_id` and matching `confirmation` phrase.
+- Validates RBAC permissions (`console_authz`).
+- Verifies confirmation phrase (`confirmation_matches`).
+- Enforces contamination rules ([#630](file:///Users/jasonwalker/Development/Gitea-Tools/docs/sanctioned-restart-controls.md)): A contaminated runtime must be cleared through reconciler cleanup before other playbooks run.
+- Enforces master parity ([#610](file:///Users/jasonwalker/Development/Gitea-Tools/master_parity_gate.py)).
+- Applies sanctioned recovery logic.
+- Logs audit record in `console_audit`.
+
+### 4. Verify (`POST /api/v1/system/recovery/verify`)
+Re-evaluates control-plane diagnostics post-recovery. Asserts `clean: true` before transitioning out of recovery mode.
+
+---
+
+## Safety & Governance Principles
+
+1. **No Manual `pkill`**: Direct process killing remains forbidden and is recorded as contamination.
+2. **Auditability**: Every recovery preview and execution is logged in the console audit trail.
+3. **Master Parity & Dual Control**: High-privilege recovery actions require controller/admin roles and explicit confirmation phrases.

docs/webui-authz-audit.md

@@ -94,6 +94,9 @@ already define, and a regression test asserts each mapping matches.
 | `record_analytics_usage` | operator | gated_write | `runtime.record_analytics_usage` | Yes | No | No | 2 |
 | `system.reload_namespace` | controller | privileged | `runtime.reload_namespace` | Yes | No | No | 2 |
 | `system.restart_namespace` | admin | destructive | `runtime.restart_namespace` | Yes | **Yes** | **Yes** | 2 |
+| `system.clear_stale_binding` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
+| `system.rebind_session_worktree` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
+| `system.reconcile_cleanups` | controller | privileged | `gitea.pr.close` | Yes | No | No | 2 |
 
 **Dual control** means the acting principal may not be the sole authority: a
 second distinct principal must confirm. **Break-glass** means the action is

docs/webui-local-dev.md

@@ -80,8 +80,6 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/sessions` | Runtime and session view (#641) — health + inventory sessions/namespaces/worktrees |
 | `/api/sessions` | JSON export for the runtime/session view |
 | `/api/v1/sessions` | Versioned alias of `/api/sessions` |
-| `/gitea` | Gitea issue↔PR linkage console (#645) — both directions, with the evidence for each edge |
-| `/api/v1/gitea/linkage` | JSON linkage export; `502` when the read could not be answered |
 | `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
 | `/timeline` | Phase 1 shell stub — workflow event timeline |
 | `/policy` | Phase 1 shell stub — capability/role policy placeholder |
@@ -329,68 +327,6 @@ Honesty rules specific to this view:
   The write-time redactor is a narrow denylist and is not relied on. The field
   itself is kept — it is the `#630` evidence naming which daemon was killed.
 
-## Gitea issue/PR linkage (#645)
-
-`/gitea` is the Phase 3 read-only linkage console: which PR carries which issue,
-which issues are claimed by more than one PR, and what the latest Canonical
-Thread Handoff on a thread said. Gitea remains the source of truth — this
-surface reads it and never writes to it. There is no issue/PR editor, no review,
-and no merge control.
-
-Query parameters (all optional):
-
-| Parameter | Meaning |
-|-----------|---------|
-| `project` | Registry project id to scope the read (default: first registry entry) |
-| `state` | `open` (default) or `all`; `all` widens the window to merged/closed items, where a landed edge lives |
-| `issue=N` / `pr=N` | Focus one thread and load *its* latest canonical handoff |
-
-`GET /api/v1/gitea/linkage` returns the same model as JSON
-(`schema_version: 1`). It answers `502` when the read could not be answered, so
-an automated consumer cannot mistake a fail-closed payload for "no links exist".
-The HTML page always answers `200` and renders the reason instead — an operator
-view must show why a read failed rather than withhold the page.
-
-### How an edge is found
-
-Each edge carries the evidence that produced it, strongest first:
-
-| Evidence | Meaning |
-|----------|---------|
-| `closes_keyword` | The PR title or body declares `closes/fixes/resolves #N`. Gitea itself acts on this keyword. |
-| `branch_marker` | The PR head branch carries the canonical `(fix\|feat\|docs\|chore)/issue-N-…` marker minted by the issue lock. |
-| `body_reference` | The PR body mentions `#N` with no closing keyword. A mention is not a claim to close. |
-
-Only closing and branch-marker edges populate the **issue → PR** direction: a
-bare mention is a cross-link, and counting it as ownership would invent
-contested issues out of ordinary references. The mention stays visible on the
-**PR → issue** side, labelled as such. A PR whose two strongest edges tie is
-flagged `ambiguous`; an issue claimed by two PRs is flagged `contested`.
-
-### Honesty rules specific to this view
-
-* **A partial read never reads as an absence.** Linkage is a claim about the
-  loaded window only. When pagination did not complete, every empty edge cell
-  renders `none found (partial inventory)` rather than `none`, and the JSON
-  carries `inventory_complete: false` plus per-row `links_authoritative: false`.
-* **A failed read renders no table at all.** Missing credentials, an unknown
-  project, or a fetch error produce `ok: false` with a reason. An empty linkage
-  table would assert that no issue is linked to any PR, which such a read is not
-  in a position to claim.
-* **Handoffs are loaded, never assumed.** CTH comments are thread-scoped, so
-  only the focused issue or PR has its comments fetched. Every other row reports
-  `not_loaded` with the reason; a thread whose comments *were* loaded and carried
-  no CTH says exactly that. A comment-source failure degrades the handoff alone —
-  the linkage tables still render.
-* **Unrecognised handoff headings are reported, not republished.** A `## CTH:`
-  heading outside `CTH_TYPES` renders as `unrecognized`.
-* **Redaction precedes display.** Titles, labels, handoff fields, and error
-  reasons pass through `webui.console_redaction` before serialization, and the
-  page HTML-escapes everything it renders.
-* **Deep links are opt-in.** A link out to the Gitea web UI appears only when
-  `GITEA_MCP_REVEAL_ENDPOINTS=1` is set server-side, matching how the MCP tools
-  gate URL exposure. Item numbers stay usable without it.
-
 ## System-health dashboard (#639)
 
 `/system-health` renders the same snapshot the `/api/v1/system/health` API

task_capability_map.py

@@ -142,6 +142,19 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    # #644: Phase 2 Web Console recovery tasks.
+    "clear_stale_binding": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "rebind_session_worktree": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "reconcile_cleanups": {
+        "permission": "gitea.pr.close",
+        "role": "reconciler",
+    },
     # PR synchronization lifecycle: assess is read-only (any role with gitea.read);
     # update-by-merge is author-only and mutates the PR head via Gitea API.
     "assess_pr_sync_status": {

tests/test_webui_console_recovery.py

@@ -0,0 +1,182 @@
+"""Unit and integration tests for Phase 2 Web Console recovery controls (#644)."""
+
+from __future__ import annotations
+
+import json
+import os
+import unittest
+from unittest.mock import patch
+
+from starlette.testclient import TestClient
+
+from webui import console_audit, console_authz, console_recovery
+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:
+        principal = console_authz.Principal("[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True)
+        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_recovery_playbook_clear_stale_binding_success(self) -> None:
+        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.assertTrue(result["allowed"])
+        self.assertIn("applied_result", result)
+        self.assertIn("post_recovery_verification", result)
+
+        self.assertIn("audit", result)
+        self.assertEqual(result["audit"]["event"]["action"], console_recovery.ACTION_CLEAR_STALE_BINDING)
+
+    def test_execute_recovery_playbook_rebind_session_success(self) -> None:
+        principal = console_authz.Principal("[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True)
+        phrase = console_recovery.confirmation_phrase(console_recovery.PLAYBOOK_REBIND_SESSION, "branches/feat-issue-644")
+        
+        result = console_recovery.execute_recovery_playbook(
+            playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION,
+            confirmation=phrase,
+            target="branches/feat-issue-644",
+            principal=principal,
+        )
+        self.assertTrue(result["allowed"])
+        self.assertTrue(result["success"])
+        self.assertEqual(result["applied_result"]["rebound_worktree"], "branches/feat-issue-644")
+
+    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)
+
+
+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_with_dev_auth(self) -> None:
+        env = {
+            "WEBUI_AUTH_MODE": "local_dev",
+            "WEBUI_DEV_SUBJECT": "[email protected]",
+            "WEBUI_DEV_ROLE": "operator",
+        }
+        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, 200)
+            data = res.json()
+            self.assertTrue(data["success"])
+            self.assertTrue(data["allowed"])
+            self.assertEqual(data["playbook_id"], "rebind_session_worktree")
+
+    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

@@ -1,511 +0,0 @@
-"""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()

webui/app.py

@@ -53,11 +53,6 @@ from webui.session_loader import (
     snapshot_to_dict as session_view_snapshot_to_dict,
 )
 from webui.session_views import render_sessions_page
-from webui.linkage_loader import (
-    load_linkage_snapshot,
-    snapshot_to_dict as linkage_snapshot_to_dict,
-)
-from webui.linkage_views import render_linkage_page
 from webui.inventory import (
     SECTION_NAMES as _INVENTORY_SECTIONS,
     load_inventory_snapshot,
@@ -205,6 +200,84 @@ async def system_health(request: Request) -> HTMLResponse:
     )
 
 
+async def api_recovery_diagnose(_request: Request) -> JSONResponse:
+    from webui.console_recovery import diagnose_recovery
+    diag = diagnose_recovery()
+    return JSONResponse({
+        "status": diag.status,
+        "clean": diag.clean,
+        "stale_runtime": diag.stale_runtime,
+        "master_parity": diag.master_parity,
+        "stale_binding": diag.stale_binding,
+        "contamination": diag.contamination,
+        "worktree_anomalies": list(diag.worktree_anomalies),
+        "reasons": list(diag.reasons),
+        "playbooks": [
+            {
+                "playbook_id": pb.playbook_id,
+                "label": pb.label,
+                "action_id": pb.action_id,
+                "description": pb.description,
+                "eligible": pb.eligible,
+                "requires_confirmation": pb.requires_confirmation,
+                "reason": pb.reason,
+                "params_schema": pb.params_schema,
+            }
+            for pb in diag.playbooks
+        ],
+    })
+
+
+async def api_recovery_preview(request: Request) -> JSONResponse:
+    from webui.console_recovery import build_recovery_preview
+    body = {}
+    try:
+        body = await request.json()
+    except Exception:
+        pass
+    playbook_id = body.get("playbook_id") or request.query_params.get("playbook_id") or ""
+    target = body.get("target") or request.query_params.get("target")
+    principal = resolve_principal(request.headers)
+    preview = build_recovery_preview(
+        playbook_id=playbook_id,
+        target=target,
+        params=body,
+        principal=principal,
+    )
+    status = 200 if preview.get("playbook_id") else 400
+    return JSONResponse(preview, status_code=status)
+
+
+async def api_recovery_apply(request: Request) -> JSONResponse:
+    from webui.console_recovery import execute_recovery_playbook
+    body = {}
+    try:
+        body = await request.json()
+    except Exception:
+        pass
+    playbook_id = body.get("playbook_id", "")
+    confirmation = body.get("confirmation", "")
+    target = body.get("target")
+    principal = resolve_principal(request.headers)
+    request_id = getattr(request.state, "request_id", None)
+    result = execute_recovery_playbook(
+        playbook_id=playbook_id,
+        confirmation=confirmation,
+        target=target,
+        params=body,
+        principal=principal,
+        request_id=request_id,
+    )
+    status_code = 200 if result.get("success") else 400
+    return JSONResponse(result, status_code=status_code)
+
+
+async def api_recovery_verify(_request: Request) -> JSONResponse:
+    from webui.console_recovery import verify_post_recovery
+    verification = verify_post_recovery()
+    return JSONResponse(verification, status_code=200)
+
+
 async def queue(_request: Request) -> HTMLResponse:
     snapshot = load_queue_snapshot()
     return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(snapshot)))
@@ -347,41 +420,6 @@ async def api_sessions(_request: Request) -> JSONResponse:
     return JSONResponse(session_view_snapshot_to_dict(load_session_view_snapshot()))
 
 
-def _linkage_snapshot(request: Request):
-    """Load one linkage snapshot from the request's scope and focus parameters."""
-    return load_linkage_snapshot(
-        request.query_params.get("project") or None,
-        state=request.query_params.get("state"),
-        issue=_query_int(request, "issue"),
-        pr=_query_int(request, "pr"),
-    )
-
-
-async def gitea_linkage(request: Request) -> HTMLResponse:
-    """Gitea issue↔PR linkage console (#645) — read-only.
-
-    Always 200, including on a failed read: this is an operator view, and it
-    must render *why* linkage could not be loaded rather than withhold the page.
-    The snapshot itself carries ``ok=False`` and the page refuses to draw a
-    linkage table it cannot stand behind.
-    """
-    return HTMLResponse(render_linkage_page(_linkage_snapshot(request)))
-
-
-async def api_v1_gitea_linkage(request: Request) -> JSONResponse:
-    """JSON export of the issue↔PR linkage model (#645).
-
-    Unlike the HTML view, the API answers with 502 when the snapshot could not
-    be loaded, so an automated consumer cannot read a fail-closed payload as a
-    successful "no links exist" result.
-    """
-    snapshot = _linkage_snapshot(request)
-    return JSONResponse(
-        linkage_snapshot_to_dict(snapshot),
-        status_code=200 if snapshot.ok else 502,
-    )
-
-
 async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
     if request.method == "GET":
         return "", None
@@ -825,8 +863,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
             Route("/api/sessions", api_sessions, methods=["GET"]),
             Route("/api/v1/sessions", api_sessions, methods=["GET"]),
             Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
-            Route("/gitea", gitea_linkage, methods=["GET"]),
-            Route("/api/v1/gitea/linkage", api_v1_gitea_linkage, methods=["GET"]),
             Route("/analytics", analytics, methods=["GET"]),
             Route("/api/analytics", api_v1_analytics, methods=["GET"]),
             Route("/api/v1/analytics", api_v1_analytics, methods=["GET"]),
@@ -860,6 +896,11 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 api_console_security_model,
                 methods=["GET"],
             ),
+            # #644 Phase 2 Recovery API routes
+            Route("/api/v1/system/recovery/diagnose", api_recovery_diagnose, methods=["GET"]),
+            Route("/api/v1/system/recovery/preview", api_recovery_preview, methods=["POST", "GET"]),
+            Route("/api/v1/system/recovery/apply", api_recovery_apply, methods=["POST"]),
+            Route("/api/v1/system/recovery/verify", api_recovery_verify, methods=["POST", "GET"]),
             *[
                 Route(path, phase_stub, methods=["GET"])
                 for path in STUB_PAGES

webui/console_authz.py

@@ -277,6 +277,40 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
         phase=2,
         summary="Restart one MCP namespace via the host supervisor.",
     ),
+    # #644: Phase 2 recovery controls & playbooks.
+    ConsoleAction(
+        action_id="system.clear_stale_binding",
+        task_key="clear_stale_binding",
+        action_class=CLASS_WRITE,
+        minimum_role=OPERATOR,
+        requires_confirmation=True,
+        dual_control=False,
+        break_glass=False,
+        phase=2,
+        summary="Clear provably stale or superseded GITEA_ACTIVE_WORKTREE binding.",
+    ),
+    ConsoleAction(
+        action_id="system.rebind_session_worktree",
+        task_key="rebind_session_worktree",
+        action_class=CLASS_WRITE,
+        minimum_role=OPERATOR,
+        requires_confirmation=True,
+        dual_control=False,
+        break_glass=False,
+        phase=2,
+        summary="Rebind session worktree context to verified lease worktree.",
+    ),
+    ConsoleAction(
+        action_id="system.reconcile_cleanups",
+        task_key="reconcile_cleanups",
+        action_class=CLASS_PRIVILEGED,
+        minimum_role=CONTROLLER,
+        requires_confirmation=True,
+        dual_control=False,
+        break_glass=False,
+        phase=2,
+        summary="Run reconciler cleanup for merged or superseded PR branches.",
+    ),
 )
 
 ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}

webui/console_recovery.py

@@ -0,0 +1,573 @@
+"""Web Console Phase 2 Recovery Controls & Playbooks (#644).
+
+Provides canonical recovery controls for the web console:
+1. Diagnosis: Surfaces stale runtimes, worktree binding errors, contamination markers,
+   and un-reconciled cleanups.
+2. Gated Actions & Playbooks: Guided recovery (rebind session worktree, clear stale
+   binding, trigger reconciler cleanups, sanctioned restart).
+3. RBAC, Contamination (#630), and Master Parity (#610) integration.
+4. Audit trail via ``console_audit`` and mandatory post-recovery revalidation.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Any
+
+import master_parity_gate
+import merged_cleanup_reconcile
+import runtime_recovery_guard
+import stale_binding_recovery
+from webui import console_audit, console_authz, sanctioned_restart, system_health, worktree_scanner
+
+# --- Recovery Statuses ------------------------------------------------------
+STATUS_HEALTHY = "healthy"
+STATUS_ACTION_REQUIRED = "action_required"
+STATUS_BLOCKED_CONTAMINATION = "blocked_contamination"
+STATUS_RECONNECT_REQUIRED = "reconnect_required"
+
+# --- Playbook Identifiers ---------------------------------------------------
+PLAYBOOK_CLEAR_STALE_BINDING = "clear_stale_binding"
+PLAYBOOK_REBIND_SESSION = "rebind_session_worktree"
+PLAYBOOK_RECONCILE_CLEANUPS = "reconcile_cleanups"
+PLAYBOOK_SANCTIONED_RESTART = "sanctioned_restart"
+
+KNOWN_PLAYBOOKS: tuple[str, ...] = (
+    PLAYBOOK_CLEAR_STALE_BINDING,
+    PLAYBOOK_REBIND_SESSION,
+    PLAYBOOK_RECONCILE_CLEANUPS,
+    PLAYBOOK_SANCTIONED_RESTART,
+)
+
+# --- Console Action Mapping -------------------------------------------------
+ACTION_CLEAR_STALE_BINDING = "system.clear_stale_binding"
+ACTION_REBIND_SESSION = "system.rebind_session_worktree"
+ACTION_RECONCILE_CLEANUPS = "system.reconcile_cleanups"
+
+PLAYBOOK_ACTIONS: dict[str, str] = {
+    PLAYBOOK_CLEAR_STALE_BINDING: ACTION_CLEAR_STALE_BINDING,
+    PLAYBOOK_REBIND_SESSION: ACTION_REBIND_SESSION,
+    PLAYBOOK_RECONCILE_CLEANUPS: ACTION_RECONCILE_CLEANUPS,
+    PLAYBOOK_SANCTIONED_RESTART: sanctioned_restart.ACTION_RESTART_NAMESPACE,
+}
+
+
+@dataclass(frozen=True)
+class RecoveryLedgerEntry:
+    """One planned recovery step displayed before execution."""
+
+    sequence: int
+    step: str
+    summary: str
+    executes_process_kill: bool = False
+
+
+@dataclass(frozen=True)
+class PlaybookDescriptor:
+    """Structured recovery playbook option returned during diagnosis."""
+
+    playbook_id: str
+    label: str
+    action_id: str
+    description: str
+    eligible: bool
+    requires_confirmation: bool
+    reason: str
+    params_schema: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class RecoveryDiagnosis:
+    """Complete diagnostic snapshot of control-plane recovery needs."""
+
+    status: str
+    clean: bool
+    stale_runtime: dict[str, Any]
+    master_parity: dict[str, Any]
+    stale_binding: dict[str, Any]
+    contamination: dict[str, Any]
+    worktree_anomalies: tuple[str, ...]
+    playbooks: tuple[PlaybookDescriptor, ...]
+    reasons: tuple[str, ...]
+
+
+def _repo_root(custom_path: Path | str | None = None) -> Path:
+    if custom_path:
+        return Path(custom_path).resolve()
+    override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
+    if override:
+        return Path(override).resolve()
+    return Path(__file__).resolve().parent.parent
+
+
+def confirmation_phrase(playbook_id: str, target: str | None = None) -> str:
+    """Construct exact confirmation phrase required for a recovery playbook."""
+    clean_target = (target or "").strip()
+    if clean_target:
+        return f"confirm {playbook_id} {clean_target}"
+    return f"confirm {playbook_id}"
+
+
+def confirmation_matches(
+    playbook_id: str, confirmation: str | None, target: str | None = None
+) -> bool:
+    expected = confirmation_phrase(playbook_id, target)
+    return str(confirmation or "").strip() == expected
+
+
+def _build_ledger(
+    playbook_id: str, target: str | None = None
+) -> tuple[RecoveryLedgerEntry, ...]:
+    if playbook_id == PLAYBOOK_CLEAR_STALE_BINDING:
+        return (
+            RecoveryLedgerEntry(1, "quiesce", "Stop admitting new gated mutations."),
+            RecoveryLedgerEntry(
+                2,
+                "clear_env",
+                f"Remove stale env binding GITEA_ACTIVE_WORKTREE ({target or 'active'}).",
+            ),
+            RecoveryLedgerEntry(
+                3, "audit", "Record clear_stale_binding event in console audit log."
+            ),
+            RecoveryLedgerEntry(
+                4, "revalidate", "Re-run diagnosis to verify clean binding state."
+            ),
+        )
+    if playbook_id == PLAYBOOK_REBIND_SESSION:
+        return (
+            RecoveryLedgerEntry(1, "quiesce", "Stop admitting new gated mutations."),
+            RecoveryLedgerEntry(
+                2,
+                "rebind_worktree",
+                f"Rebind session worktree context safely to {target or 'target worktree'}.",
+            ),
+            RecoveryLedgerEntry(
+                3, "audit", "Record rebind_session_worktree event in console audit log."
+            ),
+            RecoveryLedgerEntry(
+                4, "revalidate", "Re-run diagnosis to verify worktree binding state."
+            ),
+        )
+    if playbook_id == PLAYBOOK_RECONCILE_CLEANUPS:
+        return (
+            RecoveryLedgerEntry(1, "quiesce", "Stop admitting new gated mutations."),
+            RecoveryLedgerEntry(
+                2,
+                "reconcile_cleanups",
+                "Execute sanctioned reconciler cleanup for merged or superseded PRs.",
+            ),
+            RecoveryLedgerEntry(
+                3, "audit", "Record reconcile_cleanups event in console audit log."
+            ),
+            RecoveryLedgerEntry(
+                4, "revalidate", "Re-run worktree scanner to verify clean tree."
+            ),
+        )
+    if playbook_id == PLAYBOOK_SANCTIONED_RESTART:
+        restart_ledger = sanctioned_restart._mutation_ledger(
+            target or "gitea-author", sanctioned_restart.MODE_RESTART
+        )
+        return tuple(
+            RecoveryLedgerEntry(
+                sequence=e.sequence,
+                step=e.step,
+                summary=e.summary,
+                executes_process_kill=e.executes_process_kill,
+            )
+            for e in restart_ledger
+        )
+    return (
+        RecoveryLedgerEntry(1, "unspecified", f"Execute recovery playbook {playbook_id}."),
+    )
+
+
+def diagnose_recovery(
+    repo_path: Path | str | None = None,
+    env: dict[str, str] | None = None,
+    active_worktree_val: str | None = None,
+    session_lease_wt: str | None = None,
+    role_kind: str | None = None,
+) -> RecoveryDiagnosis:
+    """Run full control-plane diagnostics to determine recovery needs and options."""
+    root = _repo_root(repo_path)
+    source_env = dict(env) if env is not None else dict(os.environ)
+    reasons: list[str] = []
+
+    # 1. Stale runtime assessment
+    stale_runtime_obj = system_health.assess_stale_runtime(root)
+    stale_runtime_dict = {
+        "daemon_head": stale_runtime_obj.daemon_head,
+        "checkout_head": stale_runtime_obj.checkout_head,
+        "remote_head": stale_runtime_obj.remote_head,
+        "stale": stale_runtime_obj.stale,
+        "determinable": stale_runtime_obj.determinable,
+        "mutation_safe": stale_runtime_obj.mutation_safe,
+        "reasons": list(stale_runtime_obj.reasons),
+    }
+    if stale_runtime_obj.stale:
+        reasons.append("Runtime HEAD disagrees with checkout/remote HEAD.")
+
+    # 2. Master parity assessment
+    checkout_head = stale_runtime_obj.checkout_head
+    startup_dict = master_parity_gate.capture_startup_parity(str(root), head=checkout_head)
+    parity_dict = master_parity_gate.assess_master_parity(startup_dict, checkout_head)
+    if not parity_dict.get("in_parity", True):
+        reasons.append("Repository is not in master parity.")
+
+    # 3. Worktree binding classification
+    boot_bindings = stale_binding_recovery.snapshot_boot_bindings(source_env)
+    active_val = (
+        active_worktree_val
+        if active_worktree_val is not None
+        else source_env.get(stale_binding_recovery.ACTIVE_WORKTREE_ENV)
+    )
+    boot_inherited = bool(boot_bindings.get("active_worktree") and active_val == boot_bindings.get("active_worktree"))
+    
+    path_exists = None
+    if active_val:
+        path_exists = os.path.exists(os.path.realpath(active_val))
+
+    binding_class = stale_binding_recovery.classify_active_worktree_binding(
+        active_value=active_val,
+        session_lease_worktree=session_lease_wt,
+        boot_inherited=boot_inherited,
+        path_exists=path_exists,
+        role_kind=role_kind,
+    )
+
+    if binding_class.get("clear_eligible"):
+        reasons.append(
+            f"Active worktree binding is stale ({binding_class.get('classification')})."
+        )
+    elif binding_class.get("classification") == stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED:
+        reasons.append("Inherited worktree binding is unverified.")
+
+    # 4. Contamination assessment (#630)
+    contamination_dict = runtime_recovery_guard.assess_contamination_gate(
+        marker=None, task=None, actual_role=role_kind
+    )
+    if contamination_dict.get("block"):
+        reasons.append("Runtime is contaminated by manual process kill (#630).")
+
+    # 5. Worktree scanner hygiene & anomalies
+    hygiene = worktree_scanner.load_hygiene_snapshot(project_root=str(root))
+    worktree_anomalies = hygiene.anomalies
+
+    # Determine status & eligible playbooks
+    playbooks: list[PlaybookDescriptor] = []
+
+    # Playbook 1: Clear Stale Binding
+    clear_eligible = bool(binding_class.get("clear_eligible"))
+    playbooks.append(
+        PlaybookDescriptor(
+            playbook_id=PLAYBOOK_CLEAR_STALE_BINDING,
+            label="Clear Stale Worktree Binding",
+            action_id=ACTION_CLEAR_STALE_BINDING,
+            description="Clear provably stale or superseded GITEA_ACTIVE_WORKTREE environment binding.",
+            eligible=clear_eligible,
+            requires_confirmation=True,
+            reason=(
+                f"Binding classified as {binding_class.get('classification')}; clear is authorized."
+                if clear_eligible
+                else "Active worktree binding is clean, corroborated, or absent."
+            ),
+        )
+    )
+
+    # Playbook 2: Rebind Session Worktree
+    rebind_eligible = bool(
+        active_val
+        or binding_class.get("classification") == stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED
+    )
+    playbooks.append(
+        PlaybookDescriptor(
+            playbook_id=PLAYBOOK_REBIND_SESSION,
+            label="Rebind Session Worktree",
+            action_id=ACTION_REBIND_SESSION,
+            description="Rebind or synchronize session worktree binding safely with active lease.",
+            eligible=rebind_eligible,
+            requires_confirmation=True,
+            reason=(
+                "Session worktree binding can be rebound to verified lease worktree."
+                if rebind_eligible
+                else "Session worktree is properly bound."
+            ),
+            params_schema={"target_worktree": "string"},
+        )
+    )
+
+    # Playbook 3: Reconcile Cleanups
+    reconcile_eligible = bool(hygiene.anomalies or any(e.classification in {"stale-clean", "detached-review"} for e in hygiene.entries))
+    playbooks.append(
+        PlaybookDescriptor(
+            playbook_id=PLAYBOOK_RECONCILE_CLEANUPS,
+            label="Trigger Reconciler Cleanups",
+            action_id=ACTION_RECONCILE_CLEANUPS,
+            description="Run sanctioned reconciler cleanup preview and apply for merged/superseded PR branches.",
+            eligible=reconcile_eligible,
+            requires_confirmation=True,
+            reason=(
+                f"Worktree hygiene scanner detected {len(hygiene.anomalies)} anomalies and cleanups needed."
+                if reconcile_eligible
+                else "No reconciler cleanups pending."
+            ),
+        )
+    )
+
+    # Playbook 4: Sanctioned Restart
+    restart_eligible = bool(stale_runtime_obj.stale or contamination_dict.get("contaminated"))
+    playbooks.append(
+        PlaybookDescriptor(
+            playbook_id=PLAYBOOK_SANCTIONED_RESTART,
+            label="Sanctioned MCP Restart",
+            action_id=sanctioned_restart.ACTION_RESTART_NAMESPACE,
+            description="Restart MCP daemon via configured host supervisor without manual process kill.",
+            eligible=restart_eligible,
+            requires_confirmation=True,
+            reason=(
+                "Stale runtime or contamination detected; host supervisor restart available."
+                if restart_eligible
+                else "Runtime is healthy and clean."
+            ),
+            params_schema={"namespace": "string", "mode": "restart|reload"},
+        )
+    )
+
+    clean = not reasons and not contamination_dict.get("contaminated")
+    if contamination_dict.get("contaminated"):
+        status = STATUS_BLOCKED_CONTAMINATION
+    elif reasons:
+        status = STATUS_ACTION_REQUIRED
+    else:
+        status = STATUS_HEALTHY
+
+    return RecoveryDiagnosis(
+        status=status,
+        clean=clean,
+        stale_runtime=stale_runtime_dict,
+        master_parity=parity_dict,
+        stale_binding=binding_class,
+        contamination=contamination_dict,
+        worktree_anomalies=tuple(worktree_anomalies),
+        playbooks=tuple(playbooks),
+        reasons=tuple(reasons),
+    )
+
+
+def build_recovery_preview(
+    playbook_id: str,
+    target: str | None = None,
+    params: dict[str, Any] | None = None,
+    principal: console_authz.Principal | None = None,
+    env: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Generate dry-run preview & mutation ledger for a recovery playbook."""
+    if playbook_id not in KNOWN_PLAYBOOKS:
+        return {
+            "allowed": False,
+            "error": "unknown_playbook",
+            "detail": f"Playbook {playbook_id!r} is not a registered recovery playbook.",
+        }
+
+    action_id = PLAYBOOK_ACTIONS[playbook_id]
+    action = console_authz.get_action(action_id)
+    decision = console_authz.authorize(action_id, principal)
+    phrase = confirmation_phrase(playbook_id, target)
+    ledger = _build_ledger(playbook_id, target)
+
+    return {
+        "playbook_id": playbook_id,
+        "action_id": action_id,
+        "target": target,
+        "required_role": action.minimum_role if action else console_authz.OPERATOR,
+        "required_permission": action.mcp_permission if action else "gitea.read",
+        "requires_confirmation": True,
+        "confirmation_phrase": phrase,
+        "mutation_ledger": [asdict(entry) for entry in ledger],
+        "authorization": decision.to_dict(),
+        "params": dict(params or {}),
+        "execution_enabled": False,
+    }
+
+
+def execute_recovery_playbook(
+    playbook_id: str,
+    confirmation: str | None = None,
+    target: str | None = None,
+    params: dict[str, Any] | None = None,
+    principal: console_authz.Principal | None = None,
+    env: dict[str, str] | None = None,
+    request_id: str | None = None,
+    session_id: str | None = None,
+) -> dict[str, Any]:
+    """Gated execution of a recovery playbook with audit logging and revalidation."""
+    if playbook_id not in KNOWN_PLAYBOOKS:
+        return {
+            "success": False,
+            "allowed": False,
+            "error": "unknown_playbook",
+            "detail": f"Playbook {playbook_id!r} is not known.",
+        }
+
+    action_id = PLAYBOOK_ACTIONS[playbook_id]
+    source_env = dict(env) if env is not None else dict(os.environ)
+
+    # 1. Authorization check
+    decision = console_authz.authorize(action_id, principal)
+    if not decision.allowed:
+        console_audit.record_event(
+            action_id=action_id,
+            result=console_audit.RESULT_DENIED,
+            principal=principal,
+            target={"playbook_id": playbook_id, "target": target},
+            reason_code=decision.reason_code,
+            detail=decision.detail,
+            request_id=request_id,
+            session_id=session_id,
+        )
+        return {
+            "success": False,
+            "allowed": False,
+            "error": decision.reason_code,
+            "detail": decision.detail,
+        }
+
+    # 2. Confirmation phrase check
+    if not confirmation_matches(playbook_id, confirmation, target):
+        expected = confirmation_phrase(playbook_id, target)
+        detail = f"Confirmation phrase mismatch. Expected: {expected!r}"
+        console_audit.record_event(
+            action_id=action_id,
+            result=console_audit.RESULT_DENIED,
+            principal=principal,
+            target={"playbook_id": playbook_id, "target": target},
+            reason_code="confirmation_mismatch",
+            detail=detail,
+            request_id=request_id,
+            session_id=session_id,
+        )
+        return {
+            "success": False,
+            "allowed": False,
+            "error": "confirmation_mismatch",
+            "detail": detail,
+            "expected_confirmation_phrase": expected,
+        }
+
+    # 3. Contamination rule (#630) check
+    role_str = principal.role if principal else None
+    contam = runtime_recovery_guard.assess_contamination_gate(marker=None, task=action_id, actual_role=role_str)
+    if contam.get("block"):
+        if playbook_id != PLAYBOOK_RECONCILE_CLEANUPS:
+            detail = "Runtime is contaminated by a manual process kill (#630). Run reconciler cleanup playbook first."
+            console_audit.record_event(
+                action_id=action_id,
+                result=console_audit.RESULT_DENIED,
+                principal=principal,
+                target={"playbook_id": playbook_id, "target": target},
+                reason_code="contaminated_runtime",
+                detail=detail,
+                request_id=request_id,
+                session_id=session_id,
+            )
+            return {
+                "success": False,
+                "allowed": False,
+                "error": "contaminated_runtime",
+                "detail": detail,
+            }
+
+    # 4. Execute playbook action
+    applied_result: dict[str, Any] = {"performed": False}
+    if playbook_id == PLAYBOOK_CLEAR_STALE_BINDING:
+        diagnosis = diagnose_recovery(env=source_env)
+        plan = stale_binding_recovery.plan_recovery(diagnosis.stale_binding)
+        applied_result = stale_binding_recovery.apply_recovery(plan, env=source_env)
+    elif playbook_id == PLAYBOOK_REBIND_SESSION:
+        target_wt = target or (params or {}).get("target_worktree")
+        if target_wt:
+            source_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV] = target_wt
+            applied_result = {
+                "performed": True,
+                "rebound_worktree": target_wt,
+                "cleared_stale": True,
+            }
+        else:
+            applied_result = {
+                "performed": False,
+                "reason": "No target_worktree specified for rebind.",
+            }
+    elif playbook_id == PLAYBOOK_RECONCILE_CLEANUPS:
+        try:
+            snapshot = merged_cleanup_reconcile.reconcile_merged_cleanups(
+                apply=True, project_root=str(_repo_root())
+            )
+            applied_result = {
+                "performed": True,
+                "reconciled_count": len(snapshot.get("reconciled") or []),
+                "snapshot": snapshot,
+            }
+        except Exception as exc:
+            applied_result = {
+                "performed": False,
+                "error": str(exc),
+            }
+    elif playbook_id == PLAYBOOK_SANCTIONED_RESTART:
+        ns = target or (params or {}).get("namespace", "gitea-author")
+        md = (params or {}).get("mode", sanctioned_restart.MODE_RESTART)
+        restart_res = sanctioned_restart.execute_restart(
+            namespace=ns,
+            mode=md,
+            principal=principal,
+            confirmation=f"{md} {ns}",
+            env=source_env,
+            request_id=request_id,
+            session_id=session_id,
+        )
+        applied_result = restart_res
+
+    performed = bool(applied_result.get("performed") or applied_result.get("allowed"))
+
+    # 5. Record Audit Log
+    audit_record = console_audit.record_event(
+        action_id=action_id,
+        result=console_audit.RESULT_ALLOWED if performed else console_audit.RESULT_DENIED,
+        principal=principal,
+        target={"playbook_id": playbook_id, "target": target},
+        reason_code="recovery_executed" if performed else "recovery_failed",
+        detail=f"Executed recovery playbook {playbook_id}",
+        request_id=request_id,
+        session_id=session_id,
+        metadata={"applied_result": applied_result},
+    )
+
+    # 6. Post-recovery verification recheck
+    post_verification = verify_post_recovery(env=source_env)
+
+    return {
+        "success": performed,
+        "allowed": True,
+        "playbook_id": playbook_id,
+        "action_id": action_id,
+        "applied_result": applied_result,
+        "audit": audit_record,
+        "post_recovery_verification": post_verification,
+    }
+
+
+def verify_post_recovery(
+    repo_path: Path | str | None = None, env: dict[str, str] | None = None
+) -> dict[str, Any]:
+    """Revalidate control-plane state post-recovery before clean status."""
+    diag = diagnose_recovery(repo_path, env)
+    return {
+        "clean": diag.clean,
+        "status": diag.status,
+        "stale_runtime_clean": not diag.stale_runtime.get("stale"),
+        "binding_clean": not diag.stale_binding.get("clear_eligible"),
+        "contamination_clean": not diag.contamination.get("contaminated"),
+        "anomalies_count": len(diag.worktree_anomalies),
+        "reasons": list(diag.reasons),
+    }

webui/gated_actions.py

@@ -178,6 +178,13 @@ def build_action_registry() -> ActionRegistry:
         ("system.restart_namespace", "Restart MCP namespace",
          "restart_namespace", "host.supervisor_restart",
          "Restart one MCP namespace via the host supervisor."),
+        # #644: Phase 2 recovery playbooks & controls.
+        ("system.clear_stale_binding", "Clear stale binding", "clear_stale_binding",
+         "console.clear_stale_binding", "Clear provably stale or superseded env binding."),
+        ("system.rebind_session_worktree", "Rebind session worktree", "rebind_session_worktree",
+         "console.rebind_session_worktree", "Rebind session worktree to verified lease."),
+        ("system.reconcile_cleanups", "Reconcile cleanups", "reconcile_cleanups",
+         "console.reconcile_cleanups", "Run reconciler cleanup for merged or superseded PRs."),
     )
     actions = tuple(
         GatedAction(

webui/linkage_loader.py

@@ -1,771 +0,0 @@
-"""Gitea issue↔PR linkage model for the console (#645, Phase 3).
-
-Operators lose context between an issue and the PR that closes it: which PR
-carries which issue, whether two PRs claim the same issue, and what the latest
-canonical handoff on that thread said. The evidence exists in Gitea, but only
-as free text scattered across PR titles, bodies, and branch names.
-
-This module resolves that linkage into one read-only model:
-
-* :func:`resolve_linkage` is a pure function from raw Gitea issue/PR payloads to
-  a :class:`LinkageIndex`. It records *how* each edge was found (a ``Closes #N``
-  keyword, the canonical ``feat/issue-N-…`` branch marker, or a bare ``#N``
-  body reference) and never collapses several candidates into one silent guess.
-* :func:`load_linkage_snapshot` scopes that index to a registry project and
-  optionally attaches the latest Canonical Thread Handoff (CTH) summary for one
-  focused issue or PR.
-
-Design rules, matching the rest of the console:
-
-- **Read-only.** Gitea is read through the shared authenticated helpers. No
-  endpoint here mutates anything, and no write action is registered.
-- **Qualified absence.** Linkage is a claim about a *loaded* window of Gitea.
-  When pagination did not complete, when credentials were unavailable, or when
-  only open items were fetched, the snapshot says so and every "no linked PR"
-  is marked non-authoritative. An empty edge list from a partial read is not
-  evidence that no link exists.
-- **Handoff is loaded, never assumed.** CTH comments are thread-scoped, so they
-  are fetched only for an explicitly focused issue or PR. Every other row
-  reports ``not_loaded`` rather than rendering as "no handoff".
-- **Redaction at the boundary.** Titles, labels, handoff fields, and error
-  reasons are free text from Gitea and cross :mod:`webui.console_redaction`
-  before they leave this module.
-- **Deep links are opt-in.** A link to the Gitea web UI is emitted only under
-  the ``GITEA_MCP_REVEAL_ENDPOINTS`` admin opt-in, exactly as the MCP tools
-  gate their own URL exposure.
-
-Non-goals (from the issue): no issue/PR editor, no browser review or merge, no
-reimplementation of Gitea search.
-"""
-
-from __future__ import annotations
-
-import os
-import re
-from dataclasses import dataclass
-from typing import Any, Callable, Iterable, Sequence
-
-from gitea_auth import api_fetch_page, get_auth_header, gitea_url, repo_api_url
-
-from webui import console_redaction
-from webui.project_registry import ProjectRecord, load_registry
-from webui.queue_loader import (
-    PaginationMeta,
-    _fetch_issues,
-    _fetch_prs,
-    _host_from_url,
-)
-
-#: Version of the serialized linkage contract. Bump on any breaking change.
-LINKAGE_SCHEMA_VERSION = 1
-
-# --- Linkage evidence -------------------------------------------------------
-# Ordered strongest to weakest. The strength ordering is what makes an
-# ambiguous PR detectable: two candidates at the same strength are a genuine
-# ambiguity, while a weaker candidate alongside a stronger one is not.
-EVIDENCE_CLOSES = "closes_keyword"
-EVIDENCE_BRANCH = "branch_marker"
-EVIDENCE_REFERENCE = "body_reference"
-
-EVIDENCE_ORDER: tuple[str, ...] = (
-    EVIDENCE_CLOSES,
-    EVIDENCE_BRANCH,
-    EVIDENCE_REFERENCE,
-)
-_EVIDENCE_RANK = {name: rank for rank, name in enumerate(EVIDENCE_ORDER)}
-
-EVIDENCE_DESCRIPTIONS: dict[str, str] = {
-    EVIDENCE_CLOSES: (
-        "the PR title or body declares 'closes/fixes/resolves #N' — Gitea itself "
-        "acts on this keyword, so it is the strongest available evidence"
-    ),
-    EVIDENCE_BRANCH: (
-        "the PR head branch carries the canonical issue marker "
-        "'(fix|feat|docs|chore)/issue-N-…' minted by the issue lock"
-    ),
-    EVIDENCE_REFERENCE: (
-        "the PR body mentions '#N' without a closing keyword; a mention is not "
-        "a claim that the PR closes that issue"
-    ),
-}
-
-_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.IGNORECASE)
-_REFERENCE_RE = re.compile(r"#(\d+)")
-_BRANCH_MARKER_RE = re.compile(
-    r"^(?:fix|feat|docs|chore)/issue-(\d+)(?:[-/]|$)", re.IGNORECASE
-)
-
-# Handoff-source states. ``not_loaded`` is deliberately distinct from "none
-# found": a row whose comments were never fetched proves nothing about whether
-# a handoff exists on that thread.
-HANDOFF_NOT_LOADED = "not_loaded"
-HANDOFF_LOADED = "loaded"
-HANDOFF_UNAVAILABLE = "unavailable"
-
-# Which item states were fetched. Linkage claims are scoped to this window.
-STATE_OPEN = "open"
-STATE_ALL = "all"
-_SUPPORTED_STATES = (STATE_OPEN, STATE_ALL)
-
-
-def _redact(value: Any) -> Any:
-    """Redact one free-text field, failing closed to the placeholder."""
-    if value is None:
-        return None
-    return console_redaction.redact_text(str(value))
-
-
-def deep_links_enabled(env: dict[str, str] | None = None) -> bool:
-    """Whether Gitea web-UI deep links may be emitted (admin/debug opt-in)."""
-    source = env if env is not None else os.environ
-    return (source.get("GITEA_MCP_REVEAL_ENDPOINTS") or "").strip().lower() in {
-        "1",
-        "true",
-        "yes",
-        "on",
-    }
-
-
-def _deep_link(host: str, org: str, repo: str, kind: str, number: int) -> str | None:
-    """Build a Gitea web link for one item, or None when reveal is not enabled."""
-    if not deep_links_enabled() or not (host and org and repo):
-        return None
-    segment = "pulls" if kind == "pr" else "issues"
-    try:
-        return gitea_url(host, f"/{org}/{repo}/{segment}/{int(number)}")
-    except Exception:
-        return None
-
-
-# --- Pure linkage resolution -------------------------------------------------
-
-
-@dataclass(frozen=True)
-class IssueLink:
-    """One resolved edge from a PR to an issue, with the evidence that found it."""
-
-    issue_number: int
-    evidence: tuple[str, ...]
-
-    @property
-    def strength(self) -> int:
-        """Rank of the strongest evidence backing this edge (lower is stronger)."""
-        return min(
-            (_EVIDENCE_RANK.get(name, len(EVIDENCE_ORDER)) for name in self.evidence),
-            default=len(EVIDENCE_ORDER),
-        )
-
-    @property
-    def closes(self) -> bool:
-        """True only when the PR *declares* it closes the issue."""
-        return EVIDENCE_CLOSES in self.evidence
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "issue_number": self.issue_number,
-            "evidence": list(self.evidence),
-            "closes": self.closes,
-        }
-
-
-def _sorted_links(links: Iterable[IssueLink]) -> tuple[IssueLink, ...]:
-    return tuple(sorted(links, key=lambda link: (link.strength, link.issue_number)))
-
-
-def resolve_pr_links(pr: dict[str, Any]) -> tuple[IssueLink, ...]:
-    """Resolve every issue a PR points at, strongest evidence first.
-
-    Every candidate is kept. Collapsing to a single "linked issue" is what makes
-    a mislinked or double-claimed PR invisible, so the caller decides what to do
-    with several candidates rather than being handed one guess.
-    """
-    found: dict[int, set[str]] = {}
-
-    def _add(number: Any, evidence: str) -> None:
-        try:
-            issue_number = int(number)
-        except (TypeError, ValueError):
-            return
-        if issue_number <= 0:
-            return
-        found.setdefault(issue_number, set()).add(evidence)
-
-    title = str(pr.get("title") or "")
-    body = str(pr.get("body") or "")
-    for text in (title, body):
-        for match in _CLOSES_RE.finditer(text):
-            _add(match.group(1), EVIDENCE_CLOSES)
-
-    head_ref = str((pr.get("head") or {}).get("ref") or "")
-    branch_match = _BRANCH_MARKER_RE.match(head_ref.strip())
-    if branch_match:
-        _add(branch_match.group(1), EVIDENCE_BRANCH)
-
-    # The ``#N`` inside "Closes #N" is the *same* textual occurrence as the
-    # closing keyword, not a second, independent mention. Blanking the closing
-    # phrases first keeps "mention" meaning what the legend says it means: a
-    # reference the PR made without claiming to close anything.
-    for match in _REFERENCE_RE.finditer(_CLOSES_RE.sub(" ", body)):
-        _add(match.group(1), EVIDENCE_REFERENCE)
-
-    # A PR's own number appearing in its body is self-reference, not linkage.
-    try:
-        found.pop(int(pr.get("number")), None)
-    except (TypeError, ValueError):
-        pass
-
-    return _sorted_links(
-        IssueLink(
-            issue_number=number,
-            evidence=tuple(name for name in EVIDENCE_ORDER if name in evidence),
-        )
-        for number, evidence in found.items()
-    )
-
-
-@dataclass(frozen=True)
-class LinkageIndex:
-    """Resolved linkage over one loaded window of issues and PRs."""
-
-    pr_links: dict[int, tuple[IssueLink, ...]]
-    issue_prs: dict[int, tuple[int, ...]]
-
-    def primary_issue(self, pr_number: int) -> IssueLink | None:
-        """The strongest edge for a PR, or None when it points at no issue."""
-        links = self.pr_links.get(int(pr_number)) or ()
-        return links[0] if links else None
-
-    def ambiguous(self, pr_number: int) -> bool:
-        """True when two or more issues tie at the PR's strongest evidence."""
-        links = self.pr_links.get(int(pr_number)) or ()
-        if len(links) < 2:
-            return False
-        best = links[0].strength
-        return sum(1 for link in links if link.strength == best) > 1
-
-    def contested_issues(self) -> tuple[int, ...]:
-        """Issues claimed by more than one PR in the loaded window."""
-        return tuple(
-            number for number, prs in sorted(self.issue_prs.items()) if len(prs) > 1
-        )
-
-
-def resolve_linkage(prs: Sequence[dict[str, Any]]) -> LinkageIndex:
-    """Build the bidirectional linkage index for a loaded window of PRs.
-
-    Only *closing* and *branch-marker* edges populate the issue→PR direction: a
-    bare ``#N`` mention is a reference, and treating it as "this PR is the work
-    for issue N" would invent contested issues out of ordinary cross-links. The
-    weaker edge stays visible on the PR→issue side, where it is labelled.
-    """
-    pr_links: dict[int, tuple[IssueLink, ...]] = {}
-    issue_prs: dict[int, list[int]] = {}
-    for pr in prs or []:
-        try:
-            pr_number = int(pr["number"])
-        except (KeyError, TypeError, ValueError):
-            continue
-        links = resolve_pr_links(pr)
-        pr_links[pr_number] = links
-        for link in links:
-            if link.evidence == (EVIDENCE_REFERENCE,):
-                continue
-            bucket = issue_prs.setdefault(link.issue_number, [])
-            if pr_number not in bucket:
-                bucket.append(pr_number)
-    return LinkageIndex(
-        pr_links=pr_links,
-        issue_prs={number: tuple(sorted(items)) for number, items in issue_prs.items()},
-    )
-
-
-# --- Canonical handoff summary ----------------------------------------------
-
-
-@dataclass(frozen=True)
-class HandoffSummary:
-    """The latest CTH comment on one thread, redacted for display."""
-
-    comment_id: int | None
-    created_at: str | None
-    author: str | None
-    cth_type: str
-    cth_type_known: bool
-    status: str | None
-    next_owner: str | None
-    current_blocker: str | None
-    decision: str | None
-    next_action: str | None
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "comment_id": self.comment_id,
-            "created_at": self.created_at,
-            "author": self.author,
-            "cth_type": self.cth_type,
-            "cth_type_known": self.cth_type_known,
-            "status": self.status,
-            "next_owner": self.next_owner,
-            "current_blocker": self.current_blocker,
-            "decision": self.decision,
-            "next_action": self.next_action,
-        }
-
-
-@dataclass(frozen=True)
-class HandoffStatus:
-    """Why a thread's handoff summary is present, absent, or unknown."""
-
-    state: str
-    reason: str | None = None
-    target: str | None = None
-
-    @property
-    def loaded(self) -> bool:
-        return self.state == HANDOFF_LOADED
-
-    def to_dict(self) -> dict[str, Any]:
-        return {"state": self.state, "reason": self.reason, "target": self.target}
-
-
-def summarize_handoff(comments: Sequence[dict[str, Any]]) -> HandoffSummary | None:
-    """Summarize the newest CTH comment in *comments*, or None when there is none.
-
-    Every field is redacted before it is returned: a handoff body is operator
-    free text that regularly quotes commands, and it is rendered verbatim on the
-    page this feeds.
-    """
-    from canonical_thread_handoff import find_latest_cth, is_known_cth_type
-
-    try:
-        latest = find_latest_cth(list(comments or []))
-    except Exception:
-        return None
-    if not latest:
-        return None
-    fields = latest.get("fields") or {}
-    cth_type = str(latest.get("cth_type") or "").strip()
-    known = is_known_cth_type(cth_type)
-    try:
-        comment_id: int | None = int(latest.get("comment_id"))
-    except (TypeError, ValueError):
-        comment_id = None
-    return HandoffSummary(
-        comment_id=comment_id,
-        created_at=_redact(latest.get("created_at")),
-        author=_redact(latest.get("author")),
-        # An unrecognised heading is reported as such rather than republished:
-        # the heading is free text, and CTH_TYPES is the only authority for what
-        # a handoff type may be.
-        cth_type=cth_type if known else "unrecognized",
-        cth_type_known=known,
-        status=_redact(fields.get("status")),
-        next_owner=_redact(fields.get("next owner")),
-        current_blocker=_redact(fields.get("current blocker")),
-        decision=_redact(fields.get("decision")),
-        next_action=_redact(fields.get("next action")),
-    )
-
-
-CommentSource = Callable[[str, int], list[dict[str, Any]]]
-
-
-def build_comment_source(host: str, org: str, repo: str) -> CommentSource | None:
-    """Build an authenticated ``(kind, number) -> comments`` fetcher, or None.
-
-    Returns None when the console is running in offline test mode or when no
-    credential is available for *host*, so the caller reports the handoff source
-    as unavailable instead of as an empty thread.
-    """
-    if _offline_test_mode() or not (host and org and repo):
-        return None
-    auth = get_auth_header(host)
-    if not auth:
-        return None
-
-    def _fetch(kind: str, number: int) -> list[dict[str, Any]]:
-        segment = "pulls" if kind == "pr" else "issues"
-        url = f"{repo_api_url(host, org, repo)}/{segment}/{int(number)}/comments"
-        comments: list[dict[str, Any]] = []
-        page = 1
-        while page <= 20:
-            raw, meta = api_fetch_page(url, auth, page=page, limit=50)
-            comments.extend(raw)
-            if bool(meta["is_final_page"]):
-                break
-            page += 1
-        return comments
-
-    return _fetch
-
-
-# --- Snapshot ----------------------------------------------------------------
-
-
-@dataclass(frozen=True)
-class LinkageNode:
-    """One issue or PR row with its resolved links and display metadata."""
-
-    kind: str
-    number: int
-    title: str
-    state: str
-    labels: tuple[str, ...] = ()
-    links: tuple[IssueLink, ...] = ()
-    linked_prs: tuple[int, ...] = ()
-    ambiguous: bool = False
-    contested: bool = False
-    deep_link: str | None = None
-    handoff: HandoffSummary | None = None
-    handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
-    links_authoritative: bool = True
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "kind": self.kind,
-            "number": self.number,
-            "title": self.title,
-            "state": self.state,
-            "labels": list(self.labels),
-            "links": [link.to_dict() for link in self.links],
-            "linked_prs": list(self.linked_prs),
-            "ambiguous": self.ambiguous,
-            "contested": self.contested,
-            "deep_link": self.deep_link,
-            "links_authoritative": self.links_authoritative,
-            "handoff": self.handoff.to_dict() if self.handoff else None,
-            "handoff_status": self.handoff_status.to_dict(),
-        }
-
-
-@dataclass(frozen=True)
-class LinkageSnapshot:
-    """One answered linkage query over a scoped window of a Gitea repo."""
-
-    ok: bool
-    project_id: str
-    repo_label: str
-    host: str
-    state_scope: str
-    issues: tuple[LinkageNode, ...] = ()
-    prs: tuple[LinkageNode, ...] = ()
-    contested_issues: tuple[int, ...] = ()
-    focus: tuple[str, int] | None = None
-    inventory_complete: bool = False
-    deep_links_enabled: bool = False
-    handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
-    fetch_error: str | None = None
-
-    @property
-    def orphan_prs(self) -> tuple[LinkageNode, ...]:
-        """PRs in the loaded window that point at no issue at all."""
-        return tuple(node for node in self.prs if not node.links)
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "ok": self.ok,
-            "schema_version": LINKAGE_SCHEMA_VERSION,
-            "project_id": self.project_id,
-            "repo": self.repo_label,
-            "state_scope": self.state_scope,
-            "inventory_complete": self.inventory_complete,
-            "deep_links_enabled": self.deep_links_enabled,
-            "focus": (
-                None
-                if self.focus is None
-                else {"kind": self.focus[0], "number": self.focus[1]}
-            ),
-            "handoff_source": self.handoff_status.to_dict(),
-            "fetch_error": self.fetch_error,
-            "contested_issues": list(self.contested_issues),
-            "issues": [node.to_dict() for node in self.issues],
-            "prs": [node.to_dict() for node in self.prs],
-            "evidence_kinds": [
-                {"name": name, "description": EVIDENCE_DESCRIPTIONS[name]}
-                for name in EVIDENCE_ORDER
-            ],
-        }
-
-
-def snapshot_to_dict(snapshot: LinkageSnapshot) -> dict[str, Any]:
-    """JSON-serializable export for ``/api/v1/gitea/linkage``."""
-    return snapshot.to_dict()
-
-
-def _offline_test_mode() -> bool:
-    return (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {
-        "1",
-        "true",
-        "yes",
-    }
-
-
-def _labels_of(item: dict[str, Any]) -> tuple[str, ...]:
-    return tuple(
-        str(_redact(label.get("name")))
-        for label in (item.get("labels") or [])
-        if label.get("name")
-    )
-
-
-def _failed_snapshot(
-    *,
-    project_id: str,
-    repo_label: str,
-    host: str,
-    state_scope: str,
-    reason: str,
-) -> LinkageSnapshot:
-    """A read that could not be answered. Never an empty-and-healthy snapshot."""
-    return LinkageSnapshot(
-        ok=False,
-        project_id=project_id,
-        repo_label=repo_label,
-        host=host,
-        state_scope=state_scope,
-        inventory_complete=False,
-        deep_links_enabled=deep_links_enabled(),
-        handoff_status=HandoffStatus(
-            HANDOFF_UNAVAILABLE, reason="linkage inventory could not be loaded"
-        ),
-        fetch_error=str(_redact(reason)),
-    )
-
-
-def _resolve_project(project_id: str | None) -> ProjectRecord | None:
-    registry = load_registry()
-    if project_id:
-        for entry in registry.projects:
-            if entry.id == project_id:
-                return entry
-        return None
-    return registry.projects[0] if registry.projects else None
-
-
-def _normalize_state(state: str | None) -> str:
-    text = (state or STATE_OPEN).strip().lower()
-    return text if text in _SUPPORTED_STATES else STATE_OPEN
-
-
-def load_linkage_snapshot(
-    project_id: str | None = None,
-    *,
-    state: str | None = None,
-    issue: int | None = None,
-    pr: int | None = None,
-    fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
-    fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
-    comment_source: CommentSource | None = None,
-) -> LinkageSnapshot:
-    """Load issue↔PR linkage for a registry project.
-
-    ``issue``/``pr`` focus one thread: the focused row is the only one whose
-    Canonical Thread Handoff comments are fetched, because handoff comments are
-    thread-scoped and loading them for a whole queue would be one request per
-    row. Every unfocused row reports its handoff as ``not_loaded``.
-    """
-    state_scope = _normalize_state(state)
-    try:
-        project = _resolve_project(project_id)
-    except Exception as exc:  # registry invalid — fail closed with the reason
-        return _failed_snapshot(
-            project_id=project_id or "",
-            repo_label="",
-            host="",
-            state_scope=state_scope,
-            reason=f"project registry unavailable: {exc}",
-        )
-
-    if project is None:
-        return _failed_snapshot(
-            project_id=project_id or "",
-            repo_label="",
-            host="",
-            state_scope=state_scope,
-            reason=(
-                f"project {project_id!r} not found in registry"
-                if project_id
-                else "no projects registered"
-            ),
-        )
-
-    host = _host_from_url(project.remote_host)
-    repo_label = f"{project.gitea_owner}/{project.repo_name}"
-    offline_test = _offline_test_mode()
-
-    def _empty_fetch(*_args, **_kwargs):
-        return [], PaginationMeta(
-            page=1,
-            per_page=50,
-            returned_count=0,
-            has_more=False,
-            is_final_page=True,
-            # An offline stub loaded nothing; claiming a complete inventory here
-            # would let the page assert that no issue has a linked PR.
-            inventory_complete=False,
-            pages_fetched=0,
-        )
-
-    pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
-    issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
-    using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
-    auth = get_auth_header(host) if using_live_fetch else "test-auth"
-    if using_live_fetch and not auth:
-        return _failed_snapshot(
-            project_id=project.id,
-            repo_label=repo_label,
-            host=host,
-            state_scope=state_scope,
-            reason=(
-                f"Gitea credentials unavailable for {host}; linkage cannot be "
-                "loaded (fail closed — not rendering an empty linkage table)"
-            ),
-        )
-
-    try:
-        raw_prs, pr_pagination = pr_fetch(
-            host, project.gitea_owner, project.repo_name, auth, state=state_scope
-        )
-        raw_issues, issue_pagination = issue_fetch(
-            host, project.gitea_owner, project.repo_name, auth, state=state_scope
-        )
-    except Exception as exc:  # noqa: BLE001 — operator-visible fetch failure
-        return _failed_snapshot(
-            project_id=project.id,
-            repo_label=repo_label,
-            host=host,
-            state_scope=state_scope,
-            reason=f"Gitea fetch failed: {exc}",
-        )
-
-    inventory_complete = bool(
-        getattr(pr_pagination, "inventory_complete", False)
-        and getattr(issue_pagination, "inventory_complete", False)
-    )
-
-    index = resolve_linkage(raw_prs)
-    contested = index.contested_issues()
-
-    focus: tuple[str, int] | None = None
-    if pr is not None:
-        focus = ("pr", int(pr))
-    elif issue is not None:
-        focus = ("issue", int(issue))
-
-    unfocused_reason = (
-        "canonical handoff comments are thread-scoped; focus one issue or PR "
-        "to load its latest handoff"
-    )
-    handoff_status = HandoffStatus(HANDOFF_NOT_LOADED, reason=unfocused_reason)
-    focus_handoff: HandoffSummary | None = None
-    if focus is not None:
-        source = comment_source
-        if source is None and not offline_test:
-            source = build_comment_source(host, project.gitea_owner, project.repo_name)
-        target = f"{focus[0]}#{focus[1]}"
-        if source is None:
-            handoff_status = HandoffStatus(
-                HANDOFF_UNAVAILABLE,
-                reason="no authenticated comment source available for this read",
-                target=target,
-            )
-        else:
-            try:
-                focus_handoff = summarize_handoff(source(focus[0], focus[1]) or [])
-                handoff_status = HandoffStatus(HANDOFF_LOADED, target=target)
-            except Exception as exc:  # fail soft: degrade this source only
-                handoff_status = HandoffStatus(
-                    HANDOFF_UNAVAILABLE,
-                    reason=str(_redact(f"handoff fetch failed: {exc}")),
-                    target=target,
-                )
-
-    def _node_handoff(
-        kind: str, number: int
-    ) -> tuple[HandoffSummary | None, HandoffStatus]:
-        """Attach the handoff only to the focused row; qualify every other row."""
-        if focus == (kind, number):
-            return (focus_handoff, handoff_status)
-        return (
-            None,
-            HandoffStatus(
-                HANDOFF_NOT_LOADED,
-                reason=unfocused_reason if focus is None else "not the focused thread",
-            ),
-        )
-
-    def _number_of(raw: dict[str, Any]) -> int | None:
-        try:
-            return int(raw["number"])
-        except (KeyError, TypeError, ValueError):
-            return None
-
-    def _sort_key(raw: dict[str, Any]) -> int:
-        number = _number_of(raw)
-        return -1 if number is None else number
-
-    issue_nodes: list[LinkageNode] = []
-    for raw in sorted(raw_issues or [], key=_sort_key, reverse=True):
-        number = _number_of(raw)
-        if number is None:
-            continue
-        node_handoff, node_status = _node_handoff("issue", number)
-        linked_prs = index.issue_prs.get(number, ())
-        issue_nodes.append(
-            LinkageNode(
-                kind="issue",
-                number=number,
-                title=str(_redact(raw.get("title")) or ""),
-                state=str(raw.get("state") or ""),
-                labels=_labels_of(raw),
-                linked_prs=linked_prs,
-                contested=len(linked_prs) > 1,
-                deep_link=_deep_link(
-                    host, project.gitea_owner, project.repo_name, "issue", number
-                ),
-                handoff=node_handoff,
-                handoff_status=node_status,
-                links_authoritative=inventory_complete,
-            )
-        )
-
-    pr_nodes: list[LinkageNode] = []
-    for raw in sorted(raw_prs or [], key=_sort_key, reverse=True):
-        number = _number_of(raw)
-        if number is None:
-            continue
-        node_handoff, node_status = _node_handoff("pr", number)
-        links = index.pr_links.get(number, ())
-        pr_nodes.append(
-            LinkageNode(
-                kind="pr",
-                number=number,
-                title=str(_redact(raw.get("title")) or ""),
-                state=str(raw.get("state") or ""),
-                labels=_labels_of(raw),
-                links=links,
-                ambiguous=index.ambiguous(number),
-                contested=any(link.issue_number in contested for link in links),
-                deep_link=_deep_link(
-                    host, project.gitea_owner, project.repo_name, "pr", number
-                ),
-                handoff=node_handoff,
-                handoff_status=node_status,
-                links_authoritative=inventory_complete,
-            )
-        )
-
-    return LinkageSnapshot(
-        ok=True,
-        project_id=project.id,
-        repo_label=repo_label,
-        host=host,
-        state_scope=state_scope,
-        issues=tuple(issue_nodes),
-        prs=tuple(pr_nodes),
-        contested_issues=contested,
-        focus=focus,
-        inventory_complete=inventory_complete,
-        deep_links_enabled=deep_links_enabled(),
-        handoff_status=handoff_status,
-    )

webui/linkage_views.py

@@ -1,364 +0,0 @@
-"""HTML views for the Gitea issue↔PR linkage console (#645, Phase 3).
-
-Read-only renderer over :mod:`webui.linkage_loader`. The page's job is to make
-three things impossible to misread:
-
-* **why** an edge exists — every link carries its evidence badge, so a bare
-  ``#N`` mention never looks like a closing claim;
-* **what was not loaded** — a partial inventory, an unfocused thread, or an
-  unavailable handoff source renders as an explicit qualifier, never as an
-  affirmative "none";
-* **that nothing here mutates** — there is no review, merge, or edit control,
-  and the deep link out to Gitea appears only under the admin reveal opt-in.
-"""
-
-from __future__ import annotations
-
-from html import escape
-from typing import Sequence
-
-from webui.layout import render_page
-from webui.linkage_loader import (
-    EVIDENCE_BRANCH,
-    EVIDENCE_CLOSES,
-    EVIDENCE_DESCRIPTIONS,
-    EVIDENCE_ORDER,
-    EVIDENCE_REFERENCE,
-    HANDOFF_LOADED,
-    HANDOFF_NOT_LOADED,
-    HandoffSummary,
-    LinkageNode,
-    LinkageSnapshot,
-)
-
-_EVIDENCE_CSS = {
-    EVIDENCE_CLOSES: "badge-health-ok",
-    EVIDENCE_BRANCH: "badge-health-skipped",
-    EVIDENCE_REFERENCE: "badge-health-unproven",
-}
-
-_EVIDENCE_LABEL = {
-    EVIDENCE_CLOSES: "closes",
-    EVIDENCE_BRANCH: "branch",
-    EVIDENCE_REFERENCE: "mention",
-}
-
-
-def _badge(text: str, css: str) -> str:
-    return f'<span class="badge {css}">{escape(text)}</span>'
-
-
-def _labels(names: Sequence[str]) -> str:
-    if not names:
-        return '<span class="muted">—</span>'
-    return " ".join(_badge(name, "badge-health-skipped") for name in names)
-
-
-def _ref(node: LinkageNode) -> str:
-    """Render an item reference, hyperlinked only when deep links are revealed."""
-    label = f"#{node.number}"
-    if node.deep_link:
-        return f'<a href="{escape(node.deep_link)}"><code>{escape(label)}</code></a>'
-    return f"<code>{escape(label)}</code>"
-
-
-def _scope_card(snapshot: LinkageSnapshot) -> str:
-    focus = (
-        "none"
-        if snapshot.focus is None
-        else f"{snapshot.focus[0]}#{snapshot.focus[1]}"
-    )
-    completeness = (
-        _badge("complete", "badge-health-ok")
-        if snapshot.inventory_complete
-        else _badge("partial", "badge-health-degraded")
-    )
-    links_note = (
-        "Every linkage edge below is a claim about this loaded window only."
-        if snapshot.inventory_complete
-        else (
-            "Pagination did not complete for this window, so an empty link list "
-            "means <em>none found in what was loaded</em> — not that no link exists."
-        )
-    )
-    deep_links = (
-        _badge("enabled", "badge-health-ok")
-        if snapshot.deep_links_enabled
-        else _badge("hidden", "badge-health-skipped")
-    )
-    return f"""<div class="health-card">
-  <h3>Scope</h3>
-  <table class="detail">
-    <tr><th>Project</th><td><code>{escape(snapshot.project_id or "—")}</code></td></tr>
-    <tr><th>Repository</th><td><code>{escape(snapshot.repo_label or "—")}</code></td></tr>
-    <tr><th>Item state</th><td><code>{escape(snapshot.state_scope)}</code></td></tr>
-    <tr><th>Focused thread</th><td><code>{escape(focus)}</code></td></tr>
-    <tr><th>Inventory</th><td>{completeness}</td></tr>
-    <tr><th>Gitea deep links</th><td>{deep_links}</td></tr>
-  </table>
-  <p class="muted">{links_note}</p>
-</div>"""
-
-
-def _error_card(snapshot: LinkageSnapshot) -> str:
-    if snapshot.ok and not snapshot.fetch_error:
-        return ""
-    return (
-        '<div class="health-card health-stale"><strong>Linkage unavailable:</strong> '
-        f"{escape(snapshot.fetch_error or 'the linkage read did not complete')}. "
-        "No linkage table is rendered: an empty table would read as "
-        "<em>no issue is linked to any PR</em>, which this read cannot claim."
-        "</div>"
-    )
-
-
-def _contested_card(snapshot: LinkageSnapshot) -> str:
-    if not snapshot.contested_issues:
-        return ""
-    refs = ", ".join(f"<code>#{number}</code>" for number in snapshot.contested_issues)
-    return (
-        '<div class="health-card health-stale">'
-        f"<strong>Contested issues:</strong> {refs}. More than one PR in this "
-        "window claims each of these — duplicate work or a superseded PR. "
-        "Resolution stays in Gitea and the workflow; this console only reports it."
-        "</div>"
-    )
-
-
-def _evidence_badges(evidence: Sequence[str]) -> str:
-    return " ".join(
-        _badge(
-            _EVIDENCE_LABEL.get(name, name),
-            _EVIDENCE_CSS.get(name, "badge-health-skipped"),
-        )
-        for name in EVIDENCE_ORDER
-        if name in evidence
-    )
-
-
-def _handoff_inline(handoff: HandoffSummary) -> str:
-    type_css = "badge-health-ok" if handoff.cth_type_known else "badge-health-degraded"
-    return (
-        f'{_badge(handoff.cth_type or "—", type_css)}'
-        f'<div class="muted" style="font-size:0.82rem;">'
-        f'{escape(handoff.status or "—")} → {escape(handoff.next_owner or "—")}</div>'
-    )
-
-
-def _handoff_cell(node: LinkageNode) -> str:
-    """Render the handoff column, distinguishing 'none found' from 'not loaded'."""
-    status = node.handoff_status
-    if status.state == HANDOFF_LOADED:
-        if node.handoff is None:
-            return '<span class="muted">no canonical handoff on this thread</span>'
-        return _handoff_inline(node.handoff)
-    if status.state == HANDOFF_NOT_LOADED:
-        return (
-            f'{_badge("not loaded", "badge-health-skipped")}'
-            f'<div class="muted" style="font-size:0.82rem;">'
-            f'{escape(status.reason or "")}</div>'
-        )
-    return (
-        f'{_badge("unavailable", "badge-health-degraded")}'
-        f'<div class="muted" style="font-size:0.82rem;">'
-        f'{escape(status.reason or "")}</div>'
-    )
-
-
-def _issue_rows(snapshot: LinkageSnapshot) -> str:
-    rows = []
-    for node in snapshot.issues:
-        if node.linked_prs:
-            linked = ", ".join(f"<code>#{number}</code>" for number in node.linked_prs)
-            if node.contested:
-                linked += " " + _badge("contested", "badge-blocked")
-        elif node.links_authoritative:
-            linked = '<span class="muted">none</span>'
-        else:
-            # The distinction an operator needs: nothing found in a window that
-            # was not fully loaded is not the same as nothing existing.
-            linked = '<span class="muted">none found (partial inventory)</span>'
-        rows.append(
-            "<tr>"
-            f"<td>{_ref(node)}</td>"
-            f"<td>{escape(node.title)}</td>"
-            f"<td>{escape(node.state or '—')}</td>"
-            f"<td>{_labels(node.labels)}</td>"
-            f"<td>{linked}</td>"
-            f"<td>{_handoff_cell(node)}</td>"
-            "</tr>"
-        )
-    if not rows:
-        return '<tr><td colspan="6" class="muted">No issues in the loaded window.</td></tr>'
-    return "".join(rows)
-
-
-def _pr_rows(snapshot: LinkageSnapshot) -> str:
-    rows = []
-    for node in snapshot.prs:
-        if node.links:
-            linked = "".join(
-                f"<div><code>#{link.issue_number}</code> "
-                f"{_evidence_badges(link.evidence)}</div>"
-                for link in node.links
-            )
-            if node.ambiguous:
-                linked += _badge("ambiguous", "badge-blocked")
-            if node.contested:
-                linked += " " + _badge("contested", "badge-blocked")
-        elif node.links_authoritative:
-            linked = '<span class="muted">no issue reference</span>'
-        else:
-            linked = '<span class="muted">none found (partial inventory)</span>'
-        rows.append(
-            "<tr>"
-            f"<td>{_ref(node)}</td>"
-            f"<td>{escape(node.title)}</td>"
-            f"<td>{escape(node.state or '—')}</td>"
-            f"<td>{_labels(node.labels)}</td>"
-            f"<td>{linked}</td>"
-            f"<td>{_handoff_cell(node)}</td>"
-            "</tr>"
-        )
-    if not rows:
-        return (
-            '<tr><td colspan="6" class="muted">No pull requests in the loaded '
-            "window.</td></tr>"
-        )
-    return "".join(rows)
-
-
-def _focus_card(snapshot: LinkageSnapshot) -> str:
-    """Render the focused thread's latest canonical handoff, when one was loaded."""
-    if snapshot.focus is None:
-        return f"""<div class="prompt-card">
-  <h3>Canonical handoff</h3>
-  <p class="muted">{escape(snapshot.handoff_status.reason or "")}
-  Add <code>?issue=N</code> or <code>?pr=N</code> to load the latest
-  Canonical Thread Handoff for one thread.</p>
-</div>"""
-
-    kind, number = snapshot.focus
-    target = f"{kind} #{number}"
-    if not snapshot.handoff_status.loaded:
-        return f"""<div class="prompt-card">
-  <h3>Canonical handoff — {escape(target)}</h3>
-  <p class="muted">{_badge("unavailable", "badge-health-degraded")}
-  {escape(snapshot.handoff_status.reason or "handoff source did not run")}.
-  This is not evidence that the thread carries no handoff.</p>
-</div>"""
-
-    handoff = next(
-        (
-            node.handoff
-            for node in (snapshot.issues + snapshot.prs)
-            if node.kind == kind and node.number == number and node.handoff
-        ),
-        None,
-    )
-    if handoff is None:
-        return f"""<div class="prompt-card">
-  <h3>Canonical handoff — {escape(target)}</h3>
-  <p class="muted">Comments loaded; no Canonical Thread Handoff comment found on
-  this thread.</p>
-</div>"""
-
-    type_css = "badge-health-ok" if handoff.cth_type_known else "badge-health-degraded"
-    unknown_note = (
-        ""
-        if handoff.cth_type_known
-        else (
-            '<p class="muted">The comment\'s heading is not a declared CTH type, '
-            "so it is reported as unrecognized rather than republished.</p>"
-        )
-    )
-    return f"""<div class="prompt-card">
-  <h3>Canonical handoff — {escape(target)}</h3>
-  <p class="meta">{_badge(handoff.cth_type or "—", type_css)}
-  by <code>{escape(handoff.author or "unknown")}</code>
-  at <code>{escape(handoff.created_at or "unknown")}</code></p>
-  {unknown_note}
-  <table class="detail">
-    <tr><th>Status</th><td>{escape(handoff.status or "—")}</td></tr>
-    <tr><th>Next owner</th><td>{escape(handoff.next_owner or "—")}</td></tr>
-    <tr><th>Current blocker</th><td>{escape(handoff.current_blocker or "—")}</td></tr>
-    <tr><th>Decision</th><td>{escape(handoff.decision or "—")}</td></tr>
-    <tr><th>Next action</th><td>{escape(handoff.next_action or "—")}</td></tr>
-  </table>
-  <p class="muted">Full event history:
-  <a href="/api/v1/timeline?{escape(kind)}={number}"><code>/api/v1/timeline</code></a></p>
-</div>"""
-
-
-def _legend_card(snapshot: LinkageSnapshot) -> str:
-    items = "".join(
-        f"<li>{_badge(_EVIDENCE_LABEL[name], _EVIDENCE_CSS[name])} — "
-        f"{escape(EVIDENCE_DESCRIPTIONS[name])}</li>"
-        for name in EVIDENCE_ORDER
-    )
-    reveal_note = (
-        "Gitea deep links are shown because the "
-        "<code>GITEA_MCP_REVEAL_ENDPOINTS</code> admin opt-in is set."
-        if snapshot.deep_links_enabled
-        else (
-            "Gitea deep links are withheld. Set "
-            "<code>GITEA_MCP_REVEAL_ENDPOINTS=1</code> server-side to reveal "
-            "them; item numbers stay usable without them."
-        )
-    )
-    return f"""<div class="prompt-card">
-  <h3>How an edge was found</h3>
-  <ul class="reasons">{items}</ul>
-  <p class="muted">{reveal_note}</p>
-  <p class="muted">Read-only surface: no issue or PR editing, no review, and no merge.
-  JSON export: <a href="/api/v1/gitea/linkage"><code>/api/v1/gitea/linkage</code></a></p>
-</div>"""
-
-
-def render_linkage_page(snapshot: LinkageSnapshot) -> str:
-    """Render the full HTML page for the Gitea linkage console."""
-    if not snapshot.ok:
-        return render_page(
-            title="Gitea linkage",
-            body_html=f"""<h2>Gitea issue and PR linkage</h2>
-<p class="meta">Phase 3 read-only linkage console (#645).</p>
-{_error_card(snapshot)}
-{_scope_card(snapshot)}""",
-        )
-
-    body = f"""<h2>Gitea issue and PR linkage</h2>
-<p class="meta">Phase 3 read-only linkage console (#645). Gitea remains the source of
-truth; this page reads it and never writes to it.</p>
-{_scope_card(snapshot)}
-{_contested_card(snapshot)}
-
-<div class="prompt-card">
-  <h3>Issues → pull requests</h3>
-  <table class="registry">
-    <thead>
-      <tr>
-        <th>Issue</th><th>Title</th><th>State</th><th>Labels</th>
-        <th>Linked PRs</th><th>Latest handoff</th>
-      </tr>
-    </thead>
-    <tbody>{_issue_rows(snapshot)}</tbody>
-  </table>
-</div>
-
-<div class="prompt-card">
-  <h3>Pull requests → issues</h3>
-  <table class="registry">
-    <thead>
-      <tr>
-        <th>PR</th><th>Title</th><th>State</th><th>Labels</th>
-        <th>Linked issues</th><th>Latest handoff</th>
-      </tr>
-    </thead>
-    <tbody>{_pr_rows(snapshot)}</tbody>
-  </table>
-</div>
-
-{_focus_card(snapshot)}
-{_legend_card(snapshot)}
-"""
-    return render_page(title="Gitea linkage", body_html=body)

webui/nav.py

@@ -6,8 +6,7 @@ destination is a GET view or a Phase 1 placeholder. No mutation links.
 
 Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
 Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
-Insights (placeholder), joined by the Phase 3 Gitea linkage group (#645).
-Later-phase surfaces are declared as ``stub`` items and
+Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
 backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
 instead of a 404.
 """
@@ -61,9 +60,6 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
     NavGroup("Timeline", (
         NavItem("/timeline", "Timeline", "stub"),
     )),
-    NavGroup("Gitea", (
-        NavItem("/gitea", "Issue/PR linkage"),
-    )),
     NavGroup("Policy", (
         NavItem("/policy", "Policy", "stub"),
         NavItem("/prompts", "Prompts"),

webui/queue_loader.py

@@ -211,19 +211,6 @@ def _pagination_from_pages(
     )
 
 
-_SUPPORTED_FETCH_STATES = ("open", "closed", "all")
-
-
-def _safe_state(state: str | None) -> str:
-    """Constrain a caller-supplied item state before it reaches a query string.
-
-    The value is interpolated into the Gitea URL, so an unrecognised state falls
-    back to ``open`` rather than being passed through.
-    """
-    text = (state or "open").strip().lower()
-    return text if text in _SUPPORTED_FETCH_STATES else "open"
-
-
 def _fetch_prs(
     host: str,
     org: str,
@@ -231,15 +218,8 @@ def _fetch_prs(
     auth: str,
     *,
     per_page: int = 50,
-    state: str = "open",
 ) -> tuple[list[dict], PaginationMeta]:
-    """Fetch PRs in *state* (``open``, ``closed``, or ``all``).
-
-    The queue dashboard only ever wants the open window, so ``open`` stays the
-    default. The linkage console (#645) widens it, because a landed issue↔PR
-    edge lives on a merged PR.
-    """
-    url = f"{repo_api_url(host, org, repo)}/pulls?state={_safe_state(state)}"
+    url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
     all_raw: list[dict] = []
     pages_fetched = 0
     is_final = False
@@ -268,13 +248,8 @@ def _fetch_issues(
     auth: str,
     *,
     per_page: int = 50,
-    state: str = "open",
 ) -> tuple[list[dict], PaginationMeta]:
-    """Fetch issues in *state* (``open``, ``closed``, or ``all``); see :func:`_fetch_prs`."""
-    url = (
-        f"{repo_api_url(host, org, repo)}/issues"
-        f"?state={_safe_state(state)}&type=issues"
-    )
+    url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
     all_raw: list[dict] = []
     page = 1
     pages_fetched = 0

webui/system_health_views.py

@@ -270,26 +270,47 @@ def _probe_error_card(snapshot: SystemHealthSnapshot) -> str:
 
 
 def _recovery_card() -> str:
-    """Sanctioned recovery pointers only — never a manual process kill (#630)."""
-    return (
-        "<section class='health-card'>"
-        "<h3>Recovery</h3>"
-        "<p class='muted'>This dashboard is read-only. Restart and reload "
-        "controls arrive in Phase 2 (#642); until then recovery runs through "
-        "the sanctioned client reconnect / operator restart path.</p>"
-        "<ul class='reasons'>"
-        "<li><a href='/runtime'>Runtime health</a> — active profile, workflow "
-        "hashes, and shell health.</li>"
-        "<li><a href='/sessions'>Runtime and sessions</a> — namespaces, session "
-        "rows, worktree bindings, and contamination markers (#641).</li>"
-        "<li>Reconnect the MCP client from the IDE, then re-run the blocked "
-        "cycle. Never kill the daemon process manually: unmanaged kills are "
-        "recorded as runtime contamination (#630).</li>"
-        "<li>See <code>docs/webui-local-dev.md</code> for the documented "
-        "recovery sequence.</li>"
-        "</ul>"
-        "</section>"
-    )
+    """Sanctioned recovery controls & playbooks (#644, Phase 2)."""
+    try:
+        from webui import console_recovery
+        diag = console_recovery.diagnose_recovery()
+        status_badge = f"<span class='status-pill {diag.status}'>{diag.status}</span>"
+        playbook_lis = ""
+        for pb in diag.playbooks:
+            elig = "eligible" if pb.eligible else "disabled"
+            playbook_lis += (
+                f"<li><strong>{pb.label}</strong> (<code>{pb.playbook_id}</code>) — "
+                f"<span class='badge {elig}'>{elig}</span>: {pb.description} "
+                f"<em class='muted'>({pb.reason})</em></li>"
+            )
+        reasons_html = ""
+        if diag.reasons:
+            items = "".join(f"<li>{r}</li>" for r in diag.reasons)
+            reasons_html = f"<ul class='reasons'>{items}</ul>"
+        else:
+            reasons_html = "<p class='clean-note'>No recovery actions currently required. Control plane is healthy.</p>"
+
+        return (
+            "<section class='health-card recovery-card'>"
+            f"<h3>Sanctioned Recovery Controls (Phase 2 #644) {status_badge}</h3>"
+            "<p class='muted'>Guided recovery wizard: Diagnose &rarr; Preview &rarr; Confirm &rarr; Verify. "
+            "Reconnect the MCP client from the IDE, then re-run the blocked cycle. "
+            "Never kill the daemon process manually: unmanaged kills are recorded as runtime contamination (#630).</p>"
+            f"{reasons_html}"
+            "<h4>Available Recovery Playbooks</h4>"
+            f"<ul class='playbooks-list'>{playbook_lis}</ul>"
+            "<p class='meta'>APIs: <code>/api/v1/system/recovery/diagnose</code>, "
+            "<code>/api/v1/system/recovery/preview</code>, <code>/api/v1/system/recovery/apply</code>, "
+            "<code>/api/v1/system/recovery/verify</code>.</p>"
+            "</section>"
+        )
+    except Exception as exc:
+        return (
+            "<section class='health-card'>"
+            "<h3>Sanctioned Recovery Controls (Phase 2 #644)</h3>"
+            f"<p class='error'>Recovery diagnostics unavailable: {exc}</p>"
+            "</section>"
+        )
 
 
 def render_system_health_page(snapshot: SystemHealthSnapshot) -> str: