From 087732c8c60860888f6c8ac8b9e22271d5269e96 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:54:33 +0530 Subject: [PATCH] fix(telegram): widen fatal handoff to heartbeat watchdog path The wedged-recovery heartbeat watchdog (line 2526) calls _notify_fatal_error() directly from the heartbeat task. disconnect() cancels _polling_heartbeat_task unconditionally (no current_task guard, unlike _polling_error_task). Same bug class as #68406: the child disconnect cancels the heartbeat parent before the runner can queue reconnect. Widen _handoff_polling_fatal_error() to also clear _polling_heartbeat_task when it is the current task, and route the heartbeat watchdog call site through the handoff helper. Co-authored-by: Imgaojp <6065749+Imgaojp@users.noreply.github.com> --- plugins/platforms/telegram/adapter.py | 14 ++++---- .../test_telegram_network_reconnect.py | 36 +++++++++++++++++++ 2 files changed, 44 insertions(+), 6 deletions(-) diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 4bfb685b09b..5e158dec030 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -2523,7 +2523,7 @@ class TelegramAdapter(BasePlatformAdapter): "gateway reconnect." % stuck_for, retryable=True, ) - await self._notify_fatal_error() + await self._handoff_polling_fatal_error() return else: stuck_task_ref = None @@ -2988,15 +2988,17 @@ class TelegramAdapter(BasePlatformAdapter): """Notify the runner without letting child teardown cancel this owner. The runner bounds adapter cleanup in a child task. ``disconnect()`` - cancels the tracked polling-recovery task, so retaining the current - notifier in ``_polling_error_task`` would cancel the fatal callback - before the runner can finish its reconnect or shutdown decision. - Release only the current owner; unrelated recovery tasks remain under - teardown control. + cancels the tracked polling-recovery task and the heartbeat task, so + retaining the current notifier in either field would cancel the fatal + callback before the runner can finish its reconnect or shutdown + decision. Release only the current owner from whichever field tracks + it; unrelated tasks remain under teardown control. """ current_task = asyncio.current_task() if self._polling_error_task is current_task: self._polling_error_task = None + if getattr(self, "_polling_heartbeat_task", None) is current_task: + self._polling_heartbeat_task = None await self._notify_fatal_error() async def _create_dm_topic( diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index c2406e0a88e..4b30ef68839 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -259,6 +259,42 @@ async def test_retry_exhaustion_queues_reconnect_before_child_disconnect(tmp_pat assert runner._failed_platforms[Platform.TELEGRAM]["attempts"] == 0 +@pytest.mark.asyncio +async def test_heartbeat_watchdog_handoff_survives_child_disconnect(tmp_path): + """The wedged-recovery heartbeat watchdog must survive its fatal callback. + + The heartbeat loop force-escalates a stuck polling-recovery task. Like + the network/conflict terminal paths, the heartbeat task itself is the + owner that ``disconnect()`` cancels, so the fatal callback must release + ``_polling_heartbeat_task`` before notifying the runner. + """ + config = GatewayConfig( + platforms={ + Platform.TELEGRAM: PlatformConfig(enabled=True, token="test-token") + }, + sessions_dir=tmp_path / "sessions", + ) + runner = GatewayRunner(config) + adapter = _make_adapter() + adapter.set_fatal_error_handler(runner._handle_adapter_fatal_error) + runner.adapters = {Platform.TELEGRAM: adapter} + runner.delivery_router.adapters = runner.adapters + + # Simulate the heartbeat watchdog's fatal-escalation path directly. + adapter._set_fatal_error( + "telegram_network_error", + "Telegram reconnect task wedged; forcing gateway reconnect.", + retryable=True, + ) + heartbeat_task = asyncio.create_task(adapter._handoff_polling_fatal_error()) + adapter._polling_heartbeat_task = heartbeat_task + result = await asyncio.gather(heartbeat_task, return_exceptions=True) + + assert result == [None] + assert runner.adapters == {} + assert Platform.TELEGRAM in runner._failed_platforms + + # --------------------------------------------------------------------------- # Connection pool drain tests (PR #16466 salvage) # ---------------------------------------------------------------------------