merge(master): resolve #903 conflicts with #643 request initiation

Keep #644 recovery console actions and #643 initiate_workflow side by side
in console_authz and authz audit docs after merging latest master.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

docs/sanctioned-recovery-playbooks.md

@@ -0,0 +1,64 @@
+# 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](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](../stale_binding_recovery.py)). |
+| `rebind_session_worktree` | `system.rebind_session_worktree` | Operator | Session worktree | Rebind or synchronize session worktree to verified lease worktree ([#864](../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](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. Gates run in this order, and each fails closed before anything is mutated:
+
+1. **RBAC and execution phase** (`console_authz.authorize(..., for_execution=True)`). The phase branch only applies when `for_execution` is set. While `ACTIVE_PHASE` is `1`, every phase-2 recovery action is refused with `phase_not_active`, so no recovery playbook writes yet. Preview reports the same decision under `execution_authorization` / `execution_blocked_reason`.
+2. **Confirmation phrase** (`confirmation_matches`).
+3. **Contamination rules** ([#630](sanctioned-restart-controls.md)): the live marker is read from the session inventory and assessed under the gated task key `console_recovery_apply`. A contaminated runtime must be cleared through the reconciler cleanup playbook, which is the one playbook exempted from this gate because it is the designated remedy. The marker is also forwarded to `sanctioned_restart.execute_restart`, so a restart cannot launder a contaminated runtime.
+
+Apply then executes the sanctioned recovery logic against the **live** process environment — not a copy — and records an audit entry in `console_audit`. A playbook that leaves the binding unchanged reports `performed: false`; `binding_before`, `binding_after`, and `binding_changed` are returned so a no-op cannot read as success.
+
+Apply does **not** enforce master parity. Parity is reported by Diagnose ([#610](../master_parity_gate.py)) as evidence for the operator; it is not a precondition of this endpoint.
+
+### 4. Verify (`POST /api/v1/system/recovery/verify`)
+Re-evaluates control-plane diagnostics post-recovery and **reports** `clean`, `stale_runtime_clean`, `binding_clean`, `binding_classification`, and `contamination_clean`. It reports; it does not assert or block. State is read fresh rather than from the mapping a mutation just wrote. An `unverified_inherited` binding is reported as not clean, because unproven is not clean.
+
+---
+
+## 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 |
 | `initiate_workflow` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
 
 **Dual control** means the acting principal may not be the sole authority: a

stable_branch_push_guard.py

@@ -63,6 +63,11 @@ CONTAMINATION_GATED_TASKS = frozenset({
     "merge_pr",
     "delete_branch",
     "complete_issue",
+    # Web console recovery playbooks that write (#644). These mutate runtime
+    # binding and process state, so a live contamination marker must block them
+    # exactly as it blocks the Gitea-side mutations above. The reconciler
+    # cleanup playbook is the designated remedy and is exempted by its caller.
+    "console_recovery_apply",
 })
 
 CONTAMINATION_KIND = "stable_branch_push"

task_capability_map.py

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

webui/app.py

@@ -207,6 +207,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)))
@@ -976,6 +1054,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

@@ -283,6 +283,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.",
+    ),
     # #643: submit a work request — desired role, issue/PR, intent — and let
     # the allocator reserve it. This is the one Phase 2 action whose execution
     # path is actually implemented (``webui.request_service``), so it carries

webui/console_recovery.py

@@ -0,0 +1,721 @@
+"""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 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,
+}
+
+#: Task key handed to :func:`runtime_recovery_guard.assess_contamination_gate`.
+#: A console *action id* is not a task name and is not a member of
+#: ``CONTAMINATION_GATED_TASKS``, so passing one left the #630 gate inert. Every
+#: writing recovery playbook shares this one gated task key; the reconciler
+#: cleanup playbook is exempted separately because it is the designated remedy.
+CONTAMINATION_GATED_TASK = "console_recovery_apply"
+
+#: Remote whose contamination markers govern this console. Markers are written
+#: per remote, so reading the wrong one reports a contaminated runtime clean.
+REMOTE_ENV = "WEBUI_GITEA_REMOTE"
+DEFAULT_REMOTE = "prgs"
+
+
+def _console_remote(env: dict[str, str] | None = None) -> str:
+    env_map = env if env is not None else os.environ
+    return (env_map.get(REMOTE_ENV) or "").strip() or DEFAULT_REMOTE
+
+
+def load_active_contamination_marker(
+    remote: str | None = None, env: dict[str, str] | None = None
+) -> dict[str, Any] | None:
+    """Return the live #630 contamination marker payload, or ``None``.
+
+    The gate is only meaningful when it is fed a real marker: with
+    ``marker=None`` :func:`assess_contamination_gate` returns ``block: False``
+    on its first statement. The #641 session inventory already reads the durable
+    markers, so reuse that reader rather than adding a second source of truth.
+    Never raises into a diagnosis or execution path.
+    """
+    try:
+        from webui import session_loader
+    except Exception:  # noqa: BLE001 — never break recovery on an import problem
+        return None
+    try:
+        markers = session_loader._load_contamination_markers(
+            remote=remote or _console_remote(env)
+        )
+    except Exception:  # noqa: BLE001 — fail soft; the caller degrades to no marker
+        return None
+    for marker in markers:
+        payload = marker.to_dict()
+        if payload.get("active"):
+            return payload
+    return None
+
+
+def _active_binding(env_map: Any) -> str | None:
+    """Read the live worktree binding so a no-op recovery cannot report success."""
+    value = env_map.get(stale_binding_recovery.ACTIVE_WORKTREE_ENV)
+    return value if value else None
+
+
+@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
+    #
+    # The baseline is the commit the *running process* started at, which is what
+    # the parity gate is about. Capturing it from ``checkout_head`` and then
+    # comparing it against that same value made ``in_parity`` structurally
+    # incapable of being false. ``live_remote_head`` restores the #610
+    # live-remote dimension, which was previously dropped.
+    checkout_head = stale_runtime_obj.checkout_head
+    startup_dict = master_parity_gate.capture_startup_parity(
+        str(root), head=stale_runtime_obj.daemon_head
+    )
+    parity_dict = master_parity_gate.assess_master_parity(
+        startup_dict, checkout_head, stale_runtime_obj.remote_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)
+    #
+    # A real marker and a task key the gate actually gates on: with marker=None
+    # the gate short-circuits to ``block: False``, and with a console action id
+    # the task is outside CONTAMINATION_GATED_TASKS, so it could never block.
+    contamination_marker = load_active_contamination_marker(env=source_env)
+    contamination_dict = runtime_recovery_guard.assess_contamination_gate(
+        contamination_marker,
+        task=CONTAMINATION_GATED_TASK,
+        actual_role=role_kind,
+    )
+    contaminated = bool(contamination_dict.get("block"))
+    if contaminated:
+        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 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 contaminated
+    if 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)
+    # Preview and apply must answer the same question. ``execution_enabled`` was
+    # a hardcoded False beside an authorization decision taken without
+    # ``for_execution``, so the preview could not tell an operator *why*
+    # execution was disabled — and the apply path did not ask at all.
+    execution_decision = console_authz.authorize(
+        action_id, principal, for_execution=True
+    )
+    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(),
+        "execution_authorization": execution_decision.to_dict(),
+        "params": dict(params or {}),
+        "execution_enabled": bool(execution_decision.allowed),
+        "execution_blocked_reason": (
+            None if execution_decision.allowed else execution_decision.reason_code
+        ),
+    }
+
+
+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]
+    # The mapping the playbooks actually mutate. ``dict(os.environ)`` produced a
+    # throwaway copy: every env playbook wrote to it, verified against it, and
+    # left the running daemon bound to the value it claimed to have fixed.
+    mutation_env: Any = env if env is not None else os.environ
+
+    # 1. Authorization check — ``for_execution=True`` is what arms the phase
+    # gate (console_authz.authorize only applies it in that branch). Without it
+    # a phase-2 write executed while the console was in phase 1.
+    decision = console_authz.authorize(action_id, principal, for_execution=True)
+    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
+    contamination_marker = load_active_contamination_marker(env=env)
+    contam = runtime_recovery_guard.assess_contamination_gate(
+        contamination_marker,
+        task=CONTAMINATION_GATED_TASK,
+        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:
+        binding_before = _active_binding(mutation_env)
+        diagnosis = diagnose_recovery(env=env)
+        plan = stale_binding_recovery.plan_recovery(diagnosis.stale_binding)
+        applied_result = stale_binding_recovery.apply_recovery(plan, env=mutation_env)
+        binding_after = _active_binding(mutation_env)
+        applied_result = {
+            **applied_result,
+            "binding_before": binding_before,
+            "binding_after": binding_after,
+            "binding_changed": binding_before != binding_after,
+        }
+        # A clear that did not clear is not a success, whatever the plan said.
+        if not applied_result["binding_changed"]:
+            applied_result["performed"] = False
+            applied_result.setdefault("reasons", []).append(
+                "clear_stale_binding did not change the live worktree binding"
+            )
+    elif playbook_id == PLAYBOOK_REBIND_SESSION:
+        target_wt = target or (params or {}).get("target_worktree")
+        if target_wt:
+            binding_before = _active_binding(mutation_env)
+            mutation_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV] = target_wt
+            binding_after = _active_binding(mutation_env)
+            applied_result = {
+                "performed": binding_after == target_wt,
+                "rebound_worktree": target_wt,
+                "binding_before": binding_before,
+                "binding_after": binding_after,
+                "binding_changed": binding_before != binding_after,
+                "cleared_stale": binding_before != binding_after,
+            }
+            if binding_after != target_wt:
+                applied_result["reasons"] = [
+                    "rebind_session_worktree did not take effect on the live "
+                    "environment"
+                ]
+        else:
+            applied_result = {
+                "performed": False,
+                "reason": "No target_worktree specified for rebind.",
+            }
+    elif playbook_id == PLAYBOOK_RECONCILE_CLEANUPS:
+        # ``merged_cleanup_reconcile`` exposes the building blocks only; the
+        # orchestrator is the MCP tool. The previous call named a function that
+        # does not exist, and a bare ``except`` turned the AttributeError into a
+        # generic failure, so this playbook could never succeed. Imported lazily
+        # because the MCP server module is large and binds FastMCP at import.
+        try:
+            import gitea_mcp_server
+
+            snapshot = gitea_mcp_server.gitea_reconcile_merged_cleanups(
+                dry_run=False,
+                execute_confirmed=True,
+                remote=(params or {}).get("remote") or _console_remote(env),
+                org=(params or {}).get("org"),
+                repo=(params or {}).get("repo"),
+            )
+            performed_reconcile = bool(snapshot.get("success"))
+            applied_result = {
+                "performed": performed_reconcile,
+                "reconciled_count": len(snapshot.get("entries") or []),
+                "snapshot": snapshot,
+            }
+            if not performed_reconcile:
+                applied_result["reasons"] = list(snapshot.get("reasons") or [])
+        except Exception as exc:  # noqa: BLE001 — surfaced with its type
+            applied_result = {
+                "performed": False,
+                "error": str(exc),
+                "error_type": type(exc).__name__,
+            }
+    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}",
+            # Without the marker the stricter guard at sanctioned_restart.py:375
+            # never fires and a restart can launder a contaminated runtime.
+            contamination_marker=contamination_marker,
+            env=mutation_env,
+            request_id=request_id,
+            session_id=session_id,
+        )
+        applied_result = restart_res
+
+    # ``allowed`` is not ``performed``: execute_restart documents that success is
+    # False in both directions because the host supervisor still has to act.
+    performed = bool(applied_result.get("performed"))
+
+    # 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.
+    #
+    # Re-read state rather than re-reading the mapping the mutation just wrote:
+    # verifying the mutated copy confirmed changes that never reached the
+    # process. Passing ``env`` through means a caller-supplied mapping is the
+    # live one for that caller, and ``None`` re-reads ``os.environ`` fresh.
+    post_verification = verify_post_recovery(env=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)
+    classification = diag.stale_binding.get("classification")
+    # ``not clear_eligible`` also reads clean for every binding recovery is not
+    # allowed to touch — an unverified inherited binding is unproven, not clean.
+    binding_clean = (
+        not diag.stale_binding.get("clear_eligible")
+        and classification != stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED
+    )
+    return {
+        "clean": diag.clean,
+        "status": diag.status,
+        "stale_runtime_clean": not diag.stale_runtime.get("stale"),
+        "binding_clean": binding_clean,
+        "binding_classification": classification,
+        # The gate returns ``block``; it has never returned ``contaminated``, so
+        # reading that key reported every runtime clean unconditionally.
+        "contamination_clean": not diag.contamination.get("block"),
+        "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/system_health_views.py

@@ -270,26 +270,53 @@ 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()
+        # Every other card in this file escapes at the interpolation boundary.
+        # This one did not, and it is where a #630 marker's operator-supplied
+        # command_summary lands once the contamination gate is wired.
+        status_badge = (
+            f"<span class='status-pill {_esc(diag.status)}'>{_esc(diag.status)}</span>"
+        )
+        playbook_lis = ""
+        for pb in diag.playbooks:
+            elig = "eligible" if pb.eligible else "disabled"
+            playbook_lis += (
+                f"<li><strong>{_esc(pb.label)}</strong> "
+                f"(<code>{_esc(pb.playbook_id)}</code>) — "
+                f"<span class='badge {elig}'>{elig}</span>: {_esc(pb.description)} "
+                f"<em class='muted'>({_esc(pb.reason)})</em></li>"
+            )
+        reasons_html = ""
+        if diag.reasons:
+            items = "".join(f"<li>{_esc(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: {_esc(exc)}</p>"
+            "</section>"
+        )
 
 
 def render_system_health_page(snapshot: SystemHealthSnapshot) -> str: