From 9a421c8b49117baf7e9ce84ac5d922d6cf664e7e Mon Sep 17 00:00:00 2001 From: spiky02plateau Date: Fri, 29 May 2026 14:12:44 +0200 Subject: [PATCH] fix(gateway): notify home channel after service-managed restart --- gateway/run.py | 224 +++++++++++++++-- tests/gateway/test_restart_notification.py | 271 ++++++++++++++++++++- 2 files changed, 478 insertions(+), 17 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 0549d7150a18..65e11ccb563e 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -730,6 +730,38 @@ def _home_thread_env_var(platform_name: str) -> str: return f"{_home_target_env_var(platform_name)}_THREAD_ID" +_STARTUP_NOTIFY_MARKER = ".startup_notify.json" +_STARTUP_NOTIFY_TTL_SECONDS = 600 + + +def _startup_notification_path() -> Path: + """Return the service-restart startup notification marker path.""" + return _hermes_home / _STARTUP_NOTIFY_MARKER + + +def _read_startup_notification_targets(path: Path) -> list[dict]: + """Read startup-notification marker targets, tolerating legacy/malformed files. + + Returns a list of per-platform target dicts. Supports both the current + ``{"targets": [...]}`` shape (one entry per configured home channel) and the + legacy single-target ``{"platform": ..., "chat_id": ...}`` shape so a marker + written by an older build is still consumed on the next startup. Missing or + unparseable files yield an empty list rather than raising. + """ + if not path.exists(): + return [] + try: + data = json.loads(path.read_text()) + except Exception: + return [] + if isinstance(data, dict) and isinstance(data.get("targets"), list): + return [t for t in data["targets"] if isinstance(t, dict)] + if isinstance(data, dict) and data.get("platform") and data.get("chat_id"): + # Legacy single-target marker. + return [data] + return [] + + def _restart_notification_pending() -> bool: """Return True when a /restart completion marker is waiting to be delivered.""" return (_hermes_home / ".restart_notify.json").exists() @@ -3552,6 +3584,12 @@ async def _notify_active_sessions_of_shutdown(self) -> None: dedup_key = (platform.value, str(home.chat_id), str(home.thread_id) if home.thread_id else None) if dedup_key in notified: + self._write_startup_notification_marker( + platform=platform, + chat_id=str(home.chat_id), + thread_id=str(home.thread_id) if home.thread_id else None, + reason="restart" if self._restart_requested else "signal_shutdown", + ) continue try: @@ -3570,6 +3608,12 @@ async def _notify_active_sessions_of_shutdown(self) -> None: continue notified.add(dedup_key) + self._write_startup_notification_marker( + platform=platform, + chat_id=str(home.chat_id), + thread_id=str(home.thread_id) if home.thread_id else None, + reason="restart" if self._restart_requested else "signal_shutdown", + ) logger.info( "Sent shutdown notification to home channel %s:%s", platform.value, @@ -3583,6 +3627,49 @@ async def _notify_active_sessions_of_shutdown(self) -> None: e, ) + def _write_startup_notification_marker( + self, + *, + platform: Platform, + chat_id: str, + thread_id: Optional[str] = None, + reason: str = "signal_shutdown", + ) -> None: + """Record a home target to notify after the next startup. + + Stores one entry per home-channel platform so every configured home + channel — not just the last one written — receives the "back up" + message on startup. Repeated writes for the same delivery target + replace the prior entry instead of accumulating duplicates. + + Best-effort: this runs inside the shutdown teardown sequence, so a + disk / read-only / permission failure must never propagate out and + abort the rest of ``_stop_impl`` (exit code, runtime status, adapter + teardown). Errors are logged and swallowed. + """ + try: + target = { + "platform": platform.value, + "chat_id": str(chat_id), + "created_at": time.time(), + "reason": reason, + } + if thread_id: + target["thread_id"] = str(thread_id) + + path = _startup_notification_path() + key = (target["platform"], target["chat_id"], target.get("thread_id")) + targets = [ + t + for t in _read_startup_notification_targets(path) + if (t.get("platform"), t.get("chat_id"), t.get("thread_id")) != key + ] + targets.append(target) + + atomic_json_write(path, {"targets": targets}, indent=None) + except Exception as exc: + logger.warning("Failed to record startup notification marker: %s", exc) + def _finalize_shutdown_agents(self, active_agents: Dict[str, Any]) -> None: for agent in active_agents.values(): try: @@ -4403,22 +4490,7 @@ async def start(self) -> bool: if connected_count > 0: await asyncio.sleep(1.0) - # Notify the chat that initiated /restart that the gateway is back. - restart_notification_pending = _restart_notification_pending() - delivered_restart_target = await self._send_restart_notification() - - # Broadcast a lightweight "gateway is back" message to configured - # home channels only when this startup is resuming from /restart. If a - # /restart requester already received a direct completion notice in the - # same chat, skip the generic broadcast there to avoid duplicates while - # still allowing a home-channel fallback when the direct send fails. - if restart_notification_pending or delivered_restart_target is not None: - skip_home_targets = ( - {delivered_restart_target} if delivered_restart_target else None - ) - await self._send_home_channel_startup_notifications( - skip_targets=skip_home_targets, - ) + await self._send_startup_lifecycle_notifications() # Automatically continue fresh sessions that were interrupted by the # previous gateway restart/shutdown. The resume_pending flag is cleared @@ -14654,6 +14726,29 @@ async def _send_update_notification(self) -> bool: return True + async def _send_startup_lifecycle_notifications(self) -> None: + """Send queued startup lifecycle notifications without duplicating targets.""" + restart_notification_pending = _restart_notification_pending() + delivered_restart_target = await self._send_restart_notification() + skip_home_targets: set[tuple[str, str, Optional[str]]] = set() + if delivered_restart_target: + skip_home_targets.add(delivered_restart_target) + + delivered_startup_targets = await self._send_startup_notification_from_marker( + skip_targets=set(skip_home_targets), + ) + skip_home_targets.update(delivered_startup_targets) + + # Broadcast a lightweight "gateway is back" message to configured + # home channels only when this startup is resuming from /restart. If a + # /restart requester already received a direct completion notice in the + # same chat, skip the generic broadcast there to avoid duplicates while + # still allowing a home-channel fallback when the direct send fails. + if restart_notification_pending or delivered_restart_target is not None: + await self._send_home_channel_startup_notifications( + skip_targets=skip_home_targets or None, + ) + async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[str]]]: """Notify the chat that initiated /restart that the gateway is back.""" notify_path = _hermes_home / ".restart_notify.json" @@ -14717,6 +14812,103 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[ finally: notify_path.unlink(missing_ok=True) + async def _send_startup_notification_from_marker( + self, + *, + skip_targets: Optional[set[tuple[str, str, Optional[str]]]] = None, + ) -> set[tuple[str, str, Optional[str]]]: + """Notify each recorded home channel that a service-managed restart completed. + + The shutdown path records one entry per connected home channel, so every + configured home target receives the "back up" message — not just the + last platform written. Returns the set of targets actually delivered so + the caller can suppress duplicate generic home-channel broadcasts. + """ + notify_path = _startup_notification_path() + if not notify_path.exists(): + return set() + + skipped = set(skip_targets) if skip_targets else set() + delivered: set[tuple[str, str, Optional[str]]] = set() + try: + now = time.time() + for entry in _read_startup_notification_targets(notify_path): + platform_str = entry.get("platform") + chat_id = entry.get("chat_id") + thread_id = entry.get("thread_id") + created_at = entry.get("created_at") + + if not platform_str or not chat_id: + continue + if not isinstance(created_at, (int, float)): + continue + if now - float(created_at) > _STARTUP_NOTIFY_TTL_SECONDS: + logger.info( + "Startup notification marker target %s:%s expired; skipping", + platform_str, + chat_id, + ) + continue + + target = (str(platform_str), str(chat_id), str(thread_id) if thread_id else None) + if target in skipped or target in delivered: + continue + + try: + platform = Platform(platform_str) + except ValueError: + logger.debug("Startup notification skipped: unknown platform %s", platform_str) + continue + + platform_cfg = self.config.platforms.get(platform) + if platform_cfg is not None and not platform_cfg.gateway_restart_notification: + logger.info( + "Startup notification suppressed: %s has gateway_restart_notification=false", + platform_str, + ) + continue + + adapter = self.adapters.get(platform) + if not adapter: + logger.debug( + "Startup notification skipped: %s adapter not connected", + platform_str, + ) + continue + + try: + metadata = {"thread_id": thread_id} if thread_id else None + result = await adapter.send( + str(chat_id), + "♻ Gateway restarted successfully. Hermes is back and ready.", + metadata=metadata, + ) + if result is not None and getattr(result, "success", True) is False: + logger.warning( + "Startup notification to %s:%s was not delivered: %s", + platform_str, + chat_id, + getattr(result, "error", "send returned success=False"), + ) + continue + + logger.info("Sent startup notification to %s:%s", platform_str, chat_id) + delivered.add(target) + except Exception as exc: + logger.warning( + "Startup notification to %s:%s failed: %s", + platform_str, + chat_id, + exc, + ) + + return delivered + except Exception as exc: + logger.warning("Startup notification failed: %s", exc) + return delivered + finally: + notify_path.unlink(missing_ok=True) + async def _send_home_channel_startup_notifications( self, *, diff --git a/tests/gateway/test_restart_notification.py b/tests/gateway/test_restart_notification.py index e7a931f8f8ad..c866bad158d3 100644 --- a/tests/gateway/test_restart_notification.py +++ b/tests/gateway/test_restart_notification.py @@ -1,13 +1,14 @@ """Tests for /restart notification — the gateway notifies the requester on comeback.""" import json +import time from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest import gateway.run as gateway_run -from gateway.config import HomeChannel, Platform +from gateway.config import HomeChannel, Platform, PlatformConfig from gateway.platforms.base import MessageEvent, MessageType, SendResult from gateway.session import build_session_key from tests.gateway.restart_test_helpers import ( @@ -226,6 +227,67 @@ def _fake_save_env_value(key, value): # ── home-channel startup notifications ───────────────────────────────────── +@pytest.mark.asyncio +async def test_shutdown_home_notification_writes_startup_notify_marker(tmp_path, monkeypatch): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + runner, adapter = make_restart_runner() + runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel( + platform=Platform.TELEGRAM, + chat_id="home-42", + name="Ops Home", + thread_id="topic-7", + ) + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="shutdown")) + + await runner._notify_active_sessions_of_shutdown() + + notify_path = tmp_path / ".startup_notify.json" + assert notify_path.exists() + data = json.loads(notify_path.read_text()) + assert len(data["targets"]) == 1 + target = data["targets"][0] + assert target["platform"] == "telegram" + assert target["chat_id"] == "home-42" + assert target["thread_id"] == "topic-7" + assert target["reason"] == "signal_shutdown" + assert isinstance(target["created_at"], float) + + +@pytest.mark.asyncio +async def test_shutdown_home_notification_writes_startup_marker_when_home_was_active_session( + tmp_path, monkeypatch +): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + runner, adapter = make_restart_runner() + runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel( + platform=Platform.TELEGRAM, + chat_id="home-42", + name="Ops Home", + thread_id="topic-7", + ) + source = make_restart_source(chat_id="home-42", thread_id="topic-7") + session_key = build_session_key(source) + runner._running_agents[session_key] = object() + runner._cache_session_source(session_key, source) + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="shutdown")) + + await runner._notify_active_sessions_of_shutdown() + + # The active-session send already covered the home target, so shutdown + # delivery is deduped to one message — but the matching startup marker must + # still be persisted or the operator gets a half-handshake. + adapter.send.assert_awaited_once() + data = json.loads((tmp_path / ".startup_notify.json").read_text()) + assert len(data["targets"]) == 1 + target = data["targets"][0] + assert target["platform"] == "telegram" + assert target["chat_id"] == "home-42" + assert target["thread_id"] == "topic-7" + assert target["reason"] == "signal_shutdown" + + @pytest.mark.asyncio async def test_send_home_channel_startup_notification_to_configured_home(tmp_path, monkeypatch): monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) @@ -336,6 +398,121 @@ async def test_send_home_channel_startup_notification_ignores_false_send_result( adapter.send.assert_called_once() +@pytest.mark.asyncio +async def test_startup_notify_marker_sends_back_up_message_and_cleans_up(tmp_path, monkeypatch): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + notify_path = tmp_path / ".startup_notify.json" + notify_path.write_text(json.dumps({ + "platform": "telegram", + "chat_id": "home-42", + "created_at": time.time(), + "reason": "signal_shutdown", + })) + + runner, adapter = make_restart_runner() + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="startup")) + + delivered = await runner._send_startup_notification_from_marker() + + assert delivered == {("telegram", "home-42", None)} + adapter.send.assert_called_once_with( + "home-42", + "♻ Gateway restarted successfully. Hermes is back and ready.", + metadata=None, + ) + assert not notify_path.exists() + + +@pytest.mark.asyncio +async def test_startup_notify_marker_preserves_thread_metadata(tmp_path, monkeypatch): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + notify_path = tmp_path / ".startup_notify.json" + notify_path.write_text(json.dumps({ + "platform": "telegram", + "chat_id": "parent-42", + "thread_id": "topic-7", + "created_at": time.time(), + "reason": "signal_shutdown", + })) + + runner, adapter = make_restart_runner() + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="startup")) + + delivered = await runner._send_startup_notification_from_marker() + + assert delivered == {("telegram", "parent-42", "topic-7")} + adapter.send.assert_called_once_with( + "parent-42", + "♻ Gateway restarted successfully. Hermes is back and ready.", + metadata={"thread_id": "topic-7"}, + ) + assert not notify_path.exists() + + +@pytest.mark.asyncio +async def test_startup_notify_marker_skipped_when_flag_disabled(tmp_path, monkeypatch): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + notify_path = tmp_path / ".startup_notify.json" + notify_path.write_text(json.dumps({ + "platform": "telegram", + "chat_id": "home-42", + "created_at": time.time(), + "reason": "signal_shutdown", + })) + + runner, adapter = make_restart_runner() + runner.config.platforms[Platform.TELEGRAM].gateway_restart_notification = False + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="startup")) + + delivered = await runner._send_startup_notification_from_marker() + + assert delivered == set() + adapter.send.assert_not_called() + assert not notify_path.exists() + + +@pytest.mark.asyncio +async def test_startup_notify_marker_ignores_and_unlinks_stale_marker(tmp_path, monkeypatch): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + notify_path = tmp_path / ".startup_notify.json" + notify_path.write_text(json.dumps({ + "platform": "telegram", + "chat_id": "home-42", + "created_at": time.time() - gateway_run._STARTUP_NOTIFY_TTL_SECONDS - 1, + "reason": "signal_shutdown", + })) + + runner, adapter = make_restart_runner() + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="late")) + + delivered = await runner._send_startup_notification_from_marker() + + assert delivered == set() + adapter.send.assert_not_called() + assert not notify_path.exists() + + +@pytest.mark.asyncio +async def test_startup_lifecycle_notifications_consume_startup_marker_after_restart_marker( + tmp_path, monkeypatch +): + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + runner, _adapter = make_restart_runner() + runner._send_restart_notification = AsyncMock(return_value=("telegram", "42", None)) + runner._send_startup_notification_from_marker = AsyncMock(return_value={("telegram", "home-42", None)}) + runner._send_home_channel_startup_notifications = AsyncMock(return_value=set()) + + await runner._send_startup_lifecycle_notifications() + + runner._send_restart_notification.assert_awaited_once() + runner._send_startup_notification_from_marker.assert_awaited_once_with( + skip_targets={("telegram", "42", None)} + ) + runner._send_home_channel_startup_notifications.assert_awaited_once_with( + skip_targets={("telegram", "42", None), ("telegram", "home-42", None)} + ) + + # ── _send_restart_notification ─────────────────────────────────────────── @@ -622,3 +799,95 @@ async def test_shutdown_notifications_use_cached_live_thread_source_when_origin_ "⚠️ Gateway shutting down — Your current task will be interrupted.", metadata={"thread_id": "topic-7"}, ) + + +# ── multi-platform fan-out + guarded marker write ──────────────────────── + + +@pytest.mark.asyncio +async def test_shutdown_marker_fans_out_to_every_home_channel(tmp_path, monkeypatch): + """A multi-platform operator gets the back-up message on EVERY home channel. + + Regression: the marker previously stored a single target and the shutdown + loop overwrote that one file per platform, so only the last platform + written was notified on startup (last-platform-wins). + """ + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + runner, telegram_adapter = make_restart_runner() + telegram_adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="tg")) + runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel( + platform=Platform.TELEGRAM, chat_id="tg-home", name="TG Home", + ) + + discord_adapter = MagicMock() + discord_adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="dc")) + runner.config.platforms[Platform.DISCORD] = PlatformConfig(enabled=True, token="***") + runner.config.platforms[Platform.DISCORD].home_channel = HomeChannel( + platform=Platform.DISCORD, chat_id="dc-home", name="DC Home", + ) + runner.adapters[Platform.DISCORD] = discord_adapter + + # Shutdown records one marker entry per home channel (not last-write-wins). + await runner._notify_active_sessions_of_shutdown() + + data = json.loads((tmp_path / ".startup_notify.json").read_text()) + recorded = {(t["platform"], t["chat_id"]) for t in data["targets"]} + assert recorded == {("telegram", "tg-home"), ("discord", "dc-home")} + + # Isolate the startup phase from the shutdown-warning sends above. + telegram_adapter.send.reset_mock() + discord_adapter.send.reset_mock() + + # Startup fans the back-up message out to BOTH home channels exactly once. + delivered = await runner._send_startup_notification_from_marker() + + assert delivered == { + ("telegram", "tg-home", None), + ("discord", "dc-home", None), + } + telegram_adapter.send.assert_awaited_once_with( + "tg-home", + "♻ Gateway restarted successfully. Hermes is back and ready.", + metadata=None, + ) + discord_adapter.send.assert_awaited_once_with( + "dc-home", + "♻ Gateway restarted successfully. Hermes is back and ready.", + metadata=None, + ) + assert not (tmp_path / ".startup_notify.json").exists() + + +@pytest.mark.asyncio +async def test_shutdown_marker_write_failure_does_not_abort_teardown(tmp_path, monkeypatch): + """A marker-write failure must be swallowed so it can't abort _stop_impl teardown. + + Regression: the marker write sat outside the per-home try/except, and + ``_notify_active_sessions_of_shutdown`` is awaited inside ``_stop_impl`` with + no surrounding try — so a disk / read-only / permission error would + propagate out and skip the rest of teardown (exit code, + ``_update_runtime_status("stopped")``, adapter disconnects). Asserting at + this boundary (the function that held the unguarded write) is what + guarantees teardown continues: if the write still raised, the awaited call + below would raise and fail the test. + """ + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + def _boom(*_args, **_kwargs): + raise OSError("read-only file system") + + monkeypatch.setattr(gateway_run, "atomic_json_write", _boom) + + runner, adapter = make_restart_runner() + runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel( + platform=Platform.TELEGRAM, chat_id="home-42", name="Ops Home", + ) + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="shutdown")) + + # Must not raise even though the marker write fails. + await runner._notify_active_sessions_of_shutdown() + + # The shutdown warning was still delivered; the failed marker just isn't written. + adapter.send.assert_awaited_once() + assert not (tmp_path / ".startup_notify.json").exists()