"""MCP restart coordinator and impact analysis (#658). Before any sanctioned MCP restart, a central coordinator must evaluate the live control-plane state — active sessions, leases/locks, in-flight issue/PR work, mutations, worktrees, and recovery history — and produce an *impact preview* so operators (and the web console, #642/#652) can see the blast radius **before** concurrent LLM work is disrupted. Design rules (mirrors the read-only posture of ``workflow_dashboard`` / ``lease_lifecycle``): * **Pure classification.** :func:`evaluate_restart_impact` takes an already gathered inventory and returns a structured report. It never touches the network, the filesystem, or a live process, so multi-session fixtures can drive every branch in unit tests. The coordinator *never restarts anything*; a mutative apply path is a later child gated by a drain proof (non-goal here). * **Fail closed.** If the inventory is not explicitly complete, the verdict is ``unsafe`` / deny — an incomplete evaluation must never green-light a restart. * **No secrets.** Session ids, pids, and profiles are operational metadata, not credentials; nothing secret flows through this module. The single sanctioned entry point post-#657 is the MCP tool ``gitea_request_mcp_restart`` (dry-run by default), which gathers the inventory from the #613 control-plane DB and calls :func:`evaluate_restart_impact`. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Mapping, Sequence import lease_lifecycle COORDINATOR_VERSION = "1.0.0-issue-658" # Restart verdicts. Exactly the three the acceptance criteria name. VERDICT_SAFE = "safe" VERDICT_UNSAFE = "unsafe" VERDICT_OVERRIDE = "override" # Blast-radius severity bands. BLAST_NONE = "none" BLAST_LOW = "low" BLAST_MEDIUM = "medium" BLAST_HIGH = "high" # A live lease with a live owner process is treated as active in-flight work. LEASE_FRESHNESS_LIVE = "active" # Default staleness window for a session heartbeat (seconds). A session whose # last heartbeat is older than this is not counted as live even if its row is # still marked ``active`` — it is assumed dead/detached. DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS = 900 def _utc_now() -> datetime: return datetime.now(timezone.utc) def _parse_ts(value: str | None) -> datetime | None: return lease_lifecycle._parse_ts(value) @dataclass(frozen=True) class SessionImpact: """One MCP session a restart would terminate.""" session_id: str role: str | None profile: str | None pid: int | None status: str | None alive: bool | None heartbeat_stale: bool is_requester: bool live: bool def as_dict(self) -> dict[str, Any]: return { "session_id": self.session_id, "role": self.role, "profile": self.profile, "pid": self.pid, "status": self.status, "alive": self.alive, "heartbeat_stale": self.heartbeat_stale, "is_requester": self.is_requester, "live": self.live, } @dataclass(frozen=True) class LeaseImpact: """One control-plane lease a restart would disrupt.""" lease_id: str | None session_id: str | None role: str | None phase: str | None freshness: str | None work_kind: str | None work_number: int | None worktree_path: str | None disruptive: bool is_mutation: bool is_critical_section: bool def as_dict(self) -> dict[str, Any]: return { "lease_id": self.lease_id, "session_id": self.session_id, "role": self.role, "phase": self.phase, "freshness": self.freshness, "work_kind": self.work_kind, "work_number": self.work_number, "worktree_path": self.worktree_path, "disruptive": self.disruptive, "is_mutation": self.is_mutation, "is_critical_section": self.is_critical_section, } @dataclass(frozen=True) class RestartImpactReport: """Impact preview DTO returned to the console / operator (#642/#652).""" coordinator_version: str evaluated_at: str dry_run: bool restart_performed: bool inventory_complete: bool verdict: str allow_restart: bool override_would_allow: bool operator_override: bool blast_radius: str reasons: list[str] affected_sessions: list[SessionImpact] affected_leases: list[LeaseImpact] critical_sections: list[LeaseImpact] affected_issues: list[int] affected_prs: list[int] mutations: list[LeaseImpact] terminal_lock: dict[str, Any] | None ack_state: dict[str, str] prior_recovery_attempts: list[dict[str, Any]] counts: dict[str, int] audit_record: dict[str, Any] incomplete_reasons: list[str] = field(default_factory=list) def as_dict(self) -> dict[str, Any]: return { "coordinator_version": self.coordinator_version, "evaluated_at": self.evaluated_at, "dry_run": self.dry_run, "restart_performed": self.restart_performed, "inventory_complete": self.inventory_complete, "incomplete_reasons": list(self.incomplete_reasons), "verdict": self.verdict, "allow_restart": self.allow_restart, "override_would_allow": self.override_would_allow, "operator_override": self.operator_override, "blast_radius": self.blast_radius, "reasons": list(self.reasons), "affected_sessions": [s.as_dict() for s in self.affected_sessions], "affected_leases": [l.as_dict() for l in self.affected_leases], "critical_sections": [l.as_dict() for l in self.critical_sections], "affected_issues": list(self.affected_issues), "affected_prs": list(self.affected_prs), "mutations": [l.as_dict() for l in self.mutations], "terminal_lock": self.terminal_lock, "ack_state": dict(self.ack_state), "prior_recovery_attempts": list(self.prior_recovery_attempts), "counts": dict(self.counts), "audit_record": dict(self.audit_record), } def _classify_session( row: Mapping[str, Any], *, now: datetime, requesting_session_id: str | None, heartbeat_stale_seconds: int, ) -> SessionImpact: session_id = str(row.get("session_id") or "") pid = row.get("pid") status = (row.get("status") or "").strip().lower() or None alive = lease_lifecycle.is_process_alive(pid) if pid is not None else None hb = _parse_ts(row.get("last_heartbeat_at")) heartbeat_stale = bool( hb is not None and (now - hb).total_seconds() > heartbeat_stale_seconds ) live = bool(status == "active" and alive is not False and not heartbeat_stale) return SessionImpact( session_id=session_id, role=row.get("role"), profile=row.get("profile"), pid=pid, status=status, alive=alive, heartbeat_stale=heartbeat_stale, is_requester=bool( requesting_session_id and session_id == requesting_session_id ), live=live, ) # Lease phases that represent an active mutation in flight (as opposed to a # mere allocation/claim with no work committed yet). An active lease in any of # these phases is a critical section a restart must not sever. _MUTATING_PHASES = frozenset( { "implementing", "publishing", "pushing", "committing", "reviewing", "merging", "reconciling", "conflict_fix", } ) def _classify_lease(row: Mapping[str, Any]) -> LeaseImpact: freshness_obj = row.get("freshness") if isinstance(freshness_obj, Mapping): freshness = str(freshness_obj.get("freshness") or "").strip().lower() or None else: freshness = str(freshness_obj or "").strip().lower() or None phase = (row.get("phase") or "").strip().lower() or None worktree = row.get("worktree_path") disruptive = freshness == LEASE_FRESHNESS_LIVE # A live lease is a mutation-in-flight if it carries an author worktree or # its phase names a mutating step. All disruptive leases are critical # sections a restart would sever regardless. is_mutation = bool( disruptive and (bool(worktree) or (phase in _MUTATING_PHASES)) ) number = row.get("work_number") try: number = int(number) if number is not None else None except (TypeError, ValueError): number = None return LeaseImpact( lease_id=row.get("lease_id"), session_id=row.get("session_id"), role=row.get("role"), phase=phase, freshness=freshness, work_kind=(str(row.get("work_kind") or "").strip().lower() or None), work_number=number, worktree_path=worktree, disruptive=disruptive, is_mutation=is_mutation, is_critical_section=disruptive, ) def _blast_radius(*, session_count: int, work_count: int, mutation_count: int) -> str: if mutation_count > 0 or work_count >= 3 or session_count >= 3: return BLAST_HIGH if work_count > 0 or session_count == 2: return BLAST_MEDIUM if session_count == 1: return BLAST_LOW return BLAST_NONE def evaluate_restart_impact( inventory: Mapping[str, Any], *, now: datetime | None = None, operator_override: bool = False, requesting_session_id: str | None = None, dry_run: bool = True, session_heartbeat_stale_seconds: int = DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS, ) -> RestartImpactReport: """Evaluate a proposed MCP restart and return an impact preview. ``inventory`` is a mapping with: * ``sessions`` — session rows (session_id, role, profile, pid, status, last_heartbeat_at). * ``leases`` — control-plane lease rows, each ideally carrying an enriched ``freshness`` dict (as :func:`lease_lifecycle.list_active_leases` returns); a bare string freshness is also accepted. * ``terminal_lock`` — the active terminal (merge) lock, if any. * ``prior_recovery_attempts`` — narrower recovery attempts already tried (e.g. sanctioned reconnects) so the operator sees escalation history. * ``inventory_complete`` — bool. **Must** be explicitly True; a missing or falsy value forces a deny (fail closed). * ``incomplete_reasons`` — optional reasons the inventory is incomplete. The coordinator never restarts anything: ``restart_performed`` is always False and the mutative apply path is a later drain-gated child. """ moment = now or _utc_now() reasons: list[str] = [] inventory_complete = bool(inventory.get("inventory_complete", False)) incomplete_reasons = [str(r) for r in (inventory.get("incomplete_reasons") or [])] sessions_raw: Sequence[Mapping[str, Any]] = inventory.get("sessions") or [] leases_raw: Sequence[Mapping[str, Any]] = inventory.get("leases") or [] terminal_lock = inventory.get("terminal_lock") or None prior_recovery_attempts = [ dict(a) for a in (inventory.get("prior_recovery_attempts") or []) ] session_impacts = [ _classify_session( s, now=moment, requesting_session_id=requesting_session_id, heartbeat_stale_seconds=session_heartbeat_stale_seconds, ) for s in sessions_raw ] lease_impacts = [_classify_lease(l) for l in leases_raw] # Only *other* live sessions and live leases constitute blast radius: a # restart that would kill only the requesting session with no other work in # flight is safe. other_live_sessions = [ s for s in session_impacts if s.live and not s.is_requester ] disruptive_leases = [l for l in lease_impacts if l.disruptive] critical_sections = [l for l in lease_impacts if l.is_critical_section] mutations = [l for l in lease_impacts if l.is_mutation] affected_issues = sorted( { l.work_number for l in disruptive_leases if l.work_kind == "issue" and l.work_number is not None } ) affected_prs = sorted( { l.work_number for l in disruptive_leases if l.work_kind == "pr" and l.work_number is not None } ) disruptive = bool(disruptive_leases or other_live_sessions or terminal_lock) if not inventory_complete: verdict = VERDICT_UNSAFE allow_restart = False reasons.append( "inventory incomplete: restart evaluation cannot confirm blast " "radius — deny (fail closed, #658)" ) reasons.extend(incomplete_reasons) elif not disruptive: verdict = VERDICT_SAFE allow_restart = True reasons.append("no other live sessions, live leases, or terminal lock") elif operator_override: verdict = VERDICT_OVERRIDE allow_restart = True reasons.append( "live work present; operator override accepts the blast radius" ) else: verdict = VERDICT_UNSAFE allow_restart = False reasons.append( "live work would be disrupted; restart denied without operator " "override" ) if critical_sections and inventory_complete: reasons.append( f"{len(critical_sections)} critical section(s) in flight " "(active lease with a live owner)" ) if terminal_lock: reasons.append("active terminal (merge) lock present") override_would_allow = bool(inventory_complete and disruptive) blast_radius = _blast_radius( session_count=len(other_live_sessions), work_count=len(affected_issues) + len(affected_prs), mutation_count=len(mutations), ) # Acknowledgement is a later child (drain protocol); expose per-session # placeholders so the console can render the ack column now. ack_state = {s.session_id: "pending" for s in other_live_sessions} counts = { "sessions_total": len(session_impacts), "sessions_live_other": len(other_live_sessions), "leases_total": len(lease_impacts), "leases_disruptive": len(disruptive_leases), "critical_sections": len(critical_sections), "mutations": len(mutations), "affected_issues": len(affected_issues), "affected_prs": len(affected_prs), "prior_recovery_attempts": len(prior_recovery_attempts), } audit_record = { "event": "restart_impact_evaluated", "coordinator_version": COORDINATOR_VERSION, "evaluated_at": moment.isoformat(), "dry_run": dry_run, "operator_override": bool(operator_override), "requesting_session_id": requesting_session_id, "inventory_complete": inventory_complete, "verdict": verdict, "allow_restart": allow_restart, "blast_radius": blast_radius, "counts": counts, } return RestartImpactReport( coordinator_version=COORDINATOR_VERSION, evaluated_at=moment.isoformat(), dry_run=dry_run, restart_performed=False, inventory_complete=inventory_complete, verdict=verdict, allow_restart=allow_restart, override_would_allow=override_would_allow, operator_override=bool(operator_override), blast_radius=blast_radius, reasons=reasons, affected_sessions=session_impacts, affected_leases=lease_impacts, critical_sections=critical_sections, affected_issues=affected_issues, affected_prs=affected_prs, mutations=mutations, terminal_lock=dict(terminal_lock) if isinstance(terminal_lock, Mapping) else terminal_lock, ack_state=ack_state, prior_recovery_attempts=prior_recovery_attempts, counts=counts, audit_record=audit_record, incomplete_reasons=incomplete_reasons, )