hermes-agent/hermes_cli/update_cmd_fleet.py

"""Gateway fleet restart + post-update verification for ``hermes update``.

Split out of ``hermes_cli/update_cmd.py``; every name is re-imported there so
``hermes_cli.update_cmd.<name>`` keeps resolving/monkeypatching. Origin helpers are
imported lazily inside each function (no import cycle; test patches stay effective).
"""

import json
import logging
import re
from contextlib import suppress
import os
import subprocess
import sys
import time as _time
from dataclasses import dataclass, field
from pathlib import Path

from hermes_cli.update_cmd_common import _best_effort
from hermes_cli.update_inventory import _gateway_service_matches_profile

# Log-record parity with the origin module.
logger = logging.getLogger("hermes_cli.update_cmd")

# Under HERMES_HOME (not next to the venv): records the fleet-restart obligation
# after a pull advanced HEAD; cleared only when the restart completes or nothing ran.
# The existing ``.update-incomplete`` / ``.lazy-refresh-incomplete`` markers gate dependency/venv repair;
# this one is the fleet-restart obligation after a git pull that advanced HEAD (#95294).
_FLEET_RESTART_PENDING_NAME = "fleet_restart_pending"

_FRESH_RESTART_SUPERVISORS = frozenset({"systemd", "launchd", "service", "s6"})

# A supervisor can report a restarted unit active before the gateway finishes its
# bootstrap and publishes ``gateway_state.json``. Keep the readiness poll bounded,
# but allow the default systemd startup budget plus status-publication slack.
_FLEET_PROBE_SETTLE_TIMEOUT_SECONDS = 120.0

_SYSTEMD_SCOPES = (("user", ["systemctl", "--user"]), ("system", ["systemctl"]))
_LIST_GATEWAY_UNITS = ["list-units", "hermes-gateway*", "hermes-serve*", "--plain", "--no-legend", "--no-pager"]


def _write_gateway_update_exit_code(ok: bool) -> None:
    from hermes_cli.update_cmd import get_hermes_home
    path = get_hermes_home() / ".update_exit_code"
    with suppress(OSError):
        path.write_text("0" if ok else "1", encoding="utf-8")


def _fleet_restart_pending_marker_path() -> Path:
    """LEGACY per-``HERMES_HOME`` breadcrumb. Read-compat only — nothing writes it any more.

    One host runs one multiplexing gateway, so the pull→restart obligation is host-scoped
    (``hermes_cli/update_host_obligation.py``). An obligation armed by the old per-profile code
    is still read and cleared here so an in-flight update is discharged after the upgrade.
    """
    from hermes_cli.update_cmd import get_hermes_home
    return get_hermes_home() / _FLEET_RESTART_PENDING_NAME


def _write_legacy_fleet_restart_pending_marker(
    *, expected_sha: str = "", runtimes: list[dict] | None = None
) -> bool:
    """Arm the LEGACY per-``HERMES_HOME`` marker. True when written. Never raises.

    Fallback only: ``$HERMES_HOME`` is writable by construction (the updater already writes its
    receipts there), so it still carries the obligation when the host state dir cannot.
    """
    path = _fleet_restart_pending_marker_path()
    try:
        lines = [f"started={_time.time()}", f"pid={os.getpid()}"]
        if expected_sha:
            lines.append(f"expected_sha={expected_sha}")
        if runtimes is not None:
            lines.append("inventory=" + json.dumps({"version": 1, "runtimes": runtimes}))
        path.write_text("\n".join(lines) + "\n", encoding="utf-8")
        return True
    except OSError as exc:
        logger.debug("Could not write legacy fleet-restart-pending marker: %s", exc)
        return False


def _write_fleet_restart_pending_marker(*, expected_sha: str = "", runtimes: list[dict] | None = None) -> None:
    """Arm the HOST pull→restart obligation. Never raises.

    An unwritable host state dir (``HERMES_GATEWAY_LOCK_DIR`` on a read-only mount, a container
    UID that does not own ``$HOME``) must never disarm the obligation: an update interrupted
    after this point would then leave stale code running with no warning and no catch-up restart
    (#117275). The legacy per-home marker — which every reader here still honours — carries it
    instead, and a host that can write neither says so out loud.
    """
    if runtimes == []:
        # An explicit empty inventory owes no restart (e.g. Desktop-hosted `serve` with no
        # gateway services). Arming the marker here leaves a breadcrumb nothing can discharge:
        # a no-gateway host would then fail every later ``hermes update`` (#115311).
        return
    from hermes_cli.update_cmd import _m
    from hermes_cli.update_host_obligation import host_obligation_path, write_host_obligation
    if _m()._pytest_owns_live_checkout(_fleet_restart_pending_marker_path().parent):
        logger.debug("Skipping fleet-restart-pending obligation under pytest (live checkout)")
        return
    if write_host_obligation(
            expected_sha=expected_sha, runtimes=runtimes, profile=_current_profile_name()):
        return
    if _write_legacy_fleet_restart_pending_marker(expected_sha=expected_sha, runtimes=runtimes):
        logger.warning(
            "Host update-restart obligation (%s) is unwritable; armed the per-home marker %s instead.",
            host_obligation_path(), _fleet_restart_pending_marker_path())
        return
    logger.error(
        "Could not arm the update-restart obligation in %s or %s; an interrupted update will not warn.",
        host_obligation_path(), _fleet_restart_pending_marker_path())
    print(
        "  ⚠ Could not record the pending gateway-restart obligation (state dir not writable) — "
        "restart gateways with `hermes gateway restart` if this update is interrupted.",
        file=sys.stderr,
    )


def _current_profile_name() -> str:
    """Profile whose CLI armed the obligation (diagnostics only — the record is host-scoped)."""
    try:
        from hermes_cli.profiles import get_active_profile_name
        return get_active_profile_name() or "default"
    except Exception:
        return ""


def _clear_fleet_restart_pending_marker() -> None:
    """Discharge the obligation for the whole host (legacy per-home marker included). Never raises."""
    from hermes_cli.update_cmd import _m
    from hermes_cli.update_host_obligation import clear_host_obligation
    clear_host_obligation()
    _m()._clear_marker_file(_fleet_restart_pending_marker_path(), label="fleet-restart-pending")


def _fleet_restart_obligation_armed() -> bool:
    """True when this HOST owes a fleet restart — from any profile's CLI."""
    from hermes_cli.update_host_obligation import host_obligation_present
    if host_obligation_present():
        return True
    with suppress(OSError):
        return _fleet_restart_pending_marker_path().is_file()
    return False


def _obligation_fields() -> dict[str, str] | None:
    """Armed obligation as ``key=value`` fields: HOST record first, then the legacy marker.

    ``None`` means nothing armed OR a malformed record; both must leave the obligation standing.
    """
    from hermes_cli.update_host_obligation import host_obligation_present, obligation_fields
    fields = obligation_fields()
    if fields is not None:
        return fields
    if host_obligation_present():
        # The record exists but its terms are unknown (corrupt, or a NEWER CLI's version). An
        # unrelated legacy marker's inventory cannot discharge terms nobody can read: fail closed.
        return None
    try:
        text = _fleet_restart_pending_marker_path().read_text(encoding="utf-8-sig")
    except (OSError, UnicodeError):
        return None
    legacy: dict[str, str] = {}
    for line in text.splitlines():
        key, sep, value = line.partition("=")
        if not sep or key in legacy:
            return None
        legacy[key] = value
    return legacy


def _current_checkout_sha() -> str | None:
    """Current on-disk checkout HEAD, or None if it cannot be resolved."""
    from hermes_cli.update_cmd import _capture_head_sha, _m
    try:
        from hermes_cli.version_info import get_code_identity
        sha = (get_code_identity(refresh=True) or {}).get("sha")
        return str(sha) if sha else None
    except Exception:
        return _capture_head_sha(["git"], _m().PROJECT_ROOT)


def _receipt_looks_unfinished(receipt: dict) -> bool:
    """True when *receipt* is from an update that did not finish cleanly.

    The command boundary stamps a ``stop_reason`` on every receipt, including clean
    ones (``completed at command boundary``, ``sys.exit(0)``); it must not make a
    successful receipt look unfinished, or the next ``hermes update`` retriggers
    ``fleet_restart_pending`` from pre-pull plan SHAs (#98022).
    """
    exit_code = receipt.get("exit_code")
    outcome = receipt.get("outcome")
    if exit_code not in (0, None) or outcome in ("failed", "partial", "running"):
        return True
    gateway_restart = receipt.get("gateway_restart")
    if isinstance(gateway_restart, dict) and gateway_restart.get("incomplete"):
        return True
    # A stop_reason alone (update_contract refusals: outcome="refused", no exit_code)
    # counts only when nothing else vouched for success.
    succeeded = exit_code == 0 or outcome == "success"
    return bool(receipt.get("stop_reason")) and not succeeded


def _receipt_reports_stale_runtime(receipt: dict, expected_sha: str | None = None) -> bool:
    """True when ``update_receipts/latest.json`` records a runtime SHA skew.

    Prefer the post-restart ``fleet`` matrix. ``plan.runtimes[].code_sha`` is captured
    *before* the pull, so a finished update's plan always looks stale and must not
    retrigger a restart; consult it only for an unfinished receipt.

    See #95294.
    """
    from hermes_cli.update_cmd import _current_checkout_sha
    if not isinstance(receipt, dict):
        return False
    expected_sha = expected_sha or _current_checkout_sha()
    if not expected_sha:
        return False

    def _sha_mismatch(code_sha) -> bool:
        return bool(code_sha) and str(code_sha) != str(expected_sha)

    from hermes_cli.update_receipt import row_is_external

    fleet = receipt.get("fleet")
    if isinstance(fleet, list) and fleet:
        return any(
            isinstance(entry, dict)
            and not row_is_external(entry)
            and (entry.get("state") == "stale" or _sha_mismatch(entry.get("code_sha")))
            for entry in fleet
        )

    if not _receipt_looks_unfinished(receipt):
        return False
    plan = receipt.get("plan")
    if not isinstance(plan, dict):
        return False
    return any(
        isinstance(runtime, dict) and _sha_mismatch(runtime.get("code_sha"))
        for runtime in plan.get("runtimes") or []
    )


_SUPERVISED_SERVE_BACKENDS = frozenset(
    {"manual-serve", "desktop", "desktop-ssh", "systemd", "launchd", "windows-service", "service"}
)
# Backends whose supervisor restarts the process without any updater bookkeeping. ``manual-serve``
# is excluded: it owes a durable handoff (``defer_manual_serve``) before it stops counting.
# ``systemd``/``windows-service``/``service`` mirror ``_SUPERVISED_SERVE_BACKENDS`` for parity only —
# the inventory writer classifies a serve/dashboard row as exactly launchd, desktop, desktop-ssh or manual-serve
# (``update_inventory._collect_ledger_runtimes``); those three are set for gateway rows alone.
_SUPERVISOR_OWNED_SERVE_BACKENDS = _SUPERVISED_SERVE_BACKENDS - {"manual-serve"}


def _receipt_owed_gateways(receipt: dict, pending_manual: list[dict]) -> set[tuple[str, str]] | None:
    """Pure coverage classification after manual retention of this receipt snapshot.

    Empty means this receipt owes no gateways, not that an independent marker owes none. Unknown identities, unclassified serve backends and failed manual transfers make coverage unverified.
    """
    plan = receipt.get("plan") or {}
    entries: list[tuple[object, str | None]] = [(entry, None) for entry in plan.get("runtimes") or []]
    entries.extend((entry, None) for entry in receipt.get("pending_manual_serves") or [])
    entries.extend((entry, "gateway") for entry in receipt.get("fleet") or [])
    owed: set[tuple[str, str]] = set()
    unverified = False
    for entry, default_kind in entries:
        if not isinstance(entry, dict):
            unverified = True
            continue
        kind = entry.get("kind", default_kind)
        profile = entry.get("profile")
        # A serve/dashboard row is outside the gateway matrix's evidence, not evidence against the
        # gateways it does cover: a supervised backend (desktop, systemd, launchd) is its
        # supervisor's to restart, and a manual-serve row outside the retention list is the serve
        # obligation mechanism's — a host running a dashboard carries such a row in every receipt,
        # and a blanket veto made the gateway warning permanently undischargeable there (#115090).
        # Only an unclassified backend or a failed manual transfer still makes coverage unverified.
        if kind in ("serve", "dashboard") and entry.get("supervisor") in _SUPERVISED_SERVE_BACKENDS and entry not in pending_manual:
            continue
        if kind != "gateway" or not profile or profile == "unknown":
            unverified = True
            continue
        owed.add((kind, profile))
    return None if unverified else owed


