diff --git a/gateway/run_adapters.py b/gateway/run_adapters.py index e5a2e08568..203c9d330d 100644 --- a/gateway/run_adapters.py +++ b/gateway/run_adapters.py @@ -898,6 +898,13 @@ class GatewayAdapterLifecycleMixin: # would park a transiently-failed profile before the first watcher tick can retry it. for profile_name in transient_failed: self._served_profile_signatures.pop(profile_name, None) + # Cached configs follow the served set: a profile that failed to start (or stopped being + # served) keeps no home channel in the host-wide notice fan-out, where it would be owed a + # notice no transport can deliver and ``.restart_pending.json`` would never be unlinked. + configs = getattr(self, "_profile_configs", None) + if configs is not None: + for profile_name in [p for p in configs if p not in self._served_profile_signatures]: + configs.pop(profile_name, None) self._restore_secondary_completion_ledgers(profile_homes) return connected diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index 033d25c0e0..f37178078f 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -44,6 +44,31 @@ def _served_notice_target_key(profile: Optional[str], platform_value: str, chat_ platform_value if profile is None else f"{profile}:{platform_value}", chat_id, thread_id) +def _delivery_target_key(platform_value: str, chat_id, thread_id) -> tuple: + """Dedupe key for one DELIVERED chat, profile-independent. + + Two served profiles can share a single home chat (one Telegram group for the whole host); + keyed per profile they would each post their own "Gateway online" notice into it. + """ + return _notice_target_key(platform_value, chat_id, thread_id) + + +def _safe_delivery_transport(platform, config, adapters, *, profile: Optional[str] = None): + """``resolve_delivery_transport`` isolated to one target: ``None`` (logged) on failure. + + The fan-out spans every served profile, so one profile's broken adapter must not abort the + pass and starve every profile after it in dict order. + """ + from gateway.delivery import resolve_delivery_transport + try: + return resolve_delivery_transport(platform, config, adapters) + except Exception as exc: + logger.debug( + "Home-channel transport unavailable for %s%s: %s", + f"{profile}:" if profile else "", getattr(platform, "value", platform), exc) + return None + + 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:] @@ -786,12 +811,11 @@ class GatewayNotificationsMixin: def _home_channel_transports(self): """Yield ``(platform, platform_cfg, home, transport)`` for every home channel with a live transport.""" - from gateway.delivery import resolve_delivery_transport for platform, platform_cfg in self.config.platforms.items(): home = platform_cfg.home_channel if not home or not home.chat_id: continue - transport = resolve_delivery_transport(platform, self.config, self.adapters) + transport = _safe_delivery_transport(platform, self.config, self.adapters) if transport is None: continue yield platform, platform_cfg, home, transport @@ -813,7 +837,6 @@ class GatewayNotificationsMixin: 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(): @@ -822,7 +845,7 @@ class GatewayNotificationsMixin: home = platform_cfg.home_channel if not home or not home.chat_id: continue - transport = resolve_delivery_transport(platform, profile_cfg, adapters) + transport = _safe_delivery_transport(platform, profile_cfg, adapters, profile=profile) if transport is None: continue yield profile, platform, platform_cfg, home, transport @@ -917,8 +940,9 @@ class GatewayNotificationsMixin: ) -> set[tuple[str, str, Optional[str]]]: """Notify EVERY served profile's configured home channels that the gateway is back online. - 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. + Best-effort, once per home CHAT — several served profiles can share one chat (a single + Telegram group for the whole host), and one host process restarting once owes that chat + one notice. Accounting stays per profile so the marker's owed set still discharges. ``skip_targets`` lets startup avoid duplicate messages when a more specific restart notification is queued for the same chat. """ @@ -928,7 +952,14 @@ class GatewayNotificationsMixin: free_tier_line = self._free_tier_startup_line() if free_tier_line: message = f"{message}\n{free_tier_line}" - for profile, platform, platform_cfg, home, transport in self._served_home_channel_transports(): + targets = list(self._served_home_channel_transports()) + # A chat already notified for ANOTHER profile is not notified again. + notified_chats = { + _delivery_target_key(platform.value, home.chat_id, home.thread_id) + for profile, platform, _cfg, home, _transport in targets + if _served_notice_target_key(profile, platform.value, home.chat_id, home.thread_id) in skipped + } + for profile, platform, platform_cfg, home, transport in targets: if not platform_cfg.gateway_restart_notification: logger.info( "Home-channel startup notification suppressed: %s has gateway_restart_notification=false", @@ -938,9 +969,14 @@ class GatewayNotificationsMixin: target = _served_notice_target_key(profile, platform.value, home.chat_id, home.thread_id) if target in skipped or target in delivered: continue + chat = _delivery_target_key(platform.value, home.chat_id, home.thread_id) + if chat in notified_chats: + delivered.add(target) + continue if await self._send_home_channel_message( platform, home, transport, message, "Home-channel startup notification failed for %s:%s: %s", ): + notified_chats.add(chat) delivered.add(target) logger.info("Sent home-channel startup notification to %s:%s", platform.value, home.chat_id) return delivered diff --git a/gateway/run_profile_reconcile.py b/gateway/run_profile_reconcile.py index 06df26be98..3b030cf43f 100644 --- a/gateway/run_profile_reconcile.py +++ b/gateway/run_profile_reconcile.py @@ -156,6 +156,11 @@ class GatewayProfileReconcileMixin: for name in transient_failed: if isinstance(self._served_profile_signatures, dict): self._served_profile_signatures.pop(name, None) + # A cached config with no live adapters is owed a home-channel notice nothing can + # deliver, and the planned-restart marker then never clears. + configs = getattr(self, "_profile_configs", None) + if isinstance(configs, dict): + configs.pop(name, None) if added: await self._after_profiles_added([(n, current[n]) for n in added]) result["served_profiles"] = self.served_profile_names() @@ -215,7 +220,7 @@ class GatewayProfileReconcileMixin: # Its ``:`` runtime entries describe a profile that no longer exists. _write_runtime_status_quiet(drop_profile_platforms=name) for attr in ("pairing_stores", "_busy_text_modes_by_profile", "_busy_input_modes_by_profile", - "_busy_text_timing_by_profile", "_human_delay_by_profile"): + "_busy_text_timing_by_profile", "_human_delay_by_profile", "_profile_configs"): store = getattr(self, attr, None) if isinstance(store, dict): store.pop(name, None) diff --git a/hermes_cli/update_cmd_fleet.py b/hermes_cli/update_cmd_fleet.py index 4ba4c31f22..0e50ac890b 100644 --- a/hermes_cli/update_cmd_fleet.py +++ b/hermes_cli/update_cmd_fleet.py @@ -56,20 +56,63 @@ def _fleet_restart_pending_marker_path() -> Path: 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.""" + """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 write_host_obligation + 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 - write_host_obligation( - expected_sha=expected_sha, runtimes=runtimes, profile=_current_profile_name()) + 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: @@ -104,10 +147,14 @@ def _obligation_fields() -> dict[str, str] | None: ``None`` means nothing armed OR a malformed record; both must leave the obligation standing. """ - from hermes_cli.update_host_obligation import obligation_fields + 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") except (OSError, UnicodeError): @@ -566,6 +613,29 @@ def _live_fleet_current_rows() -> list[dict] | None: 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 _run_pending_fleet_restart() -> bool: """Catch-up restart for gateways left on pre-update code. Never raises. @@ -579,7 +649,7 @@ def _run_pending_fleet_restart() -> bool: """ 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() + checkout_sha = _restart_identity_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 diff --git a/hermes_cli/update_host_obligation.py b/hermes_cli/update_host_obligation.py index c8e246bae4..849269f3a9 100644 --- a/hermes_cli/update_host_obligation.py +++ b/hermes_cli/update_host_obligation.py @@ -205,7 +205,9 @@ def collapse_units_to_host_processes( for unit in units: try: pid = int(main_pid(unit) or 0) - except (TypeError, ValueError): + except Exception: + # Identity that cannot be proved keeps its own restart; a probe failure of any kind + # must never abort the whole pass. pid = 0 if pid <= 0: restart.append(unit) diff --git a/tests/gateway/test_planned_restart_notice_multiplex.py b/tests/gateway/test_planned_restart_notice_multiplex.py index 973e78976d..71202f5d1c 100644 --- a/tests/gateway/test_planned_restart_notice_multiplex.py +++ b/tests/gateway/test_planned_restart_notice_multiplex.py @@ -12,6 +12,7 @@ from unittest.mock import AsyncMock, Mock import pytest +import gateway.delivery as gateway_delivery import gateway.run as gateway_run from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig from gateway.platforms.base import SendResult @@ -92,3 +93,84 @@ async def test_marker_survives_until_a_served_profile_is_reachable(multiplex_run coder.send.assert_awaited_once() assert launch.send.await_count == 1, "a reached home is never notified twice" assert not marker.exists() + + +@pytest.mark.asyncio +async def test_profiles_sharing_one_home_chat_get_one_notice(tmp_path, monkeypatch): + """One host process restarting once owes a shared chat ONE notice, not one per profile. + + A single Telegram group as the home channel of both the launch profile and a served profile + is a common setup; keyed per profile it received two "Gateway online" messages. + """ + 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.TELEGRAM, "-100999") + runner.config.sessions_dir = tmp_path / "sessions" + launch, coder = _adapter(), _adapter() + runner.adapters = {Platform.TELEGRAM: launch} + runner._profile_configs = {"coder": _home_config(Platform.TELEGRAM, "-100999")} + runner._profile_adapters = {"coder": {Platform.TELEGRAM: 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") + + await runner._replay_pending_planned_restart_notification() + + assert launch.send.await_count + coder.send.await_count == 1, "one chat, one restart, one notice" + assert not marker.exists(), "the shared chat was reached, so every owed profile is discharged" + + +@pytest.mark.asyncio +async def test_one_broken_profile_does_not_starve_the_rest(multiplex_runner, monkeypatch): + """A profile whose transport resolution raises is skipped; the fan-out continues.""" + runner, marker = multiplex_runner + runner.adapters[Platform.DISCORD] = _adapter() + ok = _adapter() + runner._profile_configs = { + "b": _home_config(Platform.TELEGRAM, "b-home"), + "c": _home_config(Platform.SLACK, "c-home"), + } + runner._profile_adapters = {"b": {Platform.TELEGRAM: _adapter()}, "c": {Platform.SLACK: ok}} + real = gateway_delivery.resolve_delivery_transport + + def resolve(platform, config, adapters): + if platform is Platform.TELEGRAM: + raise RuntimeError("broken adapter") + return real(platform, config, adapters) + + monkeypatch.setattr(gateway_delivery, "resolve_delivery_transport", resolve) + + await runner._send_home_channel_startup_notifications() + + ok.send.assert_awaited_once(), "a profile after the broken one is still notified" + + +@pytest.mark.asyncio +async def test_unserved_profile_config_is_pruned_from_the_fan_out(tmp_path, monkeypatch): + """A profile whose adapters failed keeps no cached config, or it is owed a notice forever. + + ``owed`` is built from ``_profile_configs`` while delivery needs a live transport, so a stale + entry makes ``owed <= delivered`` permanently false and ``.restart_pending.json`` immortal. + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + runner = object.__new__(gateway_run.GatewayRunner) + runner.config = _home_config(Platform.DISCORD, "launch-home") + runner._profile_configs = {"ghost": _home_config(Platform.TELEGRAM, "-200")} + runner._profile_adapters = {} + runner._multiplex_on = Mock(return_value=True) + runner._primary_resource_claims = Mock(return_value={}) + runner._record_served_profiles = Mock() + runner._restore_secondary_completion_ledgers = Mock() + runner._start_one_profile_adapters = AsyncMock(side_effect=RuntimeError("adapters failed")) + monkeypatch.setattr(gateway_run, "_multiplex_profile_homes", lambda cfg: [("ghost", tmp_path / "ghost")]) + monkeypatch.setattr("hermes_cli.profiles.get_active_profile_name", lambda: "default") + monkeypatch.setattr( + "gateway.run_profile_reconcile.profile_serve_signature", lambda home: ("sig",)) + + await runner._start_secondary_profile_adapters() + + assert "ghost" not in runner._profile_configs + assert list(runner._served_home_channel_configs()) == [ + (None, Platform.DISCORD, runner.config.platforms[Platform.DISCORD])] diff --git a/tests/hermes_cli/test_update_host_obligation.py b/tests/hermes_cli/test_update_host_obligation.py index 402ea8d446..421455996f 100644 --- a/tests/hermes_cli/test_update_host_obligation.py +++ b/tests/hermes_cli/test_update_host_obligation.py @@ -202,3 +202,103 @@ def test_host_obligation_lives_beside_the_host_rendezvous_record(two_profiles, m assert path == tmp_path / "gateway-locks" / "host-update-restart.json" assert path.is_file() assert not (two_profiles["coder"] / "fleet_restart_pending").exists() + + +@pytest.mark.skipif(getattr(os, "geteuid", lambda: 1)() == 0, reason="root ignores directory permissions") +def test_unwritable_host_state_dir_still_arms_the_obligation(two_profiles, no_live_fleet, monkeypatch, tmp_path): + """An unwritable host state dir must never silently disarm the update→restart obligation. + + The host record moved out of ``$HERMES_HOME`` (writable by construction) into the host state + dir, which a read-only mount or a container UID mismatch can make unwritable. Losing the + obligation there is the #117275 outage shape: an interrupted update leaves stale code running + with no warning and no catch-up restart. + """ + _enter(monkeypatch, two_profiles["coder"]) + lock_dir = tmp_path / "gateway-locks" + lock_dir.mkdir(parents=True, exist_ok=True) + lock_dir.chmod(0o500) + try: + _arm("coder") + assert not host_obligation.host_obligation_present(), "precondition: the record could not be written" + assert fleet._fleet_restart_obligation_armed() is True + assert fleet._pending_fleet_restart_needed() is True + finally: + lock_dir.chmod(0o700) + + +def test_unreadable_host_record_is_never_discharged_by_the_legacy_marker(two_profiles, no_live_fleet, monkeypatch, tmp_path): + """A record whose terms are UNKNOWN cannot be settled by another record's terms. + + A foreign version (a NEWER CLI wrote it) or a corrupt record is fail-closed by contract; the + legacy per-home marker describes a different obligation and must not discharge it. + """ + _enter(monkeypatch, two_profiles["coder"]) + lock_dir = tmp_path / "gateway-locks" + lock_dir.mkdir(parents=True, exist_ok=True) + (lock_dir / host_obligation.HOST_OBLIGATION_NAME).write_text( + json.dumps({"version": 99, "expected_sha": SHA}), encoding="utf-8") + fleet._fleet_restart_pending_marker_path().write_text( + f"started=1.0\npid=1\nexpected_sha={SHA}\n" + + "inventory=" + json.dumps({"version": 1, "runtimes": []}) + "\n", + encoding="utf-8", + ) + + assert fleet._obligation_fields() is None + assert fleet._pending_fleet_restart_needed() is True + + +def test_restart_runs_once_per_host_on_a_non_git_install(two_profiles, monkeypatch, capsys): + """zip/pip/Docker installs resolve no checkout SHA; the restart-once guard must still hold. + + ``mark_host_restart_completed("")`` can never match, so every profile's ``hermes update`` + re-killed the one shared multiplexer on exactly the installs this record exists for. + """ + monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: None) + monkeypatch.setattr("hermes_cli.update_receipt.collect_fleet_versions", lambda: []) + 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) + 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"]) + assert update_cmd._run_pending_fleet_restart() is True + + assert len(kills) == 1, "the host gateway must be stopped once per update, not once per profile" + assert "already restarted for this update" in capsys.readouterr().out + + +def test_a_failing_main_pid_probe_keeps_its_own_restart(): + """Any probe error is unproven identity (its own restart), never an aborted restart pass.""" + def boom(unit): + raise RuntimeError("systemctl exploded") + + restart, covered = host_obligation.collapse_units_to_host_processes(["a.service", "b.service"], boom) + + assert restart == ["a.service", "b.service"] + assert covered == {} + + +@pytest.mark.parametrize("env", [ + {"HERMES_GATEWAY_LOCK_DIR": "/srv/override/locks"}, + {"XDG_STATE_HOME": "/srv/xdg-state"}, + {"XDG_STATE_HOME": "relative/state"}, + {}, +]) +def test_recovery_host_state_dir_matches_the_gateway_resolver(monkeypatch, env): + """``update_restart_recovery`` re-implements the lock-dir rule (it may import no Hermes code + at runtime); the duplicate must not drift from ``gateway.status._get_lock_dir``.""" + from gateway.status import _get_lock_dir + + for name in ("HERMES_GATEWAY_LOCK_DIR", "XDG_STATE_HOME"): + monkeypatch.delenv(name, raising=False) + for name, value in env.items(): + monkeypatch.setenv(name, value) + + assert recovery._host_state_dir() == str(_get_lock_dir())