"""Runtime inventory + update plan for the fleet-update pipeline.

One read-only pass answering, BEFORE any mutation: which Hermes runtimes run on this machine, how
each is deployed, which ones this update touches, and how each restarts. Every collector is a
side-effect-free probe, so ``hermes update --plan`` is safe on a live fleet.
"""

from __future__ import annotations

import logging
import shlex
import sys
from contextlib import contextmanager, suppress
from dataclasses import dataclass, field, asdict, fields as dataclass_fields
from typing import Any, Callable, Optional

logger = logging.getLogger(__name__)


@dataclass
class RuntimeRecord:
    """One running (or expected) Hermes runtime on this machine."""

    kind: str                     # gateway | dashboard | serve
    profile: str
    pid: Optional[int] = None
    supervisor: str = "manual"    # systemd | launchd | desktop | windows-service | service | manual | manual-serve
    code_sha: Optional[str] = None       # stamped running-code sha
    # See #91283.
    code_version: Optional[str] = None
    restart_via: str = ""         # mechanism id, see _RESTART_MECHANISMS
    detail: dict = field(default_factory=dict)


@dataclass
class UpdatePlan:
    """The full pre-update picture: install shape + runtimes + actions."""

    install_method: str = "unknown"       # git | docker | nix | apt | ...
    updatable_in_place: bool = True
    update_mechanism: str = "hermes update"
    expected_sha: Optional[str] = None    # current checkout HEAD (pre-pull)
    expected_version: Optional[str] = None
    profiles: list = field(default_factory=list)
    runtimes: list = field(default_factory=list)  # list[RuntimeRecord]

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)  # recursive: RuntimeRecord entries become dicts

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "UpdatePlan":
        """Inverse of :meth:`to_dict` (the plan crosses the post-swap hand-off as JSON)."""
        fields_ = {f.name for f in dataclass_fields(cls)}
        plan = cls(**{k: v for k, v in data.items() if k in fields_ and k != "runtimes"})
        record_fields = {f.name for f in dataclass_fields(RuntimeRecord)}
        plan.runtimes = [
            RuntimeRecord(**{k: v for k, v in r.items() if k in record_fields})
            for r in data.get("runtimes") or [] if isinstance(r, dict)
        ]
        return plan


def _detect_supervisor_for_pid(pid: int, service_pids: set, windows_service_pids: set | None = None) -> str:
    """Classify how a live gateway PID is supervised."""
    if windows_service_pids and pid in windows_service_pids:
        # SCM-supervised Windows gateway: the update pause machinery stops the SERVICE via sc.exe
        # instead of killing the child, so reconciliation must plan it under its own mechanism id.
        # See #91277.
        return "windows-service"
    if pid not in service_pids:
        return "manual"
    with suppress(Exception):
        from hermes_cli.gateway import is_macos, supports_systemd_services

        if supports_systemd_services():
            return "systemd"
        if is_macos():
            return "launchd"
    return "service"


# THE restart policy table: restart execution consumes these ids via match_runtime_outcomes / the
# update's restart phase, and the receipt records per-runtime outcomes against them. Display
# strings are derived by describe_restart_mechanism — never the other way around.
_RESTART_MECHANISMS = {
    "systemd": "systemd", "launchd": "launchd", "desktop": "desktop",
    "windows-service": "windows-service", "manual-serve": "respawn-argv",
    "desktop-ssh": "desktop-ssh",
}

_MECHANISM_DESCRIPTIONS = {
    "systemd": "systemctl restart (drain-first SIGUSR1 when supported)",
    "launchd": "launchctl kickstart -k (drain-first, per-label domain)",
    "desktop": "Desktop app respawns its serve backend",
    "windows-service": "sc.exe stop before venv mutation, sc.exe start after update",
    "respawn-argv": "stop before code swap, relaunch with recorded launch args",
    "desktop-ssh": "the remote Desktop that spawned it over SSH respawns it when it reconnects",
}

_SERVE_KINDS = ("serve", "dashboard")
# Serve backends a Desktop client owns and recycles: this app's own pool child (``desktop``) or one
# another machine's Desktop spawned here over SSH (``desktop-ssh``). The updater never restarts
# either; stopping one out from under its client only makes the client respawn it.
CLIENT_OWNED_SERVE_SUPERVISORS = frozenset({"desktop", "desktop-ssh"})