def _fleet_covered_gateways(fleet: list) -> set[tuple[str, str]] | None:
    """``(kind, profile)`` identities the live rows vouch for; ``None`` when any row is unidentified.

    A multiplexer's row carries the ``served_profiles`` its runtime status records (``_fleet_row``
    keeps the field only when well-formed), so one live process covers every profile it serves.
    """
    covered: set[tuple[str, str]] = set()
    for row in fleet:
        profile = row.get("profile") if isinstance(row, dict) else None
        if not profile or profile == "unknown":
            return None  # unidentified runtime: the matrix cannot vouch for it
        covered.add(("gateway", profile))
        covered.update(("gateway", served) for served in row.get("served_profiles") or [])
    return covered


def _live_fleet_covers_receipt(expected_sha: str | None, receipt: dict, owed: set[tuple[str, str]] | None, *, accept_states: tuple = ("current",)) -> bool:
    """Require a successor at the expected SHA for every owed gateway identity."""
    if not expected_sha:
        return False
    from hermes_cli.update_receipt import collect_fleet_versions, row_is_external

    try:
        if owed is None:
            return False
        if not owed:
            return bool((receipt.get("plan") or {}).get("runtimes"))
        fleet = collect_fleet_versions()
        # State labels are checkout-relative; completed restarts may accept stale rows at the pulled SHA.
        if not fleet or any(
            row.get("state") not in accept_states or row.get("code_sha") != expected_sha
            for row in fleet if not row_is_external(row)
        ):
            return False
        covered = _fleet_covered_gateways(fleet)
        return covered is not None and owed <= covered
    except Exception as exc:
        logger.debug("Could not reconcile pending fleet identities: %s", exc)
        return False


def _marker_owed_gateways(inventory: object) -> set[tuple[str, str]] | None:
    """The ``("gateway", profile)`` set a marker's inventory owes; None when it recorded none.

    Raises ValueError for a malformed or unsupported inventory, which keeps the marker.
    """
    from hermes_cli.update_cmd_fleet_gatewayless import runtime_outside_gateway_evidence

    if inventory is None:
        return None
    if not isinstance(inventory, dict) or inventory.get("version") != 1:
        raise ValueError("unsupported fleet-restart inventory")
    runtimes = inventory.get("runtimes")
    if not isinstance(runtimes, list):
        raise ValueError("fleet-restart inventory has no runtime list")
    owed: set[tuple[str, str]] = set()
    for runtime in runtimes:
        if not isinstance(runtime, dict):
            raise ValueError("fleet-restart inventory row is not an object")
        if runtime_outside_gateway_evidence(runtime):
            continue
        profile = runtime.get("profile")
        if runtime.get("kind") != "gateway" or not isinstance(profile, str) or not profile.strip() or profile == "unknown":
            raise ValueError("fleet-restart inventory row is not an identified gateway")
        owed.add(("gateway", profile))
    return owed


def _discharge_gatewayless_marker(checkout_sha: str, expected_sha: str) -> bool:
    """Settle an inventory-less marker on a host with no live gateway (#118742).

    Only when the host itself shows nothing the update could still owe a restart to, and HEAD
    still holds the code it pulled.
    """
    from hermes_cli.update_cmd_fleet_checkout import checkout_contains
    from hermes_cli.update_cmd_fleet_gatewayless import host_owes_no_gateway_restart

    try:
        gatewayless = (checkout_sha == expected_sha or checkout_contains(expected_sha)) and host_owes_no_gateway_restart()
    except Exception as exc:
        logger.debug("Gateway-less host probe failed; keeping fleet-restart-pending marker: %s", exc)
        return False
    if not gatewayless:
        return False
    _clear_fleet_restart_pending_marker()
    logger.debug("Fleet-restart-pending marker discharged: host runs no gateway at %s", checkout_sha[:10])
    return True


def _marker_only_restart_obsolete() -> bool:
    """Settle only the inventory stored with this marker's target SHA.

    Historical receipts cannot narrow this obligation. Malformed or unsupported inventories stay
    fail-closed; empty discovery never proves a stopped gateway recovered. Two shapes record no
    obligation and settle without one: an explicit empty inventory (a pull that found no gateway,
    #115311) clears outright, and an inventory-less marker (the pre-inventory writer, or a tail
    that died before its inventory was recorded, #115638) clears once every live gateway is
    current on the checkout — there is no recorded owed set, so the fleet running the code on disk
    is the whole of the evidence the marker's warning can be about, even after HEAD moved past
    ``expected_sha`` by an out-of-band pull. With no live gateway at all, the inventory-less marker
    asks the host instead (``update_cmd_fleet_gatewayless``): it clears when no profile left a
    gateway that should be running and every live runtime is supervisor-owned or handed off, so a
    Desktop-only install stops failing every later update (#118742).

    A serve/dashboard row whose supervisor owns the restart (Desktop backend, systemd/launchd
    unit, Windows service) is outside the gateway matrix's evidence, not evidence against it —
    the same boundary ``_receipt_owed_gateways`` draws for receipts (#115090) and the restart
    phase draws for the Desktop backend (#111494). Counting it made the warning permanently
    undischargeable on every host that runs a dashboard. A manual-serve row still needs its
    durable handoff (``defer_manual_serve``), and an unclassified backend stays fail-closed.
    Discharging here strands nobody: the same row is still accounted at update time by
    ``update_inventory.report_unaccounted_runtimes``, which prints it and exits 1 when the restart
    phase never touched it — this marker only stops re-warning about it on every later startup.
    """
    from hermes_cli.update_cmd_fleet_checkout import checkout_contains

    try:
        fields = _obligation_fields()
        if fields is None:
            return False
        expected_sha = fields.get("expected_sha", "").strip()
        owed = _marker_owed_gateways(json.loads(fields.get("inventory", "null")))
    except (OSError, UnicodeError, ValueError):
        return False
    if owed is not None and not owed:
        # A pull that recorded no gateway runtime owes no restart; clearing avoids the
        # stuck "Fleet restart incomplete" loop on Desktop-hosted (no-service) installs.
        _clear_fleet_restart_pending_marker()
        logger.debug("Fleet-restart-pending marker discharged: no gateway obligation recorded")
        return True
    if not expected_sha:
        return False
    checkout_sha = _current_checkout_sha()
    if owed is not None and checkout_sha != expected_sha and not checkout_contains(expected_sha):
        return False  # a newer pull moved HEAD; it owns a fresh obligation
    # HEAD may sit past ``expected_sha`` by a carried local commit (a cherry-picked hotfix) that no
    # pull made and no fresh obligation covers; the fleet is held to the code it actually runs, which
    # is what an equality gate on ``expected_sha`` could never discharge (#119367).
    target_sha = checkout_sha
    if not target_sha:
        return False
    try:
        from hermes_cli.update_receipt import collect_fleet_versions, row_is_external
        fleet = collect_fleet_versions()
    except Exception as exc:
        logger.debug("Fleet probe failed; keeping fleet-restart-pending marker: %s", exc)
        return False
    if not fleet:
        if owed is not None:
            return False  # Absence cannot prove recovery of the recorded inventory.
        return _discharge_gatewayless_marker(checkout_sha, expected_sha)
    covered = _fleet_covered_gateways(fleet)
    if covered is None:
        return False  # unidentified runtime: the matrix cannot vouch for it
    for row in fleet:
        if row_is_external(row):
            continue
        if row.get("state") != "current" or str(row.get("code_sha")) != target_sha:
            return False  # stale / down / unknown-identity row still owes the restart
    if owed is not None and not owed <= covered:
        return False  # A gateway this marker owns is absent (down) or unidentifiable.
    _clear_fleet_restart_pending_marker()
    logger.debug(
        "Fleet-restart-pending marker discharged: %d gateway(s) already serve %s",
        len(fleet), target_sha[:10],
    )
    return True


def _receipt_restart_phase_completed(receipt: dict) -> str | None:
    """Return the pulled SHA when the restart phase completed, even if a later step failed."""
    gateway_restart = receipt.get("gateway_restart")
    if not isinstance(gateway_restart, dict) or not gateway_restart:
        return None
    if gateway_restart.get("incomplete") or gateway_restart.get("phase_error"):
        return None
    post_sha = (receipt.get("post_update") or {}).get("sha")
    return str(post_sha) if post_sha else None


def _pending_fleet_restart_needed(*, receipt: dict | None = None, pending_manual: list[dict] | None = None) -> bool:
    """Require identity-matched gateways at checkout HEAD for update catch-up."""
    from hermes_cli.update_cmd import _current_checkout_sha
    from hermes_cli.update_receipt import read_latest_receipt
    from hermes_cli.update_serve_obligations import retain_receipt_manual_serves

    if receipt is None:
        receipt = read_latest_receipt() or {}
    if pending_manual is None:
        pending_manual = retain_receipt_manual_serves(receipt)
    # The HOST obligation owns its inventory; latest.json can belong to an older update.
    if _fleet_restart_obligation_armed():
        return not _marker_only_restart_obsolete()
    owed = _receipt_owed_gateways(receipt, pending_manual)
    if not _receipt_reports_stale_runtime(receipt):
        return False
    return not _live_fleet_covers_receipt(_current_checkout_sha(), receipt, owed)


def _update_owes_fleet_restart(*, receipt: dict | None = None, pending_manual: list[dict] | None = None) -> bool:
    """Hold a completed restart to the code it pulled, not a later checkout HEAD."""
    from hermes_cli.update_cmd import _current_checkout_sha
    from hermes_cli.update_receipt import read_latest_receipt
    from hermes_cli.update_serve_obligations import retain_receipt_manual_serves

    if receipt is None:
        receipt = read_latest_receipt() or {}
    if pending_manual is None:
        pending_manual = retain_receipt_manual_serves(receipt)
    # A completed older receipt cannot discharge an independent host obligation's inventory.
    if _fleet_restart_obligation_armed():
        return not _marker_only_restart_obsolete()
    owed = _receipt_owed_gateways(receipt, pending_manual)
    if not _receipt_reports_stale_runtime(receipt):
        return False
    restarted_to = _receipt_restart_phase_completed(receipt)
    # A fleet an operator has since restarted onto a moved checkout (``hermes gateway restart`` —
    # the remedy this warning names) has nothing of the update left to owe either.
    if restarted_to and _live_fleet_covers_receipt(restarted_to, receipt, owed, accept_states=("current", "stale")):
        return False
    return not _live_fleet_covers_receipt(_current_checkout_sha(), receipt, owed)


def _warn_pending_fleet_restart(*, startup: bool = False) -> None:
    """Print the specific interrupted-update fleet-restart warning."""
    stream = sys.stderr if startup else sys.stdout
    print("⚠ A previous `hermes update` pulled new code but did not restart running gateways.", file=stream)
    print("  Gateways may still be serving pre-update modules (mixed sys.modules).", file=stream)
    if startup:
        print("  Run `hermes update` or `hermes gateway restart`.", file=stream)


def _warn_pending_fleet_restart_on_startup() -> None:
    """Cheap CLI-startup hint. Never restarts; never raises."""
    from hermes_cli.update_receipt import read_latest_receipt
    from hermes_cli.update_serve_obligations import retain_receipt_manual_serves, warn_pending_manual_serves

    receipt = read_latest_receipt() or {}
    pending_manual = None
    with suppress(Exception):
        pending_manual = retain_receipt_manual_serves(receipt)
    with suppress(Exception):
        if _update_owes_fleet_restart(receipt=receipt, pending_manual=pending_manual):
            _warn_pending_fleet_restart(startup=True)
    with suppress(Exception):
        warn_pending_manual_serves(startup=True, pending_manual=pending_manual)


def _systemd_gateway_unit_listings(on_list_timeout=None):
    """Yield ``(scope, scope_cmd, list-units CompletedProcess)`` per systemd scope that answered.

    A missing systemctl skips the scope silently; a listing timeout skips it after
    ``on_list_timeout(scope, exc)`` (when given) so the other scope is still processed.
    """
    for scope, scope_cmd in _SYSTEMD_SCOPES:
        try:
            result = _systemctl(scope_cmd + _LIST_GATEWAY_UNITS, timeout=10)
        except FileNotFoundError:
            continue
        except subprocess.TimeoutExpired as exc:
            if on_list_timeout is not None:
                on_list_timeout(scope, exc)
            continue
        yield scope, scope_cmd, result


def _needs_sudo(scope: str) -> bool:
    return (
        scope == "system"
        and hasattr(os, "geteuid")
        and os.geteuid() != 0  # windows-footgun: ok — systemd path, Linux-only
    )


def _unit_main_pid(scope_cmd: list, svc_name: str) -> int:
    """Live ``MainPID`` of a unit; ``0`` when inactive, unprivileged or unreadable.

    Property reads need no manage-units privileges, and an unreadable PID is never collapsed:
    identity that cannot be proved keeps its own restart.
    """
    try:
        result = _systemctl(list(scope_cmd) + ["show", svc_name, "--property=MainPID", "--value"], timeout=10)
    except (OSError, subprocess.TimeoutExpired):
        return 0
    if getattr(result, "returncode", 1) != 0:
        return 0
    try:
        return int((getattr(result, "stdout", "") or "").strip() or 0)
    except ValueError:
        return 0


def _restart_systemd_gateway_units_best_effort(failed: list, listings) -> None:
    """Restart every hermes-gateway/serve unit ONCE PER LIVE HOST PROCESS.

    One host runs one multiplexing gateway, so leftover per-profile units
    (``hermes-gateway-<profile>.service``) all point at the SAME live ``MainPID``; restarting
    each in turn restarts the host gateway N times — a self-inflicted N-fold outage triggered
    by one update. Units that share a live main PID are collapsed to one representative and the
    others are named as LEGACY units to migrate, never silently dropped.
    """
    from hermes_cli.update_host_obligation import collapse_units_to_host_processes

    answered = set()
    targets: dict[str, tuple[str, list, str]] = {}  # "<scope>/<unit>" -> (scope, scope_cmd, unit)
    for scope, scope_cmd, result in listings:
        answered.add(scope)
        if result.returncode != 0:
            failed.append(f"systemd-{scope} (listing failed)")
            continue
        _for_each_systemd_gateway_unit(
            result.stdout,
            process_unit=lambda svc_name, _scope=scope, _cmd=scope_cmd: targets.setdefault(
                f"{_scope}/{svc_name}", (_scope, _cmd, svc_name)),
            on_unit_timeout=lambda svc_name, exc: failed.append(svc_name),
        )

    keys = list(targets)
    covered: dict[str, str] = {}
    if len(keys) > 1:
        # Only worth a `systemctl show` round when several units could be one process.
        keys, covered = collapse_units_to_host_processes(
            keys, lambda key: _unit_main_pid(targets[key][1], targets[key][2]))
    for unit_key, owner_key in covered.items():
        print(
            f"  • {unit_key} is a legacy per-profile unit sharing one host gateway process with "
            f"{owner_key}; restarting it again would restart that process twice. Fold the units "
            "together with: hermes gateway migrate"
        )

    for key in keys:
        scope, scope_cmd, svc_name = targets[key]
        if not _systemd_unit_owned_by_update(scope_cmd, svc_name):
            continue
        manage_cmd = list(scope_cmd) + ["--no-ask-password"]
        if _needs_sudo(scope):
            manage_cmd = ["sudo", "-n"] + manage_cmd
        try:
            result = _systemctl_reset_and_restart(manage_cmd, svc_name, scope_cmd=scope_cmd)
            if result.returncode != 0 or not _wait_for_service_active(scope_cmd, svc_name):
                failed.append(svc_name)
        except subprocess.TimeoutExpired:
            failed.append(svc_name)
    # A timeout or missing executable is not an empty scope.
    failed.extend(f"systemd-{scope} (listing unavailable)" for scope, _ in _SYSTEMD_SCOPES if scope not in answered)


def _live_fleet_current_rows() -> list[dict] | None:
    """The fleet matrix when the probe finds at least one gateway and every row is ``current``
    at the checkout SHA (identity known); ``None`` on any unknown/stale/down row or a failed
    probe (restart)."""
    checkout_sha = _current_checkout_sha()
    if not checkout_sha:
        return None
    try:
        from hermes_cli.update_receipt import collect_fleet_versions
        fleet = collect_fleet_versions()
    except Exception as exc:
        logger.debug("Pending fleet restart: fleet probe failed: %s", exc)
        return None
    if not fleet or _fleet_covered_gateways(fleet) is None:
        return None
    if all(row.get("state") == "current" and str(row.get("code_sha")) == checkout_sha for row in fleet):
        return fleet
    return None


def _restart_identity_sha() -> str:
    """The SHA a completed host restart is stamped with; ``""`` when nothing names the code.

    ``_current_checkout_sha()`` is ``None`` on every non-git install (zip, pip, Docker), and an
    empty stamp can never match, so the per-host restart-once guard would be inert exactly on the
    installs it exists for: each profile's ``hermes update`` would re-kill the one shared
    multiplexer. The obligation's own ``expected_sha`` — else the receipt's post-update identity —
    names the same pulled code.
    """
    sha = _current_checkout_sha()
    if sha:
        return str(sha)
    sha = ((_obligation_fields() or {}).get("expected_sha") or "").strip()
    if sha:
        return sha
    with suppress(Exception):
        from hermes_cli.update_receipt import read_latest_receipt
        post_update = (read_latest_receipt() or {}).get("post_update")
        if isinstance(post_update, dict):
            return str(post_update.get("sha") or "")
    return ""


def _fleet_restart_skip_reason(plan) -> str | None:
    """Why the completion tail may leave the fleet alone, or ``None`` when a restart is owed.

    Every route (pulled, already-current, ZIP) now finishes through the same completion
    tail, so the guards the old catch-up path carried live here: one host runs ONE
    multiplexing gateway, so a second profile's ``hermes update`` attaches to the restart the
    first one already stamped (#95294), and a fleet already serving the checkout code (a
    no-op update, a manual ``hermes gateway restart`` seconds ago) is not re-killed (#117051).

    The second guard needs BOTH the pre-update plan and the live probe: the live matrix only
    lists gateways, so a planned ``serve`` still on pre-update code (or any runtime without a
    stamped identity) keeps the restart — the reconciliation there is what surfaces it.
    """
    from hermes_cli.update_host_obligation import host_restart_already_completed
    checkout_sha = _restart_identity_sha()
    if host_restart_already_completed(checkout_sha):
        return "this host's gateway was already restarted for this update"
    if (checkout_sha and plan is not None and plan.runtimes
            and all(str(runtime.code_sha) == checkout_sha for runtime in plan.runtimes)
            and _live_fleet_current_rows() is not None):
        return "every running gateway already serves the checkout code"
    return None


def _run_pending_fleet_restart() -> bool:
    """Historical retry hook; new retries use the ordinary completion owner."""
    from hermes_cli._old_updater import stop_for_relaunch
    stop_for_relaunch(incomplete=True)


def _systemctl(cmd: list, *, timeout: float):
    """Run a systemctl (or sudo systemctl) invocation, capturing utf-8 text with a timeout."""
    return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout)


# poll() takes signed 32-bit milliseconds; keep headroom for rounding in communicate().
_SYSTEMCTL_RESTART_TIMEOUT_MAX = (2**31 - 1) // 1000 - 1


