"""Web Console Phase 2 Recovery Controls & Playbooks (#644). Provides canonical recovery controls for the web console: 1. Diagnosis: Surfaces stale runtimes, worktree binding errors, contamination markers, and un-reconciled cleanups. 2. Gated Actions & Playbooks: Guided recovery (rebind session worktree, clear stale binding, trigger reconciler cleanups, sanctioned restart). 3. RBAC, Contamination (#630), and Master Parity (#610) integration. 4. Audit trail via ``console_audit`` and mandatory post-recovery revalidation. """ from __future__ import annotations import os from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any import master_parity_gate import runtime_recovery_guard import stale_binding_recovery from webui import console_audit, console_authz, sanctioned_restart, system_health, worktree_scanner # --- Recovery Statuses ------------------------------------------------------ STATUS_HEALTHY = "healthy" STATUS_ACTION_REQUIRED = "action_required" STATUS_BLOCKED_CONTAMINATION = "blocked_contamination" STATUS_RECONNECT_REQUIRED = "reconnect_required" # --- Playbook Identifiers --------------------------------------------------- PLAYBOOK_CLEAR_STALE_BINDING = "clear_stale_binding" PLAYBOOK_REBIND_SESSION = "rebind_session_worktree" PLAYBOOK_RECONCILE_CLEANUPS = "reconcile_cleanups" PLAYBOOK_SANCTIONED_RESTART = "sanctioned_restart" KNOWN_PLAYBOOKS: tuple[str, ...] = ( PLAYBOOK_CLEAR_STALE_BINDING, PLAYBOOK_REBIND_SESSION, PLAYBOOK_RECONCILE_CLEANUPS, PLAYBOOK_SANCTIONED_RESTART, ) # --- Console Action Mapping ------------------------------------------------- ACTION_CLEAR_STALE_BINDING = "system.clear_stale_binding" ACTION_REBIND_SESSION = "system.rebind_session_worktree" ACTION_RECONCILE_CLEANUPS = "system.reconcile_cleanups" PLAYBOOK_ACTIONS: dict[str, str] = { PLAYBOOK_CLEAR_STALE_BINDING: ACTION_CLEAR_STALE_BINDING, PLAYBOOK_REBIND_SESSION: ACTION_REBIND_SESSION, PLAYBOOK_RECONCILE_CLEANUPS: ACTION_RECONCILE_CLEANUPS, PLAYBOOK_SANCTIONED_RESTART: sanctioned_restart.ACTION_RESTART_NAMESPACE, } #: Task key handed to :func:`runtime_recovery_guard.assess_contamination_gate`. #: A console *action id* is not a task name and is not a member of #: ``CONTAMINATION_GATED_TASKS``, so passing one left the #630 gate inert. Every #: writing recovery playbook shares this one gated task key; the reconciler #: cleanup playbook is exempted separately because it is the designated remedy. CONTAMINATION_GATED_TASK = "console_recovery_apply" #: Remote whose contamination markers govern this console. Markers are written #: per remote, so reading the wrong one reports a contaminated runtime clean. REMOTE_ENV = "WEBUI_GITEA_REMOTE" DEFAULT_REMOTE = "prgs" def _console_remote(env: dict[str, str] | None = None) -> str: env_map = env if env is not None else os.environ return (env_map.get(REMOTE_ENV) or "").strip() or DEFAULT_REMOTE def load_active_contamination_marker( remote: str | None = None, env: dict[str, str] | None = None ) -> dict[str, Any] | None: """Return the live #630 contamination marker payload, or ``None``. The gate is only meaningful when it is fed a real marker: with ``marker=None`` :func:`assess_contamination_gate` returns ``block: False`` on its first statement. The #641 session inventory already reads the durable markers, so reuse that reader rather than adding a second source of truth. Never raises into a diagnosis or execution path. """ try: from webui import session_loader except Exception: # noqa: BLE001 — never break recovery on an import problem return None try: markers = session_loader._load_contamination_markers( remote=remote or _console_remote(env) ) except Exception: # noqa: BLE001 — fail soft; the caller degrades to no marker return None for marker in markers: payload = marker.to_dict() if payload.get("active"): return payload return None def _active_binding(env_map: Any) -> str | None: """Read the live worktree binding so a no-op recovery cannot report success.""" value = env_map.get(stale_binding_recovery.ACTIVE_WORKTREE_ENV) return value if value else None @dataclass(frozen=True) class RecoveryLedgerEntry: """One planned recovery step displayed before execution.""" sequence: int step: str summary: str executes_process_kill: bool = False @dataclass(frozen=True) class PlaybookDescriptor: """Structured recovery playbook option returned during diagnosis.""" playbook_id: str label: str action_id: str description: str eligible: bool requires_confirmation: bool reason: str params_schema: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True) class RecoveryDiagnosis: """Complete diagnostic snapshot of control-plane recovery needs.""" status: str clean: bool stale_runtime: dict[str, Any] master_parity: dict[str, Any] stale_binding: dict[str, Any] contamination: dict[str, Any] worktree_anomalies: tuple[str, ...] playbooks: tuple[PlaybookDescriptor, ...] reasons: tuple[str, ...] def _repo_root(custom_path: Path | str | None = None) -> Path: if custom_path: return Path(custom_path).resolve() override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip() if override: return Path(override).resolve() return Path(__file__).resolve().parent.parent def confirmation_phrase(playbook_id: str, target: str | None = None) -> str: """Construct exact confirmation phrase required for a recovery playbook.""" clean_target = (target or "").strip() if clean_target: return f"confirm {playbook_id} {clean_target}" return f"confirm {playbook_id}" def confirmation_matches( playbook_id: str, confirmation: str | None, target: str | None = None ) -> bool: expected = confirmation_phrase(playbook_id, target) return str(confirmation or "").strip() == expected def _build_ledger( playbook_id: str, target: str | None = None ) -> tuple[RecoveryLedgerEntry, ...]: if playbook_id == PLAYBOOK_CLEAR_STALE_BINDING: return ( RecoveryLedgerEntry(1, "quiesce", "Stop admitting new gated mutations."), RecoveryLedgerEntry( 2, "clear_env", f"Remove stale env binding GITEA_ACTIVE_WORKTREE ({target or 'active'}).", ), RecoveryLedgerEntry( 3, "audit", "Record clear_stale_binding event in console audit log." ), RecoveryLedgerEntry( 4, "revalidate", "Re-run diagnosis to verify clean binding state." ), ) if playbook_id == PLAYBOOK_REBIND_SESSION: return ( RecoveryLedgerEntry(1, "quiesce", "Stop admitting new gated mutations."), RecoveryLedgerEntry( 2, "rebind_worktree", f"Rebind session worktree context safely to {target or 'target worktree'}.", ), RecoveryLedgerEntry( 3, "audit", "Record rebind_session_worktree event in console audit log." ), RecoveryLedgerEntry( 4, "revalidate", "Re-run diagnosis to verify worktree binding state." ), ) if playbook_id == PLAYBOOK_RECONCILE_CLEANUPS: return ( RecoveryLedgerEntry(1, "quiesce", "Stop admitting new gated mutations."), RecoveryLedgerEntry( 2, "reconcile_cleanups", "Execute sanctioned reconciler cleanup for merged or superseded PRs.", ), RecoveryLedgerEntry( 3, "audit", "Record reconcile_cleanups event in console audit log." ), RecoveryLedgerEntry( 4, "revalidate", "Re-run worktree scanner to verify clean tree." ), ) if playbook_id == PLAYBOOK_SANCTIONED_RESTART: restart_ledger = sanctioned_restart._mutation_ledger( target or "gitea-author", sanctioned_restart.MODE_RESTART ) return tuple( RecoveryLedgerEntry( sequence=e.sequence, step=e.step, summary=e.summary, executes_process_kill=e.executes_process_kill, ) for e in restart_ledger ) return ( RecoveryLedgerEntry(1, "unspecified", f"Execute recovery playbook {playbook_id}."), ) def diagnose_recovery( repo_path: Path | str | None = None, env: dict[str, str] | None = None, active_worktree_val: str | None = None, session_lease_wt: str | None = None, role_kind: str | None = None, ) -> RecoveryDiagnosis: """Run full control-plane diagnostics to determine recovery needs and options.""" root = _repo_root(repo_path) source_env = dict(env) if env is not None else dict(os.environ) reasons: list[str] = [] # 1. Stale runtime assessment stale_runtime_obj = system_health.assess_stale_runtime(root) stale_runtime_dict = { "daemon_head": stale_runtime_obj.daemon_head, "checkout_head": stale_runtime_obj.checkout_head, "remote_head": stale_runtime_obj.remote_head, "stale": stale_runtime_obj.stale, "determinable": stale_runtime_obj.determinable, "mutation_safe": stale_runtime_obj.mutation_safe, "reasons": list(stale_runtime_obj.reasons), } if stale_runtime_obj.stale: reasons.append("Runtime HEAD disagrees with checkout/remote HEAD.") # 2. Master parity assessment # # The baseline is the commit the *running process* started at, which is what # the parity gate is about. Capturing it from ``checkout_head`` and then # comparing it against that same value made ``in_parity`` structurally # incapable of being false. ``live_remote_head`` restores the #610 # live-remote dimension, which was previously dropped. checkout_head = stale_runtime_obj.checkout_head startup_dict = master_parity_gate.capture_startup_parity( str(root), head=stale_runtime_obj.daemon_head ) parity_dict = master_parity_gate.assess_master_parity( startup_dict, checkout_head, stale_runtime_obj.remote_head ) if not parity_dict.get("in_parity", True): reasons.append("Repository is not in master parity.") # 3. Worktree binding classification boot_bindings = stale_binding_recovery.snapshot_boot_bindings(source_env) active_val = ( active_worktree_val if active_worktree_val is not None else source_env.get(stale_binding_recovery.ACTIVE_WORKTREE_ENV) ) boot_inherited = bool(boot_bindings.get("active_worktree") and active_val == boot_bindings.get("active_worktree")) path_exists = None if active_val: path_exists = os.path.exists(os.path.realpath(active_val)) binding_class = stale_binding_recovery.classify_active_worktree_binding( active_value=active_val, session_lease_worktree=session_lease_wt, boot_inherited=boot_inherited, path_exists=path_exists, role_kind=role_kind, ) if binding_class.get("clear_eligible"): reasons.append( f"Active worktree binding is stale ({binding_class.get('classification')})." ) elif binding_class.get("classification") == stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED: reasons.append("Inherited worktree binding is unverified.") # 4. Contamination assessment (#630) # # A real marker and a task key the gate actually gates on: with marker=None # the gate short-circuits to ``block: False``, and with a console action id # the task is outside CONTAMINATION_GATED_TASKS, so it could never block. contamination_marker = load_active_contamination_marker(env=source_env) contamination_dict = runtime_recovery_guard.assess_contamination_gate( contamination_marker, task=CONTAMINATION_GATED_TASK, actual_role=role_kind, ) contaminated = bool(contamination_dict.get("block")) if contaminated: reasons.append("Runtime is contaminated by manual process kill (#630).") # 5. Worktree scanner hygiene & anomalies hygiene = worktree_scanner.load_hygiene_snapshot(project_root=str(root)) worktree_anomalies = hygiene.anomalies # Determine status & eligible playbooks playbooks: list[PlaybookDescriptor] = [] # Playbook 1: Clear Stale Binding clear_eligible = bool(binding_class.get("clear_eligible")) playbooks.append( PlaybookDescriptor( playbook_id=PLAYBOOK_CLEAR_STALE_BINDING, label="Clear Stale Worktree Binding", action_id=ACTION_CLEAR_STALE_BINDING, description="Clear provably stale or superseded GITEA_ACTIVE_WORKTREE environment binding.", eligible=clear_eligible, requires_confirmation=True, reason=( f"Binding classified as {binding_class.get('classification')}; clear is authorized." if clear_eligible else "Active worktree binding is clean, corroborated, or absent." ), ) ) # Playbook 2: Rebind Session Worktree rebind_eligible = bool( active_val or binding_class.get("classification") == stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED ) playbooks.append( PlaybookDescriptor( playbook_id=PLAYBOOK_REBIND_SESSION, label="Rebind Session Worktree", action_id=ACTION_REBIND_SESSION, description="Rebind or synchronize session worktree binding safely with active lease.", eligible=rebind_eligible, requires_confirmation=True, reason=( "Session worktree binding can be rebound to verified lease worktree." if rebind_eligible else "Session worktree is properly bound." ), params_schema={"target_worktree": "string"}, ) ) # Playbook 3: Reconcile Cleanups reconcile_eligible = bool(hygiene.anomalies or any(e.classification in {"stale-clean", "detached-review"} for e in hygiene.entries)) playbooks.append( PlaybookDescriptor( playbook_id=PLAYBOOK_RECONCILE_CLEANUPS, label="Trigger Reconciler Cleanups", action_id=ACTION_RECONCILE_CLEANUPS, description="Run sanctioned reconciler cleanup preview and apply for merged/superseded PR branches.", eligible=reconcile_eligible, requires_confirmation=True, reason=( f"Worktree hygiene scanner detected {len(hygiene.anomalies)} anomalies and cleanups needed." if reconcile_eligible else "No reconciler cleanups pending." ), ) ) # Playbook 4: Sanctioned Restart restart_eligible = bool(stale_runtime_obj.stale or contaminated) playbooks.append( PlaybookDescriptor( playbook_id=PLAYBOOK_SANCTIONED_RESTART, label="Sanctioned MCP Restart", action_id=sanctioned_restart.ACTION_RESTART_NAMESPACE, description="Restart MCP daemon via configured host supervisor without manual process kill.", eligible=restart_eligible, requires_confirmation=True, reason=( "Stale runtime or contamination detected; host supervisor restart available." if restart_eligible else "Runtime is healthy and clean." ), params_schema={"namespace": "string", "mode": "restart|reload"}, ) ) clean = not reasons and not contaminated if contaminated: status = STATUS_BLOCKED_CONTAMINATION elif reasons: status = STATUS_ACTION_REQUIRED else: status = STATUS_HEALTHY return RecoveryDiagnosis( status=status, clean=clean, stale_runtime=stale_runtime_dict, master_parity=parity_dict, stale_binding=binding_class, contamination=contamination_dict, worktree_anomalies=tuple(worktree_anomalies), playbooks=tuple(playbooks), reasons=tuple(reasons), ) def build_recovery_preview( playbook_id: str, target: str | None = None, params: dict[str, Any] | None = None, principal: console_authz.Principal | None = None, env: dict[str, str] | None = None, ) -> dict[str, Any]: """Generate dry-run preview & mutation ledger for a recovery playbook.""" if playbook_id not in KNOWN_PLAYBOOKS: return { "allowed": False, "error": "unknown_playbook", "detail": f"Playbook {playbook_id!r} is not a registered recovery playbook.", } action_id = PLAYBOOK_ACTIONS[playbook_id] action = console_authz.get_action(action_id) decision = console_authz.authorize(action_id, principal) # Preview and apply must answer the same question. ``execution_enabled`` was # a hardcoded False beside an authorization decision taken without # ``for_execution``, so the preview could not tell an operator *why* # execution was disabled — and the apply path did not ask at all. execution_decision = console_authz.authorize( action_id, principal, for_execution=True ) phrase = confirmation_phrase(playbook_id, target) ledger = _build_ledger(playbook_id, target) return { "playbook_id": playbook_id, "action_id": action_id, "target": target, "required_role": action.minimum_role if action else console_authz.OPERATOR, "required_permission": action.mcp_permission if action else "gitea.read", "requires_confirmation": True, "confirmation_phrase": phrase, "mutation_ledger": [asdict(entry) for entry in ledger], "authorization": decision.to_dict(), "execution_authorization": execution_decision.to_dict(), "params": dict(params or {}), "execution_enabled": bool(execution_decision.allowed), "execution_blocked_reason": ( None if execution_decision.allowed else execution_decision.reason_code ), } def execute_recovery_playbook( playbook_id: str, confirmation: str | None = None, target: str | None = None, params: dict[str, Any] | None = None, principal: console_authz.Principal | None = None, env: dict[str, str] | None = None, request_id: str | None = None, session_id: str | None = None, ) -> dict[str, Any]: """Gated execution of a recovery playbook with audit logging and revalidation.""" if playbook_id not in KNOWN_PLAYBOOKS: return { "success": False, "allowed": False, "error": "unknown_playbook", "detail": f"Playbook {playbook_id!r} is not known.", } action_id = PLAYBOOK_ACTIONS[playbook_id] # The mapping the playbooks actually mutate. ``dict(os.environ)`` produced a # throwaway copy: every env playbook wrote to it, verified against it, and # left the running daemon bound to the value it claimed to have fixed. mutation_env: Any = env if env is not None else os.environ # 1. Authorization check — ``for_execution=True`` is what arms the phase # gate (console_authz.authorize only applies it in that branch). Without it # a phase-2 write executed while the console was in phase 1. decision = console_authz.authorize(action_id, principal, for_execution=True) if not decision.allowed: console_audit.record_event( action_id=action_id, result=console_audit.RESULT_DENIED, principal=principal, target={"playbook_id": playbook_id, "target": target}, reason_code=decision.reason_code, detail=decision.detail, request_id=request_id, session_id=session_id, ) return { "success": False, "allowed": False, "error": decision.reason_code, "detail": decision.detail, } # 2. Confirmation phrase check if not confirmation_matches(playbook_id, confirmation, target): expected = confirmation_phrase(playbook_id, target) detail = f"Confirmation phrase mismatch. Expected: {expected!r}" console_audit.record_event( action_id=action_id, result=console_audit.RESULT_DENIED, principal=principal, target={"playbook_id": playbook_id, "target": target}, reason_code="confirmation_mismatch", detail=detail, request_id=request_id, session_id=session_id, ) return { "success": False, "allowed": False, "error": "confirmation_mismatch", "detail": detail, "expected_confirmation_phrase": expected, } # 3. Contamination rule (#630) check role_str = principal.role if principal else None contamination_marker = load_active_contamination_marker(env=env) contam = runtime_recovery_guard.assess_contamination_gate( contamination_marker, task=CONTAMINATION_GATED_TASK, actual_role=role_str, ) if contam.get("block"): if playbook_id != PLAYBOOK_RECONCILE_CLEANUPS: detail = "Runtime is contaminated by a manual process kill (#630). Run reconciler cleanup playbook first." console_audit.record_event( action_id=action_id, result=console_audit.RESULT_DENIED, principal=principal, target={"playbook_id": playbook_id, "target": target}, reason_code="contaminated_runtime", detail=detail, request_id=request_id, session_id=session_id, ) return { "success": False, "allowed": False, "error": "contaminated_runtime", "detail": detail, } # 4. Execute playbook action applied_result: dict[str, Any] = {"performed": False} if playbook_id == PLAYBOOK_CLEAR_STALE_BINDING: binding_before = _active_binding(mutation_env) diagnosis = diagnose_recovery(env=env) plan = stale_binding_recovery.plan_recovery(diagnosis.stale_binding) applied_result = stale_binding_recovery.apply_recovery(plan, env=mutation_env) binding_after = _active_binding(mutation_env) applied_result = { **applied_result, "binding_before": binding_before, "binding_after": binding_after, "binding_changed": binding_before != binding_after, } # A clear that did not clear is not a success, whatever the plan said. if not applied_result["binding_changed"]: applied_result["performed"] = False applied_result.setdefault("reasons", []).append( "clear_stale_binding did not change the live worktree binding" ) elif playbook_id == PLAYBOOK_REBIND_SESSION: target_wt = target or (params or {}).get("target_worktree") if target_wt: binding_before = _active_binding(mutation_env) mutation_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV] = target_wt binding_after = _active_binding(mutation_env) applied_result = { "performed": binding_after == target_wt, "rebound_worktree": target_wt, "binding_before": binding_before, "binding_after": binding_after, "binding_changed": binding_before != binding_after, "cleared_stale": binding_before != binding_after, } if binding_after != target_wt: applied_result["reasons"] = [ "rebind_session_worktree did not take effect on the live " "environment" ] else: applied_result = { "performed": False, "reason": "No target_worktree specified for rebind.", } elif playbook_id == PLAYBOOK_RECONCILE_CLEANUPS: # ``merged_cleanup_reconcile`` exposes the building blocks only; the # orchestrator is the MCP tool. The previous call named a function that # does not exist, and a bare ``except`` turned the AttributeError into a # generic failure, so this playbook could never succeed. Imported lazily # because the MCP server module is large and binds FastMCP at import. try: import gitea_mcp_server snapshot = gitea_mcp_server.gitea_reconcile_merged_cleanups( dry_run=False, execute_confirmed=True, remote=(params or {}).get("remote") or _console_remote(env), org=(params or {}).get("org"), repo=(params or {}).get("repo"), ) performed_reconcile = bool(snapshot.get("success")) applied_result = { "performed": performed_reconcile, "reconciled_count": len(snapshot.get("entries") or []), "snapshot": snapshot, } if not performed_reconcile: applied_result["reasons"] = list(snapshot.get("reasons") or []) except Exception as exc: # noqa: BLE001 — surfaced with its type applied_result = { "performed": False, "error": str(exc), "error_type": type(exc).__name__, } elif playbook_id == PLAYBOOK_SANCTIONED_RESTART: ns = target or (params or {}).get("namespace", "gitea-author") md = (params or {}).get("mode", sanctioned_restart.MODE_RESTART) restart_res = sanctioned_restart.execute_restart( namespace=ns, mode=md, principal=principal, confirmation=f"{md} {ns}", # Without the marker the stricter guard at sanctioned_restart.py:375 # never fires and a restart can launder a contaminated runtime. contamination_marker=contamination_marker, env=mutation_env, request_id=request_id, session_id=session_id, ) applied_result = restart_res # ``allowed`` is not ``performed``: execute_restart documents that success is # False in both directions because the host supervisor still has to act. performed = bool(applied_result.get("performed")) # 5. Record Audit Log audit_record = console_audit.record_event( action_id=action_id, result=console_audit.RESULT_ALLOWED if performed else console_audit.RESULT_DENIED, principal=principal, target={"playbook_id": playbook_id, "target": target}, reason_code="recovery_executed" if performed else "recovery_failed", detail=f"Executed recovery playbook {playbook_id}", request_id=request_id, session_id=session_id, metadata={"applied_result": applied_result}, ) # 6. Post-recovery verification recheck. # # Re-read state rather than re-reading the mapping the mutation just wrote: # verifying the mutated copy confirmed changes that never reached the # process. Passing ``env`` through means a caller-supplied mapping is the # live one for that caller, and ``None`` re-reads ``os.environ`` fresh. post_verification = verify_post_recovery(env=env) return { "success": performed, "allowed": True, "playbook_id": playbook_id, "action_id": action_id, "applied_result": applied_result, "audit": audit_record, "post_recovery_verification": post_verification, } def verify_post_recovery( repo_path: Path | str | None = None, env: dict[str, str] | None = None ) -> dict[str, Any]: """Revalidate control-plane state post-recovery before clean status.""" diag = diagnose_recovery(repo_path, env) classification = diag.stale_binding.get("classification") # ``not clear_eligible`` also reads clean for every binding recovery is not # allowed to touch — an unverified inherited binding is unproven, not clean. binding_clean = ( not diag.stale_binding.get("clear_eligible") and classification != stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED ) return { "clean": diag.clean, "status": diag.status, "stale_runtime_clean": not diag.stale_runtime.get("stale"), "binding_clean": binding_clean, "binding_classification": classification, # The gate returns ``block``; it has never returned ``contaminated``, so # reading that key reported every runtime clean unconditionally. "contamination_clean": not diag.contamination.get("block"), "anomalies_count": len(diag.worktree_anomalies), "reasons": list(diag.reasons), }