def _restart_mechanism(supervisor: str, profile: str) -> str:
    """Machine-readable restart mechanism id for a runtime.

    THE policy table (#91277 Phase 2): restart execution consumes these ids via
    :func:`match_runtime_outcomes` / the update's restart phase, and the receipt records per-runtime
    outcomes against them. Display strings are derived by :func:`describe_restart_mechanism` — never the
    other way around.
    """
    return _RESTART_MECHANISMS.get(supervisor, "manual")


def describe_restart_mechanism(mechanism: str, profile: str) -> str:
    """Human-readable description of a restart mechanism id."""
    return _MECHANISM_DESCRIPTIONS.get(mechanism) or (
        f"hermes -p {profile} gateway restart" if profile != "default" else "hermes gateway restart"
    )


def _runtime(
    kind: str, profile: str, pid: Optional[int], supervisor: str,
    code_sha: Any = None, code_version: Any = None, **extra: Any,
) -> RuntimeRecord:
    """A :class:`RuntimeRecord` with ``restart_via`` derived from its supervisor."""
    return RuntimeRecord(
        kind=kind, profile=profile, pid=pid, supervisor=supervisor,
        code_sha=str(code_sha) if code_sha else None, code_version=code_version,
        restart_via=_restart_mechanism(supervisor, profile), **extra,
    )


@contextmanager
def _probe(label: str):
    """Run one inventory collector; a failure is logged at debug and yields fewer rows, never an exception."""
    try:
        yield
    except Exception as exc:
        logger.debug("%s failed: %s", label, exc)


def _collect_install_shape(plan: UpdatePlan) -> None:
    with _probe("Install-method probe"):
        from hermes_cli.config import detect_install_method, get_managed_system, recommended_update_command_for_method

        method = detect_install_method()
        managed = get_managed_system()
        plan.install_method = managed or method
        plan.updatable_in_place = method in ("git", "unknown") and not managed
        # Baked image provenance is authoritative when present: a bind-mounted checkout inside a
        # container can look like `git` while the running filesystem is an immutable image.
        # Fail-closed: an invalid marker still flips the plan to not-updatable.
        with _probe("Image provenance probe"):
            # See #91277.
            from hermes_cli.image_provenance import read_image_provenance

            provenance = read_image_provenance()
            if provenance is not None:
                plan.updatable_in_place = False
                if provenance.valid and provenance.manager:
                    plan.install_method = provenance.manager
        plan.update_mechanism = recommended_update_command_for_method(method)


def _supervisor_classifier() -> Callable[[int], str]:
    """``pid -> supervisor`` over the service-PID sets; each probe degrades to an empty set."""
    service_pids: set = set()
    with _probe("Service-PID probe"):
        from hermes_cli.gateway import _get_service_pids

        service_pids = _get_service_pids(all_profiles=True) or set()
    # Windows SCM services (no-op off Windows): the update's pause phase stops these via `sc.exe
    # stop` / restarts via `sc.exe start`, so the plan must carry the matching mechanism id.
    # --- SCM-supervised gateway PIDs (Windows) ------------------------------
    # find_windows_gateway_services() maps validated gateway PIDs through process ancestry to running SCM
    # service PIDs (no-op off Windows). See #91277.
    windows_service_pids: set = set()
    with _probe("Windows SCM service-ownership probe"):
        from hermes_cli.gateway import find_windows_gateway_services

        windows_service_pids = {int(service.gateway_pid) for service in find_windows_gateway_services()}
    return lambda pid: _detect_supervisor_for_pid(pid, service_pids, windows_service_pids)


def _collect_gateway_runtimes(plan: UpdatePlan, profile_homes: list, seen: set[int]) -> None:
    """Per-profile gateways: control-socket identity first (declared by the process itself, including
    supervisor provenance — no argv/PID inference), ``gateway_state.json`` fallback, then PID-file
    mapped gateways no status record covers."""
    supervisor = _supervisor_classifier()
    with _probe("Gateway-state inventory"):
        from gateway.status import live_gateway_pid_for_home, read_runtime_status
        from hermes_cli.update_receipt import _socket_identity

        for profile, home in profile_homes:
            sock = _socket_identity(home)
            if sock is not None:
                pid, record = sock
                if pid in seen:
                    continue  # one multiplex gateway answers identify for several homes — one record per process
                seen.add(pid)
                declared = record.get("supervisor")
                sup = str(declared) if declared else supervisor(pid)
            else:
                # Verified identity, not bare PID existence: a ``stopped`` record whose PID was recycled
                # by an unrelated process fabricated a phantom gateway the restart phase could never
                # touch, so `hermes update` exited partial (#109680).
                pid = live_gateway_pid_for_home(home)
                if pid is None or pid in seen:
                    continue
                record = read_runtime_status(home / "gateway_state.json") or {}
                seen.add(pid)
                sup = supervisor(pid)
            plan.runtimes.append(_runtime("gateway", profile, pid, sup, record.get("code_sha"), record.get("code_version")))
    with _probe("PID-file gateway inventory"):
        from hermes_cli.gateway import find_profile_gateway_processes

        for proc in find_profile_gateway_processes():
            if proc.pid not in seen:
                seen.add(proc.pid)
                plan.runtimes.append(_runtime("gateway", proc.profile, proc.pid, supervisor(proc.pid)))


def _loaded_backend_launchd_jobs() -> list:
    """Loaded launchd dashboard/serve jobs for supervisor classification.

    The probe itself is darwin-gated (``[]`` on every other host); here any failure also degrades
    to ``[]`` — classification falls back to the spawner probe and never aborts the inventory.
    See #116503."""
    with suppress(Exception):
        from hermes_cli import main_dashboard as _dash

        return _dash._loaded_launchd_backend_jobs()
    return []


def _launchd_owner_for_ledger_entry(entry: dict, pid: int, jobs: list) -> "tuple[str, str, int | None] | None":
    """``(domain, label, live_pid)`` of the loaded launchd job owning this ledger row, if any.

    A KeepAlive LaunchAgent backend's recorded spawner (the bootstrap shell) is long dead, so the
    spawner probe alone misreads the row as ``manual-serve`` — and a respawn-argv restart then
    fights the job's own KeepAlive respawn. The loaded-job match (live PID, an ancestor, or the
    normalized ``ProgramArguments``) is the authoritative classification. See #116503."""
    with suppress(Exception):
        from hermes_cli import main_dashboard as _dash
        from hermes_cli.dashboard_procs import _process_ancestors

        try:
            cmdline = shlex.split(str(entry.get("argv") or "")) or None
        except ValueError:
            cmdline = None
        return _dash._launchd_job_owning_backend(pid, cmdline, jobs, ancestors=_process_ancestors(pid))
    return None


def _is_desktop_ssh_ledger_entry(entry: dict) -> bool:
    """Is this row the backend a (possibly remote) Desktop spawned over SSH? The canonical argv
    predicate also classifies rows written before the ledger carried ``isolated``, which is exactly
    the pre-update serve the first update after this change inventories."""
    from hermes_cli._startup_fast import is_desktop_ssh_backend_argv

    try:
        return is_desktop_ssh_backend_argv(shlex.split(str(entry.get("argv") or "")))
    except ValueError:
        return False


def _collect_ledger_runtimes(plan: UpdatePlan, seen: set[int]) -> None:
    """Serve/dashboard backends from the spawn ledger — runtimes the gateway collectors can never see
    (a manual `hermes serve --host <ip>` for a remote Desktop, a long-lived `hermes dashboard`).
    ledger_entries() live-verifies (pid, create_time) so PID reuse never fabricates a row. Desktop-
    supervised backends (spawner still alive) restart via the Desktop's own respawn, not ours.
    A backend owned by a loaded launchd job is classified ``launchd`` (kickstart restart, never a
    detached argv respawn) — the spawner probe cannot see that (#116503)."""
    with _probe("Serve/dashboard ledger inventory"):
        from hermes_cli.process_identity import ledger_entries, spawner_is_dead

        launchd_jobs = _loaded_backend_launchd_jobs()
        for entry in ledger_entries():
            purpose, pid = entry.get("purpose"), entry.get("pid")
            if purpose not in _SERVE_KINDS or not isinstance(pid, int) or pid in seen:
                continue
            seen.add(pid)
            # detail.create_time: process incarnation, not just the numeric PID — a post-update
            # survivor probe comparing PIDs alone calls a NEW serve that reused the number a survivor.
            detail = {
                "argv": entry.get("argv") or "", "host": entry.get("host") or "",
                "port": entry.get("port"), "create_time": entry.get("create_time"),
            }
            job = _launchd_owner_for_ledger_entry(entry, pid, launchd_jobs) if launchd_jobs else None
            if job:
                supervisor, detail["launchd_domain"], detail["launchd_label"] = "launchd", job[0], job[1]
            elif _is_desktop_ssh_ledger_entry(entry):
                # No local spawner, so the probe below would read manual-serve and file a reminder
                # nobody here can discharge; its token file and owner nonce belong to the client.
                supervisor = "desktop-ssh"
            else:
                supervisor = "desktop" if spawner_is_dead(entry) is False else "manual-serve"
            plan.runtimes.append(_runtime(
                str(purpose), str(entry.get("profile") or "default"), pid, supervisor, detail=detail,
            ))


