FLEET: one host-scoped update-restart obligation, one restart per host

One host runs one multiplexing gateway, but the update pipeline still treated the pull->restart obligation, enumerated units, recovery payloads and the planned-restart notice as per-profile. Two profiles updating meant two outages of the same process, and a served profile's channels were never told. - hermes_cli/update_host_obligation.py: new host-scoped obligation record in gateway.host_rendezvous.host_state_dir() (host-update-restart.json), plus the unit->live-MainPID collapse rule. The legacy per-home marker stays readable and clearable so an in-flight obligation is still discharged. - update_cmd_fleet: arm/clear/read the host record; the catch-up restart is idempotent per host (a completed restart onto the checkout SHA is never repeated); leftover per-profile units resolving to one MainPID restart once. - update_restart_recovery: payload profiles served by one host process are one restart target, reported under "covered". - gateway notices: owed targets and the online notice span every served profile's home channels; the marker survives until each was reached.
2026-11-06 05:17:15 +00:00 · 2026-09-21 07:14:37 -07:00
1 parent 8f7b3b48fc
commit 061195fac1
@@ -3492,6 +3492,9 @@ class GatewayRunner(
        self._session_db_init_error: Optional[str] = None
        # Non-default profiles' adapters by profile then Platform; self.adapters stays the default's map.
        self._profile_adapters: Dict[str, Dict[Platform, BasePlatformAdapter]] = {}
+        # Each SERVED profile's gateway config, as loaded once by ``_load_secondary_profile_config``.
+        # ``self.config`` is only the launch profile's: anything host-wide (restart notices) needs these.
+        self._profile_configs: Dict[str, Any] = {}
        self._warn_if_docker_media_delivery_is_risky()
        _gateway_runner_ref = _weakref.ref(self)

@@ -1047,6 +1047,12 @@ class GatewayAdapterLifecycleMixin:
        """Create+connect one profile's adapters under its runtime scope."""
        from gateway.run import _platform_has_bot_credential, _profile_runtime_scope
        profile_cfg = await self._load_secondary_profile_config(profile_name, profile_home)
+        # Keep the served profile's config: host-wide passes (planned-restart notices) must reach
+        # every served profile's home channels, and this is the only place it is loaded.
+        configs = getattr(self, "_profile_configs", None)
+        if configs is None:
+            configs = self._profile_configs = {}
+        configs[profile_name] = profile_cfg
        multiplex = self._multiplex_on()
        profile_map = self._profile_adapters.setdefault(profile_name, {})
        connected = 0
@@ -33,6 +33,17 @@ _UPDATE_FAILED_NOTICE = (
    "host to see the full error, or try /update again later.")


+def _served_notice_target_key(profile: Optional[str], platform_value: str, chat_id, thread_id) -> tuple:
+    """Notice-dedupe key for one SERVED profile's home channel.
+
+    A secondary uses the ``<profile>:<platform>`` key convention the runtime status already
+    stamps in ``gateway_state.json``; the launch profile keeps the bare platform value so a
+    marker written before this change still matches its delivered targets.
+    """
+    return _notice_target_key(
+        platform_value if profile is None else f"{profile}:{platform_value}", chat_id, thread_id)
+
+
 def _update_output_tail(output: str, limit: int) -> str:
    """Last ``limit`` chars of an update log, prefixed with an ellipsis when cut."""
    return output if len(output) <= limit else "…" + output[-limit:]
@@ -785,6 +796,37 @@ class GatewayNotificationsMixin:
                continue
            yield platform, platform_cfg, home, transport

+    def _served_home_channel_configs(self):
+        """``(profile, platform, platform_cfg)`` for every SERVED profile's configured home channel.
+
+        ``self.config`` is the launch profile's alone, but one host process multiplexes every
+        profile, so a host-wide notice built from it silently skips the others' channels. The
+        secondary configs are the ones ``_load_secondary_profile_config`` already cached at
+        adapter start; ``profile`` is ``None`` for the launch profile.
+        """
+        for platform, platform_cfg in self.config.platforms.items():
+            yield None, platform, platform_cfg
+        for profile, profile_cfg in (getattr(self, "_profile_configs", None) or {}).items():
+            for platform, platform_cfg in profile_cfg.platforms.items():
+                yield profile, platform, platform_cfg
+
+    def _served_home_channel_transports(self):
+        """``(profile, platform, platform_cfg, home, transport)`` for every served profile's home
+        channel with a live transport — the launch profile's (``profile`` ``None``) first."""
+        from gateway.delivery import resolve_delivery_transport
+        for platform, platform_cfg, home, transport in self._home_channel_transports():
+            yield None, platform, platform_cfg, home, transport
+        for profile, profile_cfg in (getattr(self, "_profile_configs", None) or {}).items():
+            adapters = (getattr(self, "_profile_adapters", None) or {}).get(profile) or {}
+            for platform, platform_cfg in profile_cfg.platforms.items():
+                home = platform_cfg.home_channel
+                if not home or not home.chat_id:
+                    continue
+                transport = resolve_delivery_transport(platform, profile_cfg, adapters)
+                if transport is None:
+                    continue
+                yield profile, platform, platform_cfg, home, transport
+
    async def _send_home_channel_message(self, platform, home, transport, message: str, failure_fmt: str) -> bool:
        """Best-effort send to one home channel; True on success, failures logged with ``failure_fmt``."""
        from gateway.run import _non_conversational_metadata
@@ -856,8 +898,9 @@ class GatewayNotificationsMixin:
                # Owed targets come from config, not live transports: a removed home or an opt-out
                # (gateway_restart_notification=false) must not keep the marker alive forever.
                owed = {
-                    _notice_target_key(platform.value, cfg.home_channel.chat_id, cfg.home_channel.thread_id)
-                    for platform, cfg in self.config.platforms.items()
+                    _served_notice_target_key(
+                        profile, platform.value, cfg.home_channel.chat_id, cfg.home_channel.thread_id)
+                    for profile, platform, cfg in self._served_home_channel_configs()
                    if cfg.home_channel and cfg.home_channel.chat_id and cfg.gateway_restart_notification
                }
                delivered |= await self._send_home_channel_startup_notifications(skip_targets=delivered)
@@ -872,10 +915,12 @@ class GatewayNotificationsMixin:
    async def _send_home_channel_startup_notifications(
        self, *, skip_targets: Optional[set[tuple[str, str, Optional[str]]]] = None
    ) -> set[tuple[str, str, Optional[str]]]:
-        """Notify configured home channels that the gateway is back online.
+        """Notify EVERY served profile's configured home channels that the gateway is back online.

-        Best-effort, once per connected platform home channel. ``skip_targets`` lets startup avoid
-        duplicate messages when a more specific restart notification is queued for the same chat.
+        Best-effort, once per (profile, platform) home channel — one host process serves them all,
+        so a notice restricted to the launch profile leaves every other profile's channel silent.
+        ``skip_targets`` lets startup avoid duplicate messages when a more specific restart
+        notification is queued for the same chat.
        """
        delivered: set[tuple[str, str, Optional[str]]] = set()
        skipped = skip_targets or set()
@@ -883,14 +928,14 @@ class GatewayNotificationsMixin:
        free_tier_line = self._free_tier_startup_line()
        if free_tier_line:
            message = f"{message}\n{free_tier_line}"
-        for platform, platform_cfg, home, transport in self._home_channel_transports():
+        for profile, platform, platform_cfg, home, transport in self._served_home_channel_transports():
            if not platform_cfg.gateway_restart_notification:
                logger.info(
                    "Home-channel startup notification suppressed: %s has gateway_restart_notification=false",
                    platform.value,
                )
                continue
-            target = _notice_target_key(platform.value, home.chat_id, home.thread_id)
+            target = _served_notice_target_key(profile, platform.value, home.chat_id, home.thread_id)
            if target in skipped or target in delivered:
                continue
            if await self._send_home_channel_message(
@@ -203,6 +203,13 @@ def _recover_gateway_restart_after_abort(
        return _all_failed()

    verified, relaunch_attempted, failed = sorted(verified), sorted(relaunch_attempted), sorted(failed)
+    covered_map = recovery_result.get("covered")
+    for owner, others in (covered_map if isinstance(covered_map, dict) else {}).items():
+        if isinstance(owner, str) and isinstance(others, list) and others:
+            print(
+                f"  • One host gateway serves {owner} and {', '.join(str(o) for o in others)} — "
+                f"restarted once through {owner}; that restart is their outcome."
+            )
    for names, text in (
        (verified, "  ✓ Restarted supervised gateway(s) in a fresh process (systemd-verified active): "),
        (relaunch_attempted, "  ⚠ Relaunch attempted in a fresh process but not"
@@ -46,40 +46,81 @@ def _write_gateway_update_exit_code(ok: bool) -> None:


 def _fleet_restart_pending_marker_path() -> Path:
-    """HERMES_HOME breadcrumb for a pull that has not yet restarted the fleet."""
+    """LEGACY per-``HERMES_HOME`` breadcrumb. Read-compat only — nothing writes it any more.
+
+    One host runs one multiplexing gateway, so the pull→restart obligation is host-scoped
+    (``hermes_cli/update_host_obligation.py``). An obligation armed by the old per-profile code
+    is still read and cleared here so an in-flight update is discharged after the upgrade.
+    """
    from hermes_cli.update_cmd import get_hermes_home
    return get_hermes_home() / _FLEET_RESTART_PENDING_NAME


 def _write_fleet_restart_pending_marker(*, expected_sha: str = "", runtimes: list[dict] | None = None) -> None:
-    """Drop the pull→restart obligation breadcrumb. Never raises."""
+    """Arm the HOST pull→restart obligation. Never raises."""
    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
-    path = _fleet_restart_pending_marker_path()
-    if _m()._pytest_owns_live_checkout(path.parent):
-        logger.debug("Skipping fleet-restart-pending marker under pytest (live checkout)")
+    from hermes_cli.update_host_obligation import 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
+    write_host_obligation(
+        expected_sha=expected_sha, runtimes=runtimes, profile=_current_profile_name())
+
+
+def _current_profile_name() -> str:
+    """Profile whose CLI armed the obligation (diagnostics only — the record is host-scoped)."""
    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")
-    except OSError as exc:
-        logger.debug("Could not write fleet-restart-pending marker: %s", exc)
+        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:
-    """Remove the pull→restart obligation breadcrumb. Never raises."""
+    """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 obligation_fields
+    fields = obligation_fields()
+    if fields is not None:
+        return fields
+    try:
+        text = _fleet_restart_pending_marker_path().read_text(encoding="utf-8")
+    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
@@ -260,12 +301,9 @@ def _marker_only_restart_obsolete() -> bool:
    from hermes_cli.update_serve_obligations import defer_manual_serve

    try:
-        fields = {}
-        for line in _fleet_restart_pending_marker_path().read_text(encoding="utf-8").splitlines():
-            key, value = line.split("=", 1)
-            if key in fields:
-                return False
-            fields[key] = value
+        fields = _obligation_fields()
+        if fields is None:
+            return False
        expected_sha = fields.get("expected_sha", "").strip()
        inventory = json.loads(fields.get("inventory", "null"))
        owed: set[tuple[str, str]] | None = None
@@ -353,10 +391,9 @@ def _pending_fleet_restart_needed(*, receipt: dict | None = None, pending_manual
        receipt = read_latest_receipt() or {}
    if pending_manual is None:
        pending_manual = retain_receipt_manual_serves(receipt)
-    # A marker owns its inventory; latest.json can belong to an older update.
-    with suppress(OSError):
-        if _fleet_restart_pending_marker_path().is_file():
-            return not _marker_only_restart_obsolete()
+    # 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
@@ -373,10 +410,9 @@ def _update_owes_fleet_restart(*, receipt: dict | None = None, pending_manual: l
        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 marker's inventory.
-    with suppress(OSError):
-        if _fleet_restart_pending_marker_path().is_file():
-            return not _marker_only_restart_obsolete()
+    # 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
@@ -439,28 +475,73 @@ def _needs_sudo(scope: str) -> bool:
    )


+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:
-    """Best-effort ``systemctl restart`` of every hermes-gateway/serve unit."""
+    """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
-
-        def process_unit(svc_name: str, _scope=scope, _cmd=scope_cmd) -> None:
-            manage_cmd = list(_cmd) + ["--no-ask-password"]
-            if _needs_sudo(_scope):
-                manage_cmd = ["sudo", "-n"] + manage_cmd
-            result = _systemctl_reset_and_restart(manage_cmd, svc_name, scope_cmd=_cmd)
-            if result.returncode != 0 or not _wait_for_service_active(_cmd, svc_name):
-                failed.append(svc_name)
-
        _for_each_systemd_gateway_unit(
            result.stdout,
-            process_unit=process_unit,
+            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]
+        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)

@@ -490,9 +571,18 @@ def _run_pending_fleet_restart() -> bool:

    True when all discovered targets recovered (or none exist); False if incomplete.

+    Idempotent per HOST: one process multiplexes every profile, so the second profile's
+    ``hermes update`` must attach to the first one's restart instead of killing the shared
+    gateway again (the obligation record carries the proof).
+
    See #95294.
    """
    from hermes_cli.update_cmd import _m
+    from hermes_cli.update_host_obligation import host_restart_already_completed, mark_host_restart_completed
+    checkout_sha = _current_checkout_sha()
+    if host_restart_already_completed(checkout_sha):
+        print("  ✓ This host's gateway was already restarted for this update — not restarting it again.")
+        return True
    print("→ Restarting gateways left on pre-update code...")
    # Warn if legacy Hermes gateway unit files are still installed. When both hermes.service (from a
    # pre-rename install) and the current hermes-gateway.service are enabled, they SIGTERM-fight for the
@@ -559,6 +649,9 @@ def _run_pending_fleet_restart() -> bool:
        if failed:
            _warn_incomplete_gateway_fleet_restart(failed)
            return False
+        # 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.
+        mark_host_restart_completed(checkout_sha or "")
        print("  ✓ Pending fleet restart completed.")
        return True
    except Exception as exc:
@@ -0,0 +1,219 @@
+"""Host-scoped update→restart obligation for ``hermes update``.
+
+Multiplex-only (Teknium ruling): exactly ONE ``hermes gateway run`` per host serves every
+profile, so "this pull still owes the fleet a restart" is a property of the HOST, not of one
+profile's ``HERMES_HOME``. The legacy ``$HERMES_HOME/fleet_restart_pending`` marker was
+per-home: ``hermes -p coder update`` armed and cleared coder's copy while restarting the
+SHARED process, and every other profile's CLI could neither see nor discharge that obligation
+— it simply armed its own and re-killed the same host process.
+
+The record therefore lives beside the host rendezvous record, in
+:func:`gateway.host_rendezvous.host_state_dir` (``$HERMES_GATEWAY_LOCK_DIR`` else
+``$XDG_STATE_HOME/hermes/gateway-locks``) — the one cross-profile, per-OS-user state root the
+tree already has. It is written once per host, read by every profile's CLI, and cleared once.
+
+The same "one host process, not one per profile" identity is what
+:func:`collapse_units_to_host_processes` applies to enumerated systemd units: leftover
+per-profile ``hermes-gateway-<p>.service`` units on a multiplexed host all point at the same
+live ``MainPID``, so restarting each one restarts the host process N times.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+from pathlib import Path
+from typing import Any, Callable, Iterable, Optional
+
+logger = logging.getLogger("hermes_cli.update_cmd")
+
+#: One file per OS user, beside ``host-gateway.json`` / ``host-serve.json``.
+HOST_OBLIGATION_NAME = "host-update-restart.json"
+
+_RECORD_VERSION = 1
+
+
+def host_obligation_path() -> Optional[Path]:
+    """Path of the host obligation record, or ``None`` when the host state dir is unresolvable."""
+    try:
+        from gateway.host_rendezvous import host_state_dir
+
+        return host_state_dir() / HOST_OBLIGATION_NAME
+    except Exception:  # pragma: no cover - import/env failure must never break the updater
+        logger.debug("Host obligation path unavailable", exc_info=True)
+        return None
+
+
+def read_host_obligation() -> Optional[dict]:
+    """The published obligation record, or ``None`` when absent/corrupt/foreign-versioned."""
+    path = host_obligation_path()
+    if path is None:
+        return None
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, ValueError):
+        return None
+    if not isinstance(payload, dict) or payload.get("version") != _RECORD_VERSION:
+        return None
+    return payload
+
+
+def host_obligation_present() -> bool:
+    """True when the record FILE exists, parseable or not.
+
+    Fail-closed: a corrupt record is an obligation whose terms are unknown, never a discharged
+    one — the restart is still owed and the reader falls back to "no recorded inventory".
+    """
+    path = host_obligation_path()
+    if path is None:
+        return False
+    try:
+        return path.is_file()
+    except OSError:
+        return False
+
+
+def amend_host_obligation(**fields: Any) -> None:
+    """Merge ``fields`` into the armed record (test/diagnostic surface). Never raises."""
+    record = read_host_obligation()
+    path = host_obligation_path()
+    if record is None or path is None:
+        return
+    record.update(fields)
+    try:
+        from utils import atomic_json_write
+
+        atomic_json_write(path, record, mode=0o600)
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.debug("Could not amend host update-restart obligation: %s", exc)
+
+
+def write_host_obligation(
+    *, expected_sha: str = "", runtimes: Optional[list] = None, profile: str = ""
+) -> bool:
+    """Arm the host obligation. True when it was written. Never raises.
+
+    Re-arming from a second profile for the SAME pulled SHA keeps the existing record (and its
+    ``restarted`` proof) instead of resetting it: the host owes one restart, not one per profile.
+    """
+    path = host_obligation_path()
+    if path is None:
+        return False
+    existing = read_host_obligation()
+    if existing is not None and expected_sha and existing.get("expected_sha") == expected_sha:
+        # Same pull, second profile: the host owes ONE restart, so keep the standing record (and
+        # any proof that the restart already happened) rather than resetting it. A later arm that
+        # carries the owed inventory still upgrades it — an inventory-less record owes no set.
+        if runtimes is None:
+            return True
+        inventory = {"version": 1, "runtimes": runtimes}
+        if existing.get("inventory") != inventory:
+            amend_host_obligation(inventory=inventory)
+        return True
+    payload: dict[str, Any] = {
+        "version": _RECORD_VERSION,
+        "started": time.time(),
+        "pid": os.getpid(),
+        "armed_by_profile": profile or "",
+        "expected_sha": expected_sha or "",
+    }
+    if runtimes is not None:
+        payload["inventory"] = {"version": 1, "runtimes": runtimes}
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        from utils import atomic_json_write
+
+        atomic_json_write(path, payload, mode=0o600)
+    except Exception as exc:
+        logger.debug("Could not write host update-restart obligation: %s", exc)
+        return False
+    return True
+
+
+def clear_host_obligation() -> None:
+    """Discharge the obligation for the whole host. Never raises."""
+    path = host_obligation_path()
+    if path is None:
+        return
+    try:
+        path.unlink(missing_ok=True)
+    except OSError as exc:
+        logger.debug("Could not clear host update-restart obligation: %s", exc)
+
+
+def obligation_fields() -> Optional[dict[str, str]]:
+    """The obligation in the legacy ``key=value`` field shape, or ``None`` when unarmed.
+
+    Keeps one parser for both sources: the fields a reader needs (``expected_sha``, the
+    serialized ``inventory``) are identical whether they came from the host record or from an
+    in-flight legacy per-home marker.
+    """
+    record = read_host_obligation()
+    if record is None:
+        return None
+    fields = {"expected_sha": str(record.get("expected_sha") or "")}
+    inventory = record.get("inventory")
+    if inventory is not None:
+        fields["inventory"] = json.dumps(inventory)
+    return fields
+
+
+def mark_host_restart_completed(sha: str) -> None:
+    """Record that the host process was restarted onto ``sha``. Never raises."""
+    record = read_host_obligation()
+    path = host_obligation_path()
+    if record is None or path is None:
+        return
+    record["restarted"] = {"sha": sha or "", "pid": os.getpid(), "at": time.time()}
+    try:
+        from utils import atomic_json_write
+
+        atomic_json_write(path, record, mode=0o600)
+    except Exception as exc:
+        logger.debug("Could not stamp host restart completion: %s", exc)
+
+
+def host_restart_already_completed(sha: Optional[str]) -> bool:
+    """True when THIS host obligation was already restarted onto ``sha``.
+
+    The guard that makes the catch-up restart idempotent per host: a second profile running
+    ``hermes update`` must attach to the first restart's outcome, never kill the shared
+    multiplexer again.
+    """
+    record = read_host_obligation()
+    if record is None or not sha:
+        return False
+    restarted = record.get("restarted")
+    return isinstance(restarted, dict) and str(restarted.get("sha") or "") == sha
+
+
+def collapse_units_to_host_processes(
+    units: Iterable[str], main_pid: Callable[[str], int]
+) -> tuple[list[str], dict[str, str]]:
+    """Split enumerated units into ``(restart, {legacy_unit: covering_unit})``.
+
+    Units resolving to the same live ``MainPID`` are ONE host process; restarting each of them
+    restarts that process N times, which on a multiplexed host is an N-fold outage triggered by
+    leftover per-profile units. A unit with no readable main PID (inactive, unprivileged scope)
+    keeps its own restart: identity that cannot be proved is never collapsed away.
+    """
+    restart: list[str] = []
+    covered: dict[str, str] = {}
+    owner_by_pid: dict[int, str] = {}
+    for unit in units:
+        try:
+            pid = int(main_pid(unit) or 0)
+        except (TypeError, ValueError):
+            pid = 0
+        if pid <= 0:
+            restart.append(unit)
+            continue
+        owner = owner_by_pid.get(pid)
+        if owner is None:
+            owner_by_pid[pid] = unit
+            restart.append(unit)
+        else:
+            covered[unit] = owner
+    return restart, covered
@@ -149,17 +149,90 @@ def _systemd_verified_active(profile: str, *, run: Callable[..., Any]) -> bool:
    )


+def _host_state_dir() -> str:
+    """The path ``gateway.host_rendezvous.host_state_dir()`` resolves, computed locally.
+
+    This module imports no Hermes code at runtime — importing the freshly pulled tree is exactly
+    what aborted the phase that calls us — so the rule is duplicated here rather than shared.
+    """
+    override = os.environ.get("HERMES_GATEWAY_LOCK_DIR")
+    if override:
+        return override
+    state_home = os.environ.get("XDG_STATE_HOME") or ""
+    if not os.path.isabs(state_home):
+        state_home = os.path.join(os.path.expanduser("~"), ".local", "state")
+    return os.path.join(state_home, "hermes", "gateway-locks")
+
+
+def _pid_is_live(pid: int) -> bool:
+    """Liveness of ``pid``: ``psutil`` when importable, else the POSIX signal-0 probe.
+
+    The signal probe is POSIX-only by construction — on Windows ``os.kill(pid, 0)`` sends a real
+    control event and can kill the target — so an unimportable psutil there means "cannot prove".
+    """
+    try:
+        import psutil
+
+        return bool(psutil.pid_exists(pid))
+    except Exception:
+        pass
+    if os.name == "nt":
+        return False
+    try:
+        os.kill(pid, 0)  # windows-footgun: ok — POSIX-only branch, guarded by os.name above
+    except PermissionError:
+        return True
+    except OSError:
+        return False
+    return True
+
+
+def _host_served_profiles() -> set[str]:
+    """Profiles the ONE live host gateway multiplexes, from its rendezvous record.
+
+    Restarting any one of them restarts the same process, so they are a single restart target.
+    Empty (no collapsing, today's per-profile behaviour) when the record is absent, unreadable,
+    dead, or when liveness cannot be probed — a missed collapse costs an extra restart, a wrong
+    one would skip a profile that really has its own process.
+    """
+    try:
+        with open(os.path.join(_host_state_dir(), "host-gateway.json"), encoding="utf-8") as handle:
+            record = json.load(handle)
+    except (OSError, UnicodeDecodeError, ValueError):
+        return set()
+    if not isinstance(record, dict) or record.get("role") != "gateway":
+        return set()
+    pid = record.get("pid")
+    profiles = record.get("profiles")
+    if not isinstance(pid, int) or pid <= 0 or not isinstance(profiles, list) or not _pid_is_live(pid):
+        return set()
+    return {name for name in profiles if isinstance(name, str) and name}
+
+
 def restart_profiles(
    profiles: Iterable[str], *, supervisors: Mapping[str, str] | None = None, run: Callable[..., Any] = subprocess.run
-) -> dict[str, list[str]]:
+) -> dict[str, Any]:
    """Restart the supplied profiles (only ones whose inventory identified a service supervisor).

+    Profiles served by the SAME host gateway process are one restart target: a host multiplexes
+    every profile, so N payload profiles meant N sequential ``gateway restart`` calls, each
+    killing the successor the previous pass had just verified. The group is restarted exactly
+    once through one representative and the rest are reported under ``covered`` with that
+    restart's outcome.
+
    A profile only lands in ``verified`` when its supervisor is systemd and ``systemctl --user is-
    active`` independently confirms the unit after the relaunch command succeeded.
    """
    supervisors = supervisors or {}
-    result: dict[str, list[str]] = {"verified": [], "relaunch_attempted": [], "failed": []}
-    for profile in sorted({p for p in profiles if isinstance(p, str) and p}):
+    result: dict[str, Any] = {"verified": [], "relaunch_attempted": [], "failed": []}
+    requested = sorted({p for p in profiles if isinstance(p, str) and p})
+    served = _host_served_profiles()
+    group = [profile for profile in requested if profile in served]
+    representative = group[0] if len(group) > 1 else None
+    covered = group[1:] if representative else []
+    for profile in requested:
+        if profile in covered:
+            continue
        if not _run_profile_restart(profile, run=run):
            bucket = "failed"
        elif supervisors.get(profile) == "systemd" and _systemd_verified_active(profile, run=run):
@@ -167,6 +240,10 @@ def restart_profiles(
        else:
            bucket = "relaunch_attempted"
        result[bucket].append(profile)
+        if profile == representative:
+            # One process: the representative's observed outcome IS these profiles' outcome.
+            result[bucket].extend(covered)
+    result["covered"] = {representative: covered} if representative else {}
    return result


@@ -534,8 +534,12 @@ def _hermetic_environment(tmp_path, monkeypatch):
    # Per-TEST host-rendezvous dir (see the session-level block at the top): the
    # host gateway/serve record is shared per OS user by design, so without this
    # one test's published owner makes the next test's lifecycle code attach to it.
-    # Skipped when the caller supplied the variable, so an explicit override still works.
+    # HOME is deliberately NOT redirected above, so an unpinned run would read and
+    # write the developer's live ~/.local/state/hermes/gateway-locks.
+    # Skipped when the caller supplied the variable, so an explicit override still
+    # works (tests of the resolution rule itself rely on that).
    if not HOST_LOCK_DIR_AT_CONFTEST_IMPORT:
+        monkeypatch.delenv("XDG_STATE_HOME", raising=False)
        monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "gateway-locks"))
    # Keep the subprocess-surviving isolation marker pointed at THIS test's
    # home (#82770): children spawned by the test inherit it by default, so
@@ -0,0 +1,94 @@
+"""A planned restart notifies EVERY served profile's home channels, not just the launch profile's.
+
+One host process multiplexes every profile, so ``self.config`` — the launch profile's — is not
+the fleet: the owed set and the online notice were both built from it alone, and a secondary
+profile's chat never heard that its gateway had restarted. The marker must also survive until
+every served profile was reached, or the missed channels are lost for good.
+"""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+import gateway.run as gateway_run
+from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig
+from gateway.platforms.base import SendResult
+
+ONLINE_NOTICE = "♻️ Gateway online — Hermes is back and ready."
+
+
+def _adapter():
+    return SimpleNamespace(
+        send_path_degraded=False,
+        send=AsyncMock(return_value=SendResult(success=True, message_id="unit-test-notice")),
+    )
+
+
+def _home_config(platform: Platform, chat_id: str) -> GatewayConfig:
+    return GatewayConfig(
+        platforms={
+            platform: PlatformConfig(
+                enabled=True,
+                gateway_restart_notification=True,
+                home_channel=HomeChannel(platform=platform, chat_id=chat_id, name=chat_id),
+            )
+        }
+    )
+
+
+@pytest.fixture
+def multiplex_runner(tmp_path, monkeypatch):
+    """A host multiplexer: launch profile on Discord, served profile ``coder`` on Telegram."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    runner = object.__new__(gateway_run.GatewayRunner)
+    runner.config = _home_config(Platform.DISCORD, "launch-home")
+    runner.config.sessions_dir = tmp_path / "sessions"
+    runner.adapters = {}
+    runner._profile_configs = {"coder": _home_config(Platform.TELEGRAM, "coder-home")}
+    runner._profile_adapters = {"coder": {}}
+    runner._free_tier_startup_line = Mock(return_value=None)
+    runner._planned_restart_notice_lock = None
+    marker = tmp_path / ".restart_pending.json"
+    marker.write_text("{}", encoding="utf-8")
+    return runner, marker
+
+
+@pytest.mark.asyncio
+async def test_planned_restart_notifies_every_served_profile(multiplex_runner):
+    runner, marker = multiplex_runner
+    launch, coder = _adapter(), _adapter()
+    runner.adapters[Platform.DISCORD] = launch
+    runner._profile_adapters["coder"][Platform.TELEGRAM] = coder
+
+    await runner._replay_pending_planned_restart_notification()
+
+    launch.send.assert_awaited_once()
+    coder.send.assert_awaited_once(), "a served profile's home channel is owed the restart notice"
+    assert coder.send.await_args.args[:2] == ("coder-home", ONLINE_NOTICE)
+    assert not marker.exists(), "every owed target was notified — the obligation is discharged"
+
+
+@pytest.mark.asyncio
+async def test_marker_survives_until_a_served_profile_is_reachable(multiplex_runner):
+    """A served profile whose platform is down at boot keeps the notice owed for its reconnect."""
+    runner, marker = multiplex_runner
+    launch = _adapter()
+    runner.adapters[Platform.DISCORD] = launch
+
+    await runner._replay_pending_planned_restart_notification()
+
+    launch.send.assert_awaited_once()
+    assert marker.exists(), "coder's channel was never notified; the marker must not be consumed"
+    delivered = json.loads(marker.read_text(encoding="utf-8"))["delivered_targets"]
+    assert [target for target in delivered if target[0] == "discord"], "the reached target is recorded"
+
+    coder = _adapter()
+    runner._profile_adapters["coder"][Platform.TELEGRAM] = coder
+    await runner._replay_pending_planned_restart_notification()
+
+    coder.send.assert_awaited_once()
+    assert launch.send.await_count == 1, "a reached home is never notified twice"
+    assert not marker.exists()
@@ -40,13 +40,13 @@ def test_manual_deferral_survives_receipt_rotation(monkeypatch, capsys, kind, co
        with pytest.raises(SystemExit) as exc:
            fleet._verify_fleet_after_update(restart, _pre_update_plan=plan, _windows_gateway_resume=None, node_failures=[], update_complete=True)
        assert exc.value.code == 1
-        assert fleet._fleet_restart_pending_marker_path().exists()
+        assert fleet._fleet_restart_obligation_armed()
        assert update_receipt.read_latest_receipt()["outcome"] == "partial"
        return
    fleet._verify_fleet_after_update(restart, _pre_update_plan=plan, _windows_gateway_resume=None, node_failures=[], update_complete=True)
    receipt = update_receipt.read_latest_receipt()
    assert receipt["runtime_outcomes"][0]["outcome"] == "deferred"
-    assert not fleet._fleet_restart_pending_marker_path().exists()
+    assert not fleet._fleet_restart_obligation_armed()
    assert "hermes-serve.service" not in capsys.readouterr().out
    update_receipt.begin_update_receipt()
    update_receipt.finalize_update_receipt("success", fleet=[])
@@ -50,14 +50,13 @@ def test_pending_marker_requires_complete_systemd_recovery(monkeypatch, tmp_path
    fleet._write_fleet_restart_pending_marker(expected_sha="pending", runtimes=[
        {"kind": "gateway", "profile": profile} for profile in ("one", "two")
    ])
-    marker = fleet._fleet_restart_pending_marker_path()
    if failure not in (None, "running"):
        with pytest.raises(SystemExit, match="1"):
            fleet._apply_pending_fleet_restart_catchup()
-        assert marker.exists()
+        assert fleet._fleet_restart_obligation_armed()
    else:
        fleet._apply_pending_fleet_restart_catchup()
-        assert not marker.exists()
+        assert not fleet._fleet_restart_obligation_armed()
        assert set(recovered) == {"hermes-gateway-one", "hermes-gateway-two"}


@@ -29,6 +29,8 @@ import hermes_cli.update_cmd_fleet as update_cmd_fleet
 import hermes_cli.update_cmd_deps as update_cmd_deps
 from hermes_cli.update_receipt import COMMAND_BOUNDARY_STOP_REASON
 from hermes_constants import get_hermes_home
+import hermes_cli.update_host_obligation as host_obligation
+from gateway import host_rendezvous


 def _make_head_moved_side_effect(pre_sha="abc123", post_sha="def456"):
@@ -156,21 +158,21 @@ def _update_args():
 # ---------------------------------------------------------------------------


-def test_marker_round_trip_under_hermes_home():
-    path = update_cmd._fleet_restart_pending_marker_path()
-    assert path.parent == get_hermes_home()
-    assert path.name == "fleet_restart_pending"
+def test_obligation_round_trip_is_host_scoped():
+    """The obligation is one record per HOST (beside the host rendezvous record), not per home."""
+    path = host_obligation.host_obligation_path()
+    assert path.parent == host_rendezvous.host_state_dir()
    assert not path.exists()

    update_cmd._write_fleet_restart_pending_marker(expected_sha="abc123")
-    assert path.is_file()
-    body = path.read_text(encoding="utf-8")
-    assert "started=" in body
-    assert "pid=" in body
-    assert "expected_sha=abc123" in body
+    assert update_cmd_fleet._fleet_restart_obligation_armed()
+    record = json.loads(path.read_text(encoding="utf-8"))
+    assert record["expected_sha"] == "abc123"
+    assert record["pid"] and record["started"]
+    assert not (get_hermes_home() / "fleet_restart_pending").exists()

    update_cmd._clear_fleet_restart_pending_marker()
-    assert not path.exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()


 def test_pending_needed_when_marker_exists():
@@ -401,14 +403,14 @@ def test_marker_written_after_pull_cleared_after_successful_restart(

    def _spy(*, expected_sha="", runtimes=None):
        orig(expected_sha=expected_sha, runtimes=runtimes)
-        wrote.append(update_cmd._fleet_restart_pending_marker_path().is_file())
+        wrote.append(update_cmd_fleet._fleet_restart_obligation_armed())

    monkeypatch.setattr(update_cmd, "_write_fleet_restart_pending_marker", _spy)

    hermes_main.cmd_update(args)

    assert wrote == [True], "marker must exist immediately after HEAD advances"
-    assert not update_cmd._fleet_restart_pending_marker_path().exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()
    out = capsys.readouterr().out
    assert "✓ Code updated!" in out

@@ -560,7 +562,7 @@ def test_clean_update_defers_desktop_owned_serve_and_clears_marker(
    assert "pid 6161" in out and "pre-update code" in out
    assert "relaunch the Desktop app" in out
    assert "Planned runtimes the restart phase never touched" not in out
-    assert not update_cmd._fleet_restart_pending_marker_path().exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()

    latest = get_hermes_home() / "logs" / "update_receipts" / "latest.json"
    receipt = json.loads(latest.read_text(encoding="utf-8"))
@@ -583,9 +585,9 @@ def test_interrupt_between_pull_and_restart_leaves_marker(
    with pytest.raises(KeyboardInterrupt):
        hermes_main.cmd_update(args)

-    marker = update_cmd._fleet_restart_pending_marker_path()
-    assert marker.is_file()
-    assert "expected_sha=def456" in marker.read_text(encoding="utf-8")
+    assert update_cmd_fleet._fleet_restart_obligation_armed()
+    record = json.loads(host_obligation.host_obligation_path().read_text(encoding="utf-8"))
+    assert record["expected_sha"] == "def456"


 def test_already_up_to_date_runs_pending_restart_when_marker_present(
@@ -612,7 +614,7 @@ def test_already_up_to_date_runs_pending_restart_when_marker_present(
    hermes_main.cmd_update(args)

    assert seen["ran"] is True
-    assert not update_cmd._fleet_restart_pending_marker_path().exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()
    out = capsys.readouterr().out
    assert "did not restart running gateways" in out

@@ -739,7 +741,7 @@ def test_startup_warn_discharged_when_fleet_current(monkeypatch, capsys):
    update_cmd._warn_pending_fleet_restart_on_startup()

    assert capsys.readouterr().err == ""
-    assert not update_cmd._fleet_restart_pending_marker_path().exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()


 def test_startup_warn_discharged_when_multiplexer_covers_owed_profiles(monkeypatch, capsys):
@@ -786,7 +788,7 @@ def test_startup_warn_discharged_when_multiplexer_covers_owed_profiles(monkeypat
    update_cmd._warn_pending_fleet_restart_on_startup()

    assert capsys.readouterr().err == ""
-    assert not update_cmd._fleet_restart_pending_marker_path().exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()
    # The same live multiplexer coverage also discharges the receipt fallback
    # after an operator has already removed the marker.
    assert update_cmd._pending_fleet_restart_needed() is False
@@ -823,7 +825,7 @@ def test_startup_warn_discharged_when_inventory_holds_supervised_serve(monkeypat
    update_cmd._warn_pending_fleet_restart_on_startup()

    assert capsys.readouterr().err == ""
-    assert not update_cmd._fleet_restart_pending_marker_path().exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()


 def test_startup_warn_kept_when_inventory_holds_unclassified_serve(monkeypatch, capsys):
@@ -847,7 +849,7 @@ def test_startup_warn_kept_when_inventory_holds_unclassified_serve(monkeypatch,
    update_cmd._warn_pending_fleet_restart_on_startup()

    assert "did not restart running gateways" in capsys.readouterr().err
-    assert update_cmd._fleet_restart_pending_marker_path().exists()
+    assert update_cmd_fleet._fleet_restart_obligation_armed()


@pytest.mark.parametrize(
@@ -870,7 +872,7 @@ def test_startup_warn_kept_without_positive_evidence(monkeypatch, capsys, disk_s
    update_cmd._warn_pending_fleet_restart_on_startup()

    assert "did not restart running gateways" in capsys.readouterr().err
-    assert update_cmd._fleet_restart_pending_marker_path().exists()
+    assert update_cmd_fleet._fleet_restart_obligation_armed()


 def test_startup_warn_kept_when_receipt_owed_gateway_is_down(monkeypatch, capsys):
@@ -904,7 +906,7 @@ def test_startup_warn_kept_when_receipt_owed_gateway_is_down(monkeypatch, capsys
    update_cmd._warn_pending_fleet_restart_on_startup()

    assert "did not restart running gateways" in capsys.readouterr().err
-    assert update_cmd._fleet_restart_pending_marker_path().exists()
+    assert update_cmd_fleet._fleet_restart_obligation_armed()


 def test_startup_warn_silent_when_failed_receipt_already_restarted_fleet(monkeypatch, capsys):
@@ -994,7 +996,7 @@ def test_startup_warn_silent_when_completed_update_fleet_restarted_onto_moved_ch
 def test_startup_warn_discharged_when_inventory_less_marker_fleet_current(monkeypatch, capsys):
    disk_sha = "e" * 40
    update_cmd._write_fleet_restart_pending_marker(expected_sha=disk_sha)
-    assert "inventory=" not in update_cmd._fleet_restart_pending_marker_path().read_text(encoding="utf-8")
+    assert "inventory" not in host_obligation.read_host_obligation()
    _patch_marker_sha(monkeypatch, disk_sha)
    monkeypatch.setattr(
        "hermes_cli.update_receipt.collect_fleet_versions",
@@ -1006,7 +1008,7 @@ def test_startup_warn_discharged_when_inventory_less_marker_fleet_current(monkey
    update_cmd._warn_pending_fleet_restart_on_startup()

    assert capsys.readouterr().err == ""
-    assert not update_cmd._fleet_restart_pending_marker_path().exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()


 def test_startup_warn_kept_when_inventory_less_marker_fleet_stale(monkeypatch, capsys):
@@ -1023,7 +1025,7 @@ def test_startup_warn_kept_when_inventory_less_marker_fleet_stale(monkeypatch, c
    update_cmd._warn_pending_fleet_restart_on_startup()

    assert "did not restart running gateways" in capsys.readouterr().err
-    assert update_cmd._fleet_restart_pending_marker_path().exists()
+    assert update_cmd_fleet._fleet_restart_obligation_armed()

 # ── Empty-inventory marker: a pull that recorded no gateway owes nothing (#115311) ──

@@ -1042,7 +1044,7 @@ def test_empty_inventory_does_not_arm_marker():
    (Desktop-hosted) install every later update would otherwise hit the unbeatable
    'Fleet restart incomplete' exit 1 (#115311)."""
    update_cmd._write_fleet_restart_pending_marker(expected_sha="e" * 40, runtimes=[])
-    assert not update_cmd._fleet_restart_pending_marker_path().exists()
+    assert not update_cmd_fleet._fleet_restart_obligation_armed()


 def test_pending_fleet_restart_cleared_instead_of_exit_1(monkeypatch, tmp_path):
@@ -0,0 +1,204 @@
+"""The update→restart obligation is HOST-scoped, and one update restarts the host gateway once.
+
+Multiplex-only (Teknium ruling): exactly one ``hermes gateway run`` per host serves every
+profile. The obligation used to live in ONE profile's ``HERMES_HOME``, so ``hermes -p coder
+update`` armed and cleared coder's copy while restarting the SHARED process; no other profile
+could see that obligation, and every profile that ran the catch-up killed the same gateway
+again. These tests pin the host-scoped contract:
+
+- an obligation armed from one profile is owed (and dischargeable) from every other profile;
+- the host gateway is restarted AT MOST ONCE per obligation, however many profiles run it;
+- enumerated systemd units that resolve to one live main PID restart that process once;
+- the fresh-process recovery restarts one host process for every profile it serves.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from types import SimpleNamespace
+
+import pytest
+
+import hermes_cli.update_cmd_fleet as fleet
+import hermes_cli.update_host_obligation as host_obligation
+import hermes_cli.update_restart_recovery as recovery
+from hermes_cli import update_cmd
+
+SHA = "a" * 40
+
+
+@pytest.fixture
+def two_profiles(tmp_path, monkeypatch):
+    """Two profile HERMES_HOMEs behind ONE host state dir — the real multiplex topology."""
+    monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "gateway-locks"))
+    homes = {}
+    for name in ("coder", "writer"):
+        home = tmp_path / "profiles" / name
+        home.mkdir(parents=True)
+        homes[name] = home
+    return homes
+
+
+def _enter(monkeypatch, home) -> None:
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+
+def _arm(profile_runtime: str) -> None:
+    fleet._write_fleet_restart_pending_marker(
+        expected_sha=SHA, runtimes=[{"kind": "gateway", "profile": profile_runtime}])
+
+
+@pytest.fixture
+def no_live_fleet(monkeypatch):
+    """No fleet matrix rows: the obligation can never be discharged by evidence in these tests."""
+    monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: SHA)
+    monkeypatch.setattr("hermes_cli.update_receipt.collect_fleet_versions", lambda: [])
+
+
+def test_obligation_armed_by_one_profile_is_owed_by_every_other(two_profiles, no_live_fleet, monkeypatch):
+    """One host, one obligation: the profile that did not pull still owes — and can discharge — it."""
+    _enter(monkeypatch, two_profiles["coder"])
+    _arm("coder")
+
+    _enter(monkeypatch, two_profiles["writer"])
+    assert fleet._pending_fleet_restart_needed() is True
+
+    fleet._clear_fleet_restart_pending_marker()
+    _enter(monkeypatch, two_profiles["coder"])
+    assert fleet._pending_fleet_restart_needed() is False
+
+
+def test_host_gateway_restarts_once_when_two_profiles_run_the_catch_up(
+    two_profiles, no_live_fleet, monkeypatch, capsys
+):
+    """``hermes -p coder update`` then ``hermes -p writer update`` stops the host gateway ONCE."""
+    monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda **k: [4242])
+    monkeypatch.setattr("hermes_cli.gateway.supports_systemd_services", lambda: False)
+    monkeypatch.setattr("hermes_cli.gateway.is_macos", lambda: False)
+    monkeypatch.setattr("hermes_cli.gateway.is_windows", lambda: False)
+    monkeypatch.setattr("hermes_cli.gateway._wait_for_gateway_exit", lambda **k: True)
+    monkeypatch.setattr(fleet, "_restart_macos_launchd_gateways", lambda *a, **k: None)
+    kills: list = []
+    monkeypatch.setattr("hermes_cli.gateway.kill_gateway_processes", lambda **k: kills.append(k))
+
+    _enter(monkeypatch, two_profiles["coder"])
+    _arm("coder")
+    assert update_cmd._run_pending_fleet_restart() is True
+
+    _enter(monkeypatch, two_profiles["writer"])
+    _arm("writer")
+    assert update_cmd._run_pending_fleet_restart() is True
+
+    assert len(kills) == 1, "the one host gateway must be stopped once per update, not once per profile"
+    assert "already restarted for this update" in capsys.readouterr().out
+
+
+def test_legacy_per_home_marker_is_still_read_and_cleared(two_profiles, no_live_fleet, monkeypatch):
+    """An obligation armed by the pre-host-scope code must still be discharged after the upgrade."""
+    _enter(monkeypatch, two_profiles["coder"])
+    legacy = fleet._fleet_restart_pending_marker_path()
+    legacy.write_text(
+        f"started=1.0\npid=1\nexpected_sha={SHA}\n"
+        + "inventory=" + json.dumps({"version": 1, "runtimes": [{"kind": "gateway", "profile": "coder"}]}) + "\n",
+        encoding="utf-8",
+    )
+
+    assert fleet._pending_fleet_restart_needed() is True
+    fleet._clear_fleet_restart_pending_marker()
+    assert not legacy.exists()
+    assert fleet._pending_fleet_restart_needed() is False
+
+
+def _listing(units: list[str]):
+    result = SimpleNamespace(returncode=0, stdout="\n".join(f"{u} loaded active running x" for u in units))
+    return [("user", ["systemctl", "--user"], result)]
+
+
+def test_leftover_per_profile_units_restart_their_one_host_process_once(monkeypatch, capsys):
+    """Three units, one live main PID = one host gateway: restart it once and name the legacy units."""
+    # raising=False keeps this usable as the red-on-base A/B (the helper is the fix).
+    monkeypatch.setattr(fleet, "_unit_main_pid", lambda scope_cmd, svc: 4242, raising=False)
+    restarted: list[str] = []
+    monkeypatch.setattr(
+        fleet, "_systemctl_reset_and_restart",
+        lambda manage_cmd, svc, scope_cmd=None: restarted.append(svc) or SimpleNamespace(returncode=0))
+    monkeypatch.setattr(fleet, "_wait_for_service_active", lambda scope_cmd, svc: True)
+    monkeypatch.setattr(fleet, "_SYSTEMD_SCOPES", (("user", ["systemctl", "--user"]),))
+
+    failed: list = []
+    fleet._restart_systemd_gateway_units_best_effort(
+        failed, _listing(["hermes-gateway.service", "hermes-gateway-coder.service", "hermes-gateway-writer.service"]))
+
+    assert restarted == ["hermes-gateway"]
+    assert failed == []
+    out = capsys.readouterr().out
+    assert "hermes-gateway-coder" in out and "legacy per-profile unit" in out
+
+
+def test_units_with_distinct_live_pids_are_each_restarted(monkeypatch):
+    """Control: genuinely separate processes are still separate restart targets."""
+    pids = {"hermes-gateway": 1, "hermes-gateway-coder": 2}
+    monkeypatch.setattr(fleet, "_unit_main_pid", lambda scope_cmd, svc: pids[svc], raising=False)
+    restarted: list[str] = []
+    monkeypatch.setattr(
+        fleet, "_systemctl_reset_and_restart",
+        lambda manage_cmd, svc, scope_cmd=None: restarted.append(svc) or SimpleNamespace(returncode=0))
+    monkeypatch.setattr(fleet, "_wait_for_service_active", lambda scope_cmd, svc: True)
+    monkeypatch.setattr(fleet, "_SYSTEMD_SCOPES", (("user", ["systemctl", "--user"]),))
+
+    fleet._restart_systemd_gateway_units_best_effort(
+        [], _listing(["hermes-gateway.service", "hermes-gateway-coder.service"]))
+
+    assert sorted(restarted) == ["hermes-gateway", "hermes-gateway-coder"]
+
+
+def _host_record(tmp_path, monkeypatch, profiles: list[str]) -> None:
+    lock_dir = tmp_path / "gateway-locks"
+    lock_dir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(lock_dir))
+    (lock_dir / "host-gateway.json").write_text(
+        json.dumps({"role": "gateway", "pid": os.getpid(), "profiles": profiles}), encoding="utf-8")
+
+
+def test_recovery_restarts_one_host_process_for_all_the_profiles_it_serves(tmp_path, monkeypatch):
+    """N payload profiles served by ONE host gateway = one relaunch, not N that kill each other."""
+    _host_record(tmp_path, monkeypatch, ["coder", "writer", "default"])
+    argvs: list[list[str]] = []
+
+    def fake_run(argv, **kwargs):
+        argvs.append(list(argv))
+        return SimpleNamespace(returncode=0, stdout="")
+
+    result = recovery.restart_profiles(["coder", "writer", "default"], run=fake_run)
+
+    relaunches = [argv for argv in argvs if argv[-2:] == ["gateway", "restart"]]
+    assert len(relaunches) == 1, "one host process must be relaunched once for every profile it serves"
+    reported = [*result["verified"], *result["relaunch_attempted"], *result["failed"]]
+    assert sorted(reported) == ["coder", "default", "writer"], "every requested profile keeps an outcome"
+    assert result["covered"] == {"coder": ["default", "writer"]}
+
+
+def test_recovery_keeps_separate_processes_separate(tmp_path, monkeypatch):
+    """Control: with no host record, each profile is its own restart target."""
+    monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "empty-locks"))
+    argvs: list[list[str]] = []
+
+    def fake_run(argv, **kwargs):
+        argvs.append(list(argv))
+        return SimpleNamespace(returncode=0, stdout="")
+
+    recovery.restart_profiles(["coder", "writer"], run=fake_run)
+
+    assert len([argv for argv in argvs if argv[-2:] == ["gateway", "restart"]]) == 2
+
+
+def test_host_obligation_lives_beside_the_host_rendezvous_record(two_profiles, monkeypatch, tmp_path):
+    """The record is written ONCE PER HOST, in the cross-profile rendezvous dir."""
+    _enter(monkeypatch, two_profiles["coder"])
+    _arm("coder")
+
+    path = host_obligation.host_obligation_path()
+    assert path == tmp_path / "gateway-locks" / "host-update-restart.json"
+    assert path.is_file()
+    assert not (two_profiles["coder"] / "fleet_restart_pending").exists()
@@ -277,6 +277,7 @@ def test_recovery_child_restarts_each_profile_with_a_fresh_main(monkeypatch):
        "verified": [],
        "relaunch_attempted": ["coder", "default"],
        "failed": [],
+        "covered": {},
    }
    assert [call[0] for call in calls] == [
        [sys.executable, "-m", "hermes_cli.main", "-p", "coder", "gateway", "restart"],
@@ -314,6 +315,7 @@ def test_recovery_child_verifies_systemd_profiles_via_is_active(monkeypatch):
        "verified": ["default"],
        "relaunch_attempted": ["coder"],
        "failed": [],
+        "covered": {},
    }
    # The launchd profile must never be probed with systemctl.
    systemctl_units = [argv[-1] for argv in calls if argv[0].endswith("systemctl")]
@@ -334,6 +336,7 @@ def test_recovery_child_treats_missing_systemctl_as_unverified(monkeypatch):
        "verified": [],
        "relaunch_attempted": ["default"],
        "failed": [],
+        "covered": {},
    }


@@ -349,6 +352,7 @@ def test_recovery_child_reports_failed_profile_without_losing_successes():
        "verified": [],
        "relaunch_attempted": ["default"],
        "failed": ["coder"],
+        "covered": {},
    }


@@ -394,6 +398,7 @@ def test_recovery_module_empty_payload_is_a_real_clean_process():
        "failed": [],
        "relaunch_attempted": [],
        "verified": [],
+        "covered": {},
        "serve_units": {"verified": [], "failed": []},
    }

@@ -475,6 +480,7 @@ def test_recovery_module_end_to_end_in_a_real_fresh_process(tmp_path):
        "failed": [],
        "relaunch_attempted": ["coder"],
        "verified": ["default"],
+        "covered": {},
        "serve_units": {"verified": [], "failed": []},
    }
    restarts = [json.loads(line) for line in ledger.read_text().splitlines()]
@@ -6,6 +6,7 @@ import pytest

 from hermes_cli import process_identity, update_cmd_fleet as fleet, update_receipt
 from hermes_constants import get_hermes_home
+import hermes_cli.update_host_obligation as host_obligation

 MANUAL = {"kind": "serve", "profile": "work", "pid": 900, "supervisor": "manual-serve", "restart_via": "respawn-argv", "code_sha": "old", "detail": {"create_time": 1000.0}}
 CURRENT = {"profile": "alpha", "state": "current", "code_sha": "new"}
@@ -56,11 +57,11 @@ def test_scoped_reconciliation_matrix(monkeypatch, capsys, name, old, marker, li
    fleet._apply_pending_fleet_restart_catchup(defer=True)
    assert ("fleet restart deferred" in capsys.readouterr().out) is pending
    assert target.read_bytes() == before
-    assert fleet._fleet_restart_pending_marker_path().exists() is (marker is not None and pending)
+    assert host_obligation.host_obligation_path().exists() is (marker is not None and pending)
    if name == "missing-sibling":
        live.append(dict(CURRENT, profile="beta"))
        assert not fleet._pending_fleet_restart_needed()
-        assert not fleet._fleet_restart_pending_marker_path().exists()
+        assert not host_obligation.host_obligation_path().exists()
        assert target.read_bytes() == before


@@ -74,7 +75,7 @@ def test_empty_marker_never_inherits_receipt_ownership(monkeypatch, capsys, aliv
    warning = capsys.readouterr().err
    assert "hermes gateway restart" in warning
    assert ("serve [work] pid 900" in warning) is (alive is not False)
-    assert fleet._fleet_restart_pending_marker_path().exists()
+    assert host_obligation.host_obligation_path().exists()
    assert target.read_bytes() == before


@@ -131,9 +132,8 @@ def test_new_marker_cannot_borrow_old_alpha_receipt(monkeypatch, capsys, complet
        old.update(post_update={"sha": "new"}, gateway_restart={"incomplete": False})
    live = [CURRENT]
    target = seed(monkeypatch, old, "new", live)
-    marker = fleet._fleet_restart_pending_marker_path()
-    with marker.open("a") as stream:
-        stream.write("inventory=" + json.dumps({"version": 1, "runtimes": [GATEWAY, dict(GATEWAY, profile="beta")]}) + "\n")
+    marker = host_obligation.host_obligation_path()
+    host_obligation.amend_host_obligation(inventory={"version": 1, "runtimes": [GATEWAY, dict(GATEWAY, profile="beta")]})
    receipt_before, marker_before = target.read_bytes(), marker.read_bytes()
    fleet._warn_pending_fleet_restart_on_startup()
    assert "hermes gateway restart" in capsys.readouterr().err
@@ -160,7 +160,7 @@ def test_legacy_marker_discharges_on_live_fleet_evidence_without_receipt(monkeyp
        old.update(post_update={"sha": "new"}, gateway_restart={"incomplete": False})
    live = [CURRENT]
    target = seed(monkeypatch, old, "new", live)
-    marker = fleet._fleet_restart_pending_marker_path()
+    marker = host_obligation.host_obligation_path()
    receipt_before = target.read_bytes()
    fleet._warn_pending_fleet_restart_on_startup()
    assert "hermes gateway restart" not in capsys.readouterr().err
@@ -177,7 +177,7 @@ def test_inventory_less_marker_settles_after_out_of_band_pull(monkeypatch, capsy
    evidence the warning can be about — a stale or absent fleet still keeps it.
    """
    seed(monkeypatch, {}, "old", live)
-    marker = fleet._fleet_restart_pending_marker_path()
+    marker = host_obligation.host_obligation_path()
    fleet._warn_pending_fleet_restart_on_startup()
    assert ("hermes gateway restart" in capsys.readouterr().err) is pending
    assert fleet._pending_fleet_restart_needed() is pending
@@ -190,9 +190,8 @@ def test_inventory_less_marker_settles_after_out_of_band_pull(monkeypatch, capsy
@pytest.mark.parametrize("inventory", [{}, [], {"version": 2, "runtimes": [GATEWAY]}, {"version": 1, "runtimes": [GATEWAY, dict(MANUAL, detail={})]}, {"version": 1, "runtimes": [None]}, {"version": 1, "runtimes": [{"kind": "gateway", "profile": "unknown"}]}, {"version": 1, "runtimes": [{"kind": "gateway", "profile": []}]}])
 def test_unverified_marker_inventory_stays_pending(monkeypatch, inventory):
    seed(monkeypatch, {"outcome": "success", "plan": {"runtimes": [GATEWAY]}}, "new", [CURRENT])
-    marker = fleet._fleet_restart_pending_marker_path()
-    with marker.open("a") as stream:
-        stream.write("inventory=" + json.dumps(inventory) + "\n")
+    marker = host_obligation.host_obligation_path()
+    host_obligation.amend_host_obligation(inventory=inventory)
    before = marker.read_bytes()
    assert fleet._pending_fleet_restart_needed()
    monkeypatch.setattr("hermes_cli.update_cmd._run_pending_fleet_restart", lambda: True)
@@ -202,12 +201,12 @@ def test_unverified_marker_inventory_stays_pending(monkeypatch, inventory):
    assert marker.read_bytes() == before


-@pytest.mark.parametrize("suffix", ['inventory={', 'inventory=null\ninventory={"version":1,"runtimes":[]}', 'broken-line'])
-def test_malformed_marker_stays_pending(monkeypatch, suffix):
+@pytest.mark.parametrize("suffix", ["{", '{"version": 1, "inventory": ', "broken-line"])
+def test_malformed_obligation_stays_pending(monkeypatch, suffix):
+    """An unparseable record is an obligation whose terms are unknown — never a discharged one."""
    seed(monkeypatch, {}, "new", [CURRENT])
-    marker = fleet._fleet_restart_pending_marker_path()
-    with marker.open("a") as stream:
-        stream.write(suffix + "\n")
+    marker = host_obligation.host_obligation_path()
+    marker.write_text(marker.read_text(encoding="utf-8") + suffix, encoding="utf-8")
    assert fleet._pending_fleet_restart_needed()
    assert marker.exists()

@@ -230,10 +229,10 @@ def test_pulled_update_marker_owns_pre_update_inventory(monkeypatch):
    monkeypatch.setattr(update_cmd, "_sweep_bytecode_after_update", interrupt)
    with pytest.raises(KeyboardInterrupt):
        update_cmd._apply_pulled_update([], "main", "old", SimpleNamespace(in_place_update=True), None, gateway_mode=False, is_fork=False, desktop_dir=None, had_desktop_app_before_update=False, pre_update_snapshot_id=None, _pre_update_plan=plan, _windows_gateway_resume=None, args=SimpleNamespace())
-    marker = fleet._fleet_restart_pending_marker_path()
-    fields = dict(line.split("=", 1) for line in marker.read_text().splitlines())
+    marker = host_obligation.host_obligation_path()
+    fields = json.loads(marker.read_text(encoding="utf-8"))
    assert fields["expected_sha"] == "new"
-    assert json.loads(fields["inventory"]) == {"version": 1, "runtimes": plan.to_dict()["runtimes"]}
+    assert fields["inventory"] == {"version": 1, "runtimes": plan.to_dict()["runtimes"]}
    assert fleet._pending_fleet_restart_needed()
    assert target.read_bytes() == before

@@ -246,7 +245,7 @@ def test_catchup_verifies_owned_fleet_after_restart(monkeypatch, successor):
    live = [CURRENT]
    target = seed(monkeypatch, old, "new", live)
    fleet._write_fleet_restart_pending_marker(expected_sha="new", runtimes=[GATEWAY, dict(GATEWAY, profile="beta")])
-    marker = fleet._fleet_restart_pending_marker_path()
+    marker = host_obligation.host_obligation_path()
    receipt_before, marker_before = target.read_bytes(), marker.read_bytes()
    restarted = []

@@ -288,7 +287,7 @@ def test_verified_restart_surviving_marker_preserves_manual_debt(monkeypatch, ca
    old = {"outcome": "partial", "post_update": {"sha": "new"}, "gateway_restart": {"incomplete": False, "phase_error": ""}, "plan": {"runtimes": [GATEWAY, *manual]}, "fleet": [CURRENT]}
    target = seed(monkeypatch, old, "new", [CURRENT])
    fleet._write_fleet_restart_pending_marker(expected_sha="new", runtimes=[GATEWAY, *manual])
-    marker = fleet._fleet_restart_pending_marker_path()
+    marker = host_obligation.host_obligation_path()
    receipt_before, marker_before = target.read_bytes(), marker.read_bytes()
    directory = get_hermes_home() / "serve_restart_pending"
    if blocked_storage:
@@ -132,7 +132,7 @@ The same inventory is embedded in every real update's receipt (`~/.hermes/logs/u

 Every `hermes update` run writes a machine-readable receipt to `~/.hermes/logs/update_receipts/` (last 20 kept, `latest.json` always points at the most recent): the pre-update fleet plan, each step taken, anything skipped and why, the gateway restart outcome, and the final fleet version matrix. The SQLite runtime repair is one of those steps (`sqlite_runtime_repair`): a failed repair records the actual reason (for example the `uv sync` error) and the SQLite version pair, a deferred or not-applicable repair lands in the skips with its reason. After the restart phase the updater compares each live gateway's running code against the freshly updated checkout and prints a per-profile matrix — a gateway still serving pre-update code is reported loudly with the exact restart command, and the update exits non-zero so automation never treats a mixed-version fleet as healthy. Both `--plan` and the fleet check ask each running gateway directly over its local control socket (`gateway.sock` in the profile's data directory, a named pipe on Windows) when available, so version and supervisor information comes from the gateway itself; gateways from older versions are still discovered through their state files as before.

-A multiplexed default gateway is one process serving several profiles, so it appears once in the matrix and vouches for every profile in its `served_profiles` record. The same coverage clears the "A previous `hermes update` pulled new code but did not restart running gateways" hint: once that gateway (or, after a manual `git pull`, every gateway an update restarted) runs the current code, `hermes gateway restart` is enough — the hint no longer waits for the next `hermes update` to write a fresh receipt. The same is true of a `fleet_restart_pending` breadcrumb left by an update that died before recording which gateways it owed (or by an older updater that never recorded them): once every live gateway runs the current checkout, the breadcrumb is retired and the hint stops. An update whose pre-update plan found no gateway at all owes nothing and leaves no breadcrumb. A backend supervised by Desktop, systemd or launchd is restarted by its supervisor and never blocks this settlement; only a manual backend whose reminder could not be saved keeps the obligation open.
+A multiplexed default gateway is one process serving several profiles, so it appears once in the matrix and vouches for every profile in its `served_profiles` record. The same coverage clears the "A previous `hermes update` pulled new code but did not restart running gateways" hint: once that gateway (or, after a manual `git pull`, every gateway an update restarted) runs the current code, `hermes gateway restart` is enough — the hint no longer waits for the next `hermes update` to write a fresh receipt. The same is true of the restart obligation left by an update that died before recording which gateways it owed (or by an older updater that never recorded them): once every live gateway runs the current checkout, the obligation is retired and the hint stops. That obligation is recorded once per HOST, in the cross-profile rendezvous directory (`$HERMES_GATEWAY_LOCK_DIR`, else `$XDG_STATE_HOME/hermes/gateway-locks`) as `host-update-restart.json`, so every profile's CLI sees the same one: `hermes -p coder update` and `hermes -p writer update` restart the shared multiplexed gateway once between them, not once each. An obligation left behind by an older per-profile updater (`fleet_restart_pending` in one profile's Hermes home) is still read and cleared. An update whose pre-update plan found no gateway at all owes nothing and leaves no breadcrumb. A backend supervised by Desktop, systemd or launchd is restarted by its supervisor and never blocks this settlement; only a manual backend whose reminder could not be saved keeps the obligation open.

 ### Manual backend restart reminders