From 2ae1630bb75ff835d011e38fa4a4d4fb484cbc3b Mon Sep 17 00:00:00 2001 From: Steven Tartakovsky Date: Tue, 15 Sep 2026 10:49:07 -0700 Subject: [PATCH] fix(gateway): persist and replay pending planned-restart notices --- gateway/run_adapters.py | 7 + gateway/run_notifications.py | 53 +++++- gateway/run_startup.py | 7 +- tests/gateway/test_restart_notice_replay.py | 189 ++++++++++++++++++++ 4 files changed, 249 insertions(+), 7 deletions(-) create mode 100644 tests/gateway/test_restart_notice_replay.py diff --git a/gateway/run_adapters.py b/gateway/run_adapters.py index 46897b6c3e..9f422d0714 100644 --- a/gateway/run_adapters.py +++ b/gateway/run_adapters.py @@ -781,6 +781,13 @@ class GatewayAdapterLifecycleMixin: logger.info("⚠ %s reconnected in degraded mode (receive path not yet confirmed)", platform.value) else: logger.info("✓ %s reconnected successfully", platform.value) + # Notification delivery must not hold up adapter recovery or other platforms' reconnects. + from gateway.run import _planned_restart_notification_pending + if _planned_restart_notification_pending(): + task = self._retain_background_task(asyncio.create_task( + self._replay_pending_planned_restart_notification(), + )) + task.add_done_callback(self._late_failure_callback("planned-restart notification replay failed")) # Responses rejected while down are owned by this live process (startup recovery cannot claim them). with _log_suppressed( logging.DEBUG, "failed-obligation redelivery after %s reconnect failed", diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index 5242a711d3..5dd898d779 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -15,7 +15,7 @@ import logging import time from contextlib import suppress from pathlib import Path -from typing import Any, Dict, Optional, cast +from typing import Any, Callable, Dict, Optional, cast from gateway.config import Platform, _BUILTIN_PLATFORM_VALUES from gateway.platforms.base import BasePlatformAdapter, _mark_notify_metadata @@ -799,13 +799,60 @@ class GatewayNotificationsMixin: return None return "Inference: Nous free tier (nous/welcome). Sign in for more: /login" + async def _replay_pending_planned_restart_notification(self) -> None: + """Checkpoint each successful home notice so unavailable targets survive boot/reconnect. + + Boot sends may outlive the restore gate and overlap reconnects. Serialize the read/send/ack + sequence; the marker also carries acknowledgments across process restarts. + """ + from gateway.run import _planned_restart_notification_path + from utils import atomic_json_write + + lock = getattr(self, "_planned_restart_notice_lock", None) + if lock is None: + lock = self._planned_restart_notice_lock = asyncio.Lock() + async with lock: + path = _planned_restart_notification_path() + if not path.exists(): + return + try: + data = json.loads(path.read_text(encoding="utf-8")) + delivered = {tuple(target) for target in data.get("delivered_targets", [])} + # Resolve obligations from configuration, never from the currently live transports. + # Removed homes and explicit notification opt-outs no longer owe a notice. + targets = { + _notice_target_key(platform.value, cfg.home_channel.chat_id, cfg.home_channel.thread_id) + for platform, cfg in self.config.platforms.items() + if cfg.home_channel and cfg.home_channel.chat_id and cfg.gateway_restart_notification + } + pending = targets - delivered + + def checkpoint(target=None): + if target is not None: + delivered.add(target) + pending.discard(target) + data["delivered_targets"] = list(delivered) + data["pending_targets"] = list(pending) + atomic_json_write(path, data) + + checkpoint() + await self._send_home_channel_startup_notifications( + skip_targets=delivered, on_delivered=checkpoint, + ) + if not pending: + path.unlink(missing_ok=True) + except Exception: + logger.warning("Planned-restart notification remains pending", exc_info=True) + async def _send_home_channel_startup_notifications( - self, *, skip_targets: Optional[set[tuple[str, str, Optional[str]]]] = None + self, *, skip_targets: Optional[set[tuple[str, str, Optional[str]]]] = None, + on_delivered: Optional[Callable[[tuple[str, str, Optional[str]]], None]] = None, ) -> set[tuple[str, str, Optional[str]]]: """Notify 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. + ``on_delivered`` persists each acknowledgment before attempting the next transport. """ delivered: set[tuple[str, str, Optional[str]]] = set() skipped = skip_targets or set() @@ -827,6 +874,8 @@ class GatewayNotificationsMixin: platform, home, transport, message, "Home-channel startup notification failed for %s:%s: %s", ): delivered.add(target) + if on_delivered is not None: + on_delivered(target) logger.info("Sent home-channel startup notification to %s:%s", platform.value, home.chat_id) return delivered diff --git a/gateway/run_startup.py b/gateway/run_startup.py index a66c17a548..30578220b5 100644 --- a/gateway/run_startup.py +++ b/gateway/run_startup.py @@ -212,16 +212,13 @@ class GatewayStartupMixin: ``_send_restart_notification`` and ``_redeliver_pending_obligations`` used to be awaited inline *before* ``_finish_startup_restore`` released the gate. See #91969. """ - from gateway.run import _clear_planned_restart_notification, _startup_restore_drain_timeout_secs + from gateway.run import _startup_restore_drain_timeout_secs claimed = await self._claim_pending_obligations() async def _boot_sends() -> None: await self._send_restart_notification() if planned_restart_notification_pending: - try: - await self._send_home_channel_startup_notifications(skip_targets=None) - finally: - _clear_planned_restart_notification() + await self._replay_pending_planned_restart_notification() await self._redeliver_claimed_obligations(claimed) boot_task = asyncio.create_task(_boot_sends()) diff --git a/tests/gateway/test_restart_notice_replay.py b/tests/gateway/test_restart_notice_replay.py new file mode 100644 index 0000000000..a9d0ef36a8 --- /dev/null +++ b/tests/gateway/test_restart_notice_replay.py @@ -0,0 +1,189 @@ +"""Regression for #112109: retain planned-restart notices until delivery succeeds. + +Uses the real boot notification pass, marker helpers, home-channel sender, and +DeliveryTransport. Patterns follow tests/gateway/test_restart_notification.py +and test_restart_resume_pending.py. Run with scripts/run_tests.sh for isolation. +""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import asyncio +import json + +import pytest + +import gateway.delivery as delivery +import gateway.run as gateway_run +from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig +from gateway.platforms.base import SendResult + + +@pytest.fixture +def boot_notice(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + # Await the boot task to completion and propagate failures deterministically. + monkeypatch.setattr(gateway_run, "_startup_restore_drain_timeout_secs", lambda: 0) + runner = object.__new__(gateway_run.GatewayRunner) + platform_config = PlatformConfig( + enabled=True, + gateway_restart_notification=True, + home_channel=HomeChannel( + platform=Platform.DISCORD, chat_id="unit-test-home", name="Test home" + ), + ) + runner.config = GatewayConfig( + platforms={Platform.DISCORD: platform_config}, + sessions_dir=tmp_path / "sessions", + ) + runner.adapters = {} + runner.delivery_router = SimpleNamespace(adapters=runner.adapters) + runner._failed_platforms = {} + runner._sync_voice_mode_state_to_adapter = Mock() + runner._bind_voice_input_callback = Mock() + runner._update_platform_runtime_status = Mock() + runner._redeliver_failed_obligations_for_platform = AsyncMock() + runner._schedule_resume_pending_sessions = Mock() + monkeypatch.setattr("gateway.channel_directory.build_channel_directory", AsyncMock()) + # Unrelated conversation recovery and optional account-status text are isolated. + runner._claim_pending_obligations = AsyncMock(return_value=[]) + runner._redeliver_claimed_obligations = AsyncMock(return_value=0) + runner._free_tier_startup_line = Mock(return_value=None) + # Keep the real requester-marker check; this case has only the planned marker. + assert not (tmp_path / ".restart_notify.json").exists() + marker = tmp_path / ".restart_pending.json" + marker.write_text("{}", encoding="utf-8") + adapter = SimpleNamespace(send_path_degraded=False, send=AsyncMock( + return_value=SendResult(success=True, message_id="unit-test-notice") + )) + return runner, platform_config, marker, adapter + + +@pytest.mark.asyncio +@pytest.mark.parametrize("live", [False, True], ids=[ + "no-live-transport-retained-and-replayed", + "live-transport-notice-sent-marker-consumed", +]) +async def test_planned_restart_boot_notice(boot_notice, monkeypatch, live): + runner, platform_config, marker, adapter = boot_notice + if live: + runner.adapters[Platform.DISCORD] = adapter + transport = ( + delivery.DeliveryTransport(adapter, platform_config, Platform.DISCORD) + if live else None + ) + # Confirm the stub represents the real resolver's result for this adapter map. + resolved = delivery.resolve_delivery_transport( + Platform.DISCORD, runner.config, runner.adapters + ) + assert resolved == transport + resolver = Mock(return_value=transport) + monkeypatch.setattr(delivery, "resolve_delivery_transport", resolver) + assert marker.exists() + assert gateway_run._planned_restart_notification_pending() + + await runner._await_startup_boot_sends( + planned_restart_notification_pending=gateway_run._planned_restart_notification_pending() + ) + + resolver.assert_called_once_with(Platform.DISCORD, runner.config, runner.adapters) + runner._claim_pending_obligations.assert_awaited_once_with() + runner._redeliver_claimed_obligations.assert_awaited_once_with([]) + if live: + adapter.send.assert_awaited_once_with( + "unit-test-home", + "♻️ Gateway online — Hermes is back and ready.", + metadata={"non_conversational": True}, + ) + else: + adapter.send.assert_not_called() + if not live: + assert marker.exists() + assert gateway_run._planned_restart_notification_pending() + resolver.return_value = delivery.DeliveryTransport(adapter, platform_config, Platform.DISCORD) + runner._failed_platforms[Platform.DISCORD] = {} + await runner._install_reconnected_adapter(Platform.DISCORD, adapter) + await asyncio.gather(*runner._background_tasks) + adapter.send.assert_awaited_once_with( + "unit-test-home", "♻️ Gateway online — Hermes is back and ready.", + metadata={"non_conversational": True}, + ) + assert not marker.exists() + assert not gateway_run._planned_restart_notification_pending() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("outage", ["unavailable", "rejected", "exception", "cancelled"]) +async def test_partial_notice_delivery_survives_restart_and_concurrent_replay(boot_notice, outage): + runner, _, marker, adapter = boot_notice + other = SimpleNamespace(send=AsyncMock(return_value=SendResult(success=True))) + # Deliver one destination first, then encounter an unavailable/failing/hung transport. + runner.config.platforms = { + Platform.TELEGRAM: PlatformConfig( + enabled=True, + home_channel=HomeChannel(platform=Platform.TELEGRAM, chat_id="other-home", thread_id="7", name="Other home"), + ), + **runner.config.platforms, + Platform.SLACK: PlatformConfig( + enabled=True, gateway_restart_notification=False, + home_channel=HomeChannel(platform=Platform.SLACK, chat_id="muted-home", name="Muted home"), + ), + } + runner.adapters[Platform.TELEGRAM] = other + started = asyncio.Event() + release = asyncio.Event() + + async def slow_send(*args, **kwargs): + started.set() + await release.wait() + return SendResult(success=True) + + if outage != "unavailable": + runner.adapters[Platform.DISCORD] = adapter + if outage == "rejected": + adapter.send.return_value = SendResult(success=False, error="temporarily unavailable") + elif outage == "exception": + adapter.send.side_effect = RuntimeError("transport disconnected") + elif outage == "cancelled": + adapter.send.side_effect = slow_send + + boot = asyncio.create_task(runner._await_startup_boot_sends(planned_restart_notification_pending=True)) + if outage == "cancelled": + await asyncio.wait_for(started.wait(), timeout=5) + # Acknowledgments must already be on disk while a later destination is hung. + assert json.loads(marker.read_text())["delivered_targets"] == [["telegram", "other-home", "7"]] + boot.cancel() + with pytest.raises(asyncio.CancelledError): + await boot + else: + await boot + data = json.loads(marker.read_text()) + assert data["delivered_targets"] == [["telegram", "other-home", "7"]] + assert data["pending_targets"] == [["discord", "unit-test-home", None]] + other.send.assert_awaited_once() + + # A new runner has no in-memory delivery history: dedupe must come from the marker. + recovered = object.__new__(gateway_run.GatewayRunner) + recovered.__dict__.update(runner.__dict__) + recovered.__dict__.pop("_planned_restart_notice_lock", None) + started.clear() + adapter.send.reset_mock() + adapter.send.side_effect = slow_send + recovered._failed_platforms[Platform.DISCORD] = {} + await asyncio.wait_for(recovered._install_reconnected_adapter(Platform.DISCORD, adapter), timeout=5) + await asyncio.wait_for(started.wait(), timeout=5) + # Installation completed even though notification delivery is still blocked. + assert marker.exists() + concurrent = asyncio.create_task(recovered._replay_pending_planned_restart_notification()) + release.set() + await asyncio.gather(concurrent, *recovered._background_tasks) + assert not marker.exists() + adapter.send.assert_awaited_once() + other.send.assert_awaited_once() + + recovered._failed_platforms[Platform.DISCORD] = {} + await recovered._install_reconnected_adapter(Platform.DISCORD, adapter) + await asyncio.gather(*recovered._background_tasks) + adapter.send.assert_awaited_once() + other.send.assert_awaited_once()