def collect_runtime_inventory() -> UpdatePlan:
    """Build the pre-update plan. Read-only; never raises — every collector degrades independently.

    The result is embeddable in the update receipt and printable via :func:`print_update_plan`.
    """
    plan = UpdatePlan()
    _collect_install_shape(plan)
    with _probe("Code-identity probe"):
        from hermes_cli.version_info import get_code_identity

        identity = get_code_identity(refresh=True)
        plan.expected_sha = identity.get("sha")
        plan.expected_version = identity.get("version")
    profile_homes: list = []
    with _probe("Profile enumeration"):
        from hermes_cli.update_receipt import _profile_homes

        profile_homes = _profile_homes()
        plan.profiles = [name for name, _ in profile_homes]
    seen: set[int] = set()
    _collect_gateway_runtimes(plan, profile_homes, seen)
    _collect_ledger_runtimes(plan, seen)
    return plan


def print_update_plan(plan: UpdatePlan) -> None:
    """Human-readable plan — what the update will touch and how."""
    print("Update plan:")
    install = f"  Install: {plan.install_method}"
    if plan.expected_version:
        install += f" (v{plan.expected_version}" + (f" @ {plan.expected_sha[:8]}" if plan.expected_sha else "") + ")"
    print(install)
    if not plan.updatable_in_place:
        print("  ⚠ This install is NOT updatable in place.")
        print(f"    Update via: {plan.update_mechanism}")
    print(f"  Profiles: {', '.join(plan.profiles) if plan.profiles else '(none found)'}")
    if not plan.runtimes:
        print("  Running Hermes services: none detected — code swap only.")
        return
    print(f"  Running services to restart ({len(plan.runtimes)}):")
    for runtime in plan.runtimes:
        sha = f" @ {runtime.code_sha[:8]}" if runtime.code_sha else ""
        print(f"    • {runtime.kind} [{runtime.profile}] pid {runtime.pid} — {runtime.supervisor}{sha}")
        print(f"      restart: {describe_restart_mechanism(runtime.restart_via, runtime.profile)}")


def _serve_unit_matches_profile(profile: str, unit: object) -> bool:
    """Does *unit* name a ``hermes-serve*``/``hermes-dashboard*`` unit for *profile*? (OWN vocabulary;
    the gateway's ``hermes-gateway*`` names never cover serve/dashboard runtimes.)

    Exact names only — ``work`` must not claim ``hermes-serve-workbench`` — and a scope prefix
    (``user/hermes-serve``) is tolerated because the restart phase records scope-qualified identities in
    some lists. See #100479.
    """
    name = str(unit).removesuffix(".service").rsplit("/", 1)[-1]
    suffix = "" if profile == "default" else f"-{profile}"
    return name in {f"hermes-serve{suffix}", f"hermes-dashboard{suffix}"}


def _gateway_service_matches_profile(profile: str, service: object) -> bool:
    """Match an exact gateway service/label (systemd/launchd/s6 shapes) to a profile.

    Never substring-match: ``foo`` must not claim ``hermes-gateway-foobar.service``.
    Launchd labels are ``ai.hermes.gateway`` / ``ai.hermes.gateway-<profile>`` — they do
    not contain the substring ``hermes-gateway``, so a successful macOS kickstart must
    still credit the planned default gateway. A scope prefix (``user/hermes-gateway``,
    ``gui/501/ai.hermes.gateway``) is stripped the same way serve units are.
    """
    name = str(service).removesuffix(".service").rsplit("/", 1)[-1]
    if profile == "default":
        return name in {"hermes-gateway", "ai.hermes.gateway", "gateway", "gateway-default"}
    return name in {f"hermes-gateway-{profile}", f"ai.hermes.gateway-{profile}", f"gateway-{profile}"}


def _gateway_named_in(r: RuntimeRecord, names: set) -> bool:
    # Gateway-only vocabulary: a serve/dashboard that merely shares the profile is a
    # different process. Exact label match (systemd + launchd + s6), not substring.
    return any(_gateway_service_matches_profile(r.profile, name) for name in names)


def match_runtime_outcomes(
    plan: "UpdatePlan", *, restarted_services: list, relaunched_profiles: list,
    externally_supervised_profiles: list, killed_pids: set, failed_units: list,
    stale_serve_pids: "set | None" = None, failed_respawn_pids: "set | None" = None,
    external_gateway_pids: "set | None" = None,
    live_gateway_pids: "dict[str, set[int]] | None" = None,
) -> list[dict[str, Any]]:
    """Reconcile the plan's runtimes against what the restart phase DID.

    The platform restart branches each re-discover their own targets, so a runtime the plan saw can
    be missed with no signal. Returns one ``{kind, profile, pid, mechanism, outcome}`` row per
    planned runtime; outcome is ``restarted``, ``stopped``, ``failed``, ``deferred``, ``external``
    or ``unaccounted`` (no bookkeeping mentions it — the blind-spot tripwire). Never raises.
    Serve/dashboard runtimes are reconciled in their OWN vocabulary and never borrow the gateway's
    outcome: with ``stale_serve_pids`` a pre-update serve whose incarnation is gone counts as
    ``restarted``, one still alive is ``unaccounted``; without the probe an untouched serve stays
    ``unaccounted``. A Desktop-supervised serve is ``deferred`` only when the survivor probe RAN
    and still lists its pid: the restart phase is forbidden to restart it out from under the app (it
    hosts the live Desktop chats), so it is handed back to its supervisor and surfaced. Without a
    probe result it remains ``unaccounted``, rather than claiming the app owns an unknown
    incarnation. ``failed_respawn_pids`` (serves the dashboard cleanup stopped and could not bring
    back) read ``failed`` before the probe, whose "gone" is exactly what a failed respawn looks like
    (#109290). The probe itself fails closed (unreadable ledger -> every planned serve is listed as
    surviving), so ``deferred`` means "not shown to be gone", not "observed alive". See #111494.

    ``live_gateway_pids`` (profile -> gateway PIDs alive for that profile AFTER the restart phase:
    control-socket identity, or a runtime-status record whose PID is still live) carries the same
    incarnation evidence for gateways: the plan identifies a gateway by the
    profile it SERVES, while the restart bookkeeping names the SERVICE it runs under, and those two
    disagree whenever a service serves a profile its name does not encode — a root-home launchd
    label running the sticky active profile, a hash-suffixed systemd unit for a custom
    ``HERMES_HOME``. No name rule can bridge that, so a planned PID that is gone while a gateway
    answers for the same profile counts as ``restarted``. A missing successor, or the planned PID
    still answering, stays ``unaccounted`` — the tripwire keeps its teeth.

    That credit needs an unambiguous baseline: a profile with SEVERAL planned gateway runtimes cannot
    be attributed to one successor (the fleet probe publishes at most one row per profile), so one
    replacement cannot have replaced two planned processes. Such a profile keeps the name-path
    verdict instead of letting an untouched sibling vanish behind a successor that can only have
    replaced one of them. That resolves conservatively on purpose: pairing a successor to the planned
    process it replaced would need a start-time identity that neither the plan record nor the fleet
    row carries, and a tripwire firing on an anomalous plan is the intended direction.

    Credit is deliberately NOT gated on restart bookkeeping either: the failure this fallback exists
    for already had bookkeeping (``restarted_services`` named the service label — only the
    profile↔label name match failed), and a gateway launchd respawned by itself has no restart-phase
    record at all while the fleet row still proves its successor runs the new code.

    See #91277.
    They never borrow the gateway's outcome: ``relaunched_profiles`` and ``hermes-gateway*`` name a
    different process that shares the profile, nothing more. See #100479.

    ``external_gateway_pids``: gateway pids the post-restart fleet probe verified as serving a
    DIFFERENT checkout than this updater (a ``profiles/<name>`` symlinked to a separate install).
    An untouched gateway in that set is ``external`` — another install owns its restart — instead
    of ``unaccounted``. No evidence keeps the tripwire armed. See #120240.
    """
    outcomes: list[dict[str, Any]] = []
    try:
        failed_set = {str(u) for u in (failed_units or [])}
        restarted_set = {str(s) for s in (restarted_services or [])}
        relaunched = set(relaunched_profiles or []) | set(externally_supervised_profiles or [])
        killed = {int(p) for p in (killed_pids or set())}
        stale_serves = {int(p) for p in stale_serve_pids} if stale_serve_pids is not None else None
        failed_respawns = {int(p) for p in (failed_respawn_pids or set())}
        external = {p for p in (external_gateway_pids or ()) if isinstance(p, int)}
        successors = (
            {str(profile): {p for p in pids if isinstance(p, int)} for profile, pids in live_gateway_pids.items()}
            if live_gateway_pids is not None
            else None
        )
        def _bookkeeping_resolves(r: RuntimeRecord) -> bool:
            """Terminal verdict already available from the restart phase's bookkeeping."""
            if r.pid is not None and r.pid in killed:
                return True
            if r.profile in relaunched:
                return True
            return _gateway_named_in(r, failed_set) or _gateway_named_in(r, restarted_set)

        # Baseline identity: the fleet probe publishes at most one row per profile, so successor
        # evidence is only attributable when ONE planned gateway for that profile still needs a
        # verdict. Serve/dashboard rows are different processes, and rows bookkeeping already resolved
        # (killed / relaunched / name-matched) are not candidates either — a stopped orphan must not
        # make its surviving sibling's successor look ambiguous.
        pending_gateways: dict[str, int] = {}
        for _planned in plan.runtimes:
            if (isinstance(_planned, RuntimeRecord) and _planned.kind == "gateway"
                    and not _bookkeeping_resolves(_planned)):
                pending_gateways[_planned.profile] = pending_gateways.get(_planned.profile, 0) + 1

        def _outcome(r: RuntimeRecord) -> str:
            killed_here = r.pid is not None and r.pid in killed
            if r.kind in _SERVE_KINDS:
                if killed_here:
                    return "stopped"
                if r.pid in failed_respawns or any(_serve_unit_matches_profile(r.profile, u) for u in failed_set):
                    return "failed"
                if stale_serves is not None and r.pid not in stale_serves:
                    # Incarnation-verified: the pre-update process is gone (replaced by its unit / the
                    # dashboard cleanup respawn / the Desktop app).
                    return "restarted"
                if r.supervisor in CLIENT_OWNED_SERVE_SUPERVISORS:
                    if stale_serves is not None:
                        # Still alive on pre-update code, but a Desktop client owns it and the restart phase
                        # must not kill it (_DESKTOP_SERVE_SKIP_REASON); only that client can pick up the new code.
                        return "deferred"
                    return "unaccounted"
                if stale_serves is not None:
                    return "unaccounted"
                return "restarted" if any(_serve_unit_matches_profile(r.profile, s) for s in restarted_set) else "unaccounted"
            if r.profile in relaunched:
                return "restarted"
            if killed_here:
                return "stopped"
            if _gateway_named_in(r, failed_set):
                return "failed"
            if _gateway_named_in(r, restarted_set):
                return "restarted"
            if successors is not None and r.pid is not None:
                # Incarnation-verified last resort: the planned process is gone and a gateway answers
                # for the same profile now. Name-independent on purpose — the supervising service's
                # label/unit may encode the install root (or a hashed home) instead of the served
                # profile, which the profile-scoped name matcher above can never credit.
                live = successors.get(r.profile)
                ambiguous = pending_gateways.get(r.profile, 0) > 1
                if live and r.pid not in live and not ambiguous:
                    return "restarted"
                if ambiguous:
                    logger.debug(
                        "%s planned gateway runtimes for profile %r still need a verdict — successor "
                        "evidence cannot attribute the restart, leaving pid %s on the name path",
                        pending_gateways[r.profile], r.profile, r.pid,
                    )
                elif not live:
                    # No row for this profile: the fallback has nothing to work with and reconciliation
                    # stays on the service-name path. Logged because the tripwire below reads identically
                    # whether the evidence was missing or the restart was actually missed.
                    logger.debug(
                        "No post-restart gateway evidence for profile %r (planned pid %s via %s) — "
                        "reconciling on restart bookkeeping alone",
                        r.profile, r.pid, r.restart_via,
                    )
            return "external" if r.pid in external else "unaccounted"

        for r in plan.runtimes:
            if isinstance(r, RuntimeRecord):
                outcomes.append(
                    {"kind": r.kind, "profile": r.profile, "pid": r.pid, "mechanism": r.restart_via, "outcome": _outcome(r)}
                )
    except Exception as exc:
        logger.debug("Runtime-outcome reconciliation failed: %s", exc)
    return outcomes


