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

Stale-runtime and runtime-mode mutation refusals previously shared the
permission-denial channel, so permission_report claimed a missing op the
active profile already held and recommended gitea_activate_profile.
Typed blocker_kind payloads report reconnect-only recovery for staleness,
omit permission_report for non-permission gates, and fail closed when a
permission_report would invent a missing permission the profile holds.

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

@@ -77,7 +77,9 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
-| `/sessions` | Phase 1 shell stub — session inventory (backed by #636) |
+| `/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` |
 | `/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 |
@@ -284,11 +286,46 @@ The header carries two read-only status badges — an **environment** badge
 a **mode: read-only** badge — plus a **Docs** link to this document. No
 privileged action controls are present in the Phase 1 shell.
 
-Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
-`/policy`, `/insights`) resolve to graceful read-only stub pages instead of
-404s; their backing views land in later child issues of #631 (the inventory
-surfaces are backed by #636). Mutating methods on stub routes still fail closed
-with `read-only-mvp`.
+Not-yet-implemented surfaces (`/inventory`, `/timeline`, `/policy`,
+`/insights`) resolve to graceful read-only stub pages instead of 404s; their
+backing views land in later child issues of #631 (the inventory surfaces are
+backed by #636). Mutating methods on stub routes still fail closed with
+`read-only-mvp`.
+
+### Runtime and sessions (#641)
+
+`/sessions` is a live Phase 1 read-only view that composes:
+
+* runtime health from `#430` (profile, role, identity, master parity, stale warning)
+* control-plane sessions / leases and filesystem locks / worktrees / namespaces from `#636`
+* durable contamination markers when detectable (`#630` runtime recovery, `#671` stable-branch push)
+
+It surfaces stale indicators (dead PID, expired lease) and never silences an
+active contamination marker. Recovery links point only at sanctioned
+reconnect/operator restart docs (`docs/mcp-namespace-eof-recovery.md`,
+`docs/mcp-namespace-health.md`, `docs/mcp-restart-path-inventory.md`, this
+document). The page does **not** restart, kill, or take over sessions; manual
+`pkill` of MCP daemons is contamination, not recovery.
+
+Honesty rules specific to this view:
+
+* **Ownership columns never assert absence they cannot prove.** When the
+  `leases` or `locks` section is degraded or unavailable, the Leases and
+  Worktree-binding cells render `unknown (inventory <status>)` with an
+  *authority unproven* badge instead of `none` / `unbound`, and a caveat names
+  the unreadable sections. A worktree binding is correlated through lease work
+  numbers, so it is unproven when *either* section fails to read.
+  `/api/sessions` carries the same facts as `ownership_authority_complete`,
+  `ownership_section_status`, and per-row `lease_authority` /
+  `worktree_authority`, so a JSON consumer can tell "holds none" from "could
+  not be read".
+* **Contamination text is redacted at the display boundary.** Marker payloads
+  (`command_summary`, `reason_class`, `session_id`, `role`) are
+  operator-supplied free text that does not arrive through inventory scrubbing,
+  so they pass through `webui.inventory.scrub_text`, which collapses `$HOME` and
+  redacts credential-shaped tokens and URL userinfo *anywhere* in the string.
+  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.
 
 ## System-health dashboard (#639)
 

gitea_mcp_server.py

@@ -9552,15 +9552,13 @@ def gitea_edit_pr(
     if closing:
         gate_reasons = _profile_operation_gate("gitea.pr.close")
         if gate_reasons:
-            return {
-                "success": False,
-                "performed": False,
-                "pr_number": pr_number,
-                "requested_state": "closed",
-                "required_permission": "gitea.pr.close",
-                "reasons": gate_reasons,
-                "permission_report": _permission_block_report("gitea.pr.close"),
-            }
+            return _build_operation_gate_refusal(
+                "gitea.pr.close",
+                gate_reasons,
+                pr_number=pr_number,
+                requested_state="closed",
+                required_permission="gitea.pr.close",
+            )
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -13823,13 +13821,17 @@ def gitea_view_issue(
 
 def _permission_block_report(required_operation: str,
                              identity: str | None = None) -> dict:
-    """Structured, LLM-safe explanation of a permission denial (#142).
+    """Structured, LLM-safe explanation of a permission denial (#142, #897).
 
     Built only after a gate has already refused; it adds guidance to the
     refusal and never widens any permission, performs network I/O, or
     raises (fail-soft: degrades to a minimal fail-closed report). Names
     configured profiles only — never auth references, tokens, endpoint
     URLs, or keychain IDs.
+
+    #897: never fabricate a missing permission when the active profile
+    already allows the operation. That path is a diagnostic defect (the
+    refusal was not a permission denial), not a cue to switch profiles.
     """
     report = {
         "requested_operation": required_operation,
@@ -13841,6 +13843,7 @@ def _permission_block_report(required_operation: str,
         "matching_configured_profiles": [],
         "runtime_switching_supported": False,
         "different_mcp_namespace_required": True,
+        "diagnostic_defect": False,
         "exact_safe_next_action": (
             "Ask the operator to fix GITEA_MCP_CONFIG/GITEA_MCP_PROFILE; "
             "the active profile could not be resolved (fail closed)."),
@@ -13855,6 +13858,32 @@ def _permission_block_report(required_operation: str,
     report["active_allowed_operations"] = (
         profile.get("allowed_operations") or [])
 
+    # #897: fail closed as a diagnostic defect when the active profile
+    # already holds the operation — callers must not invent "missing".
+    try:
+        holds, _hold_reason = gitea_config.check_operation(
+            required_operation,
+            profile.get("allowed_operations") or [],
+            profile.get("forbidden_operations") or [],
+        )
+    except Exception:
+        holds = False
+    if holds:
+        report["missing_permission"] = None
+        report["required_permission"] = required_operation
+        report["diagnostic_defect"] = True
+        report["different_mcp_namespace_required"] = False
+        report["exact_safe_next_action"] = (
+            "Diagnostic defect: the active profile already allows "
+            f"{required_operation}. This is not a permission denial — "
+            "inspect blocker_kind / reasons (stale-runtime or runtime-mode). "
+            "Do not call gitea_activate_profile or switch MCP sessions."
+        )
+        report["matching_configured_profiles"] = [
+            p for p in [profile.get("profile_name")] if p
+        ]
+        return report
+
     matching = []
     try:
         config = gitea_config.load_config() or {}
@@ -13903,6 +13932,205 @@ def _permission_block_report(required_operation: str,
     return report
 
 
+def _reason_is_stale_runtime(reason: str) -> bool:
+    """True when *reason* is a master-parity / stale-daemon refusal (#897)."""
+    r = (reason or "").lower()
+    if not r:
+        return False
+    if "stale relative to live master" in r:
+        return True
+    if "server code is stale" in r:
+        return True
+    if "daemon is stale" in r:
+        return True
+    if "started at commit" in r and "workspace master is now" in r:
+        return True
+    if "mcp server started at" in r and "stale" in r:
+        return True
+    if "restart the server to load the current capability gates" in r:
+        return True
+    if "restart/reconnect before mutating" in r:
+        return True
+    return False
+
+
+def _reason_is_runtime_mode(reason: str) -> bool:
+    """True when *reason* is a stable-control / runtime-mode refusal (#897)."""
+    r = (reason or "").lower()
+    if not r:
+        return False
+    if _reason_is_stale_runtime(reason):
+        return False
+    if "runtime mode could not be assessed" in r:
+        return True
+    if "runtime mode is" in r:
+        return True
+    if "stable control runtime" in r:
+        return True
+    if "dev-test" in r and ("runtime" in r or "production" in r):
+        return True
+    if "development worktree" in r or "dev worktree" in r:
+        return True
+    if "launched from a 'branches/" in r or "launched from a \"branches/" in r:
+        return True
+    if "process-root / active-workspace alignment" in r:
+        return True
+    if "namespace" in r and "reproof" in r:
+        return True
+    return False
+
+
+def _reason_is_permission(reason: str) -> bool:
+    """True when *reason* is a genuine profile-permission denial (#897)."""
+    r = (reason or "").lower()
+    if not r:
+        return False
+    if _reason_is_stale_runtime(reason) or _reason_is_runtime_mode(reason):
+        return False
+    if "profile could not be resolved" in r:
+        return True
+    if "profile has no configured allowed operations" in r:
+        return True
+    if "profile forbids" in r:
+        return True
+    if "profile is not allowed to" in r:
+        return True
+    if "unrecognized forbidden operation" in r:
+        return True
+    return False
+
+
+def _classify_operation_gate_reasons(reasons: list[str]) -> dict:
+    """Partition gate reasons into stale / runtime-mode / permission (#897)."""
+    stale: list[str] = []
+    runtime_mode: list[str] = []
+    permission: list[str] = []
+    other: list[str] = []
+    for reason in reasons or []:
+        if _reason_is_stale_runtime(reason):
+            stale.append(reason)
+        elif _reason_is_runtime_mode(reason):
+            runtime_mode.append(reason)
+        elif _reason_is_permission(reason):
+            permission.append(reason)
+        else:
+            other.append(reason)
+    return {
+        "stale_runtime": stale,
+        "runtime_mode": runtime_mode,
+        "permission": permission,
+        "other": other,
+    }
+
+
+def _stale_runtime_reconnect_action() -> str:
+    """Sanctioned recovery for a stale daemon — reconnect only (#685/#897)."""
+    return (
+        "Reconnect the IDE/client MCP session so the server reloads at the "
+        "current master head. Do not call gitea_activate_profile or switch "
+        "MCP role sessions — profile switching does not clear a stale daemon."
+    )
+
+
+def _build_operation_gate_refusal(
+    required_operation: str,
+    reasons: list[str],
+    **extra_fields,
+) -> dict:
+    """Structured gate refusal with typed blockers (#897).
+
+    Stale-runtime and runtime-mode refusals never attach a
+    ``permission_report`` and never recommend profile switching. True
+    permission denials still get ``permission_report``. When both apply,
+    causes are reported separately under distinct fields.
+    """
+    classified = _classify_operation_gate_reasons(reasons)
+    stale = classified["stale_runtime"]
+    runtime_mode = classified["runtime_mode"]
+    permission = classified["permission"]
+    other = classified["other"]
+
+    blocked: dict = {
+        "success": False,
+        "performed": False,
+        "reasons": list(reasons),
+        "mutation_performed": False,
+        "session_context_audit": session_ctx.mutation_context_audit_fields(),
+        "gate_reason_classes": {
+            "stale_runtime": list(stale),
+            "runtime_mode": list(runtime_mode),
+            "permission": list(permission),
+            "other": list(other),
+        },
+    }
+
+    if stale:
+        parity = _current_master_parity()
+        blocked["blocker_kind"] = "runtime_reconnect_required"
+        blocked["restart_required"] = True
+        blocked["stop_required"] = True
+        blocked["startup_head"] = parity.get("startup_head")
+        blocked["current_head"] = parity.get("current_head")
+        blocked["daemon_start_head"] = (
+            parity.get("daemon_start_head") or parity.get("startup_head")
+        )
+        blocked["local_head"] = (
+            parity.get("local_head") or parity.get("current_head")
+        )
+        blocked["live_remote_head"] = parity.get("live_remote_head")
+        blocked["live_stale"] = bool(parity.get("live_stale"))
+        blocked["live_known"] = bool(parity.get("live_known"))
+        blocked["exact_safe_next_action"] = _stale_runtime_reconnect_action()
+        if permission or other:
+            blocked["permission_block_reasons"] = list(permission) + list(other)
+            blocked["stale_runtime_reasons"] = list(stale)
+        # Never attach permission_report for a staleness refusal.
+        blocked.update(extra_fields)
+        return blocked
+
+    if runtime_mode:
+        blocked["blocker_kind"] = "runtime_mode_blocked"
+        blocked["restart_required"] = False
+        blocked["stop_required"] = True
+        blocked["exact_safe_next_action"] = (
+            "Real workflow mutations run only on the promoted stable control "
+            "runtime. Promote/reload the stable runtime; do not call "
+            "gitea_activate_profile or switch MCP role sessions to clear a "
+            "runtime-mode block."
+        )
+        if permission or other:
+            blocked["permission_block_reasons"] = list(permission) + list(other)
+            blocked["runtime_mode_reasons"] = list(runtime_mode)
+        blocked.update(extra_fields)
+        return blocked
+
+    # Pure permission (or unclassified-as-permission) denial.
+    blocked["blocker_kind"] = "permission_denied"
+    blocked["permission_report"] = _permission_block_report(required_operation)
+    blocked.update(extra_fields)
+    return blocked
+
+
+def _permission_report_for_gate_reasons(
+    required_operation: str,
+    reasons: list[str] | None,
+) -> dict | None:
+    """Attach ``permission_report`` only for true permission denials (#897).
+
+    Call sites that historically always attached a permission report after
+    ``_profile_operation_gate`` should use this so stale/runtime refusals
+    do not emit a fabricated missing-permission payload.
+    """
+    if not reasons:
+        return None
+    classified = _classify_operation_gate_reasons(reasons)
+    if classified["stale_runtime"] or classified["runtime_mode"]:
+        return None
+    if not (classified["permission"] or classified["other"]):
+        return None
+    return _permission_block_report(required_operation)
+
+
 def _role_for_operation(op: str) -> str | None:
     # Normalize op first
     try:
@@ -14079,7 +14307,7 @@ def _master_parity_block(op: str) -> list[str]:
 
 
 def _profile_operation_gate(op: str) -> list[str]:
-    """Profile permission check for a single gated operation (#126, #216, #420).
+    """Profile permission check for a single gated operation (#126, #216, #420, #897).
 
     Issue discussion comments are gated separately from the gitea.pr.*
     review/merge family: listing requires ``gitea.read``, creating requires
@@ -14092,21 +14320,26 @@ def _profile_operation_gate(op: str) -> list[str]:
     capability gate that has since been merged, and when the runtime itself is
     not the promoted stable control runtime (#615) -- a dev/test or unknown
     runtime holds production credentials but has not been promoted.
+
+    #897: collect *all* independent refusal classes (stale, runtime-mode,
+    permission) rather than short-circuiting after the first. Callers that
+    only need a boolean still treat any non-empty list as blocked; typed
+    consumers (``_build_operation_gate_refusal``) can separate causes.
     """
-    stale_reasons = _master_parity_block(op)
-    if stale_reasons:
-        return stale_reasons
-    runtime_reasons = _runtime_mode_block(op)
-    if runtime_reasons:
-        return runtime_reasons
+    reasons: list[str] = []
+    reasons.extend(_master_parity_block(op))
+    reasons.extend(_runtime_mode_block(op))
     try:
         profile = get_profile()
     except Exception as exc:
-        return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
+        reasons.append(
+            f"profile could not be resolved (fail closed): {_redact(str(exc))}"
+        )
+        return reasons
     op_ok, op_reason = gitea_config.check_operation(
         op, profile["allowed_operations"], profile["forbidden_operations"])
     if op_ok:
-        return []
+        return reasons
 
     if _try_auto_switch_for_operation(op):
         try:
@@ -14114,17 +14347,26 @@ def _profile_operation_gate(op: str) -> list[str]:
             op_ok, op_reason = gitea_config.check_operation(
                 op, profile["allowed_operations"], profile["forbidden_operations"])
             if op_ok:
-                return []
+                return reasons
         except Exception as exc:
-            return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
+            reasons.append(
+                f"profile could not be resolved (fail closed): {_redact(str(exc))}"
+            )
+            return reasons
 
     if op_reason == "no-allowed-operations":
-        return ["profile has no configured allowed operations (fail closed)"]
-    if op_reason == "forbidden":
-        return [f"profile forbids '{op}'"]
-    if op_reason == "invalid-forbidden-entry":
-        return ["profile has an unrecognized forbidden operation entry (fail closed)"]
-    return [f"profile is not allowed to {op}"]
+        reasons.append(
+            "profile has no configured allowed operations (fail closed)"
+        )
+    elif op_reason == "forbidden":
+        reasons.append(f"profile forbids '{op}'")
+    elif op_reason == "invalid-forbidden-entry":
+        reasons.append(
+            "profile has an unrecognized forbidden operation entry (fail closed)"
+        )
+    else:
+        reasons.append(f"profile is not allowed to {op}")
+    return reasons
 
 
 def _mutation_config_authority_block(required_operation: str) -> dict | None:
@@ -14344,10 +14586,14 @@ def _session_context_mutation_block(
 
 
 def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
-    """Structured permission denial for gated tools (#69, #142).
+    """Structured operation-gate denial for gated tools (#69, #142, #897).
 
     Returns a block dict when the active profile forbids *required_operation*,
-    or ``None`` when the gate passes. Never performs network I/O.
+    the daemon is stale, or the runtime mode is not mutation-safe — or
+    ``None`` when the gate passes. Never performs network I/O.
+
+    #897: stale-runtime and runtime-mode refusals are typed
+    (``blocker_kind``) and never carry a ``permission_report``.
     """
     req_role = "reviewer" if any(required_operation.startswith(p) for p in (
         "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review"
@@ -14357,15 +14603,9 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
 
     reasons = _profile_operation_gate(required_operation)
     if reasons:
-        blocked = {
-            "success": False,
-            "performed": False,
-            "reasons": reasons,
-            "permission_report": _permission_block_report(required_operation),
-            "session_context_audit": session_ctx.mutation_context_audit_fields(),
-        }
-        blocked.update(extra_fields)
-        return blocked
+        return _build_operation_gate_refusal(
+            required_operation, reasons, **extra_fields
+        )
 
     auth_block = _mutation_config_authority_block(required_operation)
     if auth_block is not None:
@@ -14494,20 +14734,14 @@ def gitea_acquire_reviewer_pr_lease(
     """Acquire a per-PR reviewer lease before review/merge mutations (#407)."""
     read_block = _profile_operation_gate("gitea.read")
     if read_block:
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
+        return _build_operation_gate_refusal(
+            "gitea.read", read_block, acquired=False
+        )
     comment_block = _profile_operation_gate("gitea.pr.comment")
     if comment_block:
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": comment_block,
-            "permission_report": _permission_block_report("gitea.pr.comment"),
-        }
+        return _build_operation_gate_refusal(
+            "gitea.pr.comment", comment_block, acquired=False
+        )
 
     # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
     # anti-stomp for the declared lease-acquire mutation inventory entry.
@@ -14623,20 +14857,14 @@ def gitea_acquire_merger_pr_lease(
     """
     read_block = _profile_operation_gate("gitea.read")
     if read_block:
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
+        return _build_operation_gate_refusal(
+            "gitea.read", read_block, acquired=False
+        )
     comment_block = _profile_operation_gate("gitea.pr.comment")
     if comment_block:
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": comment_block,
-            "permission_report": _permission_block_report("gitea.pr.comment"),
-        }
+        return _build_operation_gate_refusal(
+            "gitea.pr.comment", comment_block, acquired=False
+        )
     merge_block = _profile_operation_gate("gitea.pr.merge")
     if merge_block:
         return {
@@ -19375,14 +19603,12 @@ def gitea_update_pr_branch_by_merge(
     # Permission: author branch push / PR mutation surface.
     push_block = _profile_operation_gate("gitea.branch.push")
     if push_block:
-        return {
-            "success": False,
-            "performed": False,
-            "mutation_allowed": False,
-            "reasons": push_block,
-            "permission_report": _permission_block_report("gitea.branch.push"),
-            "role_kind": role,
-        }
+        return _build_operation_gate_refusal(
+            "gitea.branch.push",
+            push_block,
+            mutation_allowed=False,
+            role_kind=role,
+        )
 
     if role != "author":
         pre = pr_sync_status.assess_update_pr_branch_preflight(

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_issue_897_permission_stale_runtime_classification.py

@@ -0,0 +1,453 @@
+"""#897: stale-runtime / runtime-mode refusals must not look like permission denials.
+
+Acceptance criteria (issue #897):
+
+* Stale-runtime and runtime-mode refusals are typed distinctly from
+  profile-permission refusals (distinct ``blocker_kind``).
+* A refusal caused by staleness or runtime mode never emits a
+  ``permission_report`` and never names a permission the active profile holds.
+* ``_permission_block_report`` verifies the active profile actually lacks the
+  operation before reporting it missing.
+* A stale-runtime refusal reports reconnect-only recovery and never recommends
+  ``gitea_activate_profile`` or an MCP session switch.
+* The blocker payload states the observed heads (parity fields).
+* Matrix across author / reviewer / merger / reconciler profiles.
+* Regression: ``gitea_create_issue`` on a stale daemon under ``prgs-author``
+  never returns ``missing_permission: gitea.issue.create``.
+"""
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_config  # noqa: E402
+import gitea_mcp_server as mcp_server  # noqa: E402
+
+SHA_START = "7af40fb5ff7debd5e9165fe97d9c7c279358e175"
+SHA_LIVE = "2f4dec832327513118f2fe92b74da25d124a01cb"
+
+ROLE_MATRIX = (
+    (
+        "prgs-author",
+        "author",
+        "gitea.issue.create",
+        [
+            "gitea.read",
+            "gitea.issue.create",
+            "gitea.issue.comment",
+            "gitea.issue.close",
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.pr.create",
+            "gitea.pr.comment",
+            "gitea.repo.commit",
+        ],
+        ["gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes"],
+        "gitea.pr.merge",  # forbidden op for pure-permission case
+    ),
+    (
+        "prgs-reviewer",
+        "reviewer",
+        "gitea.pr.review",
+        [
+            "gitea.read",
+            "gitea.pr.review",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+        ],
+        ["gitea.branch.push", "gitea.pr.create"],
+        "gitea.branch.push",
+    ),
+    (
+        "prgs-merger",
+        "merger",
+        "gitea.pr.merge",
+        [
+            "gitea.read",
+            "gitea.pr.merge",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+        ],
+        ["gitea.pr.approve", "gitea.branch.push", "gitea.pr.create"],
+        "gitea.branch.push",
+    ),
+    (
+        "prgs-reconciler",
+        "reconciler",
+        "gitea.branch.delete",
+        [
+            "gitea.read",
+            "gitea.branch.delete",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+            "gitea.pr.close",
+            "gitea.issue.close",
+        ],
+        ["gitea.pr.approve", "gitea.pr.merge"],
+        "gitea.pr.merge",
+    ),
+)
+
+
+def _profile(name: str, role: str, allowed: list[str], forbidden: list[str]) -> dict:
+    return {
+        "profile_name": name,
+        "role": role,
+        "role_kind": role,
+        "allowed_operations": list(allowed),
+        "forbidden_operations": list(forbidden),
+        "identity": "test-user",
+    }
+
+
+def _config(profiles: dict) -> dict:
+    return {
+        "version": 2,
+        "profiles": {
+            name: {
+                "role": p["role"],
+                "allowed_operations": p["allowed_operations"],
+                "forbidden_operations": p["forbidden_operations"],
+            }
+            for name, p in profiles.items()
+        },
+        "rules": {"allow_runtime_switching": True},
+    }
+
+
+class Issue897Helpers(unittest.TestCase):
+    def test_classify_stale_reason_strings(self):
+        stale = (
+            f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
+            f"at {SHA_START[:12]}; the daemon is stale relative to live master "
+            "-- restart/reconnect before mutating"
+        )
+        classified = mcp_server._classify_operation_gate_reasons([stale])
+        self.assertEqual(classified["stale_runtime"], [stale])
+        self.assertEqual(classified["permission"], [])
+        self.assertEqual(classified["runtime_mode"], [])
+
+    def test_classify_permission_reason(self):
+        reason = "profile is not allowed to gitea.pr.merge"
+        classified = mcp_server._classify_operation_gate_reasons([reason])
+        self.assertEqual(classified["permission"], [reason])
+        self.assertEqual(classified["stale_runtime"], [])
+
+    def test_classify_runtime_mode_reason(self):
+        reason = (
+            "runtime mode is 'dev-test' and the mutation targets the "
+            "production repository; dev/test runtimes must not mutate real "
+            "issues or PRs (ADR: stable control runtime vs dev runtime)"
+        )
+        classified = mcp_server._classify_operation_gate_reasons([reason])
+        self.assertEqual(classified["runtime_mode"], [reason])
+        self.assertEqual(classified["stale_runtime"], [])
+
+
+class Issue897PermissionBlockReport(unittest.TestCase):
+    def test_holds_op_is_diagnostic_defect_not_missing_permission(self):
+        profile = _profile(
+            "prgs-author",
+            "author",
+            ["gitea.read", "gitea.issue.create", "gitea.issue.comment"],
+            [],
+        )
+        with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server.gitea_config, "load_config", return_value=_config({"prgs-author": profile})
+        ), patch.object(
+            mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=True
+        ):
+            report = mcp_server._permission_block_report("gitea.issue.create")
+        self.assertTrue(report.get("diagnostic_defect"), report)
+        self.assertIsNone(report.get("missing_permission"), report)
+        action = (report.get("exact_safe_next_action") or "").lower()
+        # Must not *recommend* profile switching; mentioning the forbidden
+        # action in a "do not call" instruction is fine.
+        self.assertNotIn("call gitea_activate_profile with", action)
+        self.assertNotIn("switch to the author mcp session", action)
+        self.assertNotIn("switch to the reviewer mcp session", action)
+        self.assertIn("diagnostic defect", action)
+
+    def test_true_missing_permission_still_reports(self):
+        profile = _profile(
+            "prgs-author",
+            "author",
+            ["gitea.read", "gitea.issue.create"],
+            ["gitea.pr.merge"],
+        )
+        reviewer = _profile(
+            "prgs-reviewer",
+            "reviewer",
+            ["gitea.read", "gitea.pr.merge", "gitea.pr.approve"],
+            [],
+        )
+        with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server.gitea_config,
+            "load_config",
+            return_value=_config({"prgs-author": profile, "prgs-reviewer": reviewer}),
+        ), patch.object(
+            mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=True
+        ):
+            report = mcp_server._permission_block_report("gitea.pr.merge")
+        self.assertFalse(report.get("diagnostic_defect"), report)
+        self.assertEqual(report.get("missing_permission"), "gitea.pr.merge")
+        self.assertIn("prgs-reviewer", report.get("matching_configured_profiles") or [])
+
+
+class Issue897GateRefusalMatrix(unittest.TestCase):
+    def _stale_parity(self) -> dict:
+        return {
+            "in_parity": True,
+            "stale": False,
+            "restart_required": True,
+            "determinable": True,
+            "startup_head": SHA_START,
+            "current_head": SHA_START,
+            "daemon_start_head": SHA_START,
+            "local_head": SHA_START,
+            "live_remote_head": SHA_LIVE,
+            "live_known": True,
+            "live_stale": True,
+            "mutation_safe": False,
+            "reasons": [
+                f"live remote master is {SHA_LIVE[:12]} but the MCP server "
+                f"started at {SHA_START[:12]}; the daemon is stale relative "
+                "to live master -- restart/reconnect before mutating"
+            ],
+        }
+
+    def test_stale_plus_permitted_op_all_roles(self):
+        for name, role, permitted_op, allowed, forbidden, _forbidden_op in ROLE_MATRIX:
+            with self.subTest(profile=name, op=permitted_op):
+                profile = _profile(name, role, allowed, forbidden)
+                parity = self._stale_parity()
+                with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+                    mcp_server, "_current_master_parity", return_value=parity
+                ), patch.object(
+                    mcp_server, "_master_parity_block", return_value=list(parity["reasons"])
+                ), patch.object(
+                    mcp_server, "_runtime_mode_block", return_value=[]
+                ), patch.object(
+                    mcp_server, "_ensure_matching_profile", return_value=None
+                ), patch.object(
+                    mcp_server.session_ctx,
+                    "mutation_context_audit_fields",
+                    return_value={"session_profile": name},
+                ):
+                    blocked = mcp_server._profile_permission_block(permitted_op)
+                self.assertIsNotNone(blocked, name)
+                assert blocked is not None
+                self.assertEqual(
+                    blocked.get("blocker_kind"),
+                    "runtime_reconnect_required",
+                    blocked,
+                )
+                self.assertNotIn("permission_report", blocked, blocked)
+                self.assertTrue(blocked.get("restart_required"), blocked)
+                self.assertEqual(blocked.get("startup_head"), SHA_START, blocked)
+                self.assertEqual(blocked.get("live_remote_head"), SHA_LIVE, blocked)
+                action = (blocked.get("exact_safe_next_action") or "").lower()
+                self.assertIn("reconnect", action)
+                self.assertNotIn("call gitea_activate_profile with", action)
+                self.assertNotIn("switch to the author mcp session", action)
+                self.assertNotIn("switch to the reviewer mcp session", action)
+
+    def test_fresh_plus_forbidden_op_all_roles(self):
+        for name, role, _permitted, allowed, forbidden, forbidden_op in ROLE_MATRIX:
+            with self.subTest(profile=name, op=forbidden_op):
+                profile = _profile(name, role, allowed, forbidden)
+                with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+                    mcp_server, "_master_parity_block", return_value=[]
+                ), patch.object(
+                    mcp_server, "_runtime_mode_block", return_value=[]
+                ), patch.object(
+                    mcp_server, "_ensure_matching_profile", return_value=None
+                ), patch.object(
+                    mcp_server.session_ctx,
+                    "mutation_context_audit_fields",
+                    return_value={"session_profile": name},
+                ), patch.object(
+                    mcp_server.gitea_config,
+                    "load_config",
+                    return_value=_config({name: profile}),
+                ), patch.object(
+                    mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=False
+                ):
+                    blocked = mcp_server._profile_permission_block(forbidden_op)
+                self.assertIsNotNone(blocked, name)
+                assert blocked is not None
+                self.assertEqual(blocked.get("blocker_kind"), "permission_denied", blocked)
+                self.assertIn("permission_report", blocked, blocked)
+                report = blocked["permission_report"]
+                self.assertEqual(report.get("missing_permission"), forbidden_op, report)
+                self.assertFalse(report.get("diagnostic_defect"), report)
+                # No runtime reconnect fields for pure permission denial
+                self.assertNotEqual(
+                    blocked.get("blocker_kind"), "runtime_reconnect_required"
+                )
+
+    def test_stale_plus_forbidden_op_both_causes_separated(self):
+        for name, role, _permitted, allowed, forbidden, forbidden_op in ROLE_MATRIX:
+            with self.subTest(profile=name, op=forbidden_op):
+                profile = _profile(name, role, allowed, forbidden)
+                parity = self._stale_parity()
+                stale_reason = parity["reasons"][0]
+                with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+                    mcp_server, "_current_master_parity", return_value=parity
+                ), patch.object(
+                    mcp_server, "_master_parity_block", return_value=[stale_reason]
+                ), patch.object(
+                    mcp_server, "_runtime_mode_block", return_value=[]
+                ), patch.object(
+                    mcp_server, "_ensure_matching_profile", return_value=None
+                ), patch.object(
+                    mcp_server.session_ctx,
+                    "mutation_context_audit_fields",
+                    return_value={"session_profile": name},
+                ):
+                    # Gate collects both classes; force permission reason too.
+                    with patch.object(
+                        mcp_server,
+                        "_profile_operation_gate",
+                        return_value=[
+                            stale_reason,
+                            f"profile is not allowed to {forbidden_op}",
+                        ],
+                    ):
+                        blocked = mcp_server._profile_permission_block(forbidden_op)
+                self.assertIsNotNone(blocked)
+                assert blocked is not None
+                self.assertEqual(
+                    blocked.get("blocker_kind"), "runtime_reconnect_required", blocked
+                )
+                self.assertNotIn("permission_report", blocked, blocked)
+                self.assertIn("permission_block_reasons", blocked, blocked)
+                self.assertIn("stale_runtime_reasons", blocked, blocked)
+                classes = blocked.get("gate_reason_classes") or {}
+                self.assertTrue(classes.get("stale_runtime"), classes)
+                self.assertTrue(classes.get("permission"), classes)
+
+    def test_runtime_mode_block_no_permission_report(self):
+        profile = _profile(
+            "prgs-author",
+            "author",
+            ["gitea.read", "gitea.issue.create"],
+            [],
+        )
+        runtime_reason = (
+            "runtime mode is 'dev-test' and the mutation targets the "
+            "production repository; dev/test runtimes must not mutate real "
+            "issues or PRs (ADR: stable control runtime vs dev runtime)"
+        )
+        with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server, "_master_parity_block", return_value=[]
+        ), patch.object(
+            mcp_server, "_runtime_mode_block", return_value=[runtime_reason]
+        ), patch.object(
+            mcp_server, "_ensure_matching_profile", return_value=None
+        ), patch.object(
+            mcp_server.session_ctx,
+            "mutation_context_audit_fields",
+            return_value={"session_profile": "prgs-author"},
+        ):
+            blocked = mcp_server._profile_permission_block("gitea.issue.create")
+        self.assertIsNotNone(blocked)
+        assert blocked is not None
+        self.assertEqual(blocked.get("blocker_kind"), "runtime_mode_blocked", blocked)
+        self.assertNotIn("permission_report", blocked, blocked)
+        action = (blocked.get("exact_safe_next_action") or "").lower()
+        self.assertNotIn("call gitea_activate_profile with", action)
+        self.assertIn("stable control runtime", action)
+
+
+class Issue897CreateIssueRegression(unittest.TestCase):
+    def test_create_issue_stale_daemon_never_missing_issue_create(self):
+        """Regression AC: stale prgs-author create_issue must not claim missing create."""
+        profile = _profile(
+            "prgs-author",
+            "author",
+            [
+                "gitea.read",
+                "gitea.issue.create",
+                "gitea.issue.comment",
+                "gitea.branch.create",
+                "gitea.branch.push",
+                "gitea.pr.create",
+                "gitea.pr.comment",
+                "gitea.repo.commit",
+            ],
+            [],
+        )
+        stale_reason = (
+            f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
+            f"at {SHA_START[:12]}; the daemon is stale relative to live master "
+            "-- restart/reconnect before mutating"
+        )
+        parity = {
+            "in_parity": True,
+            "stale": False,
+            "restart_required": True,
+            "determinable": True,
+            "startup_head": SHA_START,
+            "current_head": SHA_START,
+            "daemon_start_head": SHA_START,
+            "local_head": SHA_START,
+            "live_remote_head": SHA_LIVE,
+            "live_known": True,
+            "live_stale": True,
+            "mutation_safe": False,
+            "reasons": [stale_reason],
+        }
+
+        with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server, "_current_master_parity", return_value=parity
+        ), patch.object(
+            mcp_server, "_master_parity_block", return_value=[stale_reason]
+        ), patch.object(
+            mcp_server, "_runtime_mode_block", return_value=[]
+        ), patch.object(
+            mcp_server, "_ensure_matching_profile", return_value=None
+        ), patch.object(
+            mcp_server.session_ctx,
+            "mutation_context_audit_fields",
+            return_value={"session_profile": "prgs-author"},
+        ), patch.object(
+            mcp_server, "_mutation_config_authority_block", return_value=None
+        ), patch.object(
+            mcp_server, "_session_context_mutation_block", return_value=None
+        ):
+            blocked = mcp_server._profile_permission_block(
+                "gitea.issue.create", remote="prgs"
+            )
+
+        self.assertIsNotNone(blocked)
+        assert blocked is not None
+        self.assertEqual(blocked.get("blocker_kind"), "runtime_reconnect_required")
+        self.assertNotIn("permission_report", blocked)
+        # Even if a caller still built a raw report, holds-check must not claim missing.
+        with patch.object(mcp_server, "get_profile", return_value=profile):
+            raw = mcp_server._permission_block_report("gitea.issue.create")
+        self.assertIsNone(raw.get("missing_permission"), raw)
+        self.assertNotEqual(raw.get("missing_permission"), "gitea.issue.create")
+
+    def test_permission_report_for_gate_reasons_skips_stale(self):
+        stale = (
+            f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
+            f"at {SHA_START[:12]}; the daemon is stale relative to live master "
+            "-- restart/reconnect before mutating"
+        )
+        self.assertIsNone(
+            mcp_server._permission_report_for_gate_reasons(
+                "gitea.issue.comment", [stale]
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

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_sessions_view.py

@@ -0,0 +1,739 @@
+"""Tests for the Runtime and session view (Phase 1, #641).
+
+Covers clean and stale session rendering, contamination marker surfacing,
+worktree binding display, sanctioned recovery links (no pkill), nav/live
+status, and the JSON API export.
+
+Also pins the two invariants a reviewer found violated at head a81db754:
+degraded ownership sections must render as *unknown* rather than as an
+affirmative "none"/"unbound", and contamination payload text must be redacted
+at the display boundary rather than trusted from the write-time denylist.
+"""
+
+from __future__ import annotations
+
+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 webui.app import create_app
+from webui.inventory import (
+    AUTHORITY_CONTROL_PLANE_DB,
+    AUTHORITY_FILESYSTEM,
+    InventorySection,
+    InventorySnapshot,
+    STATUS_DEGRADED,
+    STATUS_OK,
+    STATUS_UNAVAILABLE,
+)
+from webui.nav import NAV_GROUPS, STUB_PAGES, iter_nav_items
+from webui.runtime_health import FileHash, RuntimeSnapshot
+from webui.session_loader import (
+    ContaminationMarker,
+    SessionRow,
+    SessionViewSnapshot,
+    _build_session_rows,
+    _inspect_contamination,
+    load_session_view_snapshot,
+    snapshot_to_dict,
+)
+from webui.session_views import render_sessions_page
+
+
+def _runtime(
+    *,
+    stale: str | None = None,
+    profile: str = "prgs-author",
+    role: str = "author",
+) -> RuntimeSnapshot:
+    return RuntimeSnapshot(
+        project_id="gitea-tools",
+        repo_root="/tmp/repo",
+        remote="prgs",
+        host="gitea.prgs.cc",
+        profile_name=profile,
+        role_kind=role,
+        config_model="v2-contexts",
+        profile_mode="dynamic-profile",
+        profile_source="config file profile",
+        authenticated_username="jcwalker3",
+        identity_error=None,
+        repo_sha="a" * 40,
+        remote_master_sha="a" * 40,
+        commits_behind_master=0,
+        stale_runtime_warning=stale,
+        shell_health={"shell_use_allowed": True, "consecutive_spawn_failures": 0},
+        workflow_hashes=(
+            FileHash(label="SKILL.md", path="skills/llm-project-workflow/SKILL.md", sha256="abc"),
+        ),
+        schema_hashes=(),
+        restart_guidance="docs/mcp-namespace-eof-recovery.md",
+        fetch_error=None,
+    )
+
+
+def _inventory(
+    *,
+    sessions: tuple[dict, ...] = (),
+    leases: tuple[dict, ...] = (),
+    locks: tuple[dict, ...] = (),
+    worktrees: tuple[dict, ...] = (),
+    namespaces: tuple[dict, ...] = (),
+    statuses: dict[str, str] | None = None,
+) -> InventorySnapshot:
+    """Build a snapshot; ``statuses`` degrades named sections (default all ok)."""
+    status_of = statuses or {}
+
+    def _status(name: str) -> str:
+        return status_of.get(name, STATUS_OK)
+
+    sections = (
+        InventorySection(
+            name="sessions",
+            authority=AUTHORITY_CONTROL_PLANE_DB,
+            status=_status("sessions"),
+            items=sessions,
+        ),
+        InventorySection(
+            name="leases",
+            authority=AUTHORITY_CONTROL_PLANE_DB,
+            status=_status("leases"),
+            items=leases,
+        ),
+        InventorySection(
+            name="locks",
+            authority=AUTHORITY_FILESYSTEM,
+            status=_status("locks"),
+            items=locks,
+        ),
+        InventorySection(
+            name="worktrees",
+            authority=AUTHORITY_FILESYSTEM,
+            status=_status("worktrees"),
+            items=worktrees,
+        ),
+        InventorySection(
+            name="namespaces",
+            authority=AUTHORITY_FILESYSTEM,
+            status=_status("namespaces"),
+            items=namespaces
+            or (
+                {
+                    "profile_name": "prgs-author",
+                    "role": "author",
+                    "mcp_namespace": "gitea-author",
+                    "capability_summary": {
+                        "can_author": True,
+                        "can_review": False,
+                        "can_merge": False,
+                    },
+                    "active": True,
+                },
+            ),
+            reason="only the profile serving this web process is observable",
+        ),
+    )
+    index = {section.name: section for section in sections}
+    return InventorySnapshot(
+        generated_at="2026-07-25T00:00:00+00:00",
+        sections=sections,
+        collisions=(),
+        correlations=(),
+        scan_ms=1.0,
+        _section_index=index,
+    )
+
+
+def _clean_session() -> dict:
+    return {
+        "session_id": "prgs-author-111-clean",
+        "role": "author",
+        "profile": "prgs-author",
+        "namespace": "gitea-author",
+        "pid": 1111,
+        "pid_alive": True,
+        "status": "active",
+        "started_at": "2026-07-25T00:00:00Z",
+        "last_heartbeat_at": "2026-07-25T01:00:00Z",
+    }
+
+
+def _stale_session() -> dict:
+    return {
+        "session_id": "prgs-author-222-stale",
+        "role": "author",
+        "profile": "prgs-author",
+        "namespace": "gitea-author",
+        "pid": 2222,
+        "pid_alive": False,
+        "status": "active",
+        "started_at": "2026-07-24T00:00:00Z",
+        "last_heartbeat_at": "2026-07-24T01:00:00Z",
+    }
+
+
+class TestBuildSessionRows(unittest.TestCase):
+    def test_clean_session_has_no_stale_or_contamination_flags(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            leases=(
+                {
+                    "lease_id": "lease-clean",
+                    "session_id": "prgs-author-111-clean",
+                    "status": "active",
+                    "expired": False,
+                    "work_kind": "issue",
+                    "work_number": 641,
+                },
+            ),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "branch_name": "feat/issue-641-runtime-session-view",
+                    "worktree_path": "~/Development/Gitea-Tools/branches/feat-issue-641",
+                    "live": True,
+                },
+            ),
+        )
+        rows = _build_session_rows(inventory, contamination=())
+        self.assertEqual(len(rows), 1)
+        row = rows[0]
+        self.assertEqual(row.session_id, "prgs-author-111-clean")
+        self.assertEqual(row.role, "author")
+        self.assertEqual(row.namespace, "gitea-author")
+        self.assertEqual(row.pid_alive, True)
+        self.assertEqual(row.lease_ids, ("lease-clean",))
+        self.assertEqual(row.work_refs, ("issue#641",))
+        self.assertTrue(row.worktree_paths)
+        self.assertEqual(row.stale_flags, ())
+        self.assertEqual(row.contamination_flags, ())
+
+    def test_stale_session_flags_dead_pid(self):
+        inventory = _inventory(sessions=(_stale_session(),))
+        rows = _build_session_rows(inventory, contamination=())
+        self.assertEqual(rows[0].stale_flags, ("pid-dead",))
+
+    def test_contamination_marker_binds_to_session(self):
+        inventory = _inventory(sessions=(_clean_session(),))
+        marker = ContaminationMarker(
+            kind="runtime_recovery_contamination",
+            on_disk=True,
+            has_payload=True,
+            summary="manual daemon kill",
+            reason_class="manual_daemon_kill",
+            session_id="prgs-author-111-clean",
+            role="author",
+            command_summary="pkill -f mcp_server.py",
+            cleared=False,
+        )
+        rows = _build_session_rows(inventory, contamination=(marker,))
+        self.assertIn("runtime_recovery_contamination", rows[0].contamination_flags)
+
+    def test_process_wide_contamination_surfaces_on_all_sessions(self):
+        inventory = _inventory(sessions=(_clean_session(), _stale_session()))
+        marker = ContaminationMarker(
+            kind="stable_branch_contamination",
+            on_disk=True,
+            has_payload=True,
+            summary="direct master push attempt",
+            reason_class="stable_branch_push",
+            session_id=None,
+            cleared=False,
+        )
+        rows = _build_session_rows(inventory, contamination=(marker,))
+        self.assertEqual(len(rows), 2)
+        for row in rows:
+            self.assertTrue(
+                any("stable_branch_contamination" in f for f in row.contamination_flags)
+            )
+
+
+class TestRenderSessionsPage(unittest.TestCase):
+    def _snapshot(
+        self,
+        *,
+        sessions: tuple[dict, ...],
+        contamination: tuple[ContaminationMarker, ...] = (),
+        stale_runtime: str | None = None,
+    ) -> SessionViewSnapshot:
+        inventory = _inventory(
+            sessions=sessions,
+            leases=(
+                {
+                    "lease_id": "lease-1",
+                    "session_id": sessions[0]["session_id"] if sessions else "",
+                    "status": "active",
+                    "expired": False,
+                    "work_kind": "issue",
+                    "work_number": 641,
+                },
+            )
+            if sessions
+            else (),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "worktree_path": "branches/feat-issue-641",
+                },
+            )
+            if sessions
+            else (),
+            worktrees=(
+                {
+                    "rel_path": "branches/feat-issue-641",
+                    "branch": "feat/issue-641-runtime-session-view",
+                    "classification": "active_issue_work",
+                    "registered_worktree": True,
+                    "dirty": False,
+                },
+            ),
+        )
+        rows = _build_session_rows(inventory, contamination)
+        return SessionViewSnapshot(
+            runtime=_runtime(stale=stale_runtime),
+            inventory=inventory,
+            sessions=rows,
+            contamination_markers=contamination,
+        )
+
+    def test_clean_session_render(self):
+        html = render_sessions_page(self._snapshot(sessions=(_clean_session(),)))
+        self.assertIn("Runtime and sessions", html)
+        self.assertIn("prgs-author-111-clean", html)
+        self.assertIn("gitea-author", html)
+        self.assertIn("branches/feat-issue-641", html)
+        self.assertIn("Sanctioned recovery", html)
+        self.assertIn("docs/mcp-namespace-eof-recovery.md", html)
+        # Recovery section must name reconnect and forbid manual kill.
+        recovery_idx = html.lower().find("sanctioned recovery")
+        self.assertGreaterEqual(recovery_idx, 0)
+        recovery = html[recovery_idx:].lower()
+        self.assertIn("reconnect", recovery)
+        self.assertIn("contamination", recovery)
+        self.assertIn("not recovery", recovery)
+        self.assertNotIn("run pkill", recovery)
+        self.assertNotIn("killall", recovery)
+
+    def test_stale_session_render(self):
+        html = render_sessions_page(self._snapshot(sessions=(_stale_session(),)))
+        self.assertIn("prgs-author-222-stale", html)
+        self.assertIn("pid-dead", html)
+        self.assertIn("badge-stale", html)
+
+    def test_contamination_render_is_not_silent(self):
+        marker = ContaminationMarker(
+            kind="runtime_recovery_contamination",
+            on_disk=True,
+            has_payload=True,
+            summary="manual kill",
+            reason_class="manual_daemon_kill",
+            session_id="prgs-author-111-clean",
+            command_summary="pkill -f mcp_server.py",
+            cleared=False,
+        )
+        html = render_sessions_page(
+            self._snapshot(sessions=(_clean_session(),), contamination=(marker,))
+        )
+        self.assertIn("Contamination markers", html)
+        self.assertIn("runtime_recovery_contamination", html)
+        self.assertIn("ACTIVE", html)
+        self.assertIn("badge-blocked", html)
+
+    def test_stale_runtime_banner(self):
+        html = render_sessions_page(
+            self._snapshot(
+                sessions=(_clean_session(),),
+                stale_runtime="server behind master by 3 commits",
+            )
+        )
+        self.assertIn("Stale runtime", html)
+        self.assertIn("server behind master", html)
+
+
+class TestSessionLoaderComposition(unittest.TestCase):
+    def test_load_with_injected_sources(self):
+        inventory = _inventory(sessions=(_clean_session(), _stale_session()))
+        snap = load_session_view_snapshot(
+            load_runtime=lambda: _runtime(),
+            load_inventory=lambda: inventory,
+            inspect_contamination=lambda **_k: {
+                "on_disk": False,
+                "has_payload": False,
+                "summary": "absent",
+            },
+            load_contamination_payload=lambda **_k: None,
+        )
+        self.assertEqual(len(snap.sessions), 2)
+        self.assertEqual(snap.stale_session_count, 1)
+        self.assertEqual(snap.contaminated_session_count, 0)
+        data = snapshot_to_dict(snap)
+        self.assertEqual(data["view"], "runtime-sessions")
+        self.assertEqual(data["issue"], 641)
+        self.assertTrue(data["read_only"])
+        self.assertEqual(data["session_counts"]["total"], 2)
+        self.assertEqual(data["session_counts"]["stale"], 1)
+        self.assertIn("recovery_docs", data)
+
+
+class TestSessionsRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+        inventory = _inventory(
+            sessions=(_clean_session(), _stale_session()),
+            leases=(
+                {
+                    "lease_id": "lease-x",
+                    "session_id": "prgs-author-111-clean",
+                    "status": "active",
+                    "expired": False,
+                    "work_kind": "issue",
+                    "work_number": 641,
+                },
+            ),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "worktree_path": "branches/feat-issue-641",
+                },
+            ),
+            worktrees=(
+                {
+                    "rel_path": "branches/feat-issue-641",
+                    "branch": "feat/issue-641-runtime-session-view",
+                    "classification": "active_issue_work",
+                    "registered_worktree": True,
+                    "dirty": False,
+                },
+            ),
+        )
+        rows = _build_session_rows(inventory, contamination=())
+        self.snapshot = SessionViewSnapshot(
+            runtime=_runtime(stale="stale for test"),
+            inventory=inventory,
+            sessions=rows,
+            contamination_markers=(),
+        )
+        self._patch = mock.patch(
+            "webui.app.load_session_view_snapshot",
+            return_value=self.snapshot,
+        )
+        self._patch.start()
+
+    def tearDown(self):
+        self._patch.stop()
+
+    def test_sessions_page_live(self):
+        response = self.client.get("/sessions")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Runtime and sessions", response.text)
+        self.assertIn("prgs-author-111-clean", response.text)
+        self.assertIn("prgs-author-222-stale", response.text)
+        self.assertIn("pid-dead", response.text)
+        self.assertIn("Sanctioned recovery", response.text)
+        self.assertNotIn("Phase 1 shell placeholder", response.text)
+        self.assertNotIn("child issue of #425", response.text.lower())
+
+    def test_api_sessions_json(self):
+        for path in ("/api/sessions", "/api/v1/sessions"):
+            response = self.client.get(path)
+            self.assertEqual(response.status_code, 200, path)
+            data = response.json()
+            self.assertEqual(data["view"], "runtime-sessions")
+            self.assertEqual(data["session_counts"]["total"], 2)
+            self.assertEqual(data["session_counts"]["stale"], 1)
+            self.assertTrue(data["read_only"])
+            self.assertEqual(data["mutations"], [])
+
+    def test_nav_marks_sessions_live(self):
+        sessions_items = [
+            item for item in iter_nav_items() if item.href == "/sessions"
+        ]
+        self.assertEqual(len(sessions_items), 1)
+        self.assertEqual(sessions_items[0].status, "live")
+        self.assertNotIn("/sessions", STUB_PAGES)
+        # Home page should not mark Sessions as stub.
+        home = self.client.get("/")
+        self.assertEqual(home.status_code, 200)
+        self.assertIn('href="/sessions"', home.text)
+        # Stub marker only appears next to remaining stub destinations.
+        self.assertNotIn(
+            'href="/sessions">Sessions</a> <span class="muted">(stub)</span>',
+            home.text,
+        )
+
+
+class TestDegradedOwnershipAuthority(unittest.TestCase):
+    """B1: a section that could not be read must never render as absence."""
+
+    def _snapshot(self, inventory: InventorySnapshot) -> SessionViewSnapshot:
+        return SessionViewSnapshot(
+            runtime=_runtime(),
+            inventory=inventory,
+            sessions=_build_session_rows(inventory, contamination=()),
+            contamination_markers=(),
+        )
+
+    def test_unavailable_leases_mark_row_authority_unproven(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            statuses={"leases": STATUS_UNAVAILABLE},
+        )
+        row = _build_session_rows(inventory, contamination=())[0]
+        self.assertEqual(row.lease_ids, ())
+        self.assertEqual(row.lease_authority, STATUS_UNAVAILABLE)
+        # Worktree binding is correlated through lease work numbers, so it
+        # inherits the unreadable lease section.
+        self.assertEqual(row.worktree_authority, STATUS_UNAVAILABLE)
+        self.assertFalse(row.ownership_authority_complete)
+
+    def test_readable_locks_are_not_reported_unbound_when_leases_degrade(self):
+        # The narrow variant: locks hold a real worktree_path and read cleanly,
+        # but the lease section that supplies the correlating work number does
+        # not. The row must say unknown, not "unbound".
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "worktree_path": "branches/feat-issue-641",
+                },
+            ),
+            statuses={"leases": STATUS_DEGRADED},
+        )
+        row = _build_session_rows(inventory, contamination=())[0]
+        self.assertEqual(row.worktree_paths, ())
+        self.assertEqual(row.worktree_authority, STATUS_DEGRADED)
+        self.assertFalse(row.ownership_authority_complete)
+
+    def test_degraded_render_says_unknown_not_none_or_unbound(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            statuses={"leases": STATUS_UNAVAILABLE, "locks": STATUS_UNAVAILABLE},
+        )
+        html = render_sessions_page(self._snapshot(inventory))
+        self.assertIn("unknown (inventory unavailable)", html)
+        self.assertIn("authority unproven", html)
+        self.assertIn("Ownership authority incomplete", html)
+        # The affirmative-absence strings must be gone from the row entirely.
+        self.assertNotIn(">none<", html)
+        self.assertNotIn(">unbound<", html)
+
+    def test_clean_inventory_still_renders_affirmative_absence(self):
+        # Guards against over-correcting B1 into "everything is unknown".
+        inventory = _inventory(sessions=(_clean_session(),))
+        html = render_sessions_page(self._snapshot(inventory))
+        self.assertIn(">none<", html)
+        self.assertIn(">unbound<", html)
+        # The column legend mentions "unknown (inventory …)" as static copy, so
+        # assert on the per-row marker and the concrete statuses instead.
+        self.assertNotIn("authority unproven", html)
+        self.assertNotIn("unknown (inventory unavailable)", html)
+        self.assertNotIn("unknown (inventory degraded)", html)
+        self.assertNotIn("Ownership authority incomplete", html)
+
+    def test_json_export_carries_snapshot_and_per_row_authority(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            statuses={"locks": STATUS_UNAVAILABLE},
+        )
+        data = snapshot_to_dict(self._snapshot(inventory))
+        self.assertFalse(data["ownership_authority_complete"])
+        self.assertEqual(
+            data["ownership_section_status"]["locks"], STATUS_UNAVAILABLE
+        )
+        self.assertEqual(data["ownership_section_status"]["leases"], STATUS_OK)
+        self.assertIn("unknown, not unowned", data["ownership_note"])
+
+        row = data["sessions"][0]
+        self.assertTrue(row["lease_authority_complete"])
+        self.assertFalse(row["worktree_authority_complete"])
+        self.assertEqual(row["worktree_authority"], STATUS_UNAVAILABLE)
+        self.assertFalse(row["ownership_authority_complete"])
+        self.assertIn("unknown, not unowned", row["ownership_note"])
+
+    def test_json_export_is_affirmative_when_every_source_reads(self):
+        inventory = _inventory(sessions=(_clean_session(),))
+        data = snapshot_to_dict(self._snapshot(inventory))
+        self.assertTrue(data["ownership_authority_complete"])
+        self.assertTrue(data["sessions"][0]["ownership_authority_complete"])
+
+    def test_missing_session_list_is_not_reported_as_no_sessions(self):
+        inventory = _inventory(statuses={"sessions": STATUS_UNAVAILABLE})
+        html = render_sessions_page(self._snapshot(inventory))
+        self.assertIn("could not be read", html)
+        self.assertIn("not evidence that no sessions exist", html)
+
+    def test_expired_lease_flags_row_as_stale(self):
+        inventory = _inventory(
+            sessions=(_clean_session(),),
+            leases=(
+                {
+                    "lease_id": "lease-expired-1",
+                    "session_id": "prgs-author-111-clean",
+                    "status": "active",
+                    "expired": True,
+                    "work_kind": "issue",
+                    "work_number": 641,
+                },
+            ),
+        )
+        row = _build_session_rows(inventory, contamination=())[0]
+        self.assertIn("lease-expired", row.stale_flags)
+        self.assertIn("active-lease-past-expiry", row.stale_flags)
+
+
+class TestContaminationRedaction(unittest.TestCase):
+    """B2: marker payload text is redacted at the display boundary."""
+
+    def _marker(self, payload: dict) -> ContaminationMarker:
+        return _inspect_contamination(
+            "runtime_recovery_contamination",
+            remote="prgs",
+            inspect=lambda **_k: {
+                "on_disk": True,
+                "has_payload": True,
+                "summary": "",
+            },
+            load=lambda **_k: payload,
+        )
+
+    def test_home_paths_are_collapsed(self):
+        home = os.path.expanduser("~")
+        marker = self._marker(
+            {"command_summary": f"pkill -f {home}/Development/Gitea-Tools/x.py"}
+        )
+        self.assertNotIn(home, marker.command_summary)
+        self.assertIn("~/Development/Gitea-Tools/x.py", marker.command_summary)
+
+    def test_secrets_missed_by_the_write_time_denylist_are_redacted(self):
+        # Each of these was verified in review to survive
+        # stable_branch_push_guard.redact_command untouched.
+        cases = (
+            ("curl -H 'X-Api-Key: SUPERSECRET123' https://example.invalid", "SUPERSECRET123"),
+            ("cmd --password hunter2 origin master", "hunter2"),
+            ("PRIVATE_KEY=abc123 python deploy.py", "abc123"),
+            ("fetch https://user:[email protected]/x.git", "user:pw"),
+        )
+        for raw, secret in cases:
+            with self.subTest(raw=raw):
+                marker = self._marker({"command_summary": raw})
+                self.assertNotIn(secret, marker.command_summary)
+                self.assertIn("[redacted]", marker.command_summary)
+
+    def test_command_summary_is_redacted_not_removed(self):
+        # It is legitimate #630 evidence: the operator must still see which
+        # daemon was killed.
+        marker = self._marker(
+            {
+                "command_summary": "pkill -f gitea_mcp_server.py",
+                "reason_class": "manual_daemon_kill",
+                "session_id": "prgs-author-111-clean",
+                "role": "author",
+            }
+        )
+        self.assertIn("pkill -f gitea_mcp_server.py", marker.command_summary)
+        self.assertEqual(marker.reason_class, "manual_daemon_kill")
+        self.assertEqual(marker.session_id, "prgs-author-111-clean")
+        self.assertEqual(marker.role, "author")
+
+    def test_rendered_page_exposes_no_home_path_from_a_marker(self):
+        home = os.path.expanduser("~")
+        marker = self._marker(
+            {
+                "command_summary": f"pkill -f {home}/Development/Gitea-Tools/x.py",
+                "reason_class": "manual_daemon_kill",
+            }
+        )
+        inventory = _inventory(sessions=(_clean_session(),))
+        html = render_sessions_page(
+            SessionViewSnapshot(
+                runtime=_runtime(),
+                inventory=inventory,
+                sessions=_build_session_rows(inventory, (marker,)),
+                contamination_markers=(marker,),
+            )
+        )
+        self.assertIn("Contamination markers", html)
+        self.assertNotIn(home, html)
+
+
+class TestSessionsPageEscaping(unittest.TestCase):
+    """Hostile values from every rendered source stay inert (N2)."""
+
+    HOSTILE = '<script>alert("xss")</script>'
+
+    def test_hostile_session_and_marker_values_are_escaped(self):
+        session = dict(_clean_session())
+        session["session_id"] = f"sid-{self.HOSTILE}"
+        session["role"] = self.HOSTILE
+        session["profile"] = self.HOSTILE
+        session["namespace"] = self.HOSTILE
+        session["status"] = self.HOSTILE
+        inventory = _inventory(
+            sessions=(session,),
+            leases=(
+                {
+                    "lease_id": self.HOSTILE,
+                    "session_id": session["session_id"],
+                    "status": "active",
+                    "expired": False,
+                    "work_kind": self.HOSTILE,
+                    "work_number": 641,
+                },
+            ),
+            locks=(
+                {
+                    "issue_number": 641,
+                    "worktree_path": self.HOSTILE,
+                },
+            ),
+        )
+        marker = ContaminationMarker(
+            kind="runtime_recovery_contamination",
+            on_disk=True,
+            has_payload=True,
+            summary=self.HOSTILE,
+            reason_class=self.HOSTILE,
+            session_id=session["session_id"],
+            role=self.HOSTILE,
+            command_summary=self.HOSTILE,
+            cleared=False,
+        )
+        html = render_sessions_page(
+            SessionViewSnapshot(
+                runtime=_runtime(),
+                inventory=inventory,
+                sessions=_build_session_rows(inventory, (marker,)),
+                contamination_markers=(marker,),
+            )
+        )
+        self.assertNotIn("<script>", html)
+        self.assertNotIn('alert("xss")', html)
+        self.assertIn("&lt;script&gt;", html)
+
+    def test_hostile_values_in_a_degraded_render_are_escaped(self):
+        inventory = _inventory(
+            sessions=(dict(_clean_session(), session_id=f"sid-{self.HOSTILE}"),),
+            statuses={"leases": STATUS_UNAVAILABLE, "locks": STATUS_DEGRADED},
+        )
+        html = render_sessions_page(
+            SessionViewSnapshot(
+                runtime=_runtime(),
+                inventory=inventory,
+                sessions=_build_session_rows(inventory, contamination=()),
+                contamination_markers=(),
+            )
+        )
+        self.assertNotIn("<script>", html)
+        self.assertIn("&lt;script&gt;", html)
+        self.assertIn("unknown (inventory", html)
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/app.py

@@ -48,6 +48,11 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
 from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 from webui.runtime_views import render_runtime_page
+from webui.session_loader import (
+    load_session_view_snapshot,
+    snapshot_to_dict as session_view_snapshot_to_dict,
+)
+from webui.session_views import render_sessions_page
 from webui.inventory import (
     SECTION_NAMES as _INVENTORY_SECTIONS,
     load_inventory_snapshot,
@@ -87,6 +92,7 @@ _LEGACY_PAGES = (
     ("/projects", "Projects", "registry and onboarding (#427)"),
     ("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
     ("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
+    ("/sessions", "Sessions", "runtime and session view (#641)"),
     ("/audit", "Audit", "final-report paste and validator preview (#431)"),
     ("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
     ("/leases", "Leases", "collision and lease visibility (#433)"),
@@ -194,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)))
@@ -325,6 +409,17 @@ async def api_runtime(_request: Request) -> JSONResponse:
     return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
 
 
+async def sessions(_request: Request) -> HTMLResponse:
+    """Runtime and session view (#641) — read-only composition of health + inventory."""
+    snapshot = load_session_view_snapshot()
+    return HTMLResponse(render_sessions_page(snapshot))
+
+
+async def api_sessions(_request: Request) -> JSONResponse:
+    """JSON export for the runtime/session view (#641)."""
+    return JSONResponse(session_view_snapshot_to_dict(load_session_view_snapshot()))
+
+
 async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
     if request.method == "GET":
         return "", None
@@ -764,6 +859,9 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
             Route("/api/prompts", api_prompts, methods=["GET"]),
             Route("/runtime", runtime, methods=["GET"]),
             Route("/api/runtime", api_runtime, methods=["GET"]),
+            Route("/sessions", sessions, methods=["GET"]),
+            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("/analytics", analytics, methods=["GET"]),
             Route("/api/analytics", api_v1_analytics, methods=["GET"]),
@@ -798,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/inventory.py

@@ -76,6 +76,35 @@ _CREDENTIAL_KEY_RE = re.compile(
 )
 _REDACTED = "[redacted]"
 
+#: Credential-shaped *name* as it appears inside a free-form command line. This
+#: is deliberately broader than :data:`_CREDENTIAL_KEY_RE` — it also matches a
+#: bare ``key`` component, so ``PRIVATE_KEY=`` is caught. Over-redacting a
+#: displayed string is safe; under-redacting one is not.
+_TEXT_CREDENTIAL_NAME = (
+    r"[A-Za-z0-9_.\-]*"
+    r"(?:token|secret|password|passwd|key|authorization|bearer|credential)"
+    r"[A-Za-z0-9_.\-]*"
+)
+#: A value following such a name: single-quoted, double-quoted, or bare. The
+#: bare form stops at a quote so an enclosing quote survives the redaction.
+_TEXT_CREDENTIAL_VALUE = r"'[^']*'|\"[^\"]*\"|[^\s'\"]+"
+
+_TEXT_CREDENTIAL_FLAG_RE = re.compile(
+    rf"(?P<key>(?<![\w\-])--?{_TEXT_CREDENTIAL_NAME})"
+    rf"(?P<sep>[=\s]+)"
+    rf"(?P<value>{_TEXT_CREDENTIAL_VALUE})",
+    re.IGNORECASE,
+)
+_TEXT_CREDENTIAL_ASSIGN_RE = re.compile(
+    rf"(?P<key>(?<![\w\-]){_TEXT_CREDENTIAL_NAME})"
+    rf"(?P<sep>\s*[:=]\s*)"
+    rf"(?P<value>{_TEXT_CREDENTIAL_VALUE})",
+    re.IGNORECASE,
+)
+_TEXT_URL_USERINFO_RE = re.compile(
+    r"(?P<scheme>\b[A-Za-z][A-Za-z0-9+.\-]*://)[^\s/@]+@"
+)
+
 
 @dataclass(frozen=True)
 class InventorySection:
@@ -225,6 +254,42 @@ def scrub(value: Any, *, key: str | None = None) -> Any:
     return repr(value)
 
 
+def collapse_home(text: str) -> str:
+    """Collapse every ``$HOME`` occurrence *inside* a string, not just a prefix."""
+    home = os.path.expanduser("~")
+    if not home or home == "/":
+        return text
+    return text.replace(home, "~")
+
+
+def scrub_text(value: Any) -> Any:
+    """Redact a free-form text blob such as a recorded command line.
+
+    :func:`scrub` keys off structured field *names* and whole-value prefixes,
+    which is right for inventory records but blind to a secret embedded in the
+    middle of a sentence. This collapses ``$HOME`` and redacts credential-shaped
+    tokens and URL userinfo *anywhere* in the string, so operator-supplied text
+    rendered verbatim — contamination ``command_summary`` (#630) above all — is
+    held to the same standard as every other field on the page.
+
+    Returns ``None`` unchanged so callers can keep "absent" distinct from "".
+    """
+    if value is None:
+        return None
+    text = value if isinstance(value, str) else str(value)
+    text = collapse_home(text)
+    text = _TEXT_URL_USERINFO_RE.sub(
+        lambda m: f"{m.group('scheme')}{_REDACTED}@", text
+    )
+    text = _TEXT_CREDENTIAL_FLAG_RE.sub(
+        lambda m: f"{m.group('key')}{m.group('sep')}{_REDACTED}", text
+    )
+    text = _TEXT_CREDENTIAL_ASSIGN_RE.sub(
+        lambda m: f"{m.group('key')}{m.group('sep')}{_REDACTED}", text
+    )
+    return text
+
+
 # ── control-plane database (read-only) ───────────────────────────────────────
 
 

webui/nav.py

@@ -48,7 +48,7 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
     )),
     NavGroup("Runtime/Sessions", (
         NavItem("/runtime", "Runtime health"),
-        NavItem("/sessions", "Sessions", "stub"),
+        NavItem("/sessions", "Sessions"),
     )),
     NavGroup("Projects", (
         NavItem("/projects", "Projects"),
@@ -76,11 +76,6 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
 # issues of epic #631. Each maps a path to (title, description). Routes are
 # registered so nav links resolve to a graceful, read-only stub page.
 STUB_PAGES: dict[str, tuple[str, str]] = {
-    "/sessions": (
-        "Sessions",
-        "Active session, capability, and role inventory. Backed by the unified "
-        "inventory API (#636) once it lands.",
-    ),
     "/inventory": (
         "Inventory",
         "Unified sessions, leases, locks, namespaces, and worktree inventory. "

webui/runtime_views.py

@@ -88,5 +88,7 @@ def render_runtime_page(snapshot: RuntimeSnapshot) -> str:
         "<p class='muted'>MVP is read-only — restart MCP servers from your IDE/operator "
         "workflow. Related issue: <code>#420</code>. Guidance: "
         f"<code>{html.escape(snapshot.restart_guidance)}</code></p>"
-        "<p class='muted'>This page does not expose tokens or perform MCP restarts.</p>"
-    )
+        "<p class='muted'>This page does not expose tokens or perform MCP restarts. "
+        "Correlated sessions, worktree bindings, and contamination markers: "
+        "<a href='/sessions'>/sessions</a> (#641).</p>"
+    )

webui/session_loader.py

@@ -0,0 +1,517 @@
+"""Compose runtime health + inventory into a sessions/runtime view (#641).
+
+Phase 1 is read-only. It correlates namespaces, sessions, capabilities,
+worktree bindings, lease ownership, stale flags, and contamination markers
+when they are detectable on disk (#630 / #671). It never restarts, kills, or
+takes over a session.
+
+Sources:
+
+* :mod:`webui.runtime_health` — profile, role, stale runtime, shell health.
+* :mod:`webui.inventory` — sessions, leases, locks, worktrees, namespaces
+  and collision signals from the control-plane DB + filesystem.
+* :mod:`mcp_session_state` — durable contamination markers (inspect only).
+
+Secrets are never read. Inventory-sourced values arrive already redacted by
+:func:`webui.inventory.scrub`; free-form marker text this module loads itself is
+put through :func:`webui.inventory.scrub_text`, which collapses ``$HOME`` and
+redacts credential-shaped tokens *inside* a string rather than only at its start.
+
+Ownership columns are authority-aware. A lease or lock section that could not be
+read renders as ``unknown``, never as ``none`` or ``unbound``: a lease the reader
+could not load is not an absent lease (see
+:attr:`webui.inventory.InventorySnapshot.ownership_authority_complete`).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable
+
+import mcp_session_state
+from webui.inventory import (
+    OWNERSHIP_SECTIONS,
+    STATUS_OK,
+    InventorySection,
+    InventorySnapshot,
+    load_inventory_snapshot,
+    scrub_text,
+    snapshot_to_dict as inventory_snapshot_to_dict,
+)
+from webui.runtime_health import (
+    RuntimeSnapshot,
+    load_runtime_snapshot,
+    snapshot_to_dict as runtime_snapshot_to_dict,
+)
+
+# Sanctioned recovery pointers only — never pkill / killall (#630).
+SANCTIONED_RECOVERY_DOCS: tuple[dict[str, str], ...] = (
+    {
+        "label": "MCP namespace EOF recovery (reconnect only)",
+        "path": "docs/mcp-namespace-eof-recovery.md",
+        "note": "IDE/client reconnect or operator-owned restart; never kill daemons.",
+    },
+    {
+        "label": "MCP namespace health",
+        "path": "docs/mcp-namespace-health.md",
+        "note": "client_namespace probe proves namespace health.",
+    },
+    {
+        "label": "Restart path inventory",
+        "path": "docs/mcp-restart-path-inventory.md",
+        "note": "Catalog of sanctioned reconnect/restart paths.",
+    },
+    {
+        "label": "Local web UI recovery",
+        "path": "docs/webui-local-dev.md",
+        "note": "Operator console start and documented recovery sequence.",
+    },
+)
+
+_CONTAMINATION_KINDS: tuple[str, ...] = (
+    mcp_session_state.KIND_RUNTIME_RECOVERY_CONTAMINATION,
+    mcp_session_state.KIND_STABLE_BRANCH_CONTAMINATION,
+)
+
+#: Status recorded on a row when the backing inventory section is absent
+#: entirely — distinct from a section that reported itself degraded.
+AUTHORITY_MISSING = "missing"
+
+
+def _section_status(section: InventorySection | None) -> str:
+    """Status of an ownership section, treating an absent section as missing."""
+    if section is None:
+        return AUTHORITY_MISSING
+    return section.status
+
+
+def _combined_authority(*statuses: str) -> str:
+    """Worst status of the sections a derived column depends on.
+
+    A column proved from two sections is only trustworthy when *both* read
+    cleanly, so the first non-``ok`` status wins.
+    """
+    for status in statuses:
+        if status != STATUS_OK:
+            return status
+    return STATUS_OK
+
+
+@dataclass(frozen=True)
+class ContaminationMarker:
+    """A detectable durable contamination marker (audit-safe summary)."""
+
+    kind: str
+    on_disk: bool
+    has_payload: bool
+    summary: str
+    reason_class: str | None = None
+    session_id: str | None = None
+    role: str | None = None
+    command_summary: str | None = None
+    cleared: bool = False
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "kind": self.kind,
+            "on_disk": self.on_disk,
+            "has_payload": self.has_payload,
+            "summary": self.summary,
+            "reason_class": self.reason_class,
+            "session_id": self.session_id,
+            "role": self.role,
+            "command_summary": self.command_summary,
+            "cleared": self.cleared,
+            "active": self.on_disk and self.has_payload and not self.cleared,
+        }
+
+
+@dataclass(frozen=True)
+class SessionRow:
+    """One correlated session row for the sessions table."""
+
+    session_id: str
+    role: str | None
+    profile: str | None
+    namespace: str | None
+    pid: int | None
+    pid_alive: bool | None
+    status: str | None
+    started_at: str | None
+    last_heartbeat_at: str | None
+    lease_ids: tuple[str, ...] = ()
+    work_refs: tuple[str, ...] = ()
+    worktree_paths: tuple[str, ...] = ()
+    stale_flags: tuple[str, ...] = ()
+    contamination_flags: tuple[str, ...] = ()
+    #: Status of the section backing ``lease_ids``/``work_refs``. While this is
+    #: not ``ok`` those tuples mean "could not be read", never "none held".
+    lease_authority: str = STATUS_OK
+    #: Worst status across the sections backing ``worktree_paths`` (locks are
+    #: correlated through lease work numbers, so both must read cleanly).
+    worktree_authority: str = STATUS_OK
+
+    @property
+    def ownership_authority_complete(self) -> bool:
+        """True only when this row's ownership columns are provable."""
+        return self.lease_authority == STATUS_OK and self.worktree_authority == STATUS_OK
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "session_id": self.session_id,
+            "role": self.role,
+            "profile": self.profile,
+            "namespace": self.namespace,
+            "pid": self.pid,
+            "pid_alive": self.pid_alive,
+            "status": self.status,
+            "started_at": self.started_at,
+            "last_heartbeat_at": self.last_heartbeat_at,
+            "lease_ids": list(self.lease_ids),
+            "work_refs": list(self.work_refs),
+            "worktree_paths": list(self.worktree_paths),
+            "stale_flags": list(self.stale_flags),
+            "contamination_flags": list(self.contamination_flags),
+            "is_stale": bool(self.stale_flags),
+            "is_contaminated": bool(self.contamination_flags),
+            "lease_authority": self.lease_authority,
+            "worktree_authority": self.worktree_authority,
+            "lease_authority_complete": self.lease_authority == STATUS_OK,
+            "worktree_authority_complete": self.worktree_authority == STATUS_OK,
+            "ownership_authority_complete": self.ownership_authority_complete,
+            "ownership_note": (
+                "Lease and worktree columns are proved from sections that read "
+                "cleanly."
+                if self.ownership_authority_complete
+                else "An ownership source could not be read; empty lease_ids or "
+                "worktree_paths on this row mean unknown, not unowned."
+            ),
+        }
+
+
+@dataclass(frozen=True)
+class SessionViewSnapshot:
+    """Composed runtime + session inventory view (#641)."""
+
+    runtime: RuntimeSnapshot
+    inventory: InventorySnapshot
+    sessions: tuple[SessionRow, ...]
+    contamination_markers: tuple[ContaminationMarker, ...]
+    recovery_docs: tuple[dict[str, str], ...] = SANCTIONED_RECOVERY_DOCS
+    fetch_error: str | None = None
+
+    @property
+    def stale_session_count(self) -> int:
+        return sum(1 for row in self.sessions if row.stale_flags)
+
+    @property
+    def contaminated_session_count(self) -> int:
+        return sum(1 for row in self.sessions if row.contamination_flags)
+
+    @property
+    def active_contamination(self) -> tuple[ContaminationMarker, ...]:
+        return tuple(m for m in self.contamination_markers if m.to_dict()["active"])
+
+    @property
+    def ownership_authority_complete(self) -> bool:
+        """False while any ownership section is degraded, absent, or unavailable."""
+        return self.inventory.ownership_authority_complete
+
+    @property
+    def ownership_section_status(self) -> dict[str, str]:
+        """Per-section status for the three ownership-bearing sections."""
+        return {
+            name: _section_status(self.inventory.section(name))
+            for name in OWNERSHIP_SECTIONS
+        }
+
+
+def _inspect_contamination(
+    kind: str,
+    *,
+    remote: str | None,
+    inspect: Callable[..., dict[str, Any]] | None = None,
+    load: Callable[..., dict[str, Any] | None] | None = None,
+) -> ContaminationMarker:
+    """Inspect one contamination kind; never raises into the page render path."""
+    inspect_fn = inspect or mcp_session_state.inspect_state_envelope
+    load_fn = load or mcp_session_state.load_state
+    try:
+        envelope = inspect_fn(kind=kind, remote=remote)
+    except Exception as exc:  # noqa: BLE001 — fail soft for the dashboard
+        return ContaminationMarker(
+            kind=kind,
+            on_disk=False,
+            has_payload=False,
+            summary=f"contamination inspect failed: {type(exc).__name__}",
+        )
+
+    reason_class = None
+    session_id = None
+    role = None
+    command_summary = None
+    cleared = False
+    summary = scrub_text(str(envelope.get("summary") or ""))
+
+    if envelope.get("on_disk") and envelope.get("has_payload"):
+        try:
+            payload = load_fn(kind=kind, remote=remote) or {}
+        except Exception:  # noqa: BLE001
+            payload = {}
+        if isinstance(payload, dict):
+            reason_class = payload.get("reason_class")
+            session_id = payload.get("session_id")
+            role = payload.get("role")
+            command_summary = payload.get("command_summary") or payload.get("detail")
+            cleared = bool(payload.get("cleared_by_reconciler"))
+            if not summary:
+                summary = scrub_text(
+                    f"{kind}: {reason_class or 'present'}"
+                    + (" (cleared)" if cleared else "")
+                )
+
+    # Marker payloads are operator-supplied free text and are the one thing on
+    # this page that does not arrive through inventory scrubbing. The write-time
+    # redactor is a narrow denylist, so redact again at the display boundary:
+    # it leaves $HOME paths, `-H 'X-Api-Key: …'`, `--password …`, and
+    # `PRIVATE_KEY=…` intact. The field itself stays — it is #630 evidence.
+    return ContaminationMarker(
+        kind=kind,
+        on_disk=bool(envelope.get("on_disk")),
+        has_payload=bool(envelope.get("has_payload")),
+        summary=summary or f"{kind}: not present",
+        reason_class=scrub_text(str(reason_class)) if reason_class else None,
+        session_id=scrub_text(str(session_id)) if session_id else None,
+        role=scrub_text(str(role)) if role else None,
+        command_summary=scrub_text(str(command_summary)) if command_summary else None,
+        cleared=cleared,
+    )
+
+
+def _load_contamination_markers(
+    *,
+    remote: str | None,
+    inspect: Callable[..., dict[str, Any]] | None = None,
+    load: Callable[..., dict[str, Any] | None] | None = None,
+) -> tuple[ContaminationMarker, ...]:
+    return tuple(
+        _inspect_contamination(kind, remote=remote, inspect=inspect, load=load)
+        for kind in _CONTAMINATION_KINDS
+    )
+
+
+def _build_session_rows(
+    inventory: InventorySnapshot,
+    contamination: tuple[ContaminationMarker, ...],
+) -> tuple[SessionRow, ...]:
+    sessions_section = inventory.section("sessions")
+    leases_section = inventory.section("leases")
+    locks_section = inventory.section("locks")
+
+    # Ownership columns may only assert absence when their source read cleanly.
+    lease_authority = _section_status(leases_section)
+    # Locks carry claimant profile/username, not control-plane session ids, so a
+    # worktree binding is correlated through lease work numbers: it depends on
+    # the locks *and* the leases section.
+    worktree_authority = _combined_authority(
+        lease_authority, _section_status(locks_section)
+    )
+
+    leases_by_session: dict[str, list[dict[str, Any]]] = {}
+    for lease in (leases_section.items if leases_section else ()):
+        sid = str(lease.get("session_id") or "")
+        if sid:
+            leases_by_session.setdefault(sid, []).append(lease)
+
+    active_markers = [m for m in contamination if m.to_dict()["active"]]
+    marker_session_ids = {
+        m.session_id for m in active_markers if m.session_id
+    }
+
+    rows: list[SessionRow] = []
+    for raw in sessions_section.items if sessions_section else ():
+        sid = str(raw.get("session_id") or "")
+        if not sid:
+            continue
+        session_leases = leases_by_session.get(sid, [])
+        lease_ids = tuple(
+            str(lease["lease_id"])
+            for lease in session_leases
+            if lease.get("lease_id")
+        )
+        work_refs: list[str] = []
+        work_numbers: list[int] = []
+        for lease in session_leases:
+            kind = lease.get("work_kind")
+            number = lease.get("work_number")
+            if kind and number is not None:
+                work_refs.append(f"{kind}#{number}")
+                try:
+                    work_numbers.append(int(number))
+                except (TypeError, ValueError):
+                    pass
+
+        worktree_paths: list[str] = []
+        for lock in (locks_section.items if locks_section else ()):
+            try:
+                issue_no = int(lock.get("issue_number"))
+            except (TypeError, ValueError):
+                continue
+            if issue_no in work_numbers and lock.get("worktree_path"):
+                worktree_paths.append(str(lock["worktree_path"]))
+
+        stale_flags: list[str] = []
+        if raw.get("pid_alive") is False:
+            stale_flags.append("pid-dead")
+        status = str(raw.get("status") or "").lower()
+        if status and status not in {"active", "alive", "running", "ok"}:
+            stale_flags.append(f"status:{status}")
+        for lease in session_leases:
+            if lease.get("expired") is True:
+                stale_flags.append("lease-expired")
+            if str(lease.get("status") or "").lower() == "active" and lease.get(
+                "expired"
+            ) is True:
+                stale_flags.append("active-lease-past-expiry")
+
+        contamination_flags: list[str] = []
+        if sid in marker_session_ids:
+            for marker in active_markers:
+                if marker.session_id == sid:
+                    contamination_flags.append(marker.kind)
+        # Process-wide contamination with no session binding still surfaces
+        # against every live session so it cannot be silent (#630).
+        for marker in active_markers:
+            if not marker.session_id and marker.kind not in contamination_flags:
+                contamination_flags.append(f"{marker.kind}:process-wide")
+
+        rows.append(
+            SessionRow(
+                session_id=sid,
+                role=raw.get("role"),
+                profile=raw.get("profile"),
+                namespace=raw.get("namespace"),
+                pid=raw.get("pid") if isinstance(raw.get("pid"), int) else None,
+                pid_alive=raw.get("pid_alive")
+                if isinstance(raw.get("pid_alive"), bool)
+                else None,
+                status=raw.get("status"),
+                started_at=raw.get("started_at"),
+                last_heartbeat_at=raw.get("last_heartbeat_at"),
+                lease_ids=lease_ids,
+                work_refs=tuple(work_refs),
+                worktree_paths=tuple(worktree_paths),
+                stale_flags=tuple(dict.fromkeys(stale_flags)),
+                contamination_flags=tuple(dict.fromkeys(contamination_flags)),
+                lease_authority=lease_authority,
+                worktree_authority=worktree_authority,
+            )
+        )
+
+    return tuple(rows)
+
+
+def load_session_view_snapshot(
+    *,
+    load_runtime: Callable[..., RuntimeSnapshot] | None = None,
+    load_inventory: Callable[..., InventorySnapshot] | None = None,
+    inspect_contamination: Callable[..., dict[str, Any]] | None = None,
+    load_contamination_payload: Callable[..., dict[str, Any] | None] | None = None,
+) -> SessionViewSnapshot:
+    """Build the composed sessions/runtime view. Fail-soft on partial sources."""
+    runtime_loader = load_runtime or load_runtime_snapshot
+    inventory_loader = load_inventory or load_inventory_snapshot
+    fetch_error: str | None = None
+
+    try:
+        runtime = runtime_loader()
+    except Exception as exc:  # noqa: BLE001
+        fetch_error = f"runtime snapshot failed: {type(exc).__name__}: {exc}"
+        # Minimal placeholder so the page still renders inventory + recovery.
+        from webui.runtime_health import RuntimeSnapshot as _RS
+
+        runtime = _RS(
+            project_id="unknown",
+            repo_root="",
+            remote="",
+            host="",
+            profile_name="unknown",
+            role_kind="unknown",
+            config_model="unknown",
+            profile_mode="unknown",
+            profile_source="unknown",
+            authenticated_username=None,
+            identity_error=str(exc),
+            repo_sha=None,
+            remote_master_sha=None,
+            commits_behind_master=None,
+            stale_runtime_warning=None,
+            shell_health={},
+            workflow_hashes=(),
+            schema_hashes=(),
+            restart_guidance="docs/mcp-namespace-eof-recovery.md",
+            fetch_error=str(exc),
+        )
+
+    try:
+        inventory = inventory_loader()
+    except Exception as exc:  # noqa: BLE001
+        msg = f"inventory snapshot failed: {type(exc).__name__}: {exc}"
+        fetch_error = f"{fetch_error}; {msg}" if fetch_error else msg
+        inventory = load_inventory_snapshot(
+            db_path="/nonexistent-for-fail-soft",
+            lock_dir="/nonexistent-for-fail-soft",
+        )
+
+    remote = getattr(runtime, "remote", None)
+    contamination = _load_contamination_markers(
+        remote=remote,
+        inspect=inspect_contamination,
+        load=load_contamination_payload,
+    )
+    sessions = _build_session_rows(inventory, contamination)
+    return SessionViewSnapshot(
+        runtime=runtime,
+        inventory=inventory,
+        sessions=sessions,
+        contamination_markers=contamination,
+        fetch_error=fetch_error,
+    )
+
+
+def snapshot_to_dict(snapshot: SessionViewSnapshot) -> dict[str, Any]:
+    """JSON export for ``/api/sessions`` (read-only)."""
+    return {
+        "api_version": "v1",
+        "view": "runtime-sessions",
+        "issue": 641,
+        "fetch_error": snapshot.fetch_error,
+        "runtime": runtime_snapshot_to_dict(snapshot.runtime),
+        "inventory": inventory_snapshot_to_dict(snapshot.inventory),
+        "sessions": [row.to_dict() for row in snapshot.sessions],
+        "session_counts": {
+            "total": len(snapshot.sessions),
+            "stale": snapshot.stale_session_count,
+            "contaminated": snapshot.contaminated_session_count,
+        },
+        "ownership_authority_complete": snapshot.ownership_authority_complete,
+        "ownership_section_status": snapshot.ownership_section_status,
+        "ownership_note": (
+            "Every ownership source read cleanly; a session with no lease and no "
+            "worktree path genuinely holds neither."
+            if snapshot.ownership_authority_complete
+            else "An ownership source is degraded or unavailable. Empty lease_ids "
+            "and worktree_paths mean unknown, not unowned; consult each row's "
+            "lease_authority and worktree_authority."
+        ),
+        "contamination_markers": [
+            marker.to_dict() for marker in snapshot.contamination_markers
+        ],
+        "active_contamination": [
+            marker.to_dict() for marker in snapshot.active_contamination
+        ],
+        "recovery_docs": [dict(doc) for doc in snapshot.recovery_docs],
+        "read_only": True,
+        "phase": 1,
+        "mutations": [],
+    }

webui/session_views.py

@@ -0,0 +1,403 @@
+"""HTML views for the Runtime and session view (Phase 1, #641).
+
+Read-only composition of runtime health (#430) and inventory sessions /
+namespaces / worktrees (#636). Surfaces stale and contamination indicators
+when detectable. Recovery links point only at sanctioned reconnect/restart
+docs — never at manual process kill (#630).
+"""
+
+from __future__ import annotations
+
+from html import escape
+from typing import Sequence
+
+from webui.inventory import STATUS_OK
+from webui.layout import render_page
+from webui.session_loader import (
+    ContaminationMarker,
+    SessionRow,
+    SessionViewSnapshot,
+)
+
+
+def _badge(text: str, css: str) -> str:
+    return f'<span class="badge {css}">{escape(text)}</span>'
+
+
+def _flags(flags: Sequence[str], *, css: str) -> str:
+    if not flags:
+        return '<span class="muted">—</span>'
+    return " ".join(_badge(flag, css) for flag in flags)
+
+
+def _unproven_cell(status: str) -> str:
+    """Render an ownership column whose backing inventory section failed to read.
+
+    Never "none" and never "unbound": an unreadable source proves nothing about
+    ownership, and claiming otherwise is the exact failure the
+    ``ownership_authority_complete`` invariant exists to prevent.
+    """
+    return (
+        f'<span class="muted">unknown (inventory {escape(status)})</span><br>'
+        f'{_badge("authority unproven", "badge-health-degraded")}'
+    )
+
+
+def _runtime_banner(snapshot: SessionViewSnapshot) -> str:
+    runtime = snapshot.runtime
+    stale = runtime.stale_runtime_warning
+    stale_html = ""
+    if stale:
+        stale_html = (
+            f'<div class="health-card health-stale" style="margin-top:0.75rem;">'
+            f"<strong>Stale runtime:</strong> {escape(stale)}</div>"
+        )
+    identity = runtime.authenticated_username or "unresolved"
+    if runtime.identity_error:
+        identity = f"unresolved ({runtime.identity_error})"
+
+    return f"""<div class="health-card">
+  <h3>Runtime context</h3>
+  <table class="detail">
+    <tr><th>Profile</th><td><code>{escape(runtime.profile_name)}</code></td></tr>
+    <tr><th>Role kind</th><td>{escape(runtime.role_kind)}</td></tr>
+    <tr><th>Identity</th><td>{escape(str(identity))}</td></tr>
+    <tr><th>Remote / host</th>
+      <td><code>{escape(runtime.remote)}</code> · <code>{escape(runtime.host)}</code></td>
+    </tr>
+    <tr><th>Local HEAD</th>
+      <td><code>{escape(runtime.repo_sha or "unknown")}</code></td>
+    </tr>
+    <tr><th>Remote master</th>
+      <td><code>{escape(runtime.remote_master_sha or "unknown")}</code></td>
+    </tr>
+    <tr><th>Commits behind</th>
+      <td>{escape(str(runtime.commits_behind_master if runtime.commits_behind_master is not None else "unknown"))}</td>
+    </tr>
+  </table>
+  <p class="muted">Full runtime detail: <a href="/runtime">/runtime</a> ·
+  Inventory API: <a href="/api/v1/inventory"><code>/api/v1/inventory</code></a></p>
+  {stale_html}
+</div>"""
+
+
+def _summary_bar(snapshot: SessionViewSnapshot) -> str:
+    total = len(snapshot.sessions)
+    stale = snapshot.stale_session_count
+    contaminated = snapshot.contaminated_session_count
+    active_markers = len(snapshot.active_contamination)
+    inv_status = snapshot.inventory.status
+    authority_complete = snapshot.ownership_authority_complete
+    authority_text = "complete" if authority_complete else "incomplete"
+    authority_css = "badge-health-ok" if authority_complete else "badge-health-degraded"
+    return f"""<div class="health-card" style="display:flex; flex-wrap:wrap; gap:1rem; align-items:center;">
+  <div><strong>Sessions:</strong> <span class="badge badge-health-ok">{total}</span></div>
+  <div><strong>Stale:</strong> <span class="badge badge-stale">{stale}</span></div>
+  <div><strong>Contaminated:</strong> <span class="badge badge-blocked">{contaminated}</span></div>
+  <div><strong>Active markers:</strong> <span class="badge badge-health-unproven">{active_markers}</span></div>
+  <div><strong>Inventory:</strong> <span class="badge badge-health-skipped">{escape(inv_status)}</span></div>
+  <div><strong>Ownership authority:</strong> <span class="badge {authority_css}">{authority_text}</span></div>
+</div>"""
+
+
+def _ownership_caveat(snapshot: SessionViewSnapshot) -> str:
+    """Name the unreadable ownership sections, or render nothing when all read."""
+    if snapshot.ownership_authority_complete:
+        return ""
+    degraded = ", ".join(
+        f"{name}: {status}"
+        for name, status in snapshot.ownership_section_status.items()
+        if status != STATUS_OK
+    )
+    return (
+        '<div class="health-card health-stale">'
+        "<strong>Ownership authority incomplete:</strong> "
+        f"{escape(degraded)}. Columns marked <em>unknown</em> could not be read. "
+        "No session below may be treated as holding no lease or no worktree "
+        "binding — absence of evidence is not evidence of absence.</div>"
+    )
+
+
+def _render_session_row(row: SessionRow) -> str:
+    pid = "—" if row.pid is None else str(row.pid)
+    pid_alive = "—" if row.pid_alive is None else ("alive" if row.pid_alive else "dead")
+    pid_css = (
+        "badge-health-ok"
+        if row.pid_alive is True
+        else ("badge-blocked" if row.pid_alive is False else "badge-health-skipped")
+    )
+    # An empty tuple only means "holds none" when its source read cleanly.
+    if row.lease_authority != STATUS_OK:
+        lease_cell = _unproven_cell(row.lease_authority)
+    else:
+        leases = (
+            ", ".join(f"<code>{escape(lid)}</code>" for lid in row.lease_ids)
+            if row.lease_ids
+            else '<span class="muted">none</span>'
+        )
+        work = (
+            ", ".join(escape(ref) for ref in row.work_refs)
+            if row.work_refs
+            else '<span class="muted">—</span>'
+        )
+        lease_cell = (
+            f'{leases}<div class="muted" style="font-size:0.82rem; '
+            f'margin-top:0.2rem;">{work}</div>'
+        )
+
+    if row.worktree_authority != STATUS_OK:
+        worktrees = _unproven_cell(row.worktree_authority)
+    else:
+        worktrees = (
+            "<br>".join(f"<code>{escape(path)}</code>" for path in row.worktree_paths)
+            if row.worktree_paths
+            else '<span class="muted">unbound</span>'
+        )
+    return f"""<tr>
+  <td><code>{escape(row.session_id)}</code></td>
+  <td>
+    <div><code>{escape(str(row.role or "—"))}</code> / <code>{escape(str(row.profile or "—"))}</code></div>
+    <div class="muted" style="font-size:0.82rem;">ns: <code>{escape(str(row.namespace or "—"))}</code></div>
+  </td>
+  <td>
+    <code>{escape(pid)}</code>
+    {_badge(pid_alive, pid_css)}
+  </td>
+  <td>{escape(str(row.status or "—"))}<div class="muted" style="font-size:0.82rem;">{escape(str(row.last_heartbeat_at or ""))}</div></td>
+  <td>{lease_cell}</td>
+  <td>{worktrees}</td>
+  <td>{_flags(row.stale_flags, css="badge-stale")}</td>
+  <td>{_flags(row.contamination_flags, css="badge-blocked")}</td>
+</tr>"""
+
+
+def _sessions_table(snapshot: SessionViewSnapshot) -> str:
+    rows: Sequence[SessionRow] = snapshot.sessions
+    if not rows:
+        sessions_status = snapshot.ownership_section_status.get("sessions", STATUS_OK)
+        if sessions_status != STATUS_OK:
+            return (
+                '<p class="muted">Session inventory is '
+                f"<strong>{escape(sessions_status)}</strong> — the session list "
+                "could not be read. This is not evidence that no sessions "
+                "exist.</p>"
+            )
+        return (
+            '<p class="muted">No control-plane sessions recorded. Inventory may '
+            "be unavailable, or no MCP workers have registered yet.</p>"
+        )
+    body = "".join(_render_session_row(row) for row in rows)
+    return f"""<table class="registry">
+  <thead>
+    <tr>
+      <th>Session</th>
+      <th>Role / profile / namespace</th>
+      <th>PID</th>
+      <th>Status</th>
+      <th>Leases / work</th>
+      <th>Worktree binding</th>
+      <th>Stale</th>
+      <th>Contamination</th>
+    </tr>
+  </thead>
+  <tbody>
+    {body}
+  </tbody>
+</table>"""
+
+
+def _namespaces_section(snapshot: SessionViewSnapshot) -> str:
+    section = snapshot.inventory.section("namespaces")
+    if section is None:
+        return (
+            '<div class="prompt-card"><h3>Namespaces</h3>'
+            '<p class="muted">Namespaces section not loaded.</p></div>'
+        )
+    if not section.ok:
+        return f"""<div class="prompt-card">
+  <h3>Namespaces {_badge(section.status, "badge-health-degraded")}</h3>
+  <p class="muted">{escape(section.reason or "unavailable")}</p>
+</div>"""
+
+    rows = []
+    for item in section.items:
+        caps = item.get("capability_summary") or {}
+        cap_bits = ", ".join(
+            name for name, ok in sorted(caps.items()) if ok
+        ) or "none"
+        rows.append(
+            "<tr>"
+            f"<td><code>{escape(str(item.get('mcp_namespace') or '—'))}</code></td>"
+            f"<td><code>{escape(str(item.get('profile_name') or '—'))}</code></td>"
+            f"<td>{escape(str(item.get('role') or '—'))}</td>"
+            f"<td>{escape(cap_bits)}</td>"
+            f"<td>{'yes' if item.get('active') else 'no'}</td>"
+            "</tr>"
+        )
+    reason = (
+        f'<p class="muted">{escape(section.reason)}</p>'
+        if section.reason
+        else ""
+    )
+    return f"""<div class="prompt-card">
+  <h3>Namespaces / capabilities</h3>
+  {reason}
+  <table class="registry">
+    <thead>
+      <tr>
+        <th>Namespace</th>
+        <th>Profile</th>
+        <th>Role</th>
+        <th>Capabilities</th>
+        <th>Active in process</th>
+      </tr>
+    </thead>
+    <tbody>
+      {"".join(rows) if rows else '<tr><td colspan="5" class="muted">No namespace rows.</td></tr>'}
+    </tbody>
+  </table>
+</div>"""
+
+
+def _worktrees_section(snapshot: SessionViewSnapshot) -> str:
+    section = snapshot.inventory.section("worktrees")
+    if section is None:
+        return ""
+    if not section.ok and not section.items:
+        return f"""<div class="prompt-card">
+  <h3>Worktrees {_badge(section.status, "badge-health-degraded")}</h3>
+  <p class="muted">{escape(section.reason or "unavailable")}</p>
+</div>"""
+
+    rows = []
+    for item in section.items[:50]:
+        rows.append(
+            "<tr>"
+            f"<td><code>{escape(str(item.get('rel_path') or item.get('path') or '—'))}</code></td>"
+            f"<td><code>{escape(str(item.get('branch') or '—'))}</code></td>"
+            f"<td>{escape(str(item.get('classification') or '—'))}</td>"
+            f"<td>{'yes' if item.get('registered_worktree') else 'no'}</td>"
+            f"<td>{'dirty' if item.get('dirty') else 'clean'}</td>"
+            "</tr>"
+        )
+    more = ""
+    if len(section.items) > 50:
+        more = f'<p class="muted">Showing 50 of {len(section.items)}. Full list: <a href="/worktrees">/worktrees</a>.</p>'
+    return f"""<div class="prompt-card">
+  <h3>Worktree bindings</h3>
+  <p class="muted">Registered issue worktrees under <code>branches/</code>. Hygiene detail: <a href="/worktrees">/worktrees</a>.</p>
+  <table class="registry">
+    <thead>
+      <tr>
+        <th>Path</th>
+        <th>Branch</th>
+        <th>Classification</th>
+        <th>Registered</th>
+        <th>State</th>
+      </tr>
+    </thead>
+    <tbody>
+      {"".join(rows) if rows else '<tr><td colspan="5" class="muted">No worktrees recorded.</td></tr>'}
+    </tbody>
+  </table>
+  {more}
+</div>"""
+
+
+def _contamination_section(markers: Sequence[ContaminationMarker]) -> str:
+    if not markers:
+        return (
+            '<div class="prompt-card"><h3>Contamination markers</h3>'
+            '<p class="muted">No contamination kinds inspected.</p></div>'
+        )
+    rows = []
+    for marker in markers:
+        active = marker.to_dict()["active"]
+        status = "ACTIVE" if active else ("cleared" if marker.cleared else "absent")
+        css = "badge-blocked" if active else "badge-health-ok"
+        rows.append(
+            "<tr>"
+            f"<td><code>{escape(marker.kind)}</code></td>"
+            f"<td>{_badge(status, css)}</td>"
+            f"<td>{escape(marker.reason_class or '—')}</td>"
+            f"<td><code>{escape(marker.session_id or '—')}</code></td>"
+            f"<td>{escape(marker.command_summary or marker.summary)}</td>"
+            "</tr>"
+        )
+    return f"""<div class="prompt-card">
+  <h3>Contamination markers (#630 / #671)</h3>
+  <p class="muted">Durable markers only — never silent when present. Clearance is reconciler-only.</p>
+  <table class="registry">
+    <thead>
+      <tr>
+        <th>Kind</th>
+        <th>State</th>
+        <th>Reason class</th>
+        <th>Session</th>
+        <th>Summary</th>
+      </tr>
+    </thead>
+    <tbody>
+      {"".join(rows)}
+    </tbody>
+  </table>
+</div>"""
+
+
+def _recovery_section(snapshot: SessionViewSnapshot) -> str:
+    items = []
+    for doc in snapshot.recovery_docs:
+        items.append(
+            "<li>"
+            f"<code>{escape(doc['path'])}</code> — "
+            f"<strong>{escape(doc['label'])}</strong>: {escape(doc['note'])}"
+            "</li>"
+        )
+    return f"""<div class="prompt-card">
+  <h3>Sanctioned recovery (read-only)</h3>
+  <p class="muted">This view does <strong>not</strong> restart, kill, or take over sessions.
+  Manual <code>pkill</code> / <code>kill</code> of MCP daemons is contamination (#630), not recovery.</p>
+  <ul class="reasons">
+    {"".join(items)}
+    <li>Prefer IDE/client reconnect (<code>/mcp reconnect</code>) or an operator-owned restart recorded in the restart inventory.</li>
+  </ul>
+</div>"""
+
+
+def render_sessions_page(snapshot: SessionViewSnapshot) -> str:
+    """Render the full HTML body for the runtime/session view."""
+    error_block = ""
+    if snapshot.fetch_error:
+        error_block = (
+            f'<div class="health-card health-stale"><strong>Partial load:</strong> '
+            f"{escape(snapshot.fetch_error)}</div>"
+        )
+    if snapshot.runtime.fetch_error:
+        error_block += (
+            f'<div class="health-card health-stale"><strong>Runtime note:</strong> '
+            f"{escape(snapshot.runtime.fetch_error)}</div>"
+        )
+
+    body = f"""
+{error_block}
+{_runtime_banner(snapshot)}
+{_summary_bar(snapshot)}
+
+<div class="prompt-card">
+  <h3>Sessions</h3>
+  <p class="muted">Control-plane sessions correlated with leases and worktree bindings.
+  Stale and contamination flags are fail-soft: absence of a marker is not proof of cleanliness when inventory is degraded.
+  Lease and worktree columns read <em>unknown (inventory …)</em> when their source could not be loaded.</p>
+  {_ownership_caveat(snapshot)}
+  {_sessions_table(snapshot)}
+</div>
+
+{_namespaces_section(snapshot)}
+{_worktrees_section(snapshot)}
+{_contamination_section(snapshot.contamination_markers)}
+{_recovery_section(snapshot)}
+"""
+    return render_page(title="Sessions", body_html=f"""<h2>Runtime and sessions</h2>
+<p class="meta">Phase 1 read-only view (#641). Combines runtime health (#430) with
+unified inventory sessions/namespaces/worktrees (#636). No restart or session-takeover controls.</p>
+{body}""")

webui/system_health_views.py

@@ -270,24 +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 and session view</a> — active profile, "
-        "workflow hashes, and shell health.</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: