Follow-up to the salvaged restart-wait commits: - A drain or cron timeout of .inf now means "wait indefinitely" instead of an OverflowError from the integer stop envelope, which crashed `hermes gateway restart` and made `hermes update` silently fall back to its 45s floor. The fleet "draining (up to Ns)" lines and the drain progress report format the budget instead of int()-ing it, so an unbounded wait no longer crashes them either (it did on main too). - cron_drain_timeout is required: a 0.0 default meant "cron opted out", the under-budget this fix exists to remove. - Docstrings describe what the budget actually covers (PID exit, not replacement startup). - Tests assert the observer outlasts after-turn + the supervisor stop envelope and that configured cron reaches the CLI wait, instead of re-deriving the formula; the negative wording assertion on the pending footer is dropped (change-detector).
2122 lines
107 KiB
Python
2122 lines
107 KiB
Python
"""Gateway fleet restart + post-update verification for ``hermes update``.
|
|
|
|
Split out of ``hermes_cli/update_cmd.py``; every name is re-imported there so
|
|
``hermes_cli.update_cmd.<name>`` keeps resolving/monkeypatching. Origin helpers are
|
|
imported lazily inside each function (no import cycle; test patches stay effective).
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from contextlib import suppress
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time as _time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from hermes_cli.update_cmd_common import _best_effort
|
|
from hermes_cli.update_inventory import _gateway_service_matches_profile
|
|
|
|
# Log-record parity with the origin module.
|
|
logger = logging.getLogger("hermes_cli.update_cmd")
|
|
|
|
# Under HERMES_HOME (not next to the venv): records the fleet-restart obligation
|
|
# after a pull advanced HEAD; cleared only when the restart completes or nothing ran.
|
|
# The existing ``.update-incomplete`` / ``.lazy-refresh-incomplete`` markers gate dependency/venv repair;
|
|
# this one is the fleet-restart obligation after a git pull that advanced HEAD (#95294).
|
|
_FLEET_RESTART_PENDING_NAME = "fleet_restart_pending"
|
|
|
|
_FRESH_RESTART_SUPERVISORS = frozenset({"systemd", "launchd", "service", "s6"})
|
|
|
|
# A supervisor can report a restarted unit active before the gateway finishes its
|
|
# bootstrap and publishes ``gateway_state.json``. Keep the readiness poll bounded,
|
|
# but allow the default systemd startup budget plus status-publication slack.
|
|
_FLEET_PROBE_SETTLE_TIMEOUT_SECONDS = 120.0
|
|
|
|
_SYSTEMD_SCOPES = (("user", ["systemctl", "--user"]), ("system", ["systemctl"]))
|
|
_LIST_GATEWAY_UNITS = ["list-units", "hermes-gateway*", "hermes-serve*", "--plain", "--no-legend", "--no-pager"]
|
|
|
|
|
|
def _write_gateway_update_exit_code(ok: bool) -> None:
|
|
from hermes_cli.update_cmd import get_hermes_home
|
|
path = get_hermes_home() / ".update_exit_code"
|
|
with suppress(OSError):
|
|
path.write_text("0" if ok else "1", encoding="utf-8")
|
|
|
|
|
|
def _fleet_restart_pending_marker_path() -> Path:
|
|
"""LEGACY per-``HERMES_HOME`` breadcrumb. Read-compat only — nothing writes it any more.
|
|
|
|
One host runs one multiplexing gateway, so the pull→restart obligation is host-scoped
|
|
(``hermes_cli/update_host_obligation.py``). An obligation armed by the old per-profile code
|
|
is still read and cleared here so an in-flight update is discharged after the upgrade.
|
|
"""
|
|
from hermes_cli.update_cmd import get_hermes_home
|
|
return get_hermes_home() / _FLEET_RESTART_PENDING_NAME
|
|
|
|
|
|
def _write_legacy_fleet_restart_pending_marker(
|
|
*, expected_sha: str = "", runtimes: list[dict] | None = None
|
|
) -> bool:
|
|
"""Arm the LEGACY per-``HERMES_HOME`` marker. True when written. Never raises.
|
|
|
|
Fallback only: ``$HERMES_HOME`` is writable by construction (the updater already writes its
|
|
receipts there), so it still carries the obligation when the host state dir cannot.
|
|
"""
|
|
path = _fleet_restart_pending_marker_path()
|
|
try:
|
|
lines = [f"started={_time.time()}", f"pid={os.getpid()}"]
|
|
if expected_sha:
|
|
lines.append(f"expected_sha={expected_sha}")
|
|
if runtimes is not None:
|
|
lines.append("inventory=" + json.dumps({"version": 1, "runtimes": runtimes}))
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return True
|
|
except OSError as exc:
|
|
logger.debug("Could not write legacy fleet-restart-pending marker: %s", exc)
|
|
return False
|
|
|
|
|
|
def _write_fleet_restart_pending_marker(*, expected_sha: str = "", runtimes: list[dict] | None = None) -> None:
|
|
"""Arm the HOST pull→restart obligation. Never raises.
|
|
|
|
An unwritable host state dir (``HERMES_GATEWAY_LOCK_DIR`` on a read-only mount, a container
|
|
UID that does not own ``$HOME``) must never disarm the obligation: an update interrupted
|
|
after this point would then leave stale code running with no warning and no catch-up restart
|
|
(#117275). The legacy per-home marker — which every reader here still honours — carries it
|
|
instead, and a host that can write neither says so out loud.
|
|
"""
|
|
if runtimes == []:
|
|
# An explicit empty inventory owes no restart (e.g. Desktop-hosted `serve` with no
|
|
# gateway services). Arming the marker here leaves a breadcrumb nothing can discharge:
|
|
# a no-gateway host would then fail every later ``hermes update`` (#115311).
|
|
return
|
|
from hermes_cli.update_cmd import _m
|
|
from hermes_cli.update_host_obligation import host_obligation_path, write_host_obligation
|
|
if _m()._pytest_owns_live_checkout(_fleet_restart_pending_marker_path().parent):
|
|
logger.debug("Skipping fleet-restart-pending obligation under pytest (live checkout)")
|
|
return
|
|
if write_host_obligation(
|
|
expected_sha=expected_sha, runtimes=runtimes, profile=_current_profile_name()):
|
|
return
|
|
if _write_legacy_fleet_restart_pending_marker(expected_sha=expected_sha, runtimes=runtimes):
|
|
logger.warning(
|
|
"Host update-restart obligation (%s) is unwritable; armed the per-home marker %s instead.",
|
|
host_obligation_path(), _fleet_restart_pending_marker_path())
|
|
return
|
|
logger.error(
|
|
"Could not arm the update-restart obligation in %s or %s; an interrupted update will not warn.",
|
|
host_obligation_path(), _fleet_restart_pending_marker_path())
|
|
print(
|
|
" ⚠ Could not record the pending gateway-restart obligation (state dir not writable) — "
|
|
"restart gateways with `hermes gateway restart` if this update is interrupted.",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
|
|
def _current_profile_name() -> str:
|
|
"""Profile whose CLI armed the obligation (diagnostics only — the record is host-scoped)."""
|
|
try:
|
|
from hermes_cli.profiles import get_active_profile_name
|
|
return get_active_profile_name() or "default"
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _clear_fleet_restart_pending_marker() -> None:
|
|
"""Discharge the obligation for the whole host (legacy per-home marker included). Never raises."""
|
|
from hermes_cli.update_cmd import _m
|
|
from hermes_cli.update_host_obligation import clear_host_obligation
|
|
clear_host_obligation()
|
|
_m()._clear_marker_file(_fleet_restart_pending_marker_path(), label="fleet-restart-pending")
|
|
|
|
|
|
def _fleet_restart_obligation_armed() -> bool:
|
|
"""True when this HOST owes a fleet restart — from any profile's CLI."""
|
|
from hermes_cli.update_host_obligation import host_obligation_present
|
|
if host_obligation_present():
|
|
return True
|
|
with suppress(OSError):
|
|
return _fleet_restart_pending_marker_path().is_file()
|
|
return False
|
|
|
|
|
|
def _obligation_fields() -> dict[str, str] | None:
|
|
"""Armed obligation as ``key=value`` fields: HOST record first, then the legacy marker.
|
|
|
|
``None`` means nothing armed OR a malformed record; both must leave the obligation standing.
|
|
"""
|
|
from hermes_cli.update_host_obligation import host_obligation_present, obligation_fields
|
|
fields = obligation_fields()
|
|
if fields is not None:
|
|
return fields
|
|
if host_obligation_present():
|
|
# The record exists but its terms are unknown (corrupt, or a NEWER CLI's version). An
|
|
# unrelated legacy marker's inventory cannot discharge terms nobody can read: fail closed.
|
|
return None
|
|
try:
|
|
text = _fleet_restart_pending_marker_path().read_text(encoding="utf-8-sig")
|
|
except (OSError, UnicodeError):
|
|
return None
|
|
legacy: dict[str, str] = {}
|
|
for line in text.splitlines():
|
|
key, sep, value = line.partition("=")
|
|
if not sep or key in legacy:
|
|
return None
|
|
legacy[key] = value
|
|
return legacy
|
|
|
|
|
|
def _current_checkout_sha() -> str | None:
|
|
"""Current on-disk checkout HEAD, or None if it cannot be resolved."""
|
|
from hermes_cli.update_cmd import _capture_head_sha, _m
|
|
try:
|
|
from hermes_cli.version_info import get_code_identity
|
|
sha = (get_code_identity(refresh=True) or {}).get("sha")
|
|
return str(sha) if sha else None
|
|
except Exception:
|
|
return _capture_head_sha(["git"], _m().PROJECT_ROOT)
|
|
|
|
|
|
def _receipt_looks_unfinished(receipt: dict) -> bool:
|
|
"""True when *receipt* is from an update that did not finish cleanly.
|
|
|
|
The command boundary stamps a ``stop_reason`` on every receipt, including clean
|
|
ones (``completed at command boundary``, ``sys.exit(0)``); it must not make a
|
|
successful receipt look unfinished, or the next ``hermes update`` retriggers
|
|
``fleet_restart_pending`` from pre-pull plan SHAs (#98022).
|
|
"""
|
|
exit_code = receipt.get("exit_code")
|
|
outcome = receipt.get("outcome")
|
|
if exit_code not in (0, None) or outcome in ("failed", "partial", "running"):
|
|
return True
|
|
gateway_restart = receipt.get("gateway_restart")
|
|
if isinstance(gateway_restart, dict) and gateway_restart.get("incomplete"):
|
|
return True
|
|
# A stop_reason alone (update_contract refusals: outcome="refused", no exit_code)
|
|
# counts only when nothing else vouched for success.
|
|
succeeded = exit_code == 0 or outcome == "success"
|
|
return bool(receipt.get("stop_reason")) and not succeeded
|
|
|
|
|
|
def _receipt_reports_stale_runtime(receipt: dict, expected_sha: str | None = None) -> bool:
|
|
"""True when ``update_receipts/latest.json`` records a runtime SHA skew.
|
|
|
|
Prefer the post-restart ``fleet`` matrix. ``plan.runtimes[].code_sha`` is captured
|
|
*before* the pull, so a finished update's plan always looks stale and must not
|
|
retrigger a restart; consult it only for an unfinished receipt.
|
|
|
|
See #95294.
|
|
"""
|
|
from hermes_cli.update_cmd import _current_checkout_sha
|
|
if not isinstance(receipt, dict):
|
|
return False
|
|
expected_sha = expected_sha or _current_checkout_sha()
|
|
if not expected_sha:
|
|
return False
|
|
|
|
def _sha_mismatch(code_sha) -> bool:
|
|
return bool(code_sha) and str(code_sha) != str(expected_sha)
|
|
|
|
from hermes_cli.update_receipt import row_is_external
|
|
|
|
fleet = receipt.get("fleet")
|
|
if isinstance(fleet, list) and fleet:
|
|
return any(
|
|
isinstance(entry, dict)
|
|
and not row_is_external(entry)
|
|
and (entry.get("state") == "stale" or _sha_mismatch(entry.get("code_sha")))
|
|
for entry in fleet
|
|
)
|
|
|
|
if not _receipt_looks_unfinished(receipt):
|
|
return False
|
|
plan = receipt.get("plan")
|
|
if not isinstance(plan, dict):
|
|
return False
|
|
return any(
|
|
isinstance(runtime, dict) and _sha_mismatch(runtime.get("code_sha"))
|
|
for runtime in plan.get("runtimes") or []
|
|
)
|
|
|
|
|
|
_SUPERVISED_SERVE_BACKENDS = frozenset(
|
|
{"manual-serve", "desktop", "desktop-ssh", "systemd", "launchd", "windows-service", "service"}
|
|
)
|
|
# Backends whose supervisor restarts the process without any updater bookkeeping. ``manual-serve``
|
|
# is excluded: it owes a durable handoff (``defer_manual_serve``) before it stops counting.
|
|
# ``systemd``/``windows-service``/``service`` mirror ``_SUPERVISED_SERVE_BACKENDS`` for parity only —
|
|
# the inventory writer classifies a serve/dashboard row as exactly launchd, desktop, desktop-ssh or manual-serve
|
|
# (``update_inventory._collect_ledger_runtimes``); those three are set for gateway rows alone.
|
|
_SUPERVISOR_OWNED_SERVE_BACKENDS = _SUPERVISED_SERVE_BACKENDS - {"manual-serve"}
|
|
|
|
|
|
def _receipt_owed_gateways(receipt: dict, pending_manual: list[dict]) -> set[tuple[str, str]] | None:
|
|
"""Pure coverage classification after manual retention of this receipt snapshot.
|
|
|
|
Empty means this receipt owes no gateways, not that an independent marker owes none. Unknown identities, unclassified serve backends and failed manual transfers make coverage unverified.
|
|
"""
|
|
plan = receipt.get("plan") or {}
|
|
entries: list[tuple[object, str | None]] = [(entry, None) for entry in plan.get("runtimes") or []]
|
|
entries.extend((entry, None) for entry in receipt.get("pending_manual_serves") or [])
|
|
entries.extend((entry, "gateway") for entry in receipt.get("fleet") or [])
|
|
owed: set[tuple[str, str]] = set()
|
|
unverified = False
|
|
for entry, default_kind in entries:
|
|
if not isinstance(entry, dict):
|
|
unverified = True
|
|
continue
|
|
kind = entry.get("kind", default_kind)
|
|
profile = entry.get("profile")
|
|
# A serve/dashboard row is outside the gateway matrix's evidence, not evidence against the
|
|
# gateways it does cover: a supervised backend (desktop, systemd, launchd) is its
|
|
# supervisor's to restart, and a manual-serve row outside the retention list is the serve
|
|
# obligation mechanism's — a host running a dashboard carries such a row in every receipt,
|
|
# and a blanket veto made the gateway warning permanently undischargeable there (#115090).
|
|
# Only an unclassified backend or a failed manual transfer still makes coverage unverified.
|
|
if kind in ("serve", "dashboard") and entry.get("supervisor") in _SUPERVISED_SERVE_BACKENDS and entry not in pending_manual:
|
|
continue
|
|
if kind != "gateway" or not profile or profile == "unknown":
|
|
unverified = True
|
|
continue
|
|
owed.add((kind, profile))
|
|
return None if unverified else owed
|
|
|
|
|
|
def _fleet_covered_gateways(fleet: list) -> set[tuple[str, str]] | None:
|
|
"""``(kind, profile)`` identities the live rows vouch for; ``None`` when any row is unidentified.
|
|
|
|
A multiplexer's row carries the ``served_profiles`` its runtime status records (``_fleet_row``
|
|
keeps the field only when well-formed), so one live process covers every profile it serves.
|
|
"""
|
|
covered: set[tuple[str, str]] = set()
|
|
for row in fleet:
|
|
profile = row.get("profile") if isinstance(row, dict) else None
|
|
if not profile or profile == "unknown":
|
|
return None # unidentified runtime: the matrix cannot vouch for it
|
|
covered.add(("gateway", profile))
|
|
covered.update(("gateway", served) for served in row.get("served_profiles") or [])
|
|
return covered
|
|
|
|
|
|
def _live_fleet_covers_receipt(expected_sha: str | None, receipt: dict, owed: set[tuple[str, str]] | None, *, accept_states: tuple = ("current",)) -> bool:
|
|
"""Require a successor at the expected SHA for every owed gateway identity."""
|
|
if not expected_sha:
|
|
return False
|
|
from hermes_cli.update_receipt import collect_fleet_versions, row_is_external
|
|
|
|
try:
|
|
if owed is None:
|
|
return False
|
|
if not owed:
|
|
return bool((receipt.get("plan") or {}).get("runtimes"))
|
|
fleet = collect_fleet_versions()
|
|
# State labels are checkout-relative; completed restarts may accept stale rows at the pulled SHA.
|
|
if not fleet or any(
|
|
row.get("state") not in accept_states or row.get("code_sha") != expected_sha
|
|
for row in fleet if not row_is_external(row)
|
|
):
|
|
return False
|
|
covered = _fleet_covered_gateways(fleet)
|
|
return covered is not None and owed <= covered
|
|
except Exception as exc:
|
|
logger.debug("Could not reconcile pending fleet identities: %s", exc)
|
|
return False
|
|
|
|
|
|
def _marker_owed_gateways(inventory: object) -> set[tuple[str, str]] | None:
|
|
"""The ``("gateway", profile)`` set a marker's inventory owes; None when it recorded none.
|
|
|
|
Raises ValueError for a malformed or unsupported inventory, which keeps the marker.
|
|
"""
|
|
from hermes_cli.update_cmd_fleet_gatewayless import runtime_outside_gateway_evidence
|
|
|
|
if inventory is None:
|
|
return None
|
|
if not isinstance(inventory, dict) or inventory.get("version") != 1:
|
|
raise ValueError("unsupported fleet-restart inventory")
|
|
runtimes = inventory.get("runtimes")
|
|
if not isinstance(runtimes, list):
|
|
raise ValueError("fleet-restart inventory has no runtime list")
|
|
owed: set[tuple[str, str]] = set()
|
|
for runtime in runtimes:
|
|
if not isinstance(runtime, dict):
|
|
raise ValueError("fleet-restart inventory row is not an object")
|
|
if runtime_outside_gateway_evidence(runtime):
|
|
continue
|
|
profile = runtime.get("profile")
|
|
if runtime.get("kind") != "gateway" or not isinstance(profile, str) or not profile.strip() or profile == "unknown":
|
|
raise ValueError("fleet-restart inventory row is not an identified gateway")
|
|
owed.add(("gateway", profile))
|
|
return owed
|
|
|
|
|
|
def _discharge_gatewayless_marker(checkout_sha: str, expected_sha: str) -> bool:
|
|
"""Settle an inventory-less marker on a host with no live gateway (#118742).
|
|
|
|
Only when the host itself shows nothing the update could still owe a restart to, and HEAD
|
|
still holds the code it pulled.
|
|
"""
|
|
from hermes_cli.update_cmd_fleet_checkout import checkout_contains
|
|
from hermes_cli.update_cmd_fleet_gatewayless import host_owes_no_gateway_restart
|
|
|
|
try:
|
|
gatewayless = (checkout_sha == expected_sha or checkout_contains(expected_sha)) and host_owes_no_gateway_restart()
|
|
except Exception as exc:
|
|
logger.debug("Gateway-less host probe failed; keeping fleet-restart-pending marker: %s", exc)
|
|
return False
|
|
if not gatewayless:
|
|
return False
|
|
_clear_fleet_restart_pending_marker()
|
|
logger.debug("Fleet-restart-pending marker discharged: host runs no gateway at %s", checkout_sha[:10])
|
|
return True
|
|
|
|
|
|
def _marker_only_restart_obsolete() -> bool:
|
|
"""Settle only the inventory stored with this marker's target SHA.
|
|
|
|
Historical receipts cannot narrow this obligation. Malformed or unsupported inventories stay
|
|
fail-closed; empty discovery never proves a stopped gateway recovered. Two shapes record no
|
|
obligation and settle without one: an explicit empty inventory (a pull that found no gateway,
|
|
#115311) clears outright, and an inventory-less marker (the pre-inventory writer, or a tail
|
|
that died before its inventory was recorded, #115638) clears once every live gateway is
|
|
current on the checkout — there is no recorded owed set, so the fleet running the code on disk
|
|
is the whole of the evidence the marker's warning can be about, even after HEAD moved past
|
|
``expected_sha`` by an out-of-band pull. With no live gateway at all, the inventory-less marker
|
|
asks the host instead (``update_cmd_fleet_gatewayless``): it clears when no profile left a
|
|
gateway that should be running and every live runtime is supervisor-owned or handed off, so a
|
|
Desktop-only install stops failing every later update (#118742).
|
|
|
|
A serve/dashboard row whose supervisor owns the restart (Desktop backend, systemd/launchd
|
|
unit, Windows service) is outside the gateway matrix's evidence, not evidence against it —
|
|
the same boundary ``_receipt_owed_gateways`` draws for receipts (#115090) and the restart
|
|
phase draws for the Desktop backend (#111494). Counting it made the warning permanently
|
|
undischargeable on every host that runs a dashboard. A manual-serve row still needs its
|
|
durable handoff (``defer_manual_serve``), and an unclassified backend stays fail-closed.
|
|
Discharging here strands nobody: the same row is still accounted at update time by
|
|
``update_inventory.report_unaccounted_runtimes``, which prints it and exits 1 when the restart
|
|
phase never touched it — this marker only stops re-warning about it on every later startup.
|
|
"""
|
|
from hermes_cli.update_cmd_fleet_checkout import checkout_contains
|
|
|
|
try:
|
|
fields = _obligation_fields()
|
|
if fields is None:
|
|
return False
|
|
expected_sha = fields.get("expected_sha", "").strip()
|
|
owed = _marker_owed_gateways(json.loads(fields.get("inventory", "null")))
|
|
except (OSError, UnicodeError, ValueError):
|
|
return False
|
|
if owed is not None and not owed:
|
|
# A pull that recorded no gateway runtime owes no restart; clearing avoids the
|
|
# stuck "Fleet restart incomplete" loop on Desktop-hosted (no-service) installs.
|
|
_clear_fleet_restart_pending_marker()
|
|
logger.debug("Fleet-restart-pending marker discharged: no gateway obligation recorded")
|
|
return True
|
|
if not expected_sha:
|
|
return False
|
|
checkout_sha = _current_checkout_sha()
|
|
if owed is not None and checkout_sha != expected_sha and not checkout_contains(expected_sha):
|
|
return False # a newer pull moved HEAD; it owns a fresh obligation
|
|
# HEAD may sit past ``expected_sha`` by a carried local commit (a cherry-picked hotfix) that no
|
|
# pull made and no fresh obligation covers; the fleet is held to the code it actually runs, which
|
|
# is what an equality gate on ``expected_sha`` could never discharge (#119367).
|
|
target_sha = checkout_sha
|
|
if not target_sha:
|
|
return False
|
|
try:
|
|
from hermes_cli.update_receipt import collect_fleet_versions, row_is_external
|
|
fleet = collect_fleet_versions()
|
|
except Exception as exc:
|
|
logger.debug("Fleet probe failed; keeping fleet-restart-pending marker: %s", exc)
|
|
return False
|
|
if not fleet:
|
|
if owed is not None:
|
|
return False # Absence cannot prove recovery of the recorded inventory.
|
|
return _discharge_gatewayless_marker(checkout_sha, expected_sha)
|
|
covered = _fleet_covered_gateways(fleet)
|
|
if covered is None:
|
|
return False # unidentified runtime: the matrix cannot vouch for it
|
|
for row in fleet:
|
|
if row_is_external(row):
|
|
continue
|
|
if row.get("state") != "current" or str(row.get("code_sha")) != target_sha:
|
|
return False # stale / down / unknown-identity row still owes the restart
|
|
if owed is not None and not owed <= covered:
|
|
return False # A gateway this marker owns is absent (down) or unidentifiable.
|
|
_clear_fleet_restart_pending_marker()
|
|
logger.debug(
|
|
"Fleet-restart-pending marker discharged: %d gateway(s) already serve %s",
|
|
len(fleet), target_sha[:10],
|
|
)
|
|
return True
|
|
|
|
|
|
def _receipt_restart_phase_completed(receipt: dict) -> str | None:
|
|
"""Return the pulled SHA when the restart phase completed, even if a later step failed."""
|
|
gateway_restart = receipt.get("gateway_restart")
|
|
if not isinstance(gateway_restart, dict) or not gateway_restart:
|
|
return None
|
|
if gateway_restart.get("incomplete") or gateway_restart.get("phase_error"):
|
|
return None
|
|
post_sha = (receipt.get("post_update") or {}).get("sha")
|
|
return str(post_sha) if post_sha else None
|
|
|
|
|
|
def _pending_fleet_restart_needed(*, receipt: dict | None = None, pending_manual: list[dict] | None = None) -> bool:
|
|
"""Require identity-matched gateways at checkout HEAD for update catch-up."""
|
|
from hermes_cli.update_cmd import _current_checkout_sha
|
|
from hermes_cli.update_receipt import read_latest_receipt
|
|
from hermes_cli.update_serve_obligations import retain_receipt_manual_serves
|
|
|
|
if receipt is None:
|
|
receipt = read_latest_receipt() or {}
|
|
if pending_manual is None:
|
|
pending_manual = retain_receipt_manual_serves(receipt)
|
|
# The HOST obligation owns its inventory; latest.json can belong to an older update.
|
|
if _fleet_restart_obligation_armed():
|
|
return not _marker_only_restart_obsolete()
|
|
owed = _receipt_owed_gateways(receipt, pending_manual)
|
|
if not _receipt_reports_stale_runtime(receipt):
|
|
return False
|
|
return not _live_fleet_covers_receipt(_current_checkout_sha(), receipt, owed)
|
|
|
|
|
|
def _update_owes_fleet_restart(*, receipt: dict | None = None, pending_manual: list[dict] | None = None) -> bool:
|
|
"""Hold a completed restart to the code it pulled, not a later checkout HEAD."""
|
|
from hermes_cli.update_cmd import _current_checkout_sha
|
|
from hermes_cli.update_receipt import read_latest_receipt
|
|
from hermes_cli.update_serve_obligations import retain_receipt_manual_serves
|
|
|
|
if receipt is None:
|
|
receipt = read_latest_receipt() or {}
|
|
if pending_manual is None:
|
|
pending_manual = retain_receipt_manual_serves(receipt)
|
|
# A completed older receipt cannot discharge an independent host obligation's inventory.
|
|
if _fleet_restart_obligation_armed():
|
|
return not _marker_only_restart_obsolete()
|
|
owed = _receipt_owed_gateways(receipt, pending_manual)
|
|
if not _receipt_reports_stale_runtime(receipt):
|
|
return False
|
|
restarted_to = _receipt_restart_phase_completed(receipt)
|
|
# A fleet an operator has since restarted onto a moved checkout (``hermes gateway restart`` —
|
|
# the remedy this warning names) has nothing of the update left to owe either.
|
|
if restarted_to and _live_fleet_covers_receipt(restarted_to, receipt, owed, accept_states=("current", "stale")):
|
|
return False
|
|
return not _live_fleet_covers_receipt(_current_checkout_sha(), receipt, owed)
|
|
|
|
|
|
def _warn_pending_fleet_restart(*, startup: bool = False) -> None:
|
|
"""Print the specific interrupted-update fleet-restart warning."""
|
|
stream = sys.stderr if startup else sys.stdout
|
|
print("⚠ A previous `hermes update` pulled new code but did not restart running gateways.", file=stream)
|
|
print(" Gateways may still be serving pre-update modules (mixed sys.modules).", file=stream)
|
|
if startup:
|
|
print(" Run `hermes update` or `hermes gateway restart`.", file=stream)
|
|
|
|
|
|
def _warn_pending_fleet_restart_on_startup() -> None:
|
|
"""Cheap CLI-startup hint. Never restarts; never raises."""
|
|
from hermes_cli.update_receipt import read_latest_receipt
|
|
from hermes_cli.update_serve_obligations import retain_receipt_manual_serves, warn_pending_manual_serves
|
|
|
|
receipt = read_latest_receipt() or {}
|
|
pending_manual = None
|
|
with suppress(Exception):
|
|
pending_manual = retain_receipt_manual_serves(receipt)
|
|
with suppress(Exception):
|
|
if _update_owes_fleet_restart(receipt=receipt, pending_manual=pending_manual):
|
|
_warn_pending_fleet_restart(startup=True)
|
|
with suppress(Exception):
|
|
warn_pending_manual_serves(startup=True, pending_manual=pending_manual)
|
|
|
|
|
|
def _systemd_gateway_unit_listings(on_list_timeout=None):
|
|
"""Yield ``(scope, scope_cmd, list-units CompletedProcess)`` per systemd scope that answered.
|
|
|
|
A missing systemctl skips the scope silently; a listing timeout skips it after
|
|
``on_list_timeout(scope, exc)`` (when given) so the other scope is still processed.
|
|
"""
|
|
for scope, scope_cmd in _SYSTEMD_SCOPES:
|
|
try:
|
|
result = _systemctl(scope_cmd + _LIST_GATEWAY_UNITS, timeout=10)
|
|
except FileNotFoundError:
|
|
continue
|
|
except subprocess.TimeoutExpired as exc:
|
|
if on_list_timeout is not None:
|
|
on_list_timeout(scope, exc)
|
|
continue
|
|
yield scope, scope_cmd, result
|
|
|
|
|
|
def _needs_sudo(scope: str) -> bool:
|
|
return (
|
|
scope == "system"
|
|
and hasattr(os, "geteuid")
|
|
and os.geteuid() != 0 # windows-footgun: ok — systemd path, Linux-only
|
|
)
|
|
|
|
|
|
def _unit_main_pid(scope_cmd: list, svc_name: str) -> int:
|
|
"""Live ``MainPID`` of a unit; ``0`` when inactive, unprivileged or unreadable.
|
|
|
|
Property reads need no manage-units privileges, and an unreadable PID is never collapsed:
|
|
identity that cannot be proved keeps its own restart.
|
|
"""
|
|
try:
|
|
result = _systemctl(list(scope_cmd) + ["show", svc_name, "--property=MainPID", "--value"], timeout=10)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return 0
|
|
if getattr(result, "returncode", 1) != 0:
|
|
return 0
|
|
try:
|
|
return int((getattr(result, "stdout", "") or "").strip() or 0)
|
|
except ValueError:
|
|
return 0
|
|
|
|
|
|
def _restart_systemd_gateway_units_best_effort(failed: list, listings) -> None:
|
|
"""Restart every hermes-gateway/serve unit ONCE PER LIVE HOST PROCESS.
|
|
|
|
One host runs one multiplexing gateway, so leftover per-profile units
|
|
(``hermes-gateway-<profile>.service``) all point at the SAME live ``MainPID``; restarting
|
|
each in turn restarts the host gateway N times — a self-inflicted N-fold outage triggered
|
|
by one update. Units that share a live main PID are collapsed to one representative and the
|
|
others are named as LEGACY units to migrate, never silently dropped.
|
|
"""
|
|
from hermes_cli.update_host_obligation import collapse_units_to_host_processes
|
|
|
|
answered = set()
|
|
targets: dict[str, tuple[str, list, str]] = {} # "<scope>/<unit>" -> (scope, scope_cmd, unit)
|
|
for scope, scope_cmd, result in listings:
|
|
answered.add(scope)
|
|
if result.returncode != 0:
|
|
failed.append(f"systemd-{scope} (listing failed)")
|
|
continue
|
|
_for_each_systemd_gateway_unit(
|
|
result.stdout,
|
|
process_unit=lambda svc_name, _scope=scope, _cmd=scope_cmd: targets.setdefault(
|
|
f"{_scope}/{svc_name}", (_scope, _cmd, svc_name)),
|
|
on_unit_timeout=lambda svc_name, exc: failed.append(svc_name),
|
|
)
|
|
|
|
keys = list(targets)
|
|
covered: dict[str, str] = {}
|
|
if len(keys) > 1:
|
|
# Only worth a `systemctl show` round when several units could be one process.
|
|
keys, covered = collapse_units_to_host_processes(
|
|
keys, lambda key: _unit_main_pid(targets[key][1], targets[key][2]))
|
|
for unit_key, owner_key in covered.items():
|
|
print(
|
|
f" • {unit_key} is a legacy per-profile unit sharing one host gateway process with "
|
|
f"{owner_key}; restarting it again would restart that process twice. Fold the units "
|
|
"together with: hermes gateway migrate"
|
|
)
|
|
|
|
for key in keys:
|
|
scope, scope_cmd, svc_name = targets[key]
|
|
if not _systemd_unit_owned_by_update(scope_cmd, svc_name):
|
|
continue
|
|
manage_cmd = list(scope_cmd) + ["--no-ask-password"]
|
|
if _needs_sudo(scope):
|
|
manage_cmd = ["sudo", "-n"] + manage_cmd
|
|
try:
|
|
result = _systemctl_reset_and_restart(manage_cmd, svc_name, scope_cmd=scope_cmd)
|
|
if result.returncode != 0 or not _wait_for_service_active(scope_cmd, svc_name):
|
|
failed.append(svc_name)
|
|
except subprocess.TimeoutExpired:
|
|
failed.append(svc_name)
|
|
# A timeout or missing executable is not an empty scope.
|
|
failed.extend(f"systemd-{scope} (listing unavailable)" for scope, _ in _SYSTEMD_SCOPES if scope not in answered)
|
|
|
|
|
|
def _live_fleet_current_rows() -> list[dict] | None:
|
|
"""The fleet matrix when the probe finds at least one gateway and every row is ``current``
|
|
at the checkout SHA (identity known); ``None`` on any unknown/stale/down row or a failed
|
|
probe (restart)."""
|
|
checkout_sha = _current_checkout_sha()
|
|
if not checkout_sha:
|
|
return None
|
|
try:
|
|
from hermes_cli.update_receipt import collect_fleet_versions
|
|
fleet = collect_fleet_versions()
|
|
except Exception as exc:
|
|
logger.debug("Pending fleet restart: fleet probe failed: %s", exc)
|
|
return None
|
|
if not fleet or _fleet_covered_gateways(fleet) is None:
|
|
return None
|
|
if all(row.get("state") == "current" and str(row.get("code_sha")) == checkout_sha for row in fleet):
|
|
return fleet
|
|
return None
|
|
|
|
|
|
def _restart_identity_sha() -> str:
|
|
"""The SHA a completed host restart is stamped with; ``""`` when nothing names the code.
|
|
|
|
``_current_checkout_sha()`` is ``None`` on every non-git install (zip, pip, Docker), and an
|
|
empty stamp can never match, so the per-host restart-once guard would be inert exactly on the
|
|
installs it exists for: each profile's ``hermes update`` would re-kill the one shared
|
|
multiplexer. The obligation's own ``expected_sha`` — else the receipt's post-update identity —
|
|
names the same pulled code.
|
|
"""
|
|
sha = _current_checkout_sha()
|
|
if sha:
|
|
return str(sha)
|
|
sha = ((_obligation_fields() or {}).get("expected_sha") or "").strip()
|
|
if sha:
|
|
return sha
|
|
with suppress(Exception):
|
|
from hermes_cli.update_receipt import read_latest_receipt
|
|
post_update = (read_latest_receipt() or {}).get("post_update")
|
|
if isinstance(post_update, dict):
|
|
return str(post_update.get("sha") or "")
|
|
return ""
|
|
|
|
|
|
def _fleet_restart_skip_reason(plan) -> str | None:
|
|
"""Why the completion tail may leave the fleet alone, or ``None`` when a restart is owed.
|
|
|
|
Every route (pulled, already-current, ZIP) now finishes through the same completion
|
|
tail, so the guards the old catch-up path carried live here: one host runs ONE
|
|
multiplexing gateway, so a second profile's ``hermes update`` attaches to the restart the
|
|
first one already stamped (#95294), and a fleet already serving the checkout code (a
|
|
no-op update, a manual ``hermes gateway restart`` seconds ago) is not re-killed (#117051).
|
|
|
|
The second guard needs BOTH the pre-update plan and the live probe: the live matrix only
|
|
lists gateways, so a planned ``serve`` still on pre-update code (or any runtime without a
|
|
stamped identity) keeps the restart — the reconciliation there is what surfaces it.
|
|
"""
|
|
from hermes_cli.update_host_obligation import host_restart_already_completed
|
|
checkout_sha = _restart_identity_sha()
|
|
if host_restart_already_completed(checkout_sha):
|
|
return "this host's gateway was already restarted for this update"
|
|
if (checkout_sha and plan is not None and plan.runtimes
|
|
and all(str(runtime.code_sha) == checkout_sha for runtime in plan.runtimes)
|
|
and _live_fleet_current_rows() is not None):
|
|
return "every running gateway already serves the checkout code"
|
|
return None
|
|
|
|
|
|
def _run_pending_fleet_restart() -> bool:
|
|
"""Historical retry hook; new retries use the ordinary completion owner."""
|
|
from hermes_cli._old_updater import stop_for_relaunch
|
|
stop_for_relaunch(incomplete=True)
|
|
|
|
|
|
def _systemctl(cmd: list, *, timeout: float):
|
|
"""Run a systemctl (or sudo systemctl) invocation, capturing utf-8 text with a timeout."""
|
|
return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout)
|
|
|
|
|
|
# poll() takes signed 32-bit milliseconds; keep headroom for rounding in communicate().
|
|
_SYSTEMCTL_RESTART_TIMEOUT_MAX = (2**31 - 1) // 1000 - 1
|
|
|
|
|
|
def _systemd_restart_timeout(scope_cmd: list, svc_name: str, *, start_only: bool = False) -> float:
|
|
"""Outwait the unit's stop + start budgets, not just the systemctl client.
|
|
|
|
A client timeout does not cancel the manager's queued restart. Unknown or
|
|
infinite limits use systemd's usual 90s per phase so automation stays bounded.
|
|
Custom ExecStop chains or EXTEND_TIMEOUT_USEC can still exceed this budget;
|
|
genuine timeouts must continue through the existing per-unit failure path.
|
|
"""
|
|
from gateway.shutdown_forensics import parse_systemd_duration_to_us
|
|
|
|
budgets = {"TimeoutStartUSec": 90.0}
|
|
if not start_only:
|
|
budgets["TimeoutStopUSec"] = 90.0
|
|
try:
|
|
show = _systemctl(
|
|
scope_cmd + ["show", svc_name, "--property=TimeoutStopUSec,TimeoutStartUSec"],
|
|
timeout=5,
|
|
)
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
return sum(budgets.values()) + 15.0
|
|
if show.returncode == 0:
|
|
for line in (show.stdout or "").splitlines():
|
|
key, _, raw = line.partition("=")
|
|
if key in budgets:
|
|
# The shared parser returns None for infinity/unrecognized units.
|
|
try:
|
|
raw = raw.strip()
|
|
duration = int(raw) if raw.isascii() and raw.isdigit() else parse_systemd_duration_to_us(raw)
|
|
if duration is not None and duration > 0:
|
|
budgets[key] = duration / 1_000_000
|
|
except (ValueError, OverflowError):
|
|
pass
|
|
return min(sum(budgets.values()) + 15.0, _SYSTEMCTL_RESTART_TIMEOUT_MAX)
|
|
|
|
|
|
def _systemctl_reset_and_restart(manage_cmd: list, svc_name: str, *, scope_cmd: list | None = None):
|
|
"""``reset-failed`` then ``restart``: a unit parked in failed state by systemd's own
|
|
auto-restart can wedge a plain ``restart`` against RestartSec backoff and stay dead."""
|
|
# Property reads need no manage-units privileges: narrow sudoers may permit
|
|
# restart/reset-failed but deny show. Keep the same user/system manager scope.
|
|
timeout = _systemd_restart_timeout(scope_cmd if scope_cmd is not None else manage_cmd, svc_name)
|
|
_systemctl(manage_cmd + ["reset-failed", svc_name], timeout=10)
|
|
return _systemctl(manage_cmd + ["restart", svc_name], timeout=timeout)
|
|
|
|
|
|
def _systemd_unit_owned_by_update(scope_cmd: list, svc_name: str) -> bool:
|
|
"""Gate a unit restart on the unit's home being one this update owns (#93349).
|
|
|
|
``hermes-gateway*`` is an account-wide namespace: a second install's ``hermes update`` used
|
|
to drain and restart the account's real ``hermes-gateway.service`` because the unit was
|
|
listed, not because it ran the updated code. Foreign or unreadable ownership prints a notice
|
|
and leaves the unit alone; it is not a failed restart.
|
|
"""
|
|
from hermes_cli.update_fleet_scope import describe_skipped_runtime, systemd_unit_hermes_home, home_in_update_scope
|
|
home = systemd_unit_hermes_home(scope_cmd, svc_name)
|
|
if home is not None and home_in_update_scope(home):
|
|
return True
|
|
print(describe_skipped_runtime("systemd unit", svc_name, home))
|
|
return False
|
|
|
|
|
|
def _scoped_manual_gateway_pids(pids, *, keep=(), quiet: bool = False) -> list[int]:
|
|
"""*pids* whose live home this update owns (plus *keep*, PIDs already mapped to this
|
|
install's profile PID files); every other gateway process is named and left running."""
|
|
from hermes_cli.update_fleet_scope import describe_skipped_runtime, partition_gateway_pids_by_scope
|
|
keep = set(keep)
|
|
owned, foreign = partition_gateway_pids_by_scope([pid for pid in pids if pid not in keep])
|
|
if not quiet:
|
|
for pid, home in foreign:
|
|
print(describe_skipped_runtime("gateway process", f"PID {pid}", home))
|
|
return [pid for pid in pids if pid in keep or pid in owned]
|
|
|
|
|
|
def _is_hermes_gateway_unit(unit: str) -> bool:
|
|
"""Exact base unit or hyphenated profile family only: ``startswith("hermes-serve")``
|
|
would accept ``hermes-server.service``."""
|
|
return (
|
|
# list-units is already pattern-filtered, but keep the name gate so a stray non-gateway/serve line
|
|
# cannot enter the restart path. See #83595.
|
|
unit == "hermes-gateway.service"
|
|
or unit.startswith("hermes-gateway-")
|
|
or unit == "hermes-serve.service"
|
|
or unit.startswith("hermes-serve-")
|
|
)
|
|
|
|
|
|
def _for_each_systemd_gateway_unit(list_units_stdout: str, *, process_unit, on_unit_timeout) -> None:
|
|
"""Process each hermes-gateway*/hermes-serve* unit from ``systemctl list-units``.
|
|
|
|
``TimeoutExpired`` from ``process_unit`` is isolated per unit via ``on_unit_timeout``
|
|
so one wedged systemctl call cannot abort the rest of the fleet.
|
|
|
|
See #68523.
|
|
"""
|
|
for line in (list_units_stdout or "").strip().splitlines():
|
|
parts = line.split()
|
|
if not parts:
|
|
continue
|
|
unit = parts[0]
|
|
if not unit.endswith(".service") or not _is_hermes_gateway_unit(unit):
|
|
continue
|
|
svc_name = unit.removesuffix(".service")
|
|
try:
|
|
process_unit(svc_name)
|
|
except subprocess.TimeoutExpired as exc:
|
|
on_unit_timeout(svc_name, exc)
|
|
|
|
|
|
def _service_unit_supports_graceful_sigusr1_restart(svc_name: str) -> bool:
|
|
"""Whether *svc_name* wires SIGUSR1 to a graceful drain-then-restart.
|
|
|
|
Only ``hermes-gateway*`` runs ``gateway/run.py`` (the handler); SIGUSR1 would just
|
|
kill ``hermes-serve*`` and burn the drain budget, so those go straight to the blunt
|
|
restart. Same exact/hyphenated shape as ``_for_each_systemd_gateway_unit`` so a
|
|
near-prefix unit like ``hermes-gatewayd`` is never signalled.
|
|
|
|
See #83438.
|
|
"""
|
|
return svc_name == "hermes-gateway" or svc_name.startswith("hermes-gateway-")
|
|
|
|
|
|
def _warn_incomplete_gateway_fleet_restart(failed_units: list) -> None:
|
|
"""Print an explicit incomplete-update warning for unrestarted units."""
|
|
from hermes_cli.gateway import is_macos
|
|
if not failed_units:
|
|
return
|
|
ordered = list(dict.fromkeys(failed_units)) # de-dup, discovery order
|
|
print()
|
|
print("⚠ Update incomplete — some units were not restarted:")
|
|
for name in ordered:
|
|
print(f" - {name}")
|
|
if is_macos():
|
|
# A label lands here when launchd wasn't supervising a live process after
|
|
# the restart — likely deregistered, which `launchctl kickstart` can't revive.
|
|
# See #88848.
|
|
print(" Listed services may be deregistered from launchd, or still")
|
|
print(" running pre-update code (mixed sys.modules). Recover with:")
|
|
print(" hermes gateway status")
|
|
print(" launchctl list | grep <label>")
|
|
print(" launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/<label>.plist")
|
|
return
|
|
print(" Skipped units may still be running pre-update code (mixed")
|
|
print(" sys.modules). Restart them manually, then verify:")
|
|
print(" hermes gateway status")
|
|
if any(not name.startswith("ai.hermes.") for name in ordered):
|
|
print(" systemctl --user restart <unit> # user-scope")
|
|
print(" sudo systemctl restart <unit> # system-scope")
|
|
if any(name.startswith("ai.hermes.") for name in ordered):
|
|
print(" launchctl kickstart -k gui/$UID/<label> # macOS (or user/$UID)")
|
|
|
|
|
|
def _restart_launchd_gateway_after_update(
|
|
*, supervision_verify: bool = True, self_restart_pending: set | None = None,
|
|
) -> tuple[list, list]:
|
|
"""Restart the invoking profile's launchd gateway after an update.
|
|
|
|
No ``launchctl list`` gating: a booted-out job (plist present, definition
|
|
deregistered) fails it, and it can exit non-zero while the job is alive — gating on it
|
|
silently skipped the restart yet printed "Update complete!". When the plist exists
|
|
``launchd_restart()`` always runs; every failure path is loud with a manual recovery
|
|
command. Returns ``(restarted_labels, failed_labels)``; with ``supervision_verify``
|
|
success also requires a fresh supervised PID ("the call returned" is not "supervised").
|
|
|
|
74973 (salvage #75021 by @jeff-mettel): the restart used to be gated on ``launchctl list <label>``
|
|
exiting 0. A *booted-out* job — plist present, definition deregistered from launchd (crashed helper,
|
|
manual bootout, failed prior update) — fails that check, so the whole branch silently skipped: no
|
|
restart, no message, ``KeepAlive`` unable to revive a definition launchd no longer knows, and the update
|
|
still printed "Update complete!".
|
|
See #88848.
|
|
"""
|
|
from hermes_cli.gateway import (
|
|
get_launchd_label, get_launchd_plist_path, launchd_restart, wait_for_launchd_gateway_supervision,
|
|
_is_pid_ancestor_of_current_process, _launchctl_supervised_pid,
|
|
)
|
|
current_label = get_launchd_label()
|
|
old_pid = None
|
|
try:
|
|
if not get_launchd_plist_path().exists():
|
|
return [], [] # not a launchd install — nothing to do or warn
|
|
# Snapshot BEFORE the restart: "supervising some pid" was true before too, so only a pid that
|
|
# actually changed distinguishes a restart from a no-op (the sibling loop's contract). Read-only
|
|
# and verification-only — the restart itself is never gated on `launchctl list` (#74973).
|
|
old_pid = _launchctl_supervised_pid(current_label) if supervision_verify else None
|
|
try:
|
|
launchd_restart()
|
|
except subprocess.CalledProcessError as e:
|
|
stderr = (getattr(e, "stderr", "") or "").strip()
|
|
print(
|
|
f" ⚠ Gateway restart failed: {stderr}\n"
|
|
" The gateway may be DOWN on pre-update code. "
|
|
"Recover manually: hermes gateway restart"
|
|
)
|
|
return [], [current_label]
|
|
except (FileNotFoundError, subprocess.TimeoutExpired) as e:
|
|
# A plist exists, so a gateway is SUPPOSED to be supervised; a broken/wedged
|
|
# launchctl is not proof nothing needs restarting. Count it, tell the operator.
|
|
print(
|
|
# The old code `pass`ed here (#74973's second silent variant); count it and tell the operator.
|
|
" ⚠ Could not restart the gateway "
|
|
f"({e.__class__.__name__}: {e}).\n"
|
|
" Recover manually: hermes gateway restart"
|
|
)
|
|
return [], [current_label]
|
|
|
|
if not supervision_verify:
|
|
return [current_label], []
|
|
if old_pid is not None and _is_pid_ancestor_of_current_process(old_pid):
|
|
# launchd_restart() handed the restart to the gateway this updater runs INSIDE (cron job in
|
|
# the gateway tree, #100179): it exits only after this process does, so no fresh supervised
|
|
# pid can appear while we wait. Record it as pending for the fleet matrix (#119597).
|
|
if self_restart_pending is not None:
|
|
self_restart_pending.add(old_pid)
|
|
return [current_label], []
|
|
|
|
# launchd_restart() returning only means "restart REQUESTED" (async). A helper dying
|
|
# before first bootstrap, or a bootstrap exiting 0 without registering (macOS 26.6.1),
|
|
# would otherwise reach "Update complete!" unsupervised. Verified domain-agnostically:
|
|
# domain locate fails on macOS-26 per-user domains.
|
|
# launchd_restart() returning is only "restart REQUESTED" — the self-restart branch hands work to the
|
|
# running gateway, a plist reload to a detached helper; both asynchronous. See #88848.
|
|
if wait_for_launchd_gateway_supervision(label=current_label, old_pid=old_pid):
|
|
return [current_label], []
|
|
print(
|
|
f" ✗ {current_label} restarted but launchd is not supervising a new process for it.\n"
|
|
" Check logs, then: hermes gateway restart"
|
|
)
|
|
return [], [current_label]
|
|
|
|
|
|
def _restart_macos_launchd_gateways(
|
|
restarted_services: list, failed_or_stale_units: list, drain_budget: float, *, require_supervision: bool = False,
|
|
self_restart_pending: set | None = None,
|
|
) -> None:
|
|
"""Restart every launchd-managed gateway after an update (macOS).
|
|
|
|
The pull is shared across profiles, so every ``ai.hermes.gateway*`` LaunchAgent
|
|
must reload it or siblings stay on pre-update ``sys.modules`` (systemd parity).
|
|
Invoking profile uses ``launchd_restart()``; siblings get the same drain-first
|
|
sequence with their domain (``gui/<uid>`` vs ``user/<uid>``) resolved per label so
|
|
none is kickstarted in the wrong domain. ``TimeoutExpired`` is isolated per label.
|
|
|
|
See #41403.
|
|
The invoking profile keeps the existing ``launchd_restart()`` treatment (self-restart request → graceful
|
|
drain → kickstart). ``subprocess.TimeoutExpired`` is isolated per label so one wedged launchctl call
|
|
cannot leave the rest of the fleet on old code (#68523).
|
|
"""
|
|
from hermes_cli.gateway import (
|
|
get_launchd_label, get_launchd_plist_path, launchd_gateway_labels_for_install, legacy_launchd_labels_for_install,
|
|
_graceful_restart_via_sigusr1, _launchd_kickstart,
|
|
_locate_launchd_gateway_service, _wait_for_launchd_service_pid,
|
|
)
|
|
if require_supervision:
|
|
listing = subprocess.run(["launchctl", "list"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10)
|
|
if listing.returncode != 0:
|
|
failed_or_stale_units.append("launchd (listing failed)")
|
|
return
|
|
_restarted, _failed = _restart_launchd_gateway_after_update(
|
|
supervision_verify=True, self_restart_pending=self_restart_pending)
|
|
restarted_services.extend(_restarted)
|
|
failed_or_stale_units.extend(_failed)
|
|
current_label = get_launchd_label()
|
|
|
|
derived_labels = launchd_gateway_labels_for_install()
|
|
# Units labelled before the profile-name suffix scheme (ai.hermes.gateway-<hash>) are invisible
|
|
# to the derivation; legacy_launchd_labels_for_install() credits one only when its plist is
|
|
# provably this install's, so the #41403 boundary (never touch another install's fleet) holds.
|
|
# See #115254.
|
|
legacy_labels = legacy_launchd_labels_for_install(exclude=set(derived_labels) | {current_label})
|
|
if legacy_labels:
|
|
print(f" ↻ legacy-labelled units of this install join the restart: {', '.join(legacy_labels)}")
|
|
from hermes_cli.update_fleet_scope import describe_skipped_runtime, launchd_label_foreign_home
|
|
for label in derived_labels + legacy_labels:
|
|
if label == current_label:
|
|
continue
|
|
# Labels are account-global: root B's default profile derives the same bare label root A
|
|
# installed. A plist pinning a foreign HERMES_HOME is another install's job (#93349).
|
|
if (foreign_home := launchd_label_foreign_home(label)) is not None:
|
|
print(describe_skipped_runtime("launchd job", label, foreign_home))
|
|
continue
|
|
try:
|
|
# Locate = liveness + domain in one probe; kickstart and fresh-PID checks
|
|
# reuse that domain so a sibling is never probed in one and restarted in another.
|
|
domain, old_pid = _locate_launchd_gateway_service(label)
|
|
if domain is None:
|
|
if require_supervision and get_launchd_plist_path().with_name(f"{label}.plist").exists():
|
|
failed_or_stale_units.append(label)
|
|
continue # A profile without an installed job has no restart target.
|
|
graceful_ok = False
|
|
if old_pid is not None and old_pid > 0:
|
|
print(f" → {label}: draining (up to {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 systemd-launched serve sets neither HERMES_SPAWN
|
|
# nor HERMES_PARENT_PID ("manual-serve"). Unit-backed serves are recovered by the fresh
|
|
# child's systemd pass; survivors reported by _surviving_pre_update_serve_runtimes.
|
|
_SERVE_SKIP_REASON = (
|
|
"no per-profile relaunch command reaches a serve/dashboard runtime; recovered by the fresh"
|
|
" systemd unit pass when it owns a hermes-serve* unit, else left running for explicit"
|
|
" operator restart"
|
|
)
|
|
# A launchd-owned backend (#116503): the fresh child has no per-label kickstart for serve/dashboard
|
|
# jobs, but the post-update dashboard cleanup pass kickstarts the loaded job — never a detached
|
|
# argv respawn, which would fight the job's own KeepAlive.
|
|
_LAUNCHD_SERVE_SKIP_REASON = (
|
|
"launchd job owns this backend; the post-update dashboard cleanup kickstarts the job through"
|
|
" launchd, never a detached argv respawn that would fight its KeepAlive"
|
|
)
|
|
|
|
|
|
def _gateway_recovery_partition(plan, *, skip_profiles: set[str] | None = None) -> tuple[dict[str, str], list[dict]]:
|
|
"""Partition pre-update runtimes into fresh-restart candidates and skips.
|
|
|
|
Uses only the pre-checkout inventory: re-importing ``hermes_cli.gateway`` in the
|
|
failing interpreter is what raises the original ``ImportError``. Returns
|
|
``(candidates, skipped)``: profile → supervisor for supervised gateways the fresh
|
|
process may restart; every other inventoried runtime with an explicit reason so
|
|
nothing vanishes silently. Skipped serve/dashboard is NOT unrecoverable: the fresh
|
|
child's ``hermes-serve*`` systemd pass enumerates units from systemd; leftovers are
|
|
caught by :func:`_surviving_pre_update_serve_runtimes`.
|
|
"""
|
|
skip_profiles = skip_profiles or set()
|
|
candidates: dict[str, str] = {}
|
|
skipped: list[dict] = []
|
|
with _best_effort('Could not prepare fresh gateway restart profiles: %s'):
|
|
for runtime in getattr(plan, "runtimes", ()) or ():
|
|
kind = getattr(runtime, "kind", None)
|
|
profile = getattr(runtime, "profile", None)
|
|
supervisor = getattr(runtime, "supervisor", None)
|
|
if not isinstance(profile, str) or not profile:
|
|
continue
|
|
if kind == "gateway":
|
|
if profile in skip_profiles:
|
|
continue
|
|
if supervisor in _FRESH_RESTART_SUPERVISORS:
|
|
candidates.setdefault(profile, str(supervisor))
|
|
continue
|
|
reason = _MANUAL_GATEWAY_SKIP_REASON
|
|
elif kind in ("serve", "dashboard"):
|
|
from hermes_cli.update_inventory import CLIENT_OWNED_SERVE_SUPERVISORS
|
|
|
|
if supervisor in CLIENT_OWNED_SERVE_SUPERVISORS:
|
|
reason = _DESKTOP_SERVE_SKIP_REASON
|
|
elif supervisor == "launchd":
|
|
reason = _LAUNCHD_SERVE_SKIP_REASON
|
|
else:
|
|
reason = _SERVE_SKIP_REASON
|
|
else:
|
|
continue
|
|
skipped.append({"profile": profile, "kind": str(kind), "supervisor": str(supervisor), "reason": reason})
|
|
return candidates, skipped
|
|
|
|
|
|
def _warn_gateway_restart_phase_aborted(exc: BaseException, pids) -> None:
|
|
"""Print a recovery warning when the whole restart phase raised.
|
|
|
|
Previously a blanket debug-logged ``except Exception`` erased every drain/restart
|
|
line, so "Update complete!" exited 0 while the gateway kept serving pre-update
|
|
modules and died on the next turn with an ImportError.
|
|
|
|
Issue #78574: the gateway auto-restart phase was wrapped in a blanket ``except Exception`` that only
|
|
logged at debug level, so an early failure (e.g. importing ``hermes_cli.gateway`` from the freshly
|
|
pulled checkout) erased every drain/restart line from the update output.
|
|
"""
|
|
print()
|
|
print(f"⚠ Update incomplete — gateway auto-restart failed: {exc}")
|
|
if pids:
|
|
listed = ", ".join(str(pid) for pid in pids)
|
|
print(f" Gateway process(es) still running pre-update code: {listed}")
|
|
else:
|
|
print(" Any gateway still running is serving pre-update code")
|
|
print(" (mixed sys.modules) against the updated checkout.")
|
|
print(" Restart it manually, then verify:")
|
|
print(" hermes gateway restart")
|
|
print(" hermes gateway status")
|
|
|
|
|
|
def _drain_or_signal_gateway_for_update(
|
|
pid: int, drain_budget: float, label: str, *, self_restart_pending: set | None = None,
|
|
) -> bool:
|
|
"""Three-way triage (shared by systemd and bare-process paths) for handing a
|
|
running gateway over to new code. Returns True when signalled/stopped.
|
|
|
|
1. Gateway is an ancestor of this process (auto-update cron inside the gateway
|
|
tree): waiting is circular (gateway waits on in-flight work → cron session
|
|
waits on update → update waits on gateway) and the 1800s force-drain cap burns.
|
|
So fire-and-forget: signal restart and return; it completes once THIS process exits.
|
|
The pid lands in ``self_restart_pending`` so the fleet matrix can tell "restart
|
|
deferred until the updater exits" from "restart never happened" (#119597): the
|
|
ancestor is still serving the old code when the matrix runs, by construction.
|
|
2. Event loop provably wedged: SIGUSR1 can never drain it; bounded SIGTERM→SIGKILL.
|
|
3. Live out-of-tree gateway: graceful SIGUSR1 drain up to ``drain_budget``.
|
|
|
|
The wedged-loop probe cannot break it: the cron session posts activity every ~180s (process-tool poll
|
|
return), so it is "actively waiting forever" and never marked wedged — the gateway burns the full
|
|
force-drain cap (1800s) before killing its own updater's session. See #86684.
|
|
"""
|
|
from hermes_cli.gateway import (
|
|
GATEWAY_LOOP_WEDGED, _escalate_wedged_gateway, _graceful_restart_via_sigusr1,
|
|
_is_pid_ancestor_of_current_process, _request_gateway_self_restart, probe_gateway_loop_liveness,
|
|
)
|
|
if _is_pid_ancestor_of_current_process(pid):
|
|
print(
|
|
f" → {label}: update is running inside this gateway's "
|
|
"process tree — signalling restart and letting the gateway "
|
|
"drain itself (avoids the cron-update deadlock, #100179)"
|
|
)
|
|
accepted = _request_gateway_self_restart(pid)
|
|
if accepted and self_restart_pending is not None:
|
|
self_restart_pending.add(pid)
|
|
return accepted
|
|
if probe_gateway_loop_liveness(pid) == GATEWAY_LOOP_WEDGED:
|
|
print(f" ⚠ {label}: gateway event loop is unresponsive — skipping drain, forcing a bounded stop...")
|
|
_escalate_wedged_gateway(pid)
|
|
return True
|
|
print(f" → {label}: draining (up to {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).returncode == 0:
|
|
return True
|
|
# Blanket sudo refused — a targeted NOPASSWD sudoers entry may still work.
|
|
return subprocess.run(["sudo", "-n", *targeted_probe], capture_output=True, timeout=5).returncode == 0
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return False
|
|
|
|
|
|
def _resolve_manage_cmd(cache: dict, scope_: str, scope_cmd_: list, svc_name_: str):
|
|
"""Resolve the command prefix for manage-units verbs (None ⇒ no privilege path).
|
|
|
|
Manage-units verbs on a *system* service trigger a polkit prompt for non-root
|
|
users, which flashes and dies inside our captured 10-15s subprocess. Root → plain
|
|
systemctl; else ``sudo -n`` blanket probe, then a targeted ``reset-failed`` probe
|
|
so a least-privilege sudoers entry scoped to hermes-gateway* qualifies (idempotent
|
|
no-op we run before every privileged restart anyway). On None the caller must SKIP
|
|
the restart (without draining first!). ``--no-ask-password`` prevents polkit hangs.
|
|
"""
|
|
if scope_ in cache:
|
|
return cache[scope_]
|
|
cmd = scope_cmd_ + ["--no-ask-password"]
|
|
if _needs_sudo(scope_):
|
|
sudo_cmd = ["sudo", "-n"] + cmd
|
|
cmd = sudo_cmd if _sudo_noninteractive_ok(cmd + ["reset-failed", svc_name_]) else None
|
|
cache[scope_] = cmd
|
|
return cmd
|
|
|
|
|
|
def _repair_unit_without_fatal_exit_park(svc_name: str, scope: str) -> None:
|
|
"""A unit whose restart policy predates ``RestartPreventExitStatus=78`` crash-loops on the PERMANENT
|
|
exit: a ``Restart=on-failure`` system unit restarted ~180x on a host-attach refusal while the
|
|
regenerated user units parked (#118282). The gateway rewrites its USER unit at boot; a SYSTEM unit
|
|
lives in /etc, so rewrite it here when we are root, else name the repair."""
|
|
from hermes_cli.gateway import (
|
|
_SYSTEM_UNIT_DIR, GATEWAY_FATAL_CONFIG_EXIT_CODE, get_service_name,
|
|
refresh_systemd_unit_if_needed, user_systemd_unit_dir,
|
|
)
|
|
system = scope == "system"
|
|
unit_path = (_SYSTEM_UNIT_DIR if system else user_systemd_unit_dir()) / f"{svc_name}.service"
|
|
try:
|
|
parked = re.search(rf"^RestartPreventExitStatus=.*\b{GATEWAY_FATAL_CONFIG_EXIT_CODE}\b", unit_path.read_text(encoding="utf-8-sig"), re.M)
|
|
except OSError:
|
|
return
|
|
if parked:
|
|
return
|
|
if system and not _needs_sudo(scope) and svc_name == get_service_name():
|
|
# The refresh adopts the unit's HERMES_HOME into os.environ (sudo strips it); the rest of the
|
|
# update keeps running for the invoking profile.
|
|
launch_home = os.environ.get("HERMES_HOME")
|
|
try:
|
|
refresh_systemd_unit_if_needed(system=True)
|
|
finally:
|
|
if launch_home is None:
|
|
os.environ.pop("HERMES_HOME", None)
|
|
else:
|
|
os.environ["HERMES_HOME"] = launch_home
|
|
return
|
|
print(
|
|
f" ⚠ {svc_name} lacks RestartPreventExitStatus={GATEWAY_FATAL_CONFIG_EXIT_CODE}: a permanent refusal "
|
|
f"(exit {GATEWAY_FATAL_CONFIG_EXIT_CODE}) would crash-loop it instead of parking.\n"
|
|
f" Repair: {'sudo ' if system else ''}hermes gateway install{' --system' if system else ''}"
|
|
)
|
|
|
|
|
|
def _restart_one_systemd_gateway_unit(
|
|
svc_name: str, *, scope: str, scope_cmd: list, drain_budget: float, _manage_cmd_cache: dict,
|
|
restarted_services: list, failed_or_stale_units: list, self_restart_pending: set | None = None,
|
|
) -> None:
|
|
"""Restart one active systemd gateway/serve unit: graceful SIGUSR1 drain, then forced restart.
|
|
|
|
Appends settled names to ``restarted_services`` and failures to ``failed_or_stale_units``.
|
|
"""
|
|
check = _systemctl(scope_cmd + ["is-active", svc_name], timeout=5)
|
|
if check.stdout.strip() != "active":
|
|
return
|
|
if not _systemd_unit_owned_by_update(scope_cmd, svc_name):
|
|
return
|
|
_repair_unit_without_fatal_exit_park(svc_name, scope)
|
|
|
|
# None ⇒ no non-interactive privilege path; avoid manage-units verbs
|
|
# entirely or polkit prompts inside the captured subprocess.
|
|
_manage_cmd = _resolve_manage_cmd(_manage_cmd_cache, scope, scope_cmd, svc_name)
|
|
|
|
# Graceful SIGUSR1 first so in-flight runs drain: handler → request_restart(via_service=True)
|
|
# → drain → exit, Restart=always respawns. hermes-serve has no handler → blunt restart below.
|
|
_main_pid = 0
|
|
if _service_unit_supports_graceful_sigusr1_restart(svc_name):
|
|
try:
|
|
_show = _systemctl(scope_cmd + ["show", svc_name, "--property=MainPID", "--value"], timeout=5)
|
|
_main_pid = int((_show.stdout or "").strip() or 0)
|
|
except (ValueError, subprocess.TimeoutExpired, FileNotFoundError):
|
|
_main_pid = 0
|
|
|
|
# Three-way triage (ancestor / wedged / graceful drain).
|
|
_graceful_ok = _main_pid > 0 and _drain_or_signal_gateway_for_update(
|
|
_main_pid, drain_budget, svc_name, self_restart_pending=self_restart_pending)
|
|
|
|
if _graceful_ok:
|
|
# ``Restart=always`` respawns only after RestartSec (60s in our unit; dead time for a
|
|
# voluntary restart). ``reset-failed`` + ``start`` skips it (~1-3s); if RestartSec already
|
|
# elapsed, ``start`` is a no-op and we fall through to the poll. Needs manage-units
|
|
# privileges; without them auto-restart still fires after RestartSec.
|
|
if _manage_cmd is not None:
|
|
_systemctl(_manage_cmd + ["reset-failed", svc_name], timeout=10)
|
|
_systemctl(
|
|
_manage_cmd + ["start", svc_name],
|
|
timeout=_systemd_restart_timeout(scope_cmd, svc_name, start_only=True),
|
|
)
|
|
if _wait_for_service_active(scope_cmd, svc_name, timeout=10.0):
|
|
restarted_services.append(svc_name)
|
|
return
|
|
# Passive poll: auto-restart fires after RestartSec regardless of
|
|
# privileges — primary when _manage_cmd is None, fallback otherwise.
|
|
_restart_sec = _service_restart_sec(scope_cmd, svc_name, default=0.0)
|
|
if _manage_cmd is None and _restart_sec > 5.0:
|
|
print(
|
|
f" → {svc_name}: waiting for systemd "
|
|
f"auto-restart (~{int(_restart_sec)}s; "
|
|
"no root for an immediate restart)..."
|
|
)
|
|
if _wait_for_service_active(scope_cmd, svc_name, timeout=max(10.0, _restart_sec + 10.0)):
|
|
restarted_services.append(svc_name)
|
|
return
|
|
# Exited but not respawned (older unit without Restart=on-failure /
|
|
# RestartForceExitStatus=75); fall through to forced restart.
|
|
print(f" ⚠ {svc_name} drained but didn't relaunch — forcing restart")
|
|
|
|
# Forcing needs manage-units privileges; without a non-interactive path
|
|
# polkit would prompt inside the captured subprocess — skip, instruct.
|
|
if _manage_cmd is None:
|
|
failed_or_stale_units.append(svc_name)
|
|
print(
|
|
f" ⚠ {svc_name} is a system service and restarting it needs root.\n"
|
|
f" Restart it manually to load the new version:\n"
|
|
f" sudo systemctl restart {svc_name}\n"
|
|
f" To let `hermes update` restart it automatically, allow\n"
|
|
f" passwordless sudo for systemctl, or run updates with sudo."
|
|
)
|
|
return
|
|
|
|
# Blunt restart — only when the graceful path failed (no SIGUSR1 wiring, drain over
|
|
# budget, restart-policy mismatch). Mirrors `hermes gateway restart` (`systemd_restart()`).
|
|
restart = _systemctl_reset_and_restart(_manage_cmd, svc_name, scope_cmd=scope_cmd)
|
|
if restart.returncode != 0:
|
|
failed_or_stale_units.append(svc_name)
|
|
print(f" ⚠ Failed to restart {svc_name}: {restart.stderr.strip()}")
|
|
return
|
|
# restart returns 0 even if the new process crashes at once — verify.
|
|
if _wait_for_service_active(scope_cmd, svc_name, timeout=10.0):
|
|
restarted_services.append(svc_name)
|
|
return
|
|
# Retry once — transient startup failures (stale module cache,
|
|
# import race) often clear; reset-failed so the retry isn't blocked.
|
|
print(f" ⚠ {svc_name} died after restart, retrying...")
|
|
_systemctl_reset_and_restart(_manage_cmd, svc_name, scope_cmd=scope_cmd)
|
|
if _wait_for_service_active(scope_cmd, svc_name, timeout=10.0):
|
|
restarted_services.append(svc_name)
|
|
print(f" ✓ {svc_name} recovered on retry")
|
|
return
|
|
failed_or_stale_units.append(svc_name)
|
|
_scope_flag = "--user " if scope == "user" else ""
|
|
_sudo_hint = "sudo " if scope == "system" else ""
|
|
print(
|
|
f" ✗ {svc_name} failed to stay running after restart.\n"
|
|
f" Check logs: {_sudo_hint}journalctl {_scope_flag}-u {svc_name} --since '2 min ago'\n"
|
|
f" Recover manually:\n"
|
|
f" {_sudo_hint}systemctl {_scope_flag}reset-failed {svc_name}\n"
|
|
f" {_sudo_hint}systemctl {_scope_flag}restart {svc_name}"
|
|
)
|
|
|
|
|
|
def _restart_systemd_gateway_units(
|
|
restarted_services, failed_or_stale_units, restarted_scoped_units, drain_budget, self_restart_pending=None,
|
|
):
|
|
"""Restart every active hermes-gateway*/hermes-serve* systemd unit (user + system).
|
|
|
|
Settled units → ``restarted_services`` (bare) and ``restarted_scoped_units``
|
|
(``scope/name``); failures → ``failed_or_stale_units``. Per-unit timeouts isolated.
|
|
"""
|
|
from hermes_cli.gateway import supports_systemd_services, _ensure_user_systemd_env
|
|
if not supports_systemd_services():
|
|
return
|
|
_manage_cmd_cache: dict = {}
|
|
with suppress(Exception):
|
|
_ensure_user_systemd_env()
|
|
|
|
def _on_list_timeout(scope: str, exc: subprocess.TimeoutExpired) -> None:
|
|
# Discovery timeout — skip this scope, keep the other.
|
|
print(
|
|
f" ⚠ systemctl timed out listing {scope}-scope "
|
|
f"gateway units ({exc.cmd if exc.cmd else 'unknown command'}). "
|
|
f"Check the gateway with: hermes gateway status"
|
|
)
|
|
|
|
def _on_unit_timeout(svc_name: str, exc: subprocess.TimeoutExpired) -> None:
|
|
# Isolate to this unit; a scope-wide handler used to abort every
|
|
# later gateway and leave the fleet on mixed code.
|
|
failed_or_stale_units.append(svc_name)
|
|
print(
|
|
# See #68523.
|
|
f" ⚠ systemctl timed out restarting {svc_name} "
|
|
f"({exc.cmd if exc.cmd else 'unknown command'}); "
|
|
f"continuing with remaining gateways"
|
|
)
|
|
|
|
# Enumerate every scope first: leftover per-profile units (``hermes-gateway-<profile>``)
|
|
# all point at the SAME live MainPID on a multiplexed host, and restarting each in turn
|
|
# is an N-fold outage from one update. Units sharing a live PID collapse to one restart;
|
|
# the others are named as legacy units to migrate, never silently dropped.
|
|
from hermes_cli.update_host_obligation import collapse_units_to_host_processes
|
|
targets: dict[str, tuple[str, list, str]] = {} # "<scope>/<unit>" -> (scope, scope_cmd, unit)
|
|
for scope, scope_cmd, result in _systemd_gateway_unit_listings(_on_list_timeout):
|
|
_for_each_systemd_gateway_unit(
|
|
result.stdout,
|
|
process_unit=lambda svc_name, _scope=scope, _cmd=scope_cmd: targets.setdefault(
|
|
f"{_scope}/{svc_name}", (_scope, _cmd, svc_name)),
|
|
on_unit_timeout=_on_unit_timeout,
|
|
)
|
|
keys = list(targets)
|
|
covered: dict[str, str] = {}
|
|
if len(keys) > 1:
|
|
# Only worth a `systemctl show` round when several units could be one process.
|
|
keys, covered = collapse_units_to_host_processes(
|
|
keys, lambda key: _unit_main_pid(targets[key][1], targets[key][2]))
|
|
for unit_key, owner_key in covered.items():
|
|
print(
|
|
f" • {unit_key} is a legacy per-profile unit sharing one host gateway process with "
|
|
f"{owner_key}; restarting it again would restart that process twice. Fold the units "
|
|
"together with: hermes gateway migrate"
|
|
)
|
|
|
|
for key in keys:
|
|
scope, scope_cmd, svc_name = targets[key]
|
|
# Scope-qualify before the next unit; ``finally`` so a mid-pass abort keeps settled units.
|
|
_scope_mark = len(restarted_services)
|
|
try:
|
|
_restart_one_systemd_gateway_unit(
|
|
svc_name,
|
|
scope=scope,
|
|
scope_cmd=scope_cmd,
|
|
drain_budget=drain_budget,
|
|
_manage_cmd_cache=_manage_cmd_cache,
|
|
restarted_services=restarted_services,
|
|
failed_or_stale_units=failed_or_stale_units,
|
|
self_restart_pending=self_restart_pending,
|
|
)
|
|
except subprocess.TimeoutExpired as exc:
|
|
_on_unit_timeout(svc_name, exc)
|
|
finally:
|
|
restarted_scoped_units.update(f"{scope}/{name}" for name in restarted_services[_scope_mark:])
|
|
|
|
|
|
def _unit_main_pid(scope_cmd: list, svc_name: str) -> int:
|
|
"""Live ``MainPID`` of a unit; ``0`` when inactive, unprivileged or unreadable.
|
|
|
|
Property reads need no manage-units privileges, and an unreadable PID is never collapsed:
|
|
identity that cannot be proved keeps its own restart.
|
|
"""
|
|
try:
|
|
result = _systemctl(list(scope_cmd) + ["show", svc_name, "--property=MainPID", "--value"], timeout=10)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return 0
|
|
if getattr(result, "returncode", 1) != 0:
|
|
return 0
|
|
try:
|
|
return int((getattr(result, "stdout", "") or "").strip() or 0)
|
|
except ValueError:
|
|
return 0
|
|
|
|
|
|
@dataclass
|
|
class _GatewayRestartOutcome:
|
|
"""Restart-phase bookkeeping. ``restarted_services`` keeps bare unit names (fleet
|
|
probe, receipt, summary read it); ``incomplete`` ⇒ a gateway may still be stale."""
|
|
|
|
incomplete: bool
|
|
phase_errors: list
|
|
pre_restart_gateway_pids: "list | None"
|
|
restarted_services: list
|
|
failed_or_stale_units: list
|
|
relaunched_profiles: list
|
|
externally_supervised_profiles: list
|
|
killed_pids: set
|
|
#: Gateways stopped with NO successor (no profile mapping / relaunch could not be armed);
|
|
#: the summary tells the user to restart them by hand, so the fleet probe must not expect
|
|
#: a row for them.
|
|
stopped_unmapped_pids: set = field(default_factory=set)
|
|
#: ``scope/name`` of every settled systemd unit; the fleet probe stops waiting for a state stamp
|
|
#: once none of them is active or activating any more (the successor died, nothing will publish).
|
|
restarted_scoped_units: set = field(default_factory=set)
|
|
#: Gateways that are ANCESTORS of this updater and accepted a self-restart request
|
|
#: (``_drain_or_signal_gateway_for_update`` branch 1): they restart only after this process
|
|
#: exits, so the fleet matrix renders them as pending instead of STALE (#119597).
|
|
self_restart_pending_pids: set = field(default_factory=set)
|
|
|
|
def fleet_probe_signals(self) -> tuple:
|
|
"""``(pre_restart_pids, killed_pids)`` with the unmapped stops removed — the signals that
|
|
legitimately predict a fleet-matrix row."""
|
|
pre = self.pre_restart_gateway_pids
|
|
if pre is not None:
|
|
pre = [pid for pid in pre if pid not in self.stopped_unmapped_pids]
|
|
return pre, self.killed_pids - self.stopped_unmapped_pids
|
|
|
|
def record_receipt(self, **extra) -> None:
|
|
"""Best-effort ``record_gateway_restart`` from the current bookkeeping."""
|
|
with suppress(Exception):
|
|
from hermes_cli.update_receipt import record_gateway_restart
|
|
record_gateway_restart(
|
|
restarted_services=self.restarted_services, relaunched_profiles=self.relaunched_profiles,
|
|
externally_supervised_profiles=self.externally_supervised_profiles,
|
|
killed_pids=sorted(self.killed_pids), failed_units=self.failed_or_stale_units,
|
|
incomplete=self.incomplete, **extra,
|
|
)
|
|
|
|
|
|
def _restart_manual_gateways(out: _GatewayRestartOutcome, _drain_budget) -> None:
|
|
"""Drain/stop every manual (non-service) gateway and print the restart summary.
|
|
|
|
Mutates ``out`` in place; raises so the caller's abort recovery fires.
|
|
"""
|
|
import signal as _signal
|
|
from hermes_cli.gateway import (
|
|
find_gateway_pids, find_profile_gateway_processes, _prepare_profile_gateway_update_restart, _get_service_pids,
|
|
_wait_for_gateway_exit,
|
|
)
|
|
# Exclude just-restarted service PIDs so we don't kill what systemd/launchd spawned.
|
|
service_pids = _get_service_pids(all_profiles=True)
|
|
manual_pids = find_gateway_pids(exclude_pids=service_pids, all_profiles=True)
|
|
profile_processes = {
|
|
proc.pid: proc
|
|
for proc in find_profile_gateway_processes(exclude_pids=service_pids)
|
|
if proc.pid in manual_pids
|
|
}
|
|
# ``all_profiles`` is host-wide: a sibling install's gateway matches too. Only this update's
|
|
# homes are stopped; the profile-mapped PIDs come from this install's own PID files (#93349).
|
|
manual_pids = _scoped_manual_gateway_pids(manual_pids, keep=profile_processes)
|
|
# Profile gateways we couldn't arm a relaunch for must NOT keep running stale:
|
|
# the unmapped sweep below stops them and lists them under "Restart manually".
|
|
# These must NOT be left running: their modules are the pre-update ones and every lazy import from here
|
|
# on mixes versions against the new code on disk (#88654). Handing them to the unmapped sweep below
|
|
# stops them and surfaces them in the "Stopped N manual gateway process(es) / Restart manually" summary,
|
|
# which is the contract already used for gateways with no profile mapping.
|
|
unrestartable_pids = set()
|
|
for pid, proc in profile_processes.items():
|
|
restart_mode = _prepare_profile_gateway_update_restart(proc.profile, pid)
|
|
if restart_mode is None:
|
|
# A bare ``continue`` here left it serving stale modules with no signal.
|
|
print(
|
|
f" ⚠ {proc.profile}: could not arm an automatic "
|
|
f"gateway restart for PID {pid} — stopping it instead "
|
|
"so it cannot keep running pre-update code"
|
|
)
|
|
unrestartable_pids.add(pid)
|
|
continue
|
|
# SIGUSR1 drain first, SIGTERM fallback if unsupported/over budget — the watcher
|
|
# relaunches either way. The helper announces its choice first because a silent
|
|
# full-budget wait reads as a hung update.
|
|
if not _drain_or_signal_gateway_for_update(
|
|
pid, _drain_budget, proc.profile, self_restart_pending=out.self_restart_pending_pids):
|
|
with suppress(ProcessLookupError, PermissionError):
|
|
os.kill(pid, _signal.SIGTERM)
|
|
# Wait ≤5s for exit: Telegram keeps the old getUpdates session ~30s; a new gateway
|
|
# inside that window gets a 409 (_handle_polling_conflict retries, but a brief
|
|
# wait avoids it on fast machines).
|
|
_wait_for_gateway_exit(timeout=5.0, force_after=None)
|
|
out.killed_pids.add(pid)
|
|
if restart_mode == "external-supervisor":
|
|
out.externally_supervised_profiles.append(proc.profile)
|
|
else:
|
|
out.relaunched_profiles.append(proc.profile)
|
|
|
|
for pid in manual_pids:
|
|
if pid in profile_processes and pid not in unrestartable_pids:
|
|
continue
|
|
with suppress(ProcessLookupError, PermissionError):
|
|
os.kill(pid, _signal.SIGTERM)
|
|
out.killed_pids.add(pid)
|
|
out.stopped_unmapped_pids.add(pid)
|
|
|
|
if out.restarted_services or out.killed_pids:
|
|
print()
|
|
for svc in out.restarted_services:
|
|
print(f" ✓ Restarted {svc}")
|
|
if out.relaunched_profiles:
|
|
print(f" ✓ Restarting manual gateway profile(s): {', '.join(out.relaunched_profiles)}")
|
|
if out.externally_supervised_profiles:
|
|
names = ", ".join(out.externally_supervised_profiles)
|
|
print(f" ✓ Handed gateway profile(s) back to their external supervisor: {names}")
|
|
unmapped_count = (len(out.killed_pids) - len(out.relaunched_profiles) - len(out.externally_supervised_profiles))
|
|
if unmapped_count:
|
|
print(f" → Stopped {unmapped_count} manual gateway process(es)")
|
|
print(" Restart manually: hermes gateway run")
|
|
if unmapped_count > 1:
|
|
print(" (or: hermes -p <profile> gateway run for each profile)")
|
|
|
|
|
|
def _force_kill_stuck_gateways(killed_pids) -> None:
|
|
"""Survivor sweep: gateways ignoring SIGTERM (stuck drain, blocked I/O, zombie) never
|
|
exit, so the watcher never respawns and ImportErrors persist. Give graceful paths a
|
|
moment, then SIGKILL remaining pre-update PIDs."""
|
|
with _best_effort('Post-restart survivor sweep failed: %s'):
|
|
from hermes_cli.gateway import find_gateway_pids, _get_service_pids
|
|
# --- Post-restart survivor sweep ----------------------------- Issue #17648: some gateways ignore
|
|
# SIGTERM (stuck drain, blocked I/O, PID dead but zombie). The detached profile watchers wait 120s
|
|
# for the old PID to exit — if it never does, no respawn happens and the user keeps hitting
|
|
# ImportError against a stale sys.modules.
|
|
_time.sleep(3.0)
|
|
_surviving = find_gateway_pids(exclude_pids=_get_service_pids(all_profiles=True), all_profiles=True)
|
|
# Only PIDs we already tried to kill; newer ones are left alone.
|
|
_stuck = [pid for pid in _surviving if pid in killed_pids]
|
|
if _stuck:
|
|
print()
|
|
print(f" ⚠ {len(_stuck)} gateway process(es) ignored SIGTERM — force-killing")
|
|
from gateway.status import get_process_start_time, terminate_pid
|
|
for pid in _stuck:
|
|
with suppress(ProcessLookupError, PermissionError, OSError):
|
|
# taskkill /T /F on Windows (no SIGKILL there), SIGKILL on POSIX.
|
|
terminate_pid(pid, force=True, expected_start_time=get_process_start_time(pid))
|
|
# Let the OS reap so watchers see the exit and respawn.
|
|
_time.sleep(1.5)
|
|
|
|
|
|
def _recover_after_restart_phase_abort(
|
|
e, _pre_update_plan, out: _GatewayRestartOutcome, *, gateway_mode, restarted_scoped_units
|
|
) -> None:
|
|
"""Phase-abort recovery: fresh-child restart + fail-closed verdict; updates ``out`` in place."""
|
|
from hermes_cli.update_abort_recovery import _owed_stale_serve_rows
|
|
from hermes_cli.update_cmd import (
|
|
_abort_recovery_is_complete, _recover_gateway_restart_after_abort, _surviving_pre_update_serve_runtimes,
|
|
_warn_stale_serve_runtimes, _write_gateway_update_exit_code,
|
|
)
|
|
logger.debug("Gateway restart during update failed: %s", e)
|
|
out.phase_errors.append(str(e))
|
|
# Restart output never printed: assume stale unless provably no gateway runs.
|
|
# Empty ``_surviving`` proves safety only if nothing ran beforehand; a gone
|
|
# pre-restart gateway was stopped without verified replacement → fail closed.
|
|
# An exception escaping the whole phase means the drain/restart output the user relies on never printed.
|
|
# Don't let that pass for a clean update: surface it and treat the fleet as stale unless we can
|
|
# positively prove no gateway is running (#78574). A positive-empty ``_surviving`` is only
|
|
# proof-of-safety when nothing was running before we touched anything. If a gateway was discovered
|
|
# pre-restart and none survive now, it was stopped and its replacement was never verified — the same
|
|
# fail-open contract this fix closes — so we must still fail closed on ``[]``.
|
|
_surviving = _surviving_gateway_pids_after_failed_restart()
|
|
_planned_gateway_profiles = {
|
|
runtime.profile
|
|
for runtime in getattr(_pre_update_plan, "runtimes", ()) or ()
|
|
if getattr(runtime, "kind", None) == "gateway"
|
|
and isinstance(getattr(runtime, "profile", None), str)
|
|
}
|
|
_already_restarted_profiles = set(out.relaunched_profiles) | set(out.externally_supervised_profiles)
|
|
_already_restarted_profiles.update(
|
|
profile
|
|
for profile in _planned_gateway_profiles
|
|
if any(_gateway_service_matches_profile(profile, service) for service in out.restarted_services)
|
|
)
|
|
_recovery_result = _recover_gateway_restart_after_abort(
|
|
_pre_update_plan, gateway_mode=gateway_mode, skip_profiles=_already_restarted_profiles,
|
|
skip_units=set(restarted_scoped_units),
|
|
)
|
|
_serve_units_failed = list((_recovery_result.get("serve_units") or {}).get("failed") or [])
|
|
# Deliberately NOT merged into ``restarted_services`` (gateway vocabulary feeding the
|
|
# fleet probe); serve coverage lives in the recovery result/receipt. A serve/dashboard
|
|
# still the SAME pre-update process is live on old code (unreachable by `gateway
|
|
# restart`): recovery may not claim success while one remains, and must never kill
|
|
# one (manual/Desktop serves have no relaunch authority).
|
|
_stale_runtime_rows = _surviving_pre_update_serve_runtimes(_pre_update_plan)
|
|
_recovery_result["stale_runtimes"] = _stale_runtime_rows
|
|
# Only systemd-VERIFIED outcomes claim coverage; a relaunch that merely exited 0
|
|
# ("relaunch_attempted") was never observed and must not clear incomplete.
|
|
_recovery_verified = set(_recovery_result.get("verified") or [])
|
|
out.relaunched_profiles.extend(
|
|
profile for profile in sorted(_recovery_verified) if profile not in out.relaunched_profiles
|
|
)
|
|
if _abort_recovery_is_complete(
|
|
planned_gateway_profiles=_planned_gateway_profiles,
|
|
covered_gateway_profiles=_already_restarted_profiles | _recovery_verified,
|
|
recovery_result=_recovery_result,
|
|
stale_runtime_rows=_stale_runtime_rows,
|
|
):
|
|
# Fresh child is terminal; the fleet-version matrix stays the authoritative
|
|
# read-back before success is declared. Desktop-owned survivors (the only rows
|
|
# that can remain here) are named, not owed. See #111494.
|
|
out.incomplete = False
|
|
_warn_stale_serve_runtimes(_stale_runtime_rows)
|
|
elif (
|
|
_restart_phase_failure_is_incomplete(_surviving, out.pre_restart_gateway_pids)
|
|
or _owed_stale_serve_rows(_stale_runtime_rows)
|
|
or _serve_units_failed
|
|
):
|
|
out.incomplete = True
|
|
_warn_gateway_restart_phase_aborted(e, _surviving)
|
|
_warn_stale_serve_runtimes(_stale_runtime_rows)
|
|
if gateway_mode:
|
|
_write_gateway_update_exit_code(False)
|
|
out.record_receipt(phase_error=str(e), fresh_recovery=_recovery_result)
|
|
|
|
|
|
def _gateway_drain_budget() -> float:
|
|
"""Seconds a drain-first (SIGUSR1) restart may wait for a gateway to exit; 45s floor."""
|
|
try:
|
|
from hermes_cli.gateway import _get_restart_exit_wait_budget
|
|
return max(float(_get_restart_exit_wait_budget()), 45.0)
|
|
except Exception:
|
|
return 45.0
|
|
|
|
|
|
def _restart_gateway_fleet_after_update(_pre_update_plan, gateway_mode: bool):
|
|
"""Restart every running gateway (systemd, launchd, manual) onto the pulled code.
|
|
|
|
Never raises: a phase abort runs fresh-child recovery and fails closed unless
|
|
every planned gateway is verifiably covered.
|
|
"""
|
|
from hermes_cli.update_cmd import _m, _write_gateway_update_exit_code
|
|
# All bookkeeping is declared before the try so abort recovery and fleet reconciliation
|
|
# can read it even if the phase raises early. ``pre_restart_gateway_pids`` stays empty
|
|
# until we are about to stop/drain, so an early exception has nothing to fail closed on,
|
|
# while a failure after stopping a discovered gateway fails closed on an empty survivor probe.
|
|
out = _GatewayRestartOutcome(
|
|
incomplete=False, phase_errors=[], pre_restart_gateway_pids=[], restarted_services=[], failed_or_stale_units=[],
|
|
relaunched_profiles=[], externally_supervised_profiles=[], killed_pids=set(),
|
|
)
|
|
# Scope-qualified twin (``user/hermes-serve`` vs ``system/hermes-serve`` are different
|
|
# processes; abort recovery needs WHICH settled). Bare names stay in
|
|
# ``restarted_services`` for the fleet probe, receipt and summary.
|
|
# Snapshot of gateways running before we touch anything. Stays empty until we successfully import the
|
|
# probe and are about to stop/drain — so an exception raised before we touch any gateway keeps this
|
|
# empty (nothing to fail closed on), while a failure after we have stopped a discovered gateway lets the
|
|
# handler fail closed on an empty survivor probe rather than reporting a clean update (#78574).
|
|
# Declared outside the restart try/except below (and never reset to None) so it's always safe to read
|
|
# afterwards even if that block raises before reaching its own restart bookkeeping — needed to forward
|
|
# already-restarted units to ``_refresh_dashboard_after_update`` (review on #83595).
|
|
restarted_scoped_units: set = set()
|
|
|
|
try:
|
|
# Every gateway helper the phase needs is imported up front so a broken gateway
|
|
# module aborts into recovery BEFORE any unit is touched.
|
|
from hermes_cli.gateway import ( # noqa: F401
|
|
is_macos,
|
|
find_gateway_pids,
|
|
find_profile_gateway_processes,
|
|
_prepare_profile_gateway_update_restart,
|
|
_get_service_pids,
|
|
_wait_for_gateway_exit,
|
|
)
|
|
# Drain budget covers ``restart_after_turn_timeout`` and stop()'s
|
|
# ``restart_drain_timeout`` so a gateway waiting on a turn isn't hard-killed;
|
|
# units without SIGUSR1 wiring just time out into ``systemctl restart``.
|
|
_drain_budget = _gateway_drain_budget()
|
|
|
|
# Snapshot before any stop/drain so an empty survivor probe reads as "stopped
|
|
# and never came back", not "nothing was running"; None fails closed.
|
|
try:
|
|
out.pre_restart_gateway_pids = _scoped_manual_gateway_pids(find_gateway_pids(all_profiles=True), quiet=True)
|
|
except Exception:
|
|
out.pre_restart_gateway_pids = None
|
|
|
|
_restart_systemd_gateway_units(
|
|
out.restarted_services, out.failed_or_stale_units, restarted_scoped_units, _drain_budget,
|
|
out.self_restart_pending_pids,
|
|
)
|
|
|
|
# macOS: EVERY ai.hermes.gateway* LaunchAgent (systemd parity).
|
|
if is_macos():
|
|
with suppress(FileNotFoundError, ImportError):
|
|
_restart_macos_launchd_gateways(
|
|
out.restarted_services, out.failed_or_stale_units, _drain_budget,
|
|
self_restart_pending=out.self_restart_pending_pids,
|
|
)
|
|
|
|
_restart_manual_gateways(out, _drain_budget)
|
|
|
|
if out.failed_or_stale_units:
|
|
out.incomplete = True
|
|
if gateway_mode:
|
|
_write_gateway_update_exit_code(False)
|
|
_warn_incomplete_gateway_fleet_restart(out.failed_or_stale_units)
|
|
out.record_receipt()
|
|
_force_kill_stuck_gateways(out.killed_pids)
|
|
|
|
except Exception as e:
|
|
_recover_after_restart_phase_abort(
|
|
e, _pre_update_plan, out, gateway_mode=gateway_mode, restarted_scoped_units=restarted_scoped_units
|
|
)
|
|
|
|
out.restarted_scoped_units = set(restarted_scoped_units)
|
|
if not out.incomplete:
|
|
# Stamp the HOST obligation so every other profile's CLI knows this update's restart
|
|
# already happened; without it each profile re-kills the one shared multiplexer.
|
|
from hermes_cli.update_host_obligation import mark_host_restart_completed
|
|
mark_host_restart_completed(_restart_identity_sha())
|
|
return out
|
|
|
|
|
|
def _print_legacy_units_warning() -> None:
|
|
"""Legacy hermes.service fights hermes-gateway.service over the bot token; warn on
|
|
every update until migrated."""
|
|
from hermes_cli.gateway import (has_legacy_hermes_units, _find_legacy_hermes_units, supports_systemd_services)
|
|
if not (supports_systemd_services() and has_legacy_hermes_units()):
|
|
return
|
|
print()
|
|
print("⚠ Legacy Hermes gateway unit(s) detected:")
|
|
for name, path, is_sys in _find_legacy_hermes_units():
|
|
scope = "system" if is_sys else "user"
|
|
print(f" {path} ({scope} scope)")
|
|
print()
|
|
print(" These pre-rename units (hermes.service) fight the current")
|
|
print(" hermes-gateway.service for the bot token and cause SIGTERM")
|
|
print(" flap loops. Remove them with:")
|
|
print()
|
|
print(" hermes gateway migrate-legacy")
|
|
print()
|
|
print(" (add `sudo` if any are in system scope)")
|
|
|
|
|
|
def _collect_fleet_snapshot(restart, rows_expected: bool) -> list:
|
|
"""Fleet version rows, polled over a bounded settle window when runtimes are expected.
|
|
|
|
Gateways need time to rewrite gateway_state.json; Windows resumes DETACHED (~10s boot),
|
|
so a single 2s sleep reported "no rows" on healthy resumes. A "down" row may be a
|
|
detached replacement still booting: poll until none remain or the deadline passes.
|
|
Pre-restart PIDs make a gateway stopped WITHOUT verified replacement a DOWN row (exit 1)
|
|
instead of no row at all. An ``unknown`` row whose pid is NOT a pre-restart pid is a successor
|
|
that has not published its code identity yet (a relaunched gateway can sit ~10s between process
|
|
start and its first runtime-status write, #112634) — keep polling; at the deadline it is flagged
|
|
``identity_pending`` so the matrix does not call it a pre-stamping gateway.
|
|
"""
|
|
from hermes_cli.update_receipt import collect_fleet_versions
|
|
pending = getattr(restart, "self_restart_pending_pids", None) or None
|
|
if not rows_expected:
|
|
return collect_fleet_versions(
|
|
pre_restart_pids=restart.pre_restart_gateway_pids, self_restart_pending=pending)
|
|
pre_pids = restart.pre_restart_gateway_pids
|
|
_fleet_deadline = _time.monotonic() + _FLEET_PROBE_SETTLE_TIMEOUT_SECONDS
|
|
while True:
|
|
_time.sleep(2.0)
|
|
snapshot = collect_fleet_versions(pre_restart_pids=pre_pids, self_restart_pending=pending)
|
|
unstamped = [row for row in snapshot if _fleet_row_identity_pending(row, pre_pids)]
|
|
if snapshot and not unstamped and not any(row.get("state") == "down" for row in snapshot):
|
|
return snapshot
|
|
if _time.monotonic() >= _fleet_deadline or _restarted_units_gone(
|
|
getattr(restart, "restarted_scoped_units", ())):
|
|
for row in unstamped:
|
|
row["identity_pending"] = True
|
|
return snapshot
|
|
|
|
|
|
def _fleet_row_identity_pending(row: dict, pre_restart_pids) -> bool:
|
|
"""An ``unknown`` row with no sha from a pid that did not exist at update start: a relaunched
|
|
gateway still booting, not a gateway that predates version stamping. A surviving pre-restart pid
|
|
(or no pid snapshot at all) is settled as-is — waiting cannot change what it publishes."""
|
|
if row.get("state") != "unknown" or row.get("code_sha"):
|
|
return False
|
|
if pre_restart_pids is None:
|
|
return False
|
|
return row.get("pid") not in {int(p) for p in pre_restart_pids if isinstance(p, int)}
|
|
|
|
|
|
def _restarted_units_gone(scoped_units) -> bool:
|
|
"""True when every restarted systemd unit is LOADED in its scope and neither active nor
|
|
activating: the successor died, nothing will publish a state stamp, so the settle poll should fail
|
|
closed now instead of at the deadline. Anything inconclusive keeps waiting: no units, systemctl
|
|
missing/slow, or ``LoadState=not-found`` — a unit name asked in a scope that does not own it
|
|
answers ``inactive`` exactly like a dead unit (#112466), so only a loaded unit can prove death."""
|
|
if not scoped_units:
|
|
return False
|
|
scope_cmds = dict(_SYSTEMD_SCOPES)
|
|
for scoped in scoped_units:
|
|
scope, _, name = scoped.partition("/")
|
|
try:
|
|
stdout = _systemctl(scope_cmds[scope] + ["show", "-p", "LoadState,ActiveState", name], timeout=5).stdout
|
|
except (KeyError, FileNotFoundError, subprocess.TimeoutExpired):
|
|
return False
|
|
props = dict(line.split("=", 1) for line in stdout.splitlines() if "=" in line)
|
|
if props.get("LoadState") != "loaded":
|
|
return False
|
|
if props.get("ActiveState") in ("active", "activating", "reloading"):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _verify_fleet_after_update(restart, *, _pre_update_plan, _windows_gateway_resume, update_complete):
|
|
"""Post-restart verification: legacy-unit warning, dashboard cleanup, stale serve
|
|
probe, fleet version matrix, plan-vs-execution reconciliation, receipt finalize.
|
|
|
|
Exits 1 (leaving ``fleet_restart_pending`` for the next catch-up) when any gateway
|
|
may still be stale; otherwise clears the marker. A failed SQLite verdict also
|
|
exits 1, without retaining a fulfilled fleet-restart obligation.
|
|
"""
|
|
from hermes_cli.update_cmd import (
|
|
_m, _surviving_pre_update_serve_runtimes, _warn_stale_serve_runtimes,
|
|
)
|
|
from hermes_cli.update_cmd_maint import _refresh_dashboard_after_update
|
|
with _best_effort('Legacy unit check during update failed: %s'):
|
|
_print_legacy_units_warning()
|
|
|
|
# Restart a managed dashboard via systemd or stop stale manual ones (raw-killing
|
|
# a systemd-owned PID reads as clean stop and leaves the Cloudflare origin dead).
|
|
# Already-restarted units aren't redone.
|
|
_refresh_dashboard_after_update(already_restarted_units=set(restart.restarted_services))
|
|
|
|
# Success-path twin of the abort-recovery probe: the restart phase only touches
|
|
# units, so a unit-less `hermes serve` keeps stale sys.modules. Runs AFTER
|
|
# dashboard cleanup so a respawned manual dashboard isn't a survivor. Rows feed
|
|
# reconciliation (survivor → exit 1); ``None`` = probe failed, stays fail-closed.
|
|
# Check if any pre-update serve/dashboard runtimes survived on pre-update code generations (#100479).
|
|
# This is the SUCCESS-path twin of the abort-recovery probe above: the restart phase only restarts
|
|
# units, so an sshd-spawned `serve --isolated` or a manual `hermes serve` (no unit) is left running its
|
|
# pre-update sys.modules graph — and its cron ticker keeps firing agent jobs that ImportError on every
|
|
# symbol added in the pulled range. The rows also feed the plan-vs-execution reconciliation below, so a
|
|
# survivor is escalated (exit 1) instead of merely printed.
|
|
_stale_serve_rows: "list | None" = None
|
|
with _best_effort('Failed to check for surviving serve runtimes: %s'):
|
|
_stale_serve_rows = _surviving_pre_update_serve_runtimes(_pre_update_plan)
|
|
if _stale_serve_rows:
|
|
_warn_stale_serve_runtimes(_stale_serve_rows)
|
|
|
|
print()
|
|
print("Tip: You can now select a provider and model:")
|
|
print(" hermes model # Select provider and model")
|
|
|
|
# Compare every live gateway's stamped code_sha against the fresh checkout
|
|
# instead of assuming the restart phase worked.
|
|
# Phase 1 (#91277): post-update fleet version verification.
|
|
_fleet_snapshot: list = []
|
|
with _best_effort('Fleet version verification failed: %s'):
|
|
from hermes_cli.update_receipt import print_fleet_version_matrix
|
|
# Cross-platform "rows expected" signal: (restarted_services or killed_pids)
|
|
# never fires on Windows (pause/resume populates neither), so a healthy
|
|
# resumed gateway yielded zero rows and exit 0.
|
|
# See #93406.
|
|
# A gateway stopped WITHOUT a successor ("Restart manually") publishes no row by design,
|
|
# so it must not count as an expected one — otherwise an update whose only live gateways
|
|
# were unmapped exits 1 with "no rows" after correctly stopping them.
|
|
_pre_restart, _killed = restart.fleet_probe_signals()
|
|
_fleet_rows_expected = _m()._fleet_probe_expected_runtimes(
|
|
_pre_update_plan, _pre_restart, _windows_gateway_resume, restart.restarted_services, _killed,
|
|
)
|
|
_fleet_snapshot = _collect_fleet_snapshot(restart, _fleet_rows_expected)
|
|
if print_fleet_version_matrix(_fleet_snapshot):
|
|
restart.incomplete = True
|
|
# A proven-stale survivor must not keep running (its ticker yields every tick and
|
|
# nothing else restarts it, #117275): hand it to the drain-first restart path.
|
|
from hermes_cli.update_cmd_stale_survivors import signal_stale_fleet_survivors
|
|
signal_stale_fleet_survivors(_fleet_snapshot, restart, _gateway_drain_budget())
|
|
elif not _fleet_snapshot and _fleet_rows_expected:
|
|
# collect_fleet_versions() swallows every failure, so zero rows with
|
|
# expected runtimes is indistinguishable from health — fail (partial, exit 1).
|
|
print(
|
|
# Fleet probe returned zero rows even though at least one gateway runtime was (or may have
|
|
# been) live pre-update — POSIX restart bookkeeping, the pre-restart PID snapshot, the
|
|
# pre-update plan inventory, or the Windows pause/resume token all count as that signal.
|
|
# Every failure path inside collect_fleet_versions() is swallowed via logger.debug(), so an
|
|
# empty list is indistinguishable from a healthy fleet in the current output. Treat it as
|
|
# verification failure so the receipt records "partial" and the exit code is 1 (#93406).
|
|
"\n⚠ Fleet version check returned no rows even though"
|
|
" gateway runtimes were expected — verification incomplete."
|
|
)
|
|
restart.incomplete = True
|
|
|
|
# Every runtime the PLAN saw must appear in restart bookkeeping; an
|
|
# unaccounted one is a silent miss and escalates like a STALE/DOWN row.
|
|
with _best_effort('Runtime-outcome reconciliation failed: %s'):
|
|
# An unaccounted runtime is the silent-miss class (a platform branch re-discovered its own targets
|
|
# and skipped one the inventory knew about) — escalate it exactly like a STALE/DOWN fleet row. See
|
|
# #91277.
|
|
if _pre_update_plan is not None and _pre_update_plan.runtimes:
|
|
from hermes_cli.update_inventory import (match_runtime_outcomes, report_unaccounted_runtimes)
|
|
_runtime_outcomes = match_runtime_outcomes(
|
|
_pre_update_plan,
|
|
restarted_services=restart.restarted_services,
|
|
relaunched_profiles=restart.relaunched_profiles,
|
|
externally_supervised_profiles=restart.externally_supervised_profiles,
|
|
killed_pids=restart.killed_pids,
|
|
failed_units=restart.failed_or_stale_units,
|
|
# Serve/dashboard reconcile by incarnation liveness, not unit names.
|
|
# See #100479.
|
|
stale_serve_pids=(
|
|
{row.get("pid") for row in _stale_serve_rows}
|
|
if _stale_serve_rows is not None
|
|
else None
|
|
),
|
|
)
|
|
from dataclasses import asdict
|
|
from hermes_cli.update_serve_obligations import defer_manual_serve
|
|
|
|
for runtime, outcome in zip(_pre_update_plan.runtimes, _runtime_outcomes):
|
|
if outcome["outcome"] == "unaccounted" and defer_manual_serve(asdict(runtime), require_alive=True):
|
|
outcome["outcome"] = "deferred"
|
|
if report_unaccounted_runtimes(_runtime_outcomes):
|
|
restart.incomplete = True
|
|
with suppress(Exception):
|
|
import hermes_cli.update_receipt as _ur
|
|
_active = _ur._current.get()
|
|
if _active is not None:
|
|
_active.data["runtime_outcomes"] = _runtime_outcomes
|
|
|
|
with _best_effort('Update receipt finalize failed: %s'):
|
|
from hermes_cli.update_receipt import finalize_update_receipt
|
|
_receipt_path = finalize_update_receipt(
|
|
"partial" if restart.incomplete or not update_complete else "success",
|
|
fleet=_fleet_snapshot,
|
|
)
|
|
if _receipt_path is not None:
|
|
logger.info("Update receipt written: %s", _receipt_path)
|
|
|
|
if restart.incomplete:
|
|
# Code updated but a gateway may still run stale modules: fail so automation
|
|
# doesn't treat the fleet as healthy; leave the pending marker for catch-up.
|
|
sys.exit(1)
|
|
_clear_fleet_restart_pending_marker()
|
|
if not update_complete:
|
|
# Fleet caught up, but the independently checked SQLite runtime is unsafe.
|
|
sys.exit(1)
|
|
# Fleet is healthy on the new code: fold per-profile gateways into one multiplexer when nothing
|
|
# blocks it (deterministic; never prompts), else print the blockers and the one-liner to run later.
|
|
with _best_effort('Multiplex auto-migration after update failed: %s'):
|
|
from hermes_cli.gateway_migrate import maybe_auto_migrate_after_update
|
|
maybe_auto_migrate_after_update()
|
|
|
|
|
|
def _restart_phase_failure_is_incomplete(surviving, pre_restart_pids) -> bool:
|
|
"""Whether an escaped restart-phase exception must fail the update.
|
|
|
|
Fail closed unless provably safe: ``surviving`` None (unprobeable) or non-empty →
|
|
stale. ``[]`` proves safety ONLY if nothing ran beforehand; a pre-restart gateway
|
|
(``pre_restart_pids`` non-empty or None) now gone was stopped unverified.
|
|
|
|
* ``surviving is None`` — the survivor probe could not determine state (typically the freshly-pulled
|
|
``hermes_cli.gateway`` no longer imports, one of the ways the phase aborts). That is proof-of-safety
|
|
ONLY when nothing was running before we touched anything. If a gateway was discovered pre-restart
|
|
(``pre_restart_pids`` non-empty, or ``None`` meaning the pre-state could not be read), it was stopped
|
|
without a verified replacement, so we still fail closed (#78574).
|
|
"""
|
|
if surviving is None or surviving:
|
|
return True
|
|
return pre_restart_pids is None or bool(pre_restart_pids)
|
|
|
|
|
|
def _fleet_probe_expected_runtimes(
|
|
pre_update_plan, pre_restart_pids, windows_resume_token, restarted_services, killed_pids,
|
|
) -> bool:
|
|
"""Whether the post-update fleet probe should have produced rows.
|
|
|
|
``collect_fleet_versions()`` swallows every failure and an empty matrix prints as
|
|
healthy, so zero rows is only proof-of-safety when NOTHING says a gateway existed
|
|
pre-update. Signals: ``restarted_services``/``killed_pids``; ``pre_restart_pids``
|
|
non-empty or None (same contract as ``_restart_phase_failure_is_incomplete``); plan
|
|
inventoried ≥1 ``kind == "gateway"`` runtime. ``windows_resume_token`` is deliberately EXCLUDED: it is
|
|
pause/resume bookkeeping, not an inventory, and its entries don't map to probe rows
|
|
(``unmapped`` Scheduled-Task gateways never publish gateway_state.json; a paused
|
|
profile resumes DETACHED). Counting it made every Windows update that paused a
|
|
gateway exit 1 after a long silent wait; a live pre-update Windows gateway is already
|
|
covered by ``pre_restart_pids`` and the plan. The same condition gates the settle sleep.
|
|
|
|
See #93406.
|
|
See #78574.
|
|
See #93406.
|
|
"""
|
|
del windows_resume_token # excluded on purpose — see docstring
|
|
# See #93406.
|
|
if restarted_services or killed_pids:
|
|
return True
|
|
if pre_restart_pids is None or pre_restart_pids:
|
|
return True
|
|
with suppress(Exception):
|
|
# Gateway-kind only: serve/dashboard plan records never publish a gateway_state.json
|
|
# row, so a dashboard-only plan cannot ground a rows-expected verdict (#97332).
|
|
if pre_update_plan is not None and any(
|
|
getattr(runtime, "kind", None) == "gateway" for runtime in pre_update_plan.runtimes
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _wait_for_service_active(scope_cmd_: list, svc_name_: str, timeout: float = 10.0) -> bool:
|
|
"""Poll ``systemctl is-active`` (0.5s) up to ``timeout``: the Stopped -> Started
|
|
transition isn't instantaneous, so a one-shot check falsely reports down."""
|
|
deadline = _time.monotonic() + max(timeout, 0.5)
|
|
while True:
|
|
with suppress(FileNotFoundError, subprocess.TimeoutExpired):
|
|
_verify = _systemctl(scope_cmd_ + ["is-active", svc_name_], timeout=5)
|
|
if _verify.stdout.strip() == "active":
|
|
return True
|
|
if _time.monotonic() >= deadline:
|
|
return False
|
|
_time.sleep(0.5)
|
|
|
|
|
|
_RESTART_SEC_UNITS = (("ms", 0.001), ("us", 0.000001), ("min", 60.0), ("s", 1.0))
|
|
|
|
|
|
def _service_restart_sec(scope_cmd_: list, svc_name_: str, default: float = 0.0) -> float:
|
|
"""Read the unit's ``RestartUSec`` in seconds. ``is-active`` pollers must wait
|
|
>= RestartSec + slack or they give up *during* the cooldown and misreport."""
|
|
try:
|
|
_show = _systemctl(scope_cmd_ + ["show", svc_name_, "--property=RestartUSec", "--value"], timeout=5)
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
return default
|
|
raw = (_show.stdout or "").strip()
|
|
# Values like "30s", "100ms", "1min 30s", "infinity"; on any miss return default.
|
|
if not raw or raw == "infinity":
|
|
return default
|
|
total = 0.0
|
|
matched = False
|
|
for part in raw.split():
|
|
for _suf, _mult in _RESTART_SEC_UNITS:
|
|
if part.endswith(_suf):
|
|
with suppress(ValueError):
|
|
total += float(part[: -len(_suf)]) * _mult
|
|
matched = True
|
|
break
|
|
return total if matched else default
|