"""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_host_obligation import PROFILE_MARKER_NAME
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 = PROFILE_MARKER_NAME

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

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


def _write_gateway_update_exit_code(ok: bool) -> None:
    from hermes_cli.update_cmd import get_hermes_home
    from hermes_cli.update_receipt import committed_success
    from pm.filesystem import durable_write_bytes

    # A run whose receipt already closed as a success is never reported failed (C3, review
    # regression 3): the command boundary's catch-alls call this with False.
    ok = ok or committed_success()
    path = get_hermes_home() / ".update_exit_code"
    with suppress(OSError):
        # Atomic: the gateway watcher polls this file and reads an empty one as exit 1.
        durable_write_bytes(path, b"0" if ok else b"1")


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_fleet_restart_pending_marker(*, expected_sha: str = "", runtimes: list[dict] | None = None, owner: str = "") -> bool:
    """Arm the HOST pull→restart obligation. Never raises; False when no store every profile reads took it.

    An unwritable host state dir falls back to this profile's per-home marker, which settles the
    debt only on an install proven to have no other profile; otherwise the arm returns False and
    the commit point refuses to move (``update_host_obligation.arm_host_obligation``, review S2).
    """
    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 True
    from hermes_cli.update_cmd import _m
    from hermes_cli.update_host_obligation import arm_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 True
    return arm_host_obligation(_fleet_restart_pending_marker_path(), expected_sha=expected_sha, runtimes=runtimes,
                               profile=_current_profile_name(), owner=owner)


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.
# The inventory writer classifies a serve/dashboard row as launchd, desktop, desktop-ssh, systemd (a unit's
# MainPID, #141021) or manual-serve (``update_inventory._collect_ledger_runtimes``); ``windows-service`` and
# ``service`` mirror ``_SUPERVISED_SERVE_BACKENDS`` for parity only — they 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, None) for entry in receipt.get("carried_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


# A gateway the restart phase stopped without knowing its profile: no live row can be matched to it,
# so only a live, current gateway of this checkout settles it, never an empty host (""=no profile).
_UNMAPPED_GATEWAY = ("gateway", "")


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
        if runtime.get("kind") == "gateway" and runtime.get("stopped_unmapped"):
            owed.add(_UNMAPPED_GATEWAY)
            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 — and so does an inventory-less record armed with no
    SHA at all (a no-op update whose head capture failed, #125952): with no owed set and no SHA,
    the checkout is the only code it can be held to. 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 records an owed restart 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 owed is not None and not expected_sha:
        return False  # an inventoried obligation without its SHA can never be proven
    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 or (not expected_sha and all(row_is_external(row) for row in fleet)):
        if owed is not None or not expected_sha:
            # Absence cannot prove recovery of the recorded inventory / unnamed code; a fleet
            # whose every row serves ANOTHER checkout root is absence too, not evidence.
            return False
        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 - {_UNMAPPED_GATEWAY} <= covered:
        return False  # A gateway this marker owns is absent (down) or unidentifiable.
    if owed and _UNMAPPED_GATEWAY in owed and not (_fleet_covered_gateways(
            [row for row in fleet if not row_is_external(row)]) or set()) - owed:
        # An unmapped stopped gateway has no identity to match, so only a local gateway that no
        # mapped owed one explains can be its successor; another checkout's gateway never is.
        return False
    _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, check=False)


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.
    from hermes_cli.update_cmd_fleet_unit_budget import systemd_restart_timeout
    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_serve_or_dashboard_unit(unit: str) -> bool:
    """``hermes-serve*``/``hermes-dashboard*`` (``.service``) by the same exact-or-hyphenated shape."""
    return (
        unit == "hermes-serve.service"
        or unit.startswith("hermes-serve-")
        # #125297: ``hermes-dashboard*`` units are systemd-supervised dashboard backends — the
        # same fleet this pass restarts. Leaving them out meant a successful update reported
        # the dashboard ``deferred`` (still on pre-update code) while nothing ever restarted it.
        or unit == "hermes-dashboard.service"
        or unit.startswith("hermes-dashboard-")
    )


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 _is_serve_or_dashboard_unit(unit)
    )


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, check=False)
        if listing.returncode != 0:
            failed_or_stale_units.append("launchd (listing failed)")
            return
    from hermes_cli.update_cmd_posix_pause import already_restarted
    current_label = get_launchd_label()
    resumed = already_restarted()["labels"]  # paused for the update, restarted on the new code
    _restarted, _failed = ([], []) if current_label in resumed else _restart_launchd_gateway_after_update(
        supervision_verify=True, self_restart_pending=self_restart_pending)
    restarted_services.extend(_restarted)
    failed_or_stale_units.extend(_failed)

    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 or label in resumed:
            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 {drain_budget:.0f}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 unit's MainPID row is classified "systemd" (#141021),
# and "manual-serve" still covers a process inside some unit's cgroup it is not MainPID of, or
# unreadable ownership. 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 a hermes-serve*/hermes-dashboard* unit owns it, 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 {drain_budget:.0f}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, check=False).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, check=False).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.MULTILINE)
    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:
        from hermes_cli.update_cmd_fleet_unit_budget import service_restart_sec, systemd_restart_timeout
        # ``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"
        )

    from hermes_cli.update_cmd_posix_pause import already_restarted
    for key in keys:
        scope, scope_cmd, svc_name = targets[key]
        if key in already_restarted()["units"]:  # paused for the update, restarted on the new code
            restarted_scoped_units.add(key)
            continue
        # 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.
    from hermes_cli.update_cmd_posix_pause import already_restarted
    # Plus the gateways this update paused and already restarted on the new code.
    service_pids = _get_service_pids(all_profiles=True) | already_restarted()["pids"]
    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,
    )
    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
    ):
        # Owed restart (C3): verification records the follow-up; the /update marker stays committed.
        out.incomplete = True
        _warn_gateway_restart_phase_aborted(e, _surviving)
        _warn_stale_serve_runtimes(_stale_runtime_rows)
    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
    # 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 (
            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:
            # Owed restart (C3): verification records the follow-up; the /update marker stays committed.
            out.incomplete = True
        _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 _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)