"""Sanctioned MCP restart and graceful reload controls (#642, Phase 2). Sessions have historically recovered MCP connectivity by killing the host daemon (``pkill -f mcp_server.py``, #630). That path stays forbidden: it kills every namespace on the host, contaminates the surviving session, and leaves no audit trail. This module is the sanctioned replacement. A restart is modelled as a *gated action*, never as a command: 1. **Capability** — the console action resolves through ``console_authz`` against ``task_capability_map``, so the console cannot invent an authority the MCP layer does not already define. 2. **Preview** — :func:`build_restart_preview` renders a mutation ledger and the exact confirmation phrase. It never returns a shell command. 3. **Confirmation** — the operator echoes a phrase naming the exact namespace and mode. A phrase for one namespace never authorizes another. 4. **Operator authorization** — host daemon maintenance is authorized out of band through the environment (#630, and #710 finding F1: a worker session cannot set an env var for an already-running daemon, so this cannot be self-asserted the way a tool argument could). 5. **Execution** — :func:`execute_restart` never spawns a process. Once every gate passes it hands the request to the configured host-managed restart hook; with no hook configured it fails closed. 6. **Health recheck** — :func:`verify_post_restart_health` requires live client-namespace probe evidence before any post-restart clean claim. Manual ``pkill`` remains forbidden and is classified as contamination by :func:`classify_restart_command`, which blocks clean claims (#630 AC3). This module performs no I/O beyond reading its own environment configuration, imports no MCP client, and holds no credential. """ from __future__ import annotations import os from dataclasses import asdict, dataclass from typing import Any import mcp_namespace_health import runtime_recovery_guard from webui import console_audit, console_authz # --- Operations ------------------------------------------------------------- MODE_RESTART = "restart" MODE_RELOAD = "reload" MODES: tuple[str, ...] = (MODE_RESTART, MODE_RELOAD) ACTION_RESTART_NAMESPACE = "system.restart_namespace" ACTION_RELOAD_NAMESPACE = "system.reload_namespace" ACTION_FOR_MODE: dict[str, str] = { MODE_RESTART: ACTION_RESTART_NAMESPACE, MODE_RELOAD: ACTION_RELOAD_NAMESPACE, } # Namespaces the console may target. An unlisted name fails closed rather than # being passed through to a host hook. KNOWN_NAMESPACES: tuple[str, ...] = tuple( sorted( set(mcp_namespace_health.DEFAULT_NAMESPACES) | {"gitea-author", "gitea-reviewer", "gitea-merger", "gitea-reconciler", "gitea-controller"} ) ) # Scope tokens that would mean "everything at once". Explicit non-goal: the # console never offers a fleet-wide restart, because that is the blast radius # `pkill -f mcp_server.py` already had. _FLEET_TOKENS = frozenset({"*", "all", "fleet", "any", ""}) # --- Environment configuration ---------------------------------------------- # Read server-side only; the value is an opaque host hook reference (e.g. a # launchd label), never a command line, and is never rendered to a client. RESTART_HOOK_ENV = "GITEA_SANCTIONED_RESTART_HOOK" # --- Reason codes ----------------------------------------------------------- DENY_UNKNOWN_MODE = "unknown_mode" DENY_UNKNOWN_NAMESPACE = "unknown_namespace" DENY_FLEET_SCOPE = "fleet_scope_not_permitted" DENY_UNAUTHORIZED = "unauthorized" DENY_CONFIRMATION_MISSING = "confirmation_required" DENY_CONFIRMATION_MISMATCH = "confirmation_mismatch" DENY_OPERATOR_AUTHORIZATION = "operator_authorization_missing" DENY_HOOK_NOT_CONFIGURED = "restart_hook_not_configured" DENY_CONTAMINATED_RUNTIME = "contaminated_runtime" ALLOW_HOST_ACTION_REQUIRED = "host_action_required" # Post-restart verification outcomes. HEALTH_CLEAN = "clean" HEALTH_UNPROVEN = "unproven" HEALTH_UNHEALTHY = "unhealthy" def _clean(value: Any) -> str: return str(value or "").strip() # --- Mutation ledger -------------------------------------------------------- @dataclass(frozen=True) class RestartLedgerEntry: """One planned step, shown before anything is asked of the host.""" sequence: int step: str summary: str executes_process_kill: bool = False def _mutation_ledger(namespace: str, mode: str) -> tuple[RestartLedgerEntry, ...]: if mode == MODE_RELOAD: middle = RestartLedgerEntry( sequence=2, step="host_graceful_reload", summary=( f"Ask the configured host supervisor to reload {namespace} " "in place, draining in-flight requests. The console does not " "signal the process itself." ), ) else: middle = RestartLedgerEntry( sequence=2, step="host_restart_hook", summary=( f"Ask the configured host supervisor to restart {namespace}. " "The console never sends a signal and never runs a kill." ), ) return ( RestartLedgerEntry( sequence=1, step="quiesce", summary=( f"Stop admitting new gated mutations for {namespace} and " "record the intent before anything restarts." ), ), middle, RestartLedgerEntry( sequence=3, step="health_recheck", summary=( f"Re-probe {namespace} through the live client namespace and " "prove the required tool is callable again." ), ), RestartLedgerEntry( sequence=4, step="audit", summary=( "Append actor, target namespace, mode, and result to the " "console audit log." ), ), ) # --- Confirmation ----------------------------------------------------------- def confirmation_phrase(namespace: str, mode: str) -> str: """Exact phrase an operator must echo, naming the namespace and mode. Binding the namespace into the phrase is the point: a confirmation typed for ``gitea-author`` cannot be replayed against ``gitea-merger``. """ return f"{_clean(mode)} {_clean(namespace)}" def confirmation_matches( namespace: str, mode: str, confirmation: str | None ) -> bool: """Compare *confirmation* to the required phrase (exact, whitespace-trimmed).""" return _clean(confirmation) == confirmation_phrase(namespace, mode) # --- Scope validation ------------------------------------------------------- def _validate_scope(namespace: str, mode: str) -> tuple[str, str] | None: """Return ``(reason_code, detail)`` when the scope is refused.""" ns = _clean(namespace) md = _clean(mode) if md not in MODES: return ( DENY_UNKNOWN_MODE, f"Mode {md!r} is not one of {', '.join(MODES)}.", ) if ns.lower() in _FLEET_TOKENS: return ( DENY_FLEET_SCOPE, ( "Fleet-wide restart is an explicit non-goal: it reproduces the " "blast radius of `pkill -f mcp_server.py` (#630). Restart one " "namespace at a time." ), ) if ns not in KNOWN_NAMESPACES: return ( DENY_UNKNOWN_NAMESPACE, f"Namespace {ns!r} is not a known MCP namespace.", ) return None # --- Host hook -------------------------------------------------------------- def restart_hook(env: dict[str, str] | None = None) -> dict[str, Any]: """Report the configured host-managed restart hook. The hook is a reference the *host* resolves (a supervisor label), not a command this process runs. ``configured=False`` fails restart closed. """ source = env if env is not None else os.environ reference = _clean(source.get(RESTART_HOOK_ENV)) return { "configured": bool(reference), "reference": reference or None, "source": RESTART_HOOK_ENV if reference else None, "self_assertable": False, "console_executes_process": False, } # --- Preview ---------------------------------------------------------------- def build_restart_preview( namespace: str, mode: str = MODE_RESTART, *, principal: console_authz.Principal | None = None, env: dict[str, str] | None = None, ) -> dict[str, Any]: """Render the dry-run preview for a restart/reload request. Read-only: no authorization is granted, no host is contacted, and the result never contains a shell command. """ ns = _clean(namespace) md = _clean(mode) action_id = ACTION_FOR_MODE.get(md, ACTION_RESTART_NAMESPACE) action = console_authz.get_action(action_id) decision = console_authz.authorize(action_id, principal) scope_error = _validate_scope(ns, md) hook = restart_hook(env) operator = runtime_recovery_guard.operator_authorization(env) return { "action_id": action_id, "namespace": ns, "mode": md, "scope_valid": scope_error is None, "scope_reason_code": scope_error[0] if scope_error else None, "scope_detail": scope_error[1] if scope_error else None, "required_role": action.minimum_role if action else None, "required_permission": action.mcp_permission if action else None, "action_class": action.action_class if action else None, "dual_control": action.dual_control if action else True, "break_glass": action.break_glass if action else True, "requires_confirmation": True, "confirmation_phrase": confirmation_phrase(ns, md), "mutation_ledger": [asdict(entry) for entry in _mutation_ledger(ns, md)], "authorization": decision.to_dict(), "operator_authorization": operator, "restart_hook": hook, "execution_enabled": False, "raw_process_kill_exposed": False, "known_namespaces": list(KNOWN_NAMESPACES), "post_restart_verification_required": True, } # --- Gate ------------------------------------------------------------------- def assess_restart_request( namespace: str, mode: str = MODE_RESTART, *, principal: console_authz.Principal | None = None, confirmation: str | None = None, contamination_marker: dict[str, Any] | None = None, env: dict[str, str] | None = None, ) -> dict[str, Any]: """Decide whether a restart request may proceed to the host hook. Every gate must pass. The first failure wins and is reported with a stable reason code; a pass never means "restarted", only "may be handed to the configured host hook". """ ns = _clean(namespace) md = _clean(mode) action_id = ACTION_FOR_MODE.get(md, ACTION_RESTART_NAMESPACE) preview = build_restart_preview(ns, md, principal=principal, env=env) def refuse(reason_code: str, detail: str) -> dict[str, Any]: return { "allowed": False, "gates_passed": False, "reason_code": reason_code, "detail": detail, "action_id": action_id, "namespace": ns, "mode": md, "preview": preview, "execution_enabled": False, } scope_error = _validate_scope(ns, md) if scope_error is not None: return refuse(*scope_error) # Authority is checked as an authorization decision, not an execution # grant. ``for_execution=True`` asks "may the console perform this write?", # and the answer here is permanently no: step 2 of the ledger is a request # to the host supervisor, so the console's Phase 2 execution gate is not # the relevant gate. Every branch below keeps ``execution_enabled`` False # and :func:`execute_restart` never touches a process. decision = console_authz.authorize(action_id, principal) if not decision.allowed: return refuse(DENY_UNAUTHORIZED, decision.detail) if not _clean(confirmation): return refuse( DENY_CONFIRMATION_MISSING, ( "Type the confirmation phrase " f"{preview['confirmation_phrase']!r} to proceed." ), ) if not confirmation_matches(ns, md, confirmation): return refuse( DENY_CONFIRMATION_MISMATCH, ( "Confirmation does not name this namespace and mode; expected " f"{preview['confirmation_phrase']!r}." ), ) operator = preview["operator_authorization"] if not operator["authorized"]: return refuse( DENY_OPERATOR_AUTHORIZATION, ( "Host daemon maintenance requires out-of-band operator " "authorization via " f"{runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV}." ), ) # #630's task-scoped gate deliberately lets a contaminated worker keep # commenting and handing off. Restart is stricter and unconditional: a # runtime already contaminated by a manual kill must be reconciled before # it is restarted again, or the restart just launders the contamination. if contamination_marker and not contamination_marker.get( "cleared_by_reconciler" ): return refuse( DENY_CONTAMINATED_RUNTIME, ( "A live contamination marker is present; clear it through the " "reconciler path before restarting." ), ) hook = preview["restart_hook"] if not hook["configured"]: return refuse( DENY_HOOK_NOT_CONFIGURED, ( "No host-managed restart hook is configured " f"({RESTART_HOOK_ENV}). The console will not fall back to a " "process kill." ), ) return { "allowed": True, "gates_passed": True, "reason_code": ALLOW_HOST_ACTION_REQUIRED, "detail": ( "Every gate passed. The restart must be performed by the " "configured host supervisor; the console does not signal the " "process." ), "action_id": action_id, "namespace": ns, "mode": md, "preview": preview, "execution_enabled": False, "console_executes": False, "console_active_phase": console_authz.ACTIVE_PHASE, } # --- Execution -------------------------------------------------------------- def execute_restart( namespace: str, mode: str = MODE_RESTART, *, principal: console_authz.Principal | None = None, confirmation: str | None = None, contamination_marker: dict[str, Any] | None = None, env: dict[str, str] | None = None, request_id: str | None = None, session_id: str | None = None, ) -> dict[str, Any]: """Run every gate, audit the outcome, and hand off to the host. This function never spawns a process, never sends a signal, and never builds a command line. ``success`` is False in both directions: a refused request is refused, and an authorized request still requires the host supervisor to act. """ assessment = assess_restart_request( namespace, mode, principal=principal, confirmation=confirmation, contamination_marker=contamination_marker, env=env, ) action_id = assessment["action_id"] allowed = assessment["allowed"] audit = console_audit.record_event( action_id=action_id, result=( console_audit.RESULT_ALLOWED if allowed else console_audit.RESULT_DENIED ), principal=principal, target={"namespace": assessment["namespace"], "mode": assessment["mode"]}, reason_code=assessment["reason_code"], detail=assessment["detail"], request_id=request_id, session_id=session_id, metadata={ "gates_passed": assessment["gates_passed"], "process_kill_executed": False, "post_restart_verification_required": True, }, ) return { "success": False, "outcome": ( ALLOW_HOST_ACTION_REQUIRED if allowed else assessment["reason_code"] ), "allowed": allowed, "detail": assessment["detail"], "namespace": assessment["namespace"], "mode": assessment["mode"], "action_id": action_id, "process_kill_executed": False, "host_hook": assessment["preview"]["restart_hook"], "next_action": ( "Have the host supervisor perform the restart, then call " "verify_post_restart_health with live client-namespace evidence " "before claiming a clean session." if allowed else assessment["detail"] ), "assessment": assessment, "audit": audit, } # --- Contamination classification ------------------------------------------- def classify_restart_command( command: str | None, *, mcp_pids: list[Any] | tuple[Any, ...] | None = None, session_id: str | None = None, remote: str | None = None, role: str | None = None, ) -> dict[str, Any]: """Classify an operator-proposed recovery command (#630 AC2). A manual ``pkill``/``kill`` of the MCP daemon is contamination, not a restart. When contaminating, a durable marker is returned so downstream gated mutations and clean claims fail closed. """ classification = runtime_recovery_guard.classify_recovery_command( command, mcp_pids=mcp_pids ) contaminating = bool(classification.get("contamination")) marker = None if contaminating: marker = runtime_recovery_guard.build_contamination_record( reason_class=( classification.get("reason_class") or runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL ), command_redacted=classification.get("redacted_command"), session_id=session_id, remote=remote, role=role, detail=( "Manual daemon kill is forbidden; use the sanctioned " f"{ACTION_RESTART_NAMESPACE} gated action instead." ), ) return { "contamination": contaminating, "sanctioned": not contaminating and not classification.get("process_kill"), "clean_claim_allowed": not contaminating, "reason_class": classification.get("reason_class"), "redacted_command": classification.get("redacted_command"), "classification": classification, "contamination_marker": marker, "sanctioned_alternative": ACTION_RESTART_NAMESPACE, } # --- Post-restart health verification --------------------------------------- def verify_post_restart_health( namespace: str, *, probe_result: dict[str, Any] | None = None, probe_source: str | None = None, registered_tools: list[str] | tuple[str, ...] | None = None, required_tool: str | None = None, profile: str | None = None, ) -> dict[str, Any]: """Require live proof a namespace is callable before any clean claim (AC3). Static registration is not proof and neither is an offline subprocess probe: only ``probe_source=client_namespace`` evidence can clear a post-restart session for mutations. """ ns = _clean(namespace) health = mcp_namespace_health.classify_namespace_probe( ns, required_tool=required_tool, registered_tools=registered_tools, probe_result=probe_result, profile=profile, probe_source=probe_source, ) healthy = bool(health.get("healthy")) proven = bool(health.get("ide_namespace_proven")) if healthy and proven: status = HEALTH_CLEAN elif healthy: status = HEALTH_UNPROVEN else: status = HEALTH_UNHEALTHY reasons = list(health.get("reasons") or []) if status == HEALTH_UNPROVEN: reasons.append( "Namespace reported healthy without live client-namespace " "evidence; a post-restart clean claim requires " f"probe_source={mcp_namespace_health.PROBE_SOURCE_CLIENT!r}." ) return { "namespace": ns, "status": status, "healthy": healthy, "ide_namespace_proven": proven, "clean_claim_allowed": status == HEALTH_CLEAN, "mutations_allowed": status == HEALTH_CLEAN, "reasons": reasons, "health": health, } # --- Policy surface --------------------------------------------------------- def restart_policy() -> dict[str, Any]: """Machine-readable description of the sanctioned restart contract.""" return { "policy_version": 1, "modes": list(MODES), "actions": [ACTION_RESTART_NAMESPACE, ACTION_RELOAD_NAMESPACE], "known_namespaces": list(KNOWN_NAMESPACES), "fleet_scope_permitted": False, "console_executes_process_kill": False, "raw_kill_exposed": False, "requires_confirmation": True, "confirmation_binds_namespace": True, "operator_authorization_env": ( runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV ), "restart_hook_env": RESTART_HOOK_ENV, "manual_kill_classified_as": runtime_recovery_guard.CONTAMINATION_KIND, "post_restart_clean_claim_requires": ( mcp_namespace_health.PROBE_SOURCE_CLIENT ), "audit_required": True, "silent_auto_restart_permitted": False, }