From 9e8fe42f667f8d7a39e826a9a59acfcefa16d2f7 Mon Sep 17 00:00:00 2001 From: Frowtek Date: Mon, 1 Jun 2026 03:29:03 +0300 Subject: [PATCH] fix(gateway): preserve pending /restart confirmation until the target platform reconnects When /restart fires, the requester's routing info is persisted to .restart_notify.json and delivered once at startup. If the requester's platform adapter hadn't reconnected yet, the startup attempt returned early but the finally block still deleted the marker, and the reconnect watcher never retried -- so the "Gateway restarted successfully" confirmation promised in the slash-command docs was permanently lost. Preserve the marker when the platform is still queued for reconnect (_failed_platforms), and retry _send_restart_notification() on the reconnect success path. The marker is consumed only after a successful or terminal delivery, so no duplicate notifications are sent. Platforms that will never reconnect still consume the marker as before, so no stale marker leaks into a later restart. Adds regression tests for the deferred-then-delivered flow and the reconnect-watcher retry wiring. --- gateway/run.py | 42 ++++++++-- tests/gateway/restart_test_helpers.py | 1 + tests/gateway/test_platform_reconnect.py | 40 ++++++++++ tests/gateway/test_restart_notification.py | 91 +++++++++++++++++++++- 4 files changed, 167 insertions(+), 7 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 933e88af3c1e..f930305db7e1 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5997,6 +5997,20 @@ async def _platform_reconnect_watcher(self) -> None: await build_channel_directory(self.adapters) except Exception: pass + + # A /restart confirmation may have been deferred at + # startup because this platform wasn't connected yet. + # Now that it's back, retry delivery — the marker is + # consumed only once the send succeeds (or terminally + # fails), so this is safe to call on every reconnect. + try: + await self._send_restart_notification() + except Exception: + logger.debug( + "Restart notification retry after %s reconnect failed", + platform.value, + exc_info=True, + ) # Check if the failure is non-retryable elif adapter.has_fatal_error and not adapter.fatal_error_retryable: self._update_platform_runtime_status( @@ -14903,6 +14917,15 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[ if not notify_path.exists(): return None + # Preserve the marker (instead of consuming it in the finally block) + # only when the requester's platform is still pending reconnect, so the + # confirmation can be retried once the platform comes back — the + # reconnect watcher calls this again on a successful reconnect. Every + # other outcome (delivered, suppressed, malformed, terminal send + # failure, or a platform that will never reconnect) consumes the marker + # so the same chat is never notified twice and no stale marker leaks + # into a later, unrelated restart. + preserve_marker = False try: data = json.loads(notify_path.read_text()) platform_str = data.get("platform") @@ -14916,10 +14939,18 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[ platform = Platform(platform_str) adapter = self.adapters.get(platform) if not adapter: - logger.debug( - "Restart notification skipped: %s adapter not connected", - platform_str, - ) + if platform in self._failed_platforms: + logger.info( + "Restart notification deferred: %s not connected yet, " + "will retry after reconnect", + platform_str, + ) + preserve_marker = True + else: + logger.debug( + "Restart notification skipped: %s adapter not connected", + platform_str, + ) return None platform_cfg = self.config.platforms.get(platform) @@ -14965,7 +14996,8 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[ logger.warning("Restart notification failed: %s", e) return None finally: - notify_path.unlink(missing_ok=True) + if not preserve_marker: + notify_path.unlink(missing_ok=True) async def _send_home_channel_startup_notifications( self, diff --git a/tests/gateway/restart_test_helpers.py b/tests/gateway/restart_test_helpers.py index a91816c4e207..3c8c57dc8951 100644 --- a/tests/gateway/restart_test_helpers.py +++ b/tests/gateway/restart_test_helpers.py @@ -60,6 +60,7 @@ def make_restart_runner( runner._exit_code = None runner._running_agents = {} runner._running_agents_ts = {} + runner._failed_platforms = {} runner._pending_messages = {} runner._pending_approvals = {} runner._pending_model_notes = {} diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py index 3cd507550c5b..1fe4a74d1bad 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -217,6 +217,46 @@ async def fake_sleep(n): assert Platform.TELEGRAM not in runner._failed_platforms assert Platform.TELEGRAM in runner.adapters + @pytest.mark.asyncio + async def test_reconnect_retries_restart_notification(self): + """Regression: after a platform reconnects, a /restart confirmation that + was deferred at startup (because the platform wasn't connected yet) is + retried. Without this the requester never learns the restart finished.""" + runner = _make_runner() + runner._sync_voice_mode_state_to_adapter = MagicMock() + runner._send_restart_notification = AsyncMock() + + platform_config = PlatformConfig(enabled=True, token="test") + runner._failed_platforms[Platform.TELEGRAM] = { + "config": platform_config, + "attempts": 1, + "next_retry": time.monotonic() - 1, # Already past retry time + } + + succeed_adapter = StubAdapter(succeed=True) + real_sleep = asyncio.sleep + + with patch.object(runner, "_create_adapter", return_value=succeed_adapter): + with patch("gateway.run.build_channel_directory", create=True): + async def run_one_iteration(): + runner._running = True + call_count = 0 + + async def fake_sleep(n): + nonlocal call_count + call_count += 1 + if call_count > 1: + runner._running = False + await real_sleep(0) + + with patch("asyncio.sleep", side_effect=fake_sleep): + await runner._platform_reconnect_watcher() + + await run_one_iteration() + + assert Platform.TELEGRAM in runner.adapters + runner._send_restart_notification.assert_awaited_once() + @pytest.mark.asyncio async def test_reconnect_nonretryable_removed_from_queue(self): """Non-retryable errors should remove the platform from the retry queue.""" diff --git a/tests/gateway/test_restart_notification.py b/tests/gateway/test_restart_notification.py index 56be2337031c..31491c021b6a 100644 --- a/tests/gateway/test_restart_notification.py +++ b/tests/gateway/test_restart_notification.py @@ -7,7 +7,7 @@ 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 ( @@ -422,7 +422,9 @@ async def test_send_restart_notification_noop_when_no_file(tmp_path, monkeypatch @pytest.mark.asyncio async def test_send_restart_notification_skips_when_adapter_missing(tmp_path, monkeypatch): - """If the requester's platform isn't connected, clean up without crashing.""" + """If the requester's platform isn't connected AND isn't queued for + reconnect, clean up without crashing — nothing will ever retry it, so a + preserved marker would only leak into a later, unrelated restart.""" monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) notify_path = tmp_path / ".restart_notify.json" @@ -432,6 +434,8 @@ async def test_send_restart_notification_skips_when_adapter_missing(tmp_path, mo })) runner, _adapter = make_restart_runner() + # discord is not pending reconnect (empty _failed_platforms), so there is + # no retry coming — the marker must be consumed. await runner._send_restart_notification() @@ -439,6 +443,89 @@ async def test_send_restart_notification_skips_when_adapter_missing(tmp_path, mo assert not notify_path.exists() +@pytest.mark.asyncio +async def test_send_restart_notification_deferred_when_platform_pending_reconnect( + tmp_path, monkeypatch +): + """Regression: if the requester's platform is still reconnecting at startup, + the marker must be PRESERVED so the confirmation can be retried — not + deleted (which permanently dropped the /restart confirmation).""" + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + notify_path = tmp_path / ".restart_notify.json" + notify_path.write_text(json.dumps({ + "platform": "discord", # not connected yet, queued for reconnect + "chat_id": "42", + })) + + runner, _adapter = make_restart_runner() # only telegram adapter connected + runner._failed_platforms = { + Platform.DISCORD: { + "config": PlatformConfig(enabled=True, token="t"), + "attempts": 1, + "next_retry": 0.0, + } + } + + delivered_target = await runner._send_restart_notification() + + # Nothing delivered yet, but the marker survives for the reconnect retry. + assert delivered_target is None + assert notify_path.exists() + + +@pytest.mark.asyncio +async def test_send_restart_notification_retried_after_reconnect_delivers( + tmp_path, monkeypatch +): + """Regression (end-to-end at the notification layer): a confirmation that + was deferred because the platform was reconnecting is delivered to the + requester's chat/thread once the platform comes back, then cleaned up.""" + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + notify_path = tmp_path / ".restart_notify.json" + notify_path.write_text(json.dumps({ + "platform": "telegram", + "chat_id": "42", + "chat_type": "dm", + "thread_id": "777", + })) + + runner, adapter = make_restart_runner() + # Simulate startup: telegram is not connected yet, queued for reconnect. + runner.adapters = {} + runner._failed_platforms = { + Platform.TELEGRAM: { + "config": PlatformConfig(enabled=True, token="t"), + "attempts": 1, + "next_retry": 0.0, + } + } + + # First attempt (startup): adapter missing → marker preserved, no send. + first = await runner._send_restart_notification() + assert first is None + assert notify_path.exists() + + # Platform reconnects: adapter is back and no longer in the retry queue. + adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="m-1")) + runner.adapters = {Platform.TELEGRAM: adapter} + runner._failed_platforms = {} + + # Retry (driven by the reconnect watcher) delivers to the requester thread. + second = await runner._send_restart_notification() + assert second == ("telegram", "42", "777") + adapter.send.assert_called_once() + call_args = adapter.send.call_args + assert call_args[0][0] == "42" # chat_id + assert call_args[1]["metadata"] == { + "thread_id": "777", + "telegram_dm_topic_reply_fallback": True, + "direct_messages_topic_id": "777", + } + assert not notify_path.exists() + + @pytest.mark.asyncio async def test_send_restart_notification_cleans_up_on_send_failure( tmp_path, monkeypatch