Gitea-Tools/restart_coordinator.py

"""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,
    )