def _systemd_restart_timeout(scope_cmd: list, svc_name: str, *, start_only: bool = False) -> float:
    """Outwait the unit's stop + start budgets, not just the systemctl client.

    A client timeout does not cancel the manager's queued restart. Unknown or
    infinite limits use systemd's usual 90s per phase so automation stays bounded.
    Custom ExecStop chains or EXTEND_TIMEOUT_USEC can still exceed this budget;
    genuine timeouts must continue through the existing per-unit failure path.
    """
    from gateway.shutdown_forensics import parse_systemd_duration_to_us

    budgets = {"TimeoutStartUSec": 90.0}
    if not start_only:
        budgets["TimeoutStopUSec"] = 90.0
    try:
        show = _systemctl(
            scope_cmd + ["show", svc_name, "--property=TimeoutStopUSec,TimeoutStartUSec"],
            timeout=5,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return sum(budgets.values()) + 15.0
    if show.returncode == 0:
        for line in (show.stdout or "").splitlines():
            key, _, raw = line.partition("=")
            if key in budgets:
                # The shared parser returns None for infinity/unrecognized units.
                try:
                    raw = raw.strip()
                    duration = int(raw) if raw.isascii() and raw.isdigit() else parse_systemd_duration_to_us(raw)
                    if duration is not None and duration > 0:
                        budgets[key] = duration / 1_000_000
                except (ValueError, OverflowError):
                    pass
    return min(sum(budgets.values()) + 15.0, _SYSTEMCTL_RESTART_TIMEOUT_MAX)


def _systemctl_reset_and_restart(manage_cmd: list, svc_name: str, *, scope_cmd: list | None = None):
    """``reset-failed`` then ``restart``: a unit parked in failed state by systemd's own
    auto-restart can wedge a plain ``restart`` against RestartSec backoff and stay dead."""
    # Property reads need no manage-units privileges: narrow sudoers may permit
    # restart/reset-failed but deny show. Keep the same user/system manager scope.
    timeout = _systemd_restart_timeout(scope_cmd if scope_cmd is not None else manage_cmd, svc_name)
    _systemctl(manage_cmd + ["reset-failed", svc_name], timeout=10)
    return _systemctl(manage_cmd + ["restart", svc_name], timeout=timeout)


def _systemd_unit_owned_by_update(scope_cmd: list, svc_name: str) -> bool:
    """Gate a unit restart on the unit's home being one this update owns (#93349).

    ``hermes-gateway*`` is an account-wide namespace: a second install's ``hermes update`` used
    to drain and restart the account's real ``hermes-gateway.service`` because the unit was
    listed, not because it ran the updated code. Foreign or unreadable ownership prints a notice
    and leaves the unit alone; it is not a failed restart.
    """
    from hermes_cli.update_fleet_scope import describe_skipped_runtime, systemd_unit_hermes_home, home_in_update_scope
    home = systemd_unit_hermes_home(scope_cmd, svc_name)
    if home is not None and home_in_update_scope(home):
        return True
    print(describe_skipped_runtime("systemd unit", svc_name, home))
    return False


def _scoped_manual_gateway_pids(pids, *, keep=(), quiet: bool = False) -> list[int]:
    """*pids* whose live home this update owns (plus *keep*, PIDs already mapped to this
    install's profile PID files); every other gateway process is named and left running."""
    from hermes_cli.update_fleet_scope import describe_skipped_runtime, partition_gateway_pids_by_scope
    keep = set(keep)
    owned, foreign = partition_gateway_pids_by_scope([pid for pid in pids if pid not in keep])
    if not quiet:
        for pid, home in foreign:
            print(describe_skipped_runtime("gateway process", f"PID {pid}", home))
    return [pid for pid in pids if pid in keep or pid in owned]


def _is_hermes_gateway_unit(unit: str) -> bool:
    """Exact base unit or hyphenated profile family only: ``startswith("hermes-serve")``
    would accept ``hermes-server.service``."""
    return (
        # list-units is already pattern-filtered, but keep the name gate so a stray non-gateway/serve line
        # cannot enter the restart path. See #83595.
        unit == "hermes-gateway.service"
        or unit.startswith("hermes-gateway-")
        or unit == "hermes-serve.service"
        or unit.startswith("hermes-serve-")
    )


def _for_each_systemd_gateway_unit(list_units_stdout: str, *, process_unit, on_unit_timeout) -> None:
    """Process each hermes-gateway*/hermes-serve* unit from ``systemctl list-units``.

    ``TimeoutExpired`` from ``process_unit`` is isolated per unit via ``on_unit_timeout``
    so one wedged systemctl call cannot abort the rest of the fleet.

    See #68523.
    """
    for line in (list_units_stdout or "").strip().splitlines():
        parts = line.split()
        if not parts:
            continue
        unit = parts[0]
        if not unit.endswith(".service") or not _is_hermes_gateway_unit(unit):
            continue
        svc_name = unit.removesuffix(".service")
        try:
            process_unit(svc_name)
        except subprocess.TimeoutExpired as exc:
            on_unit_timeout(svc_name, exc)


def _service_unit_supports_graceful_sigusr1_restart(svc_name: str) -> bool:
    """Whether *svc_name* wires SIGUSR1 to a graceful drain-then-restart.

    Only ``hermes-gateway*`` runs ``gateway/run.py`` (the handler); SIGUSR1 would just
    kill ``hermes-serve*`` and burn the drain budget, so those go straight to the blunt
    restart. Same exact/hyphenated shape as ``_for_each_systemd_gateway_unit`` so a
    near-prefix unit like ``hermes-gatewayd`` is never signalled.

    See #83438.
    """
    return svc_name == "hermes-gateway" or svc_name.startswith("hermes-gateway-")


def _warn_incomplete_gateway_fleet_restart(failed_units: list) -> None:
    """Print an explicit incomplete-update warning for unrestarted units."""
    from hermes_cli.gateway import is_macos
    if not failed_units:
        return
    ordered = list(dict.fromkeys(failed_units))  # de-dup, discovery order
    print()
    print("⚠ Update incomplete — some units were not restarted:")
    for name in ordered:
        print(f"    - {name}")
    if is_macos():
        # A label lands here when launchd wasn't supervising a live process after
        # the restart — likely deregistered, which `launchctl kickstart` can't revive.
        # See #88848.
        print("  Listed services may be deregistered from launchd, or still")
        print("  running pre-update code (mixed sys.modules). Recover with:")
        print("    hermes gateway status")
        print("    launchctl list | grep <label>")
        print("    launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/<label>.plist")
        return
    print("  Skipped units may still be running pre-update code (mixed")
    print("  sys.modules). Restart them manually, then verify:")
    print("    hermes gateway status")
    if any(not name.startswith("ai.hermes.") for name in ordered):
        print("    systemctl --user restart <unit>   # user-scope")
        print("    sudo systemctl restart <unit>     # system-scope")
    if any(name.startswith("ai.hermes.") for name in ordered):
        print("    launchctl kickstart -k gui/$UID/<label>   # macOS (or user/$UID)")


def _restart_launchd_gateway_after_update(
    *, supervision_verify: bool = True, self_restart_pending: set | None = None,
) -> tuple[list, list]:
    """Restart the invoking profile's launchd gateway after an update.

    No ``launchctl list`` gating: a booted-out job (plist present, definition
    deregistered) fails it, and it can exit non-zero while the job is alive — gating on it
    silently skipped the restart yet printed "Update complete!". When the plist exists
    ``launchd_restart()`` always runs; every failure path is loud with a manual recovery
    command. Returns ``(restarted_labels, failed_labels)``; with ``supervision_verify``
    success also requires a fresh supervised PID ("the call returned" is not "supervised").

    74973 (salvage #75021 by @jeff-mettel): the restart used to be gated on ``launchctl list <label>``
    exiting 0. A *booted-out* job — plist present, definition deregistered from launchd (crashed helper,
    manual bootout, failed prior update) — fails that check, so the whole branch silently skipped: no
    restart, no message, ``KeepAlive`` unable to revive a definition launchd no longer knows, and the update
    still printed "Update complete!".
    See #88848.
    """
    from hermes_cli.gateway import (
        get_launchd_label, get_launchd_plist_path, launchd_restart, wait_for_launchd_gateway_supervision,
        _is_pid_ancestor_of_current_process, _launchctl_supervised_pid,
    )
    current_label = get_launchd_label()
    old_pid = None
    try:
        if not get_launchd_plist_path().exists():
            return [], []  # not a launchd install — nothing to do or warn
        # Snapshot BEFORE the restart: "supervising some pid" was true before too, so only a pid that
        # actually changed distinguishes a restart from a no-op (the sibling loop's contract). Read-only
        # and verification-only — the restart itself is never gated on `launchctl list` (#74973).
        old_pid = _launchctl_supervised_pid(current_label) if supervision_verify else None
        try:
            launchd_restart()
        except subprocess.CalledProcessError as e:
            stderr = (getattr(e, "stderr", "") or "").strip()
            print(
                f"  ⚠ Gateway restart failed: {stderr}\n"
                "    The gateway may be DOWN on pre-update code. "
                "Recover manually: hermes gateway restart"
            )
            return [], [current_label]
    except (FileNotFoundError, subprocess.TimeoutExpired) as e:
        # A plist exists, so a gateway is SUPPOSED to be supervised; a broken/wedged
        # launchctl is not proof nothing needs restarting. Count it, tell the operator.
        print(
            # The old code `pass`ed here (#74973's second silent variant); count it and tell the operator.
            "  ⚠ Could not restart the gateway "
            f"({e.__class__.__name__}: {e}).\n"
            "    Recover manually: hermes gateway restart"
        )
        return [], [current_label]

    if not supervision_verify:
        return [current_label], []
    if old_pid is not None and _is_pid_ancestor_of_current_process(old_pid):
        # launchd_restart() handed the restart to the gateway this updater runs INSIDE (cron job in
        # the gateway tree, #100179): it exits only after this process does, so no fresh supervised
        # pid can appear while we wait. Record it as pending for the fleet matrix (#119597).
        if self_restart_pending is not None:
            self_restart_pending.add(old_pid)
        return [current_label], []

    # launchd_restart() returning only means "restart REQUESTED" (async). A helper dying
    # before first bootstrap, or a bootstrap exiting 0 without registering (macOS 26.6.1),
    # would otherwise reach "Update complete!" unsupervised. Verified domain-agnostically:
    # domain locate fails on macOS-26 per-user domains.
    # launchd_restart() returning is only "restart REQUESTED" — the self-restart branch hands work to the
    # running gateway, a plist reload to a detached helper; both asynchronous. See #88848.
    if wait_for_launchd_gateway_supervision(label=current_label, old_pid=old_pid):
        return [current_label], []
    print(
        f"  ✗ {current_label} restarted but launchd is not supervising a new process for it.\n"
        "    Check logs, then: hermes gateway restart"
    )
    return [], [current_label]


def _restart_macos_launchd_gateways(
    restarted_services: list, failed_or_stale_units: list, drain_budget: float, *, require_supervision: bool = False,
    self_restart_pending: set | None = None,
) -> None:
    """Restart every launchd-managed gateway after an update (macOS).

    The pull is shared across profiles, so every ``ai.hermes.gateway*`` LaunchAgent
    must reload it or siblings stay on pre-update ``sys.modules`` (systemd parity).
    Invoking profile uses ``launchd_restart()``; siblings get the same drain-first
    sequence with their domain (``gui/<uid>`` vs ``user/<uid>``) resolved per label so
    none is kickstarted in the wrong domain. ``TimeoutExpired`` is isolated per label.

    See #41403.
    The invoking profile keeps the existing ``launchd_restart()`` treatment (self-restart request → graceful
    drain → kickstart). ``subprocess.TimeoutExpired`` is isolated per label so one wedged launchctl call
    cannot leave the rest of the fleet on old code (#68523).
    """
    from hermes_cli.gateway import (
        get_launchd_label, get_launchd_plist_path, launchd_gateway_labels_for_install, legacy_launchd_labels_for_install,
        _graceful_restart_via_sigusr1, _launchd_kickstart,
        _locate_launchd_gateway_service, _wait_for_launchd_service_pid,
    )
    if require_supervision:
        listing = subprocess.run(["launchctl", "list"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10)
        if listing.returncode != 0:
            failed_or_stale_units.append("launchd (listing failed)")
            return
    _restarted, _failed = _restart_launchd_gateway_after_update(
        supervision_verify=True, self_restart_pending=self_restart_pending)
    restarted_services.extend(_restarted)
    failed_or_stale_units.extend(_failed)
    current_label = get_launchd_label()

    derived_labels = launchd_gateway_labels_for_install()
    # Units labelled before the profile-name suffix scheme (ai.hermes.gateway-<hash>) are invisible
    # to the derivation; legacy_launchd_labels_for_install() credits one only when its plist is
    # provably this install's, so the #41403 boundary (never touch another install's fleet) holds.
    # See #115254.
    legacy_labels = legacy_launchd_labels_for_install(exclude=set(derived_labels) | {current_label})
    if legacy_labels:
        print(f"  ↻ legacy-labelled units of this install join the restart: {', '.join(legacy_labels)}")
    from hermes_cli.update_fleet_scope import describe_skipped_runtime, launchd_label_foreign_home
    for label in derived_labels + legacy_labels:
        if label == current_label:
            continue
        # Labels are account-global: root B's default profile derives the same bare label root A
        # installed. A plist pinning a foreign HERMES_HOME is another install's job (#93349).
        if (foreign_home := launchd_label_foreign_home(label)) is not None:
            print(describe_skipped_runtime("launchd job", label, foreign_home))
            continue
        try:
            # Locate = liveness + domain in one probe; kickstart and fresh-PID checks
            # reuse that domain so a sibling is never probed in one and restarted in another.
            domain, old_pid = _locate_launchd_gateway_service(label)
            if domain is None:
                if require_supervision and get_launchd_plist_path().with_name(f"{label}.plist").exists():
                    failed_or_stale_units.append(label)
                continue  # A profile without an installed job has no restart target.
            graceful_ok = False
            if old_pid is not None and old_pid > 0:
                print(f"  → {label}: draining (up to {int(drain_budget)}s)...")
                from hermes_cli.update_cmd_drain_report import drain_progress_reporter
                graceful_ok = _graceful_restart_via_sigusr1(
                    old_pid, drain_timeout=drain_budget,
                    on_progress=drain_progress_reporter(_gateway_home_for_pid(old_pid), budget_s=drain_budget))
            if graceful_ok and _wait_for_launchd_service_pid(label, old_pid=old_pid, timeout=10.0, domain=domain):
                # KeepAlive already respawned it on new code — a kickstart would kill it.
                restarted_services.append(label)
                continue
            try:
                _launchd_kickstart(label, domain)
            except subprocess.CalledProcessError as e:
                stderr = (getattr(e, "stderr", "") or "").strip()
                failed_or_stale_units.append(label)
                print(
                    f"  ⚠ Failed to restart {label}: {stderr}\n"
                    f"    Recover manually: launchctl kickstart -k {domain}/{label}"
                )
                continue
            if _wait_for_launchd_service_pid(label, old_pid=old_pid, timeout=15.0, domain=domain):
                restarted_services.append(label)
            else:
                failed_or_stale_units.append(label)
                print(
                    f"  ✗ {label} failed to come back after restart.\n"
                    f"    Check logs, then: launchctl kickstart -k {domain}/{label}"
                )
        except subprocess.TimeoutExpired:
            failed_or_stale_units.append(label)
            print(f"  ⚠ launchctl timed out restarting {label}; continuing with remaining gateways")


def _surviving_gateway_pids_after_failed_restart():
    """Best-effort PIDs of gateways still running after the restart phase died.

    ``None`` when undeterminable (notably ``hermes_cli.gateway`` no longer importing
    under the replaced checkout). Callers treat ``None`` and non-empty as "assume
    stale"; only a positive empty result proves nothing needs restarting.
    """
    try:
        from hermes_cli.gateway import find_gateway_pids
        return _scoped_manual_gateway_pids(find_gateway_pids(all_profiles=True), quiet=True)
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("Could not probe for surviving gateways after update: %s", exc)
        return None


_MANUAL_GATEWAY_SKIP_REASON = (
    "manual gateway has no supervisor relaunch authority; left running for explicit operator restart"
)
_DESKTOP_SERVE_SKIP_REASON = (
    "desktop app owns and respawns this serve backend;"
    " the recovery pass must not restart it out from under its supervisor"
)
# NOT a claim that no supervisor exists: a systemd-launched serve sets neither HERMES_SPAWN
# nor HERMES_PARENT_PID ("manual-serve"). Unit-backed serves are recovered by the fresh
# child's systemd pass; survivors reported by _surviving_pre_update_serve_runtimes.
_SERVE_SKIP_REASON = (
    "no per-profile relaunch command reaches a serve/dashboard runtime; recovered by the fresh"
    " systemd unit pass when it owns a hermes-serve* unit, else left running for explicit"
    " operator restart"
)
# A launchd-owned backend (#116503): the fresh child has no per-label kickstart for serve/dashboard
# jobs, but the post-update dashboard cleanup pass kickstarts the loaded job — never a detached
# argv respawn, which would fight the job's own KeepAlive.
_LAUNCHD_SERVE_SKIP_REASON = (
    "launchd job owns this backend; the post-update dashboard cleanup kickstarts the job through"
    " launchd, never a detached argv respawn that would fight its KeepAlive"
)


def _gateway_recovery_partition(plan, *, skip_profiles: set[str] | None = None) -> tuple[dict[str, str], list[dict]]:
    """Partition pre-update runtimes into fresh-restart candidates and skips.

    Uses only the pre-checkout inventory: re-importing ``hermes_cli.gateway`` in the
    failing interpreter is what raises the original ``ImportError``. Returns
    ``(candidates, skipped)``: profile → supervisor for supervised gateways the fresh
    process may restart; every other inventoried runtime with an explicit reason so
    nothing vanishes silently. Skipped serve/dashboard is NOT unrecoverable: the fresh
    child's ``hermes-serve*`` systemd pass enumerates units from systemd; leftovers are
    caught by :func:`_surviving_pre_update_serve_runtimes`.
    """
    skip_profiles = skip_profiles or set()
    candidates: dict[str, str] = {}
    skipped: list[dict] = []
    with _best_effort('Could not prepare fresh gateway restart profiles: %s'):
        for runtime in getattr(plan, "runtimes", ()) or ():
            kind = getattr(runtime, "kind", None)
            profile = getattr(runtime, "profile", None)
            supervisor = getattr(runtime, "supervisor", None)
            if not isinstance(profile, str) or not profile:
                continue
            if kind == "gateway":
                if profile in skip_profiles:
                    continue
                if supervisor in _FRESH_RESTART_SUPERVISORS:
                    candidates.setdefault(profile, str(supervisor))
                    continue
                reason = _MANUAL_GATEWAY_SKIP_REASON
            elif kind in ("serve", "dashboard"):
                from hermes_cli.update_inventory import CLIENT_OWNED_SERVE_SUPERVISORS

                if supervisor in CLIENT_OWNED_SERVE_SUPERVISORS:
                    reason = _DESKTOP_SERVE_SKIP_REASON
                elif supervisor == "launchd":
                    reason = _LAUNCHD_SERVE_SKIP_REASON
                else:
                    reason = _SERVE_SKIP_REASON
            else:
                continue
            skipped.append({"profile": profile, "kind": str(kind), "supervisor": str(supervisor), "reason": reason})
    return candidates, skipped


def _warn_gateway_restart_phase_aborted(exc: BaseException, pids) -> None:
    """Print a recovery warning when the whole restart phase raised.

    Previously a blanket debug-logged ``except Exception`` erased every drain/restart
    line, so "Update complete!" exited 0 while the gateway kept serving pre-update
    modules and died on the next turn with an ImportError.

    Issue #78574: the gateway auto-restart phase was wrapped in a blanket ``except Exception`` that only
    logged at debug level, so an early failure (e.g. importing ``hermes_cli.gateway`` from the freshly
    pulled checkout) erased every drain/restart line from the update output.
    """
    print()
    print(f"⚠ Update incomplete — gateway auto-restart failed: {exc}")
    if pids:
        listed = ", ".join(str(pid) for pid in pids)
        print(f"  Gateway process(es) still running pre-update code: {listed}")
    else:
        print("  Any gateway still running is serving pre-update code")
        print("  (mixed sys.modules) against the updated checkout.")
    print("  Restart it manually, then verify:")
    print("    hermes gateway restart")
    print("    hermes gateway status")


def _drain_or_signal_gateway_for_update(
    pid: int, drain_budget: float, label: str, *, self_restart_pending: set | None = None,
) -> bool:
    """Three-way triage (shared by systemd and bare-process paths) for handing a
    running gateway over to new code. Returns True when signalled/stopped.

    1. Gateway is an ancestor of this process (auto-update cron inside the gateway
       tree): waiting is circular (gateway waits on in-flight work → cron session
       waits on update → update waits on gateway) and the 1800s force-drain cap burns.
       So fire-and-forget: signal restart and return; it completes once THIS process exits.
       The pid lands in ``self_restart_pending`` so the fleet matrix can tell "restart
       deferred until the updater exits" from "restart never happened" (#119597): the
       ancestor is still serving the old code when the matrix runs, by construction.
    2. Event loop provably wedged: SIGUSR1 can never drain it; bounded SIGTERM→SIGKILL.
    3. Live out-of-tree gateway: graceful SIGUSR1 drain up to ``drain_budget``.

    The wedged-loop probe cannot break it: the cron session posts activity every ~180s (process-tool poll
    return), so it is "actively waiting forever" and never marked wedged — the gateway burns the full
    force-drain cap (1800s) before killing its own updater's session. See #86684.
    """
    from hermes_cli.gateway import (
        GATEWAY_LOOP_WEDGED, _escalate_wedged_gateway, _graceful_restart_via_sigusr1,
        _is_pid_ancestor_of_current_process, _request_gateway_self_restart, probe_gateway_loop_liveness,
    )
    if _is_pid_ancestor_of_current_process(pid):
        print(
            f"  → {label}: update is running inside this gateway's "
            "process tree — signalling restart and letting the gateway "
            "drain itself (avoids the cron-update deadlock, #100179)"
        )
        accepted = _request_gateway_self_restart(pid)
        if accepted and self_restart_pending is not None:
            self_restart_pending.add(pid)
        return accepted
    if probe_gateway_loop_liveness(pid) == GATEWAY_LOOP_WEDGED:
        print(f"  ⚠ {label}: gateway event loop is unresponsive — skipping drain, forcing a bounded stop...")
        _escalate_wedged_gateway(pid)
        return True
    print(f"  → {label}: draining (up to {int(drain_budget)}s)...")
    from hermes_cli.update_cmd_drain_report import drain_progress_reporter
    return _graceful_restart_via_sigusr1(
        pid, drain_timeout=drain_budget,
        on_progress=drain_progress_reporter(_gateway_home_for_pid(pid), budget_s=drain_budget))


def _gateway_home_for_pid(pid: int):
    """HERMES_HOME of the gateway ``pid`` per the fleet inventory, else None (own profile's file)."""
    with suppress(Exception):
        from hermes_cli.update_receipt import _profile_homes
        from gateway.status import read_runtime_status
        for _profile, home in _profile_homes():
            record = read_runtime_status(home / "gateway_state.json") or {}
            if record.get("pid") == pid:
                return home
    return None


def _sudo_noninteractive_ok(targeted_probe: list) -> bool:
    """True when this user can elevate without a prompt.

    ``sudo -n true`` first; a refusal is inconclusive because a NOPASSWD sudoers entry scoped
    to one command (the hardened shape) rejects the blanket probe, so fall back to running
    ``sudo -n <targeted_probe>`` — callers pass a non-destructive stand-in for the argv they
    are about to elevate.
    """
    try:
        if subprocess.run(["sudo", "-n", "true"], capture_output=True, timeout=5).returncode == 0:
            return True
        # Blanket sudo refused — a targeted NOPASSWD sudoers entry may still work.
        return subprocess.run(["sudo", "-n", *targeted_probe], capture_output=True, timeout=5).returncode == 0
    except (OSError, subprocess.TimeoutExpired):
        return False


def _resolve_manage_cmd(cache: dict, scope_: str, scope_cmd_: list, svc_name_: str):
    """Resolve the command prefix for manage-units verbs (None ⇒ no privilege path).

    Manage-units verbs on a *system* service trigger a polkit prompt for non-root
    users, which flashes and dies inside our captured 10-15s subprocess. Root → plain
    systemctl; else ``sudo -n`` blanket probe, then a targeted ``reset-failed`` probe
    so a least-privilege sudoers entry scoped to hermes-gateway* qualifies (idempotent
    no-op we run before every privileged restart anyway). On None the caller must SKIP
    the restart (without draining first!). ``--no-ask-password`` prevents polkit hangs.
    """
    if scope_ in cache:
        return cache[scope_]
    cmd = scope_cmd_ + ["--no-ask-password"]
    if _needs_sudo(scope_):
        sudo_cmd = ["sudo", "-n"] + cmd
        cmd = sudo_cmd if _sudo_noninteractive_ok(cmd + ["reset-failed", svc_name_]) else None
    cache[scope_] = cmd
    return cmd


def _repair_unit_without_fatal_exit_park(svc_name: str, scope: str) -> None:
    """A unit whose restart policy predates ``RestartPreventExitStatus=78`` crash-loops on the PERMANENT
    exit: a ``Restart=on-failure`` system unit restarted ~180x on a host-attach refusal while the
    regenerated user units parked (#118282). The gateway rewrites its USER unit at boot; a SYSTEM unit
    lives in /etc, so rewrite it here when we are root, else name the repair."""
    from hermes_cli.gateway import (
        _SYSTEM_UNIT_DIR, GATEWAY_FATAL_CONFIG_EXIT_CODE, get_service_name,
        refresh_systemd_unit_if_needed, user_systemd_unit_dir,
    )
    system = scope == "system"
    unit_path = (_SYSTEM_UNIT_DIR if system else user_systemd_unit_dir()) / f"{svc_name}.service"
    try:
        parked = re.search(rf"^RestartPreventExitStatus=.*\b{GATEWAY_FATAL_CONFIG_EXIT_CODE}\b", unit_path.read_text(encoding="utf-8-sig"), re.M)
    except OSError:
        return
    if parked:
        return
    if system and not _needs_sudo(scope) and svc_name == get_service_name():
        # The refresh adopts the unit's HERMES_HOME into os.environ (sudo strips it); the rest of the
        # update keeps running for the invoking profile.
        launch_home = os.environ.get("HERMES_HOME")
        try:
            refresh_systemd_unit_if_needed(system=True)
        finally:
            if launch_home is None:
                os.environ.pop("HERMES_HOME", None)
            else:
                os.environ["HERMES_HOME"] = launch_home
        return
    print(
        f"  ⚠ {svc_name} lacks RestartPreventExitStatus={GATEWAY_FATAL_CONFIG_EXIT_CODE}: a permanent refusal "
        f"(exit {GATEWAY_FATAL_CONFIG_EXIT_CODE}) would crash-loop it instead of parking.\n"
        f"    Repair: {'sudo ' if system else ''}hermes gateway install{' --system' if system else ''}"
    )


def _restart_one_systemd_gateway_unit(
    svc_name: str, *, scope: str, scope_cmd: list, drain_budget: float, _manage_cmd_cache: dict,
    restarted_services: list, failed_or_stale_units: list, self_restart_pending: set | None = None,
) -> None:
    """Restart one active systemd gateway/serve unit: graceful SIGUSR1 drain, then forced restart.

    Appends settled names to ``restarted_services`` and failures to ``failed_or_stale_units``.
    """
    check = _systemctl(scope_cmd + ["is-active", svc_name], timeout=5)
    if check.stdout.strip() != "active":
        return
    if not _systemd_unit_owned_by_update(scope_cmd, svc_name):
        return
    _repair_unit_without_fatal_exit_park(svc_name, scope)

    # None ⇒ no non-interactive privilege path; avoid manage-units verbs
    # entirely or polkit prompts inside the captured subprocess.
    _manage_cmd = _resolve_manage_cmd(_manage_cmd_cache, scope, scope_cmd, svc_name)

    # Graceful SIGUSR1 first so in-flight runs drain: handler → request_restart(via_service=True)
    # → drain → exit, Restart=always respawns. hermes-serve has no handler → blunt restart below.
    _main_pid = 0
    if _service_unit_supports_graceful_sigusr1_restart(svc_name):
        try:
            _show = _systemctl(scope_cmd + ["show", svc_name, "--property=MainPID", "--value"], timeout=5)
            _main_pid = int((_show.stdout or "").strip() or 0)
        except (ValueError, subprocess.TimeoutExpired, FileNotFoundError):
            _main_pid = 0

    # Three-way triage (ancestor / wedged / graceful drain).
    _graceful_ok = _main_pid > 0 and _drain_or_signal_gateway_for_update(
        _main_pid, drain_budget, svc_name, self_restart_pending=self_restart_pending)

    if _graceful_ok:
        # ``Restart=always`` respawns only after RestartSec (60s in our unit; dead time for a
        # voluntary restart). ``reset-failed`` + ``start`` skips it (~1-3s); if RestartSec already
        # elapsed, ``start`` is a no-op and we fall through to the poll. Needs manage-units
        # privileges; without them auto-restart still fires after RestartSec.
        if _manage_cmd is not None:
            _systemctl(_manage_cmd + ["reset-failed", svc_name], timeout=10)
            _systemctl(
                _manage_cmd + ["start", svc_name],
                timeout=_systemd_restart_timeout(scope_cmd, svc_name, start_only=True),
            )
            if _wait_for_service_active(scope_cmd, svc_name, timeout=10.0):
                restarted_services.append(svc_name)
                return
        # Passive poll: auto-restart fires after RestartSec regardless of
        # privileges — primary when _manage_cmd is None, fallback otherwise.
        _restart_sec = _service_restart_sec(scope_cmd, svc_name, default=0.0)
        if _manage_cmd is None and _restart_sec > 5.0:
            print(
                f"  → {svc_name}: waiting for systemd "
                f"auto-restart (~{int(_restart_sec)}s; "
                "no root for an immediate restart)..."
            )
        if _wait_for_service_active(scope_cmd, svc_name, timeout=max(10.0, _restart_sec + 10.0)):
            restarted_services.append(svc_name)
            return
        # Exited but not respawned (older unit without Restart=on-failure /
        # RestartForceExitStatus=75); fall through to forced restart.
        print(f"  ⚠ {svc_name} drained but didn't relaunch — forcing restart")

    # Forcing needs manage-units privileges; without a non-interactive path
    # polkit would prompt inside the captured subprocess — skip, instruct.
    if _manage_cmd is None:
        failed_or_stale_units.append(svc_name)
        print(
            f"  ⚠ {svc_name} is a system service and restarting it needs root.\n"
            f"    Restart it manually to load the new version:\n"
            f"      sudo systemctl restart {svc_name}\n"
            f"    To let `hermes update` restart it automatically, allow\n"
            f"    passwordless sudo for systemctl, or run updates with sudo."
        )
        return

    # Blunt restart — only when the graceful path failed (no SIGUSR1 wiring, drain over
    # budget, restart-policy mismatch). Mirrors `hermes gateway restart` (`systemd_restart()`).
    restart = _systemctl_reset_and_restart(_manage_cmd, svc_name, scope_cmd=scope_cmd)
    if restart.returncode != 0:
        failed_or_stale_units.append(svc_name)
        print(f"  ⚠ Failed to restart {svc_name}: {restart.stderr.strip()}")
        return
    # restart returns 0 even if the new process crashes at once — verify.
    if _wait_for_service_active(scope_cmd, svc_name, timeout=10.0):
        restarted_services.append(svc_name)
        return
    # Retry once — transient startup failures (stale module cache,
    # import race) often clear; reset-failed so the retry isn't blocked.
    print(f"  ⚠ {svc_name} died after restart, retrying...")
    _systemctl_reset_and_restart(_manage_cmd, svc_name, scope_cmd=scope_cmd)
    if _wait_for_service_active(scope_cmd, svc_name, timeout=10.0):
        restarted_services.append(svc_name)
        print(f"  ✓ {svc_name} recovered on retry")
        return
    failed_or_stale_units.append(svc_name)
    _scope_flag = "--user " if scope == "user" else ""
    _sudo_hint = "sudo " if scope == "system" else ""
    print(
        f"  ✗ {svc_name} failed to stay running after restart.\n"
        f"    Check logs: {_sudo_hint}journalctl {_scope_flag}-u {svc_name} --since '2 min ago'\n"
        f"    Recover manually:\n"
        f"      {_sudo_hint}systemctl {_scope_flag}reset-failed {svc_name}\n"
        f"      {_sudo_hint}systemctl {_scope_flag}restart {svc_name}"
    )


def _restart_systemd_gateway_units(
    restarted_services, failed_or_stale_units, restarted_scoped_units, drain_budget, self_restart_pending=None,
):
    """Restart every active hermes-gateway*/hermes-serve* systemd unit (user + system).

    Settled units → ``restarted_services`` (bare) and ``restarted_scoped_units``
    (``scope/name``); failures → ``failed_or_stale_units``. Per-unit timeouts isolated.
    """
    from hermes_cli.gateway import supports_systemd_services, _ensure_user_systemd_env
    if not supports_systemd_services():
        return
    _manage_cmd_cache: dict = {}
    with suppress(Exception):
        _ensure_user_systemd_env()

    def _on_list_timeout(scope: str, exc: subprocess.TimeoutExpired) -> None:
        # Discovery timeout — skip this scope, keep the other.
        print(
            f"  ⚠ systemctl timed out listing {scope}-scope "
            f"gateway units ({exc.cmd if exc.cmd else 'unknown command'}). "
            f"Check the gateway with: hermes gateway status"
        )

    def _on_unit_timeout(svc_name: str, exc: subprocess.TimeoutExpired) -> None:
        # Isolate to this unit; a scope-wide handler used to abort every
        # later gateway and leave the fleet on mixed code.
        failed_or_stale_units.append(svc_name)
        print(
            # See #68523.
            f"  ⚠ systemctl timed out restarting {svc_name} "
            f"({exc.cmd if exc.cmd else 'unknown command'}); "
            f"continuing with remaining gateways"
        )

    # Enumerate every scope first: leftover per-profile units (``hermes-gateway-<profile>``)
    # all point at the SAME live MainPID on a multiplexed host, and restarting each in turn
    # is an N-fold outage from one update. Units sharing a live PID collapse to one restart;
    # the others are named as legacy units to migrate, never silently dropped.
    from hermes_cli.update_host_obligation import collapse_units_to_host_processes
    targets: dict[str, tuple[str, list, str]] = {}  # "<scope>/<unit>" -> (scope, scope_cmd, unit)
    for scope, scope_cmd, result in _systemd_gateway_unit_listings(_on_list_timeout):
        _for_each_systemd_gateway_unit(
            result.stdout,
            process_unit=lambda svc_name, _scope=scope, _cmd=scope_cmd: targets.setdefault(
                f"{_scope}/{svc_name}", (_scope, _cmd, svc_name)),
            on_unit_timeout=_on_unit_timeout,
        )
    keys = list(targets)
    covered: dict[str, str] = {}
    if len(keys) > 1:
        # Only worth a `systemctl show` round when several units could be one process.
        keys, covered = collapse_units_to_host_processes(
            keys, lambda key: _unit_main_pid(targets[key][1], targets[key][2]))
    for unit_key, owner_key in covered.items():
        print(
            f"  • {unit_key} is a legacy per-profile unit sharing one host gateway process with "
            f"{owner_key}; restarting it again would restart that process twice. Fold the units "
            "together with: hermes gateway migrate"
        )

    for key in keys:
        scope, scope_cmd, svc_name = targets[key]
        # Scope-qualify before the next unit; ``finally`` so a mid-pass abort keeps settled units.
        _scope_mark = len(restarted_services)
        try:
            _restart_one_systemd_gateway_unit(
                svc_name,
                scope=scope,
                scope_cmd=scope_cmd,
                drain_budget=drain_budget,
                _manage_cmd_cache=_manage_cmd_cache,
                restarted_services=restarted_services,
                failed_or_stale_units=failed_or_stale_units,
                self_restart_pending=self_restart_pending,
            )
        except subprocess.TimeoutExpired as exc:
            _on_unit_timeout(svc_name, exc)
        finally:
            restarted_scoped_units.update(f"{scope}/{name}" for name in restarted_services[_scope_mark:])


def _unit_main_pid(scope_cmd: list, svc_name: str) -> int:
    """Live ``MainPID`` of a unit; ``0`` when inactive, unprivileged or unreadable.

    Property reads need no manage-units privileges, and an unreadable PID is never collapsed:
    identity that cannot be proved keeps its own restart.
    """
    try:
        result = _systemctl(list(scope_cmd) + ["show", svc_name, "--property=MainPID", "--value"], timeout=10)
    except (OSError, subprocess.TimeoutExpired):
        return 0
    if getattr(result, "returncode", 1) != 0:
        return 0
    try:
        return int((getattr(result, "stdout", "") or "").strip() or 0)
    except ValueError:
        return 0


@dataclass
class _GatewayRestartOutcome:
    """Restart-phase bookkeeping. ``restarted_services`` keeps bare unit names (fleet
    probe, receipt, summary read it); ``incomplete`` ⇒ a gateway may still be stale."""

    incomplete: bool
    phase_errors: list
    pre_restart_gateway_pids: "list | None"
    restarted_services: list
    failed_or_stale_units: list
    relaunched_profiles: list
    externally_supervised_profiles: list
    killed_pids: set
    #: Gateways stopped with NO successor (no profile mapping / relaunch could not be armed);
    #: the summary tells the user to restart them by hand, so the fleet probe must not expect
    #: a row for them.
    stopped_unmapped_pids: set = field(default_factory=set)
    #: ``scope/name`` of every settled systemd unit; the fleet probe stops waiting for a state stamp
    #: once none of them is active or activating any more (the successor died, nothing will publish).
    restarted_scoped_units: set = field(default_factory=set)
    #: Gateways that are ANCESTORS of this updater and accepted a self-restart request
    #: (``_drain_or_signal_gateway_for_update`` branch 1): they restart only after this process
    #: exits, so the fleet matrix renders them as pending instead of STALE (#119597).
    self_restart_pending_pids: set = field(default_factory=set)

    def fleet_probe_signals(self) -> tuple:
        """``(pre_restart_pids, killed_pids)`` with the unmapped stops removed — the signals that
        legitimately predict a fleet-matrix row."""
        pre = self.pre_restart_gateway_pids
        if pre is not None:
            pre = [pid for pid in pre if pid not in self.stopped_unmapped_pids]
        return pre, self.killed_pids - self.stopped_unmapped_pids

    def record_receipt(self, **extra) -> None:
        """Best-effort ``record_gateway_restart`` from the current bookkeeping."""
        with suppress(Exception):
            from hermes_cli.update_receipt import record_gateway_restart
            record_gateway_restart(
                restarted_services=self.restarted_services, relaunched_profiles=self.relaunched_profiles,
                externally_supervised_profiles=self.externally_supervised_profiles,
                killed_pids=sorted(self.killed_pids), failed_units=self.failed_or_stale_units,
                incomplete=self.incomplete, **extra,
            )


def _restart_manual_gateways(out: _GatewayRestartOutcome, _drain_budget) -> None:
    """Drain/stop every manual (non-service) gateway and print the restart summary.

    Mutates ``out`` in place; raises so the caller's abort recovery fires.
    """
    import signal as _signal
    from hermes_cli.gateway import (
        find_gateway_pids, find_profile_gateway_processes, _prepare_profile_gateway_update_restart, _get_service_pids,
        _wait_for_gateway_exit,
    )
    # Exclude just-restarted service PIDs so we don't kill what systemd/launchd spawned.
    service_pids = _get_service_pids(all_profiles=True)
    manual_pids = find_gateway_pids(exclude_pids=service_pids, all_profiles=True)
    profile_processes = {
        proc.pid: proc
        for proc in find_profile_gateway_processes(exclude_pids=service_pids)
        if proc.pid in manual_pids
    }
    # ``all_profiles`` is host-wide: a sibling install's gateway matches too. Only this update's
    # homes are stopped; the profile-mapped PIDs come from this install's own PID files (#93349).
    manual_pids = _scoped_manual_gateway_pids(manual_pids, keep=profile_processes)
    # Profile gateways we couldn't arm a relaunch for must NOT keep running stale:
    # the unmapped sweep below stops them and lists them under "Restart manually".
    # These must NOT be left running: their modules are the pre-update ones and every lazy import from here
    # on mixes versions against the new code on disk (#88654). Handing them to the unmapped sweep below
    # stops them and surfaces them in the "Stopped N manual gateway process(es) / Restart manually" summary,
    # which is the contract already used for gateways with no profile mapping.
    unrestartable_pids = set()
    for pid, proc in profile_processes.items():
        restart_mode = _prepare_profile_gateway_update_restart(proc.profile, pid)
        if restart_mode is None:
            # A bare ``continue`` here left it serving stale modules with no signal.
            print(
                f"  ⚠ {proc.profile}: could not arm an automatic "
                f"gateway restart for PID {pid} — stopping it instead "
                "so it cannot keep running pre-update code"
            )
            unrestartable_pids.add(pid)
            continue
        # SIGUSR1 drain first, SIGTERM fallback if unsupported/over budget — the watcher
        # relaunches either way. The helper announces its choice first because a silent
        # full-budget wait reads as a hung update.
        if not _drain_or_signal_gateway_for_update(
                pid, _drain_budget, proc.profile, self_restart_pending=out.self_restart_pending_pids):
            with suppress(ProcessLookupError, PermissionError):
                os.kill(pid, _signal.SIGTERM)
        # Wait ≤5s for exit: Telegram keeps the old getUpdates session ~30s; a new gateway
        # inside that window gets a 409 (_handle_polling_conflict retries, but a brief
        # wait avoids it on fast machines).
        _wait_for_gateway_exit(timeout=5.0, force_after=None)
        out.killed_pids.add(pid)
        if restart_mode == "external-supervisor":
            out.externally_supervised_profiles.append(proc.profile)
        else:
            out.relaunched_profiles.append(proc.profile)

    for pid in manual_pids:
        if pid in profile_processes and pid not in unrestartable_pids:
            continue
        with suppress(ProcessLookupError, PermissionError):
            os.kill(pid, _signal.SIGTERM)
            out.killed_pids.add(pid)
            out.stopped_unmapped_pids.add(pid)

    if out.restarted_services or out.killed_pids:
        print()
        for svc in out.restarted_services:
            print(f"  ✓ Restarted {svc}")
        if out.relaunched_profiles:
            print(f"  ✓ Restarting manual gateway profile(s): {', '.join(out.relaunched_profiles)}")
        if out.externally_supervised_profiles:
            names = ", ".join(out.externally_supervised_profiles)
            print(f"  ✓ Handed gateway profile(s) back to their external supervisor: {names}")
        unmapped_count = (len(out.killed_pids) - len(out.relaunched_profiles) - len(out.externally_supervised_profiles))
        if unmapped_count:
            print(f"  → Stopped {unmapped_count} manual gateway process(es)")
            print("    Restart manually: hermes gateway run")
            if unmapped_count > 1:
                print("    (or: hermes -p <profile> gateway run  for each profile)")


def _force_kill_stuck_gateways(killed_pids) -> None:
    """Survivor sweep: gateways ignoring SIGTERM (stuck drain, blocked I/O, zombie) never
    exit, so the watcher never respawns and ImportErrors persist. Give graceful paths a
    moment, then SIGKILL remaining pre-update PIDs."""
    with _best_effort('Post-restart survivor sweep failed: %s'):
        from hermes_cli.gateway import find_gateway_pids, _get_service_pids
        # --- Post-restart survivor sweep ----------------------------- Issue #17648: some gateways ignore
        # SIGTERM (stuck drain, blocked I/O, PID dead but zombie). The detached profile watchers wait 120s
        # for the old PID to exit — if it never does, no respawn happens and the user keeps hitting
        # ImportError against a stale sys.modules.
        _time.sleep(3.0)
        _surviving = find_gateway_pids(exclude_pids=_get_service_pids(all_profiles=True), all_profiles=True)
        # Only PIDs we already tried to kill; newer ones are left alone.
        _stuck = [pid for pid in _surviving if pid in killed_pids]
        if _stuck:
            print()
            print(f"  ⚠ {len(_stuck)} gateway process(es) ignored SIGTERM — force-killing")
            from gateway.status import get_process_start_time, terminate_pid
            for pid in _stuck:
                with suppress(ProcessLookupError, PermissionError, OSError):
                    # taskkill /T /F on Windows (no SIGKILL there), SIGKILL on POSIX.
                    terminate_pid(pid, force=True, expected_start_time=get_process_start_time(pid))
            # Let the OS reap so watchers see the exit and respawn.
            _time.sleep(1.5)


def _recover_after_restart_phase_abort(
    e, _pre_update_plan, out: _GatewayRestartOutcome, *, gateway_mode, restarted_scoped_units
) -> None:
    """Phase-abort recovery: fresh-child restart + fail-closed verdict; updates ``out`` in place."""
    from hermes_cli.update_abort_recovery import _owed_stale_serve_rows
    from hermes_cli.update_cmd import (
        _abort_recovery_is_complete, _recover_gateway_restart_after_abort, _surviving_pre_update_serve_runtimes,
        _warn_stale_serve_runtimes, _write_gateway_update_exit_code,
    )
    logger.debug("Gateway restart during update failed: %s", e)
    out.phase_errors.append(str(e))
    # Restart output never printed: assume stale unless provably no gateway runs.
    # Empty ``_surviving`` proves safety only if nothing ran beforehand; a gone
    # pre-restart gateway was stopped without verified replacement → fail closed.
    # An exception escaping the whole phase means the drain/restart output the user relies on never printed.
    # Don't let that pass for a clean update: surface it and treat the fleet as stale unless we can
    # positively prove no gateway is running (#78574). A positive-empty ``_surviving`` is only
    # proof-of-safety when nothing was running before we touched anything. If a gateway was discovered
    # pre-restart and none survive now, it was stopped and its replacement was never verified — the same
    # fail-open contract this fix closes — so we must still fail closed on ``[]``.
    _surviving = _surviving_gateway_pids_after_failed_restart()
    _planned_gateway_profiles = {
        runtime.profile
        for runtime in getattr(_pre_update_plan, "runtimes", ()) or ()
        if getattr(runtime, "kind", None) == "gateway"
        and isinstance(getattr(runtime, "profile", None), str)
    }
    _already_restarted_profiles = set(out.relaunched_profiles) | set(out.externally_supervised_profiles)
    _already_restarted_profiles.update(
        profile
        for profile in _planned_gateway_profiles
        if any(_gateway_service_matches_profile(profile, service) for service in out.restarted_services)
    )
    _recovery_result = _recover_gateway_restart_after_abort(
        _pre_update_plan, gateway_mode=gateway_mode, skip_profiles=_already_restarted_profiles,
        skip_units=set(restarted_scoped_units),
    )
    _serve_units_failed = list((_recovery_result.get("serve_units") or {}).get("failed") or [])
    # Deliberately NOT merged into ``restarted_services`` (gateway vocabulary feeding the
    # fleet probe); serve coverage lives in the recovery result/receipt. A serve/dashboard
    # still the SAME pre-update process is live on old code (unreachable by `gateway
    # restart`): recovery may not claim success while one remains, and must never kill
    # one (manual/Desktop serves have no relaunch authority).
    _stale_runtime_rows = _surviving_pre_update_serve_runtimes(_pre_update_plan)
    _recovery_result["stale_runtimes"] = _stale_runtime_rows
    # Only systemd-VERIFIED outcomes claim coverage; a relaunch that merely exited 0
    # ("relaunch_attempted") was never observed and must not clear incomplete.
    _recovery_verified = set(_recovery_result.get("verified") or [])
    out.relaunched_profiles.extend(
        profile for profile in sorted(_recovery_verified) if profile not in out.relaunched_profiles
    )
    if _abort_recovery_is_complete(
        planned_gateway_profiles=_planned_gateway_profiles,
        covered_gateway_profiles=_already_restarted_profiles | _recovery_verified,
        recovery_result=_recovery_result,
        stale_runtime_rows=_stale_runtime_rows,
    ):
        # Fresh child is terminal; the fleet-version matrix stays the authoritative
        # read-back before success is declared. Desktop-owned survivors (the only rows
        # that can remain here) are named, not owed. See #111494.
        out.incomplete = False
        _warn_stale_serve_runtimes(_stale_runtime_rows)
    elif (
        _restart_phase_failure_is_incomplete(_surviving, out.pre_restart_gateway_pids)
        or _owed_stale_serve_rows(_stale_runtime_rows)
        or _serve_units_failed
    ):
        out.incomplete = True
        _warn_gateway_restart_phase_aborted(e, _surviving)
        _warn_stale_serve_runtimes(_stale_runtime_rows)
        if gateway_mode:
            _write_gateway_update_exit_code(False)
    out.record_receipt(phase_error=str(e), fresh_recovery=_recovery_result)


def _gateway_drain_budget() -> float:
    """Seconds a drain-first (SIGUSR1) restart may wait for a gateway to exit; 45s floor."""
    try:
        from hermes_cli.gateway import _get_restart_exit_wait_budget
        return max(float(_get_restart_exit_wait_budget()), 45.0)
    except Exception:
        return 45.0


def _restart_gateway_fleet_after_update(_pre_update_plan, gateway_mode: bool):
    """Restart every running gateway (systemd, launchd, manual) onto the pulled code.

    Never raises: a phase abort runs fresh-child recovery and fails closed unless
    every planned gateway is verifiably covered.
    """
    from hermes_cli.update_cmd import _m, _write_gateway_update_exit_code
    # All bookkeeping is declared before the try so abort recovery and fleet reconciliation
    # can read it even if the phase raises early. ``pre_restart_gateway_pids`` stays empty
    # until we are about to stop/drain, so an early exception has nothing to fail closed on,
    # while a failure after stopping a discovered gateway fails closed on an empty survivor probe.
    out = _GatewayRestartOutcome(
        incomplete=False, phase_errors=[], pre_restart_gateway_pids=[], restarted_services=[], failed_or_stale_units=[],
        relaunched_profiles=[], externally_supervised_profiles=[], killed_pids=set(),
    )
    # Scope-qualified twin (``user/hermes-serve`` vs ``system/hermes-serve`` are different
    # processes; abort recovery needs WHICH settled). Bare names stay in
    # ``restarted_services`` for the fleet probe, receipt and summary.
    # Snapshot of gateways running before we touch anything. Stays empty until we successfully import the
    # probe and are about to stop/drain — so an exception raised before we touch any gateway keeps this
    # empty (nothing to fail closed on), while a failure after we have stopped a discovered gateway lets the
    # handler fail closed on an empty survivor probe rather than reporting a clean update (#78574).
    # Declared outside the restart try/except below (and never reset to None) so it's always safe to read
    # afterwards even if that block raises before reaching its own restart bookkeeping — needed to forward
    # already-restarted units to ``_refresh_dashboard_after_update`` (review on #83595).
    restarted_scoped_units: set = set()

    try:
        # Every gateway helper the phase needs is imported up front so a broken gateway
        # module aborts into recovery BEFORE any unit is touched.
        from hermes_cli.gateway import (  # noqa: F401
            is_macos,
            find_gateway_pids,
            find_profile_gateway_processes,
            _prepare_profile_gateway_update_restart,
            _get_service_pids,
            _wait_for_gateway_exit,
        )
        # Drain budget covers ``restart_after_turn_timeout`` and stop()'s
        # ``restart_drain_timeout`` so a gateway waiting on a turn isn't hard-killed;
        # units without SIGUSR1 wiring just time out into ``systemctl restart``.
        _drain_budget = _gateway_drain_budget()

        # Snapshot before any stop/drain so an empty survivor probe reads as "stopped
        # and never came back", not "nothing was running"; None fails closed.
        try:
            out.pre_restart_gateway_pids = _scoped_manual_gateway_pids(find_gateway_pids(all_profiles=True), quiet=True)
        except Exception:
            out.pre_restart_gateway_pids = None

        _restart_systemd_gateway_units(
            out.restarted_services, out.failed_or_stale_units, restarted_scoped_units, _drain_budget,
            out.self_restart_pending_pids,
        )

        # macOS: EVERY ai.hermes.gateway* LaunchAgent (systemd parity).
        if is_macos():
            with suppress(FileNotFoundError, ImportError):
                _restart_macos_launchd_gateways(
                    out.restarted_services, out.failed_or_stale_units, _drain_budget,
                    self_restart_pending=out.self_restart_pending_pids,
                )

        _restart_manual_gateways(out, _drain_budget)

        if out.failed_or_stale_units:
            out.incomplete = True
            if gateway_mode:
                _write_gateway_update_exit_code(False)
        _warn_incomplete_gateway_fleet_restart(out.failed_or_stale_units)
        out.record_receipt()
        _force_kill_stuck_gateways(out.killed_pids)

    except Exception as e:
        _recover_after_restart_phase_abort(
            e, _pre_update_plan, out, gateway_mode=gateway_mode, restarted_scoped_units=restarted_scoped_units
        )

    out.restarted_scoped_units = set(restarted_scoped_units)
    if not out.incomplete:
        # Stamp the HOST obligation so every other profile's CLI knows this update's restart
        # already happened; without it each profile re-kills the one shared multiplexer.
        from hermes_cli.update_host_obligation import mark_host_restart_completed
        mark_host_restart_completed(_restart_identity_sha())
    return out


def _print_legacy_units_warning() -> None:
    """Legacy hermes.service fights hermes-gateway.service over the bot token; warn on
    every update until migrated."""
    from hermes_cli.gateway import (has_legacy_hermes_units, _find_legacy_hermes_units, supports_systemd_services)
    if not (supports_systemd_services() and has_legacy_hermes_units()):
        return
    print()
    print("⚠ Legacy Hermes gateway unit(s) detected:")
    for name, path, is_sys in _find_legacy_hermes_units():
        scope = "system" if is_sys else "user"
        print(f"    {path}  ({scope} scope)")
    print()
    print("  These pre-rename units (hermes.service) fight the current")
    print("  hermes-gateway.service for the bot token and cause SIGTERM")
    print("  flap loops. Remove them with:")
    print()
    print("    hermes gateway migrate-legacy")
    print()
    print("  (add `sudo` if any are in system scope)")


def _collect_fleet_snapshot(restart, rows_expected: bool) -> list:
    """Fleet version rows, polled over a bounded settle window when runtimes are expected.

    Gateways need time to rewrite gateway_state.json; Windows resumes DETACHED (~10s boot),
    so a single 2s sleep reported "no rows" on healthy resumes. A "down" row may be a
    detached replacement still booting: poll until none remain or the deadline passes.
    Pre-restart PIDs make a gateway stopped WITHOUT verified replacement a DOWN row (exit 1)
    instead of no row at all. An ``unknown`` row whose pid is NOT a pre-restart pid is a successor
    that has not published its code identity yet (a relaunched gateway can sit ~10s between process
    start and its first runtime-status write, #112634) — keep polling; at the deadline it is flagged
    ``identity_pending`` so the matrix does not call it a pre-stamping gateway.
    """
    from hermes_cli.update_receipt import collect_fleet_versions
    pending = getattr(restart, "self_restart_pending_pids", None) or None
    if not rows_expected:
        return collect_fleet_versions(
            pre_restart_pids=restart.pre_restart_gateway_pids, self_restart_pending=pending)
    pre_pids = restart.pre_restart_gateway_pids
    _fleet_deadline = _time.monotonic() + _FLEET_PROBE_SETTLE_TIMEOUT_SECONDS
    while True:
        _time.sleep(2.0)
        snapshot = collect_fleet_versions(pre_restart_pids=pre_pids, self_restart_pending=pending)
        unstamped = [row for row in snapshot if _fleet_row_identity_pending(row, pre_pids)]
        if snapshot and not unstamped and not any(row.get("state") == "down" for row in snapshot):
            return snapshot
        if _time.monotonic() >= _fleet_deadline or _restarted_units_gone(
                getattr(restart, "restarted_scoped_units", ())):
            for row in unstamped:
                row["identity_pending"] = True
            return snapshot


def _fleet_row_identity_pending(row: dict, pre_restart_pids) -> bool:
    """An ``unknown`` row with no sha from a pid that did not exist at update start: a relaunched
    gateway still booting, not a gateway that predates version stamping. A surviving pre-restart pid
    (or no pid snapshot at all) is settled as-is — waiting cannot change what it publishes."""
    if row.get("state") != "unknown" or row.get("code_sha"):
        return False
    if pre_restart_pids is None:
        return False
    return row.get("pid") not in {int(p) for p in pre_restart_pids if isinstance(p, int)}


def _restarted_units_gone(scoped_units) -> bool:
    """True when every restarted systemd unit is LOADED in its scope and neither active nor
    activating: the successor died, nothing will publish a state stamp, so the settle poll should fail
    closed now instead of at the deadline. Anything inconclusive keeps waiting: no units, systemctl
    missing/slow, or ``LoadState=not-found`` — a unit name asked in a scope that does not own it
    answers ``inactive`` exactly like a dead unit (#112466), so only a loaded unit can prove death."""
    if not scoped_units:
        return False
    scope_cmds = dict(_SYSTEMD_SCOPES)
    for scoped in scoped_units:
        scope, _, name = scoped.partition("/")
        try:
            stdout = _systemctl(scope_cmds[scope] + ["show", "-p", "LoadState,ActiveState", name], timeout=5).stdout
        except (KeyError, FileNotFoundError, subprocess.TimeoutExpired):
            return False
        props = dict(line.split("=", 1) for line in stdout.splitlines() if "=" in line)
        if props.get("LoadState") != "loaded":
            return False
        if props.get("ActiveState") in ("active", "activating", "reloading"):
            return False
    return True


def _verify_fleet_after_update(restart, *, _pre_update_plan, _windows_gateway_resume, update_complete):
    """Post-restart verification: legacy-unit warning, dashboard cleanup, stale serve
    probe, fleet version matrix, plan-vs-execution reconciliation, receipt finalize.

    Exits 1 (leaving ``fleet_restart_pending`` for the next catch-up) when any gateway
    may still be stale; otherwise clears the marker. A failed SQLite verdict also
    exits 1, without retaining a fulfilled fleet-restart obligation.
    """
    from hermes_cli.update_cmd import (
        _m, _surviving_pre_update_serve_runtimes, _warn_stale_serve_runtimes,
    )
    from hermes_cli.update_cmd_maint import _refresh_dashboard_after_update
    with _best_effort('Legacy unit check during update failed: %s'):
        _print_legacy_units_warning()

    # Restart a managed dashboard via systemd or stop stale manual ones (raw-killing
    # a systemd-owned PID reads as clean stop and leaves the Cloudflare origin dead).
    # Already-restarted units aren't redone.
    _refresh_dashboard_after_update(already_restarted_units=set(restart.restarted_services))

    # Success-path twin of the abort-recovery probe: the restart phase only touches
    # units, so a unit-less `hermes serve` keeps stale sys.modules. Runs AFTER
    # dashboard cleanup so a respawned manual dashboard isn't a survivor. Rows feed
    # reconciliation (survivor → exit 1); ``None`` = probe failed, stays fail-closed.
    # Check if any pre-update serve/dashboard runtimes survived on pre-update code generations (#100479).
    # This is the SUCCESS-path twin of the abort-recovery probe above: the restart phase only restarts
    # units, so an sshd-spawned `serve --isolated` or a manual `hermes serve` (no unit) is left running its
    # pre-update sys.modules graph — and its cron ticker keeps firing agent jobs that ImportError on every
    # symbol added in the pulled range. The rows also feed the plan-vs-execution reconciliation below, so a
    # survivor is escalated (exit 1) instead of merely printed.
    _stale_serve_rows: "list | None" = None
    with _best_effort('Failed to check for surviving serve runtimes: %s'):
        _stale_serve_rows = _surviving_pre_update_serve_runtimes(_pre_update_plan)
        if _stale_serve_rows:
            _warn_stale_serve_runtimes(_stale_serve_rows)

    print()
    print("Tip: You can now select a provider and model:")
    print("  hermes model              # Select provider and model")

    # Compare every live gateway's stamped code_sha against the fresh checkout
    # instead of assuming the restart phase worked.
    # Phase 1 (#91277): post-update fleet version verification.
    _fleet_snapshot: list = []
    with _best_effort('Fleet version verification failed: %s'):
        from hermes_cli.update_receipt import print_fleet_version_matrix
        # Cross-platform "rows expected" signal: (restarted_services or killed_pids)
        # never fires on Windows (pause/resume populates neither), so a healthy
        # resumed gateway yielded zero rows and exit 0.
        # See #93406.
        # A gateway stopped WITHOUT a successor ("Restart manually") publishes no row by design,
        # so it must not count as an expected one — otherwise an update whose only live gateways
        # were unmapped exits 1 with "no rows" after correctly stopping them.
        _pre_restart, _killed = restart.fleet_probe_signals()
        _fleet_rows_expected = _m()._fleet_probe_expected_runtimes(
            _pre_update_plan, _pre_restart, _windows_gateway_resume, restart.restarted_services, _killed,
        )
        _fleet_snapshot = _collect_fleet_snapshot(restart, _fleet_rows_expected)
        if print_fleet_version_matrix(_fleet_snapshot):
            restart.incomplete = True
            # A proven-stale survivor must not keep running (its ticker yields every tick and
            # nothing else restarts it, #117275): hand it to the drain-first restart path.
            from hermes_cli.update_cmd_stale_survivors import signal_stale_fleet_survivors
            signal_stale_fleet_survivors(_fleet_snapshot, restart, _gateway_drain_budget())
        elif not _fleet_snapshot and _fleet_rows_expected:
            # collect_fleet_versions() swallows every failure, so zero rows with
            # expected runtimes is indistinguishable from health — fail (partial, exit 1).
            print(
                # Fleet probe returned zero rows even though at least one gateway runtime was (or may have
                # been) live pre-update — POSIX restart bookkeeping, the pre-restart PID snapshot, the
                # pre-update plan inventory, or the Windows pause/resume token all count as that signal.
                # Every failure path inside collect_fleet_versions() is swallowed via logger.debug(), so an
                # empty list is indistinguishable from a healthy fleet in the current output. Treat it as
                # verification failure so the receipt records "partial" and the exit code is 1 (#93406).
                "\n⚠ Fleet version check returned no rows even though"
                " gateway runtimes were expected — verification incomplete."
            )
            restart.incomplete = True

    # Every runtime the PLAN saw must appear in restart bookkeeping; an
    # unaccounted one is a silent miss and escalates like a STALE/DOWN row.
    with _best_effort('Runtime-outcome reconciliation failed: %s'):
        # An unaccounted runtime is the silent-miss class (a platform branch re-discovered its own targets
        # and skipped one the inventory knew about) — escalate it exactly like a STALE/DOWN fleet row. See
        # #91277.
        if _pre_update_plan is not None and _pre_update_plan.runtimes:
            from hermes_cli.update_inventory import (match_runtime_outcomes, report_unaccounted_runtimes)
            _runtime_outcomes = match_runtime_outcomes(
                _pre_update_plan,
                restarted_services=restart.restarted_services,
                relaunched_profiles=restart.relaunched_profiles,
                externally_supervised_profiles=restart.externally_supervised_profiles,
                killed_pids=restart.killed_pids,
                failed_units=restart.failed_or_stale_units,
                # Serve/dashboard reconcile by incarnation liveness, not unit names.
                # See #100479.
                stale_serve_pids=(
                    {row.get("pid") for row in _stale_serve_rows}
                    if _stale_serve_rows is not None
                    else None
                ),
            )
            from dataclasses import asdict
            from hermes_cli.update_serve_obligations import defer_manual_serve

            for runtime, outcome in zip(_pre_update_plan.runtimes, _runtime_outcomes):
                if outcome["outcome"] == "unaccounted" and defer_manual_serve(asdict(runtime), require_alive=True):
                    outcome["outcome"] = "deferred"
            if report_unaccounted_runtimes(_runtime_outcomes):
                restart.incomplete = True
            with suppress(Exception):
                import hermes_cli.update_receipt as _ur
                _active = _ur._current.get()
                if _active is not None:
                    _active.data["runtime_outcomes"] = _runtime_outcomes

    with _best_effort('Update receipt finalize failed: %s'):
        from hermes_cli.update_receipt import finalize_update_receipt
        _receipt_path = finalize_update_receipt(
            "partial" if restart.incomplete or not update_complete else "success",
            fleet=_fleet_snapshot,
        )
        if _receipt_path is not None:
            logger.info("Update receipt written: %s", _receipt_path)

    if restart.incomplete:
        # Code updated but a gateway may still run stale modules: fail so automation
        # doesn't treat the fleet as healthy; leave the pending marker for catch-up.
        sys.exit(1)
    _clear_fleet_restart_pending_marker()
    if not update_complete:
        # Fleet caught up, but the independently checked SQLite runtime is unsafe.
        sys.exit(1)
    # Fleet is healthy on the new code: fold per-profile gateways into one multiplexer when nothing
    # blocks it (deterministic; never prompts), else print the blockers and the one-liner to run later.
    with _best_effort('Multiplex auto-migration after update failed: %s'):
        from hermes_cli.gateway_migrate import maybe_auto_migrate_after_update
        maybe_auto_migrate_after_update()


def _restart_phase_failure_is_incomplete(surviving, pre_restart_pids) -> bool:
    """Whether an escaped restart-phase exception must fail the update.

    Fail closed unless provably safe: ``surviving`` None (unprobeable) or non-empty →
    stale. ``[]`` proves safety ONLY if nothing ran beforehand; a pre-restart gateway
    (``pre_restart_pids`` non-empty or None) now gone was stopped unverified.

    * ``surviving is None`` — the survivor probe could not determine state (typically the freshly-pulled
    ``hermes_cli.gateway`` no longer imports, one of the ways the phase aborts). That is proof-of-safety
    ONLY when nothing was running before we touched anything. If a gateway was discovered pre-restart
    (``pre_restart_pids`` non-empty, or ``None`` meaning the pre-state could not be read), it was stopped
    without a verified replacement, so we still fail closed (#78574).
    """
    if surviving is None or surviving:
        return True
    return pre_restart_pids is None or bool(pre_restart_pids)


def _fleet_probe_expected_runtimes(
    pre_update_plan, pre_restart_pids, windows_resume_token, restarted_services, killed_pids,
) -> bool:
    """Whether the post-update fleet probe should have produced rows.

    ``collect_fleet_versions()`` swallows every failure and an empty matrix prints as
    healthy, so zero rows is only proof-of-safety when NOTHING says a gateway existed
    pre-update. Signals: ``restarted_services``/``killed_pids``; ``pre_restart_pids``
    non-empty or None (same contract as ``_restart_phase_failure_is_incomplete``); plan
    inventoried ≥1 ``kind == "gateway"`` runtime. ``windows_resume_token`` is deliberately EXCLUDED: it is
    pause/resume bookkeeping, not an inventory, and its entries don't map to probe rows
    (``unmapped`` Scheduled-Task gateways never publish gateway_state.json; a paused
    profile resumes DETACHED). Counting it made every Windows update that paused a
    gateway exit 1 after a long silent wait; a live pre-update Windows gateway is already
    covered by ``pre_restart_pids`` and the plan. The same condition gates the settle sleep.

    See #93406.
    See #78574.
    See #93406.
    """
    del windows_resume_token  # excluded on purpose — see docstring
    # See #93406.
    if restarted_services or killed_pids:
        return True
    if pre_restart_pids is None or pre_restart_pids:
        return True
    with suppress(Exception):
        # Gateway-kind only: serve/dashboard plan records never publish a gateway_state.json
        # row, so a dashboard-only plan cannot ground a rows-expected verdict (#97332).
        if pre_update_plan is not None and any(
            getattr(runtime, "kind", None) == "gateway" for runtime in pre_update_plan.runtimes
        ):
            return True
    return False


def _wait_for_service_active(scope_cmd_: list, svc_name_: str, timeout: float = 10.0) -> bool:
    """Poll ``systemctl is-active`` (0.5s) up to ``timeout``: the Stopped -> Started
    transition isn't instantaneous, so a one-shot check falsely reports down."""
    deadline = _time.monotonic() + max(timeout, 0.5)
    while True:
        with suppress(FileNotFoundError, subprocess.TimeoutExpired):
            _verify = _systemctl(scope_cmd_ + ["is-active", svc_name_], timeout=5)
            if _verify.stdout.strip() == "active":
                return True
        if _time.monotonic() >= deadline:
            return False
        _time.sleep(0.5)


_RESTART_SEC_UNITS = (("ms", 0.001), ("us", 0.000001), ("min", 60.0), ("s", 1.0))


def _service_restart_sec(scope_cmd_: list, svc_name_: str, default: float = 0.0) -> float:
    """Read the unit's ``RestartUSec`` in seconds. ``is-active`` pollers must wait
    >= RestartSec + slack or they give up *during* the cooldown and misreport."""
    try:
        _show = _systemctl(scope_cmd_ + ["show", svc_name_, "--property=RestartUSec", "--value"], timeout=5)
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return default
    raw = (_show.stdout or "").strip()
    # Values like "30s", "100ms", "1min 30s", "infinity"; on any miss return default.
    if not raw or raw == "infinity":
        return default
    total = 0.0
    matched = False
    for part in raw.split():
        for _suf, _mult in _RESTART_SEC_UNITS:
            if part.endswith(_suf):
                with suppress(ValueError):
                    total += float(part[: -len(_suf)]) * _mult
                    matched = True
                break
    return total if matched else default