def report_unaccounted_runtimes(outcomes: list[dict[str, Any]]) -> bool:
    """Print a loud warning for runtimes the restart phase never touched.

    Returns True when at least one planned runtime is unaccounted; the caller escalates like a
    STALE/DOWN fleet row (exit 1) — a promised restart silently missed is the class this phase
    exists to kill.
    """
    manual = [o for o in outcomes if o.get("outcome") == "deferred" and o.get("mechanism") == "respawn-argv"]
    if manual:
        print()
        print("  ⚠ Manual serve restarts deferred to their owner (reminders retained until the old processes exit):")
        for o in manual:
            print(f"    • {o['kind']} [{o['profile']}] pid {o['pid']}: relaunch `hermes serve` / `hermes dashboard`, or reconnect Desktop for an SSH backend")
    deferred = [o for o in outcomes if o.get("outcome") == "deferred" and o.get("mechanism") != "respawn-argv"]
    if deferred:
        # Surfaced but not escalated: the updater has no authority over these, so holding
        # ``fleet_restart_pending`` for them would never be discharged. See #111494.
        print()
        print("  ℹ Left to the Desktop app (still on pre-update code until it is relaunched):")
        for o in deferred:
            action = ("owned by a Desktop connected over SSH; it picks up the update when that Desktop reconnects"
                      if o.get("mechanism") == "desktop-ssh" else "relaunch the Desktop app to pick up the update")
            print(f"    • {o['kind']} [{o['profile']}] pid {o['pid']} — {action}")
    external = [o for o in outcomes if o.get("outcome") == "external"]
    if external:
        # Surfaced but not escalated: another install's updater owns these restarts. See #120240.
        print()
        print("  ℹ Left to their own install (separate checkout, not restarted by this update):")
        for o in external:
            print(f"    • {o['kind']} [{o['profile']}] pid {o['pid']}")
    missed = [o for o in outcomes if o.get("outcome") == "unaccounted"]
    if not missed:
        return False
    print()
    print("  ⚠ Planned runtimes the restart phase never touched:")
    for o in missed:
        print(f"    ✗ {o['kind']} [{o['profile']}] pid {o['pid']} — planned mechanism: {o['mechanism']}")
    print("    Restart them manually, then verify:")
    if any(o.get("kind") not in _SERVE_KINDS for o in missed):
        print("      hermes gateway restart                # active profile")
        print("      hermes -p <profile> gateway restart   # named profile")
    if any(o.get("kind") in _SERVE_KINDS for o in missed):
        # A serve/dashboard is not reachable by any `gateway restart` command: name the process, not the wrong verb.
        # See #100479.
        if sys.platform == "linux":
            print("      systemctl --user restart hermes-serve.service   # unit-managed serve")
        print("      relaunch `hermes serve` / `hermes dashboard`")
    return True


def record_plan_in_receipt(plan: UpdatePlan) -> None:
    """Attach the inventory to the active update receipt. Never raises."""
    try:
        import hermes_cli.update_receipt as ur

        current = ur._current.get()
        if current is not None:
            current.data["plan"] = plan.to_dict()
    except Exception as exc:
        logger.debug("Could not record plan in receipt: %s", exc)