fix(telegram): add cause-agnostic wedged-recovery watchdog so the reconnect ladder can't freeze silently (#66377)

The Telegram gateway could go silently deaf for hours: the reconnect ladder
stalled mid-way (e.g. "attempt 4/10, reconnecting in 40s" then nothing) while
the process stayed active(running), so Restart=always never fired.

Root class: every recovery path — the ladder's re-entry
(_schedule_polling_recovery), the pending-update probe (_probe_pending_updates),
and PTB's error callback — gates new recovery on _polling_error_task.done(). If
that single task wedges on any hung await, all recovery returns early forever
and nothing retries.

The heartbeat loop is a separate task, so make it an independent, cause-agnostic
watchdog: if the same recovery task stays in-flight past
_POLLING_ERROR_TASK_STUCK_TIMEOUT (300s — well beyond a healthy ladder attempt's
bounded stop+drain+start+backoff), force a retryable-fatal so the background
reconnector rebuilds the adapter instead of relying on the frozen ladder. This
guarantees progress regardless of *where* the stall is (issue direction #1),
tracked locally so no task-assignment site needs to change.

Also salvages @koduri-mahesh-bhushan-chowdary's #66492 (drain-await timeout),
which closes the one concrete wedge vector documented in the incident
(_drain_polling_connections' unbounded shutdown()/initialize() on a wedged
CLOSE-WAIT pool). The watchdog covers the rest of the class.

Co-authored-by: Koduri Mahesh Bhushan Chowdary <mkoduri73@gmail.com>

plugins/platforms/telegram/adapter.py

@@ -17,6 +17,7 @@ import os
 import html as _html
 import re
 import threading
+import time
 from contextvars import ContextVar
 from datetime import datetime, timezone
 from typing import Dict, List, Optional, Set, Any
@@ -551,6 +552,16 @@ _UPDATER_START_TIMEOUT = 30.0
 # so the ladder always advances toward the fatal-restart escalation. Matches
 # _UPDATER_STOP_TIMEOUT. Refs: NousResearch/hermes-agent#66377
 _DRAIN_TIMEOUT = 15.0
+# Cause-agnostic wedged-recovery watchdog (#66377). Every recovery path (the
+# reconnect ladder's re-entry, the pending-update probe, PTB's error callback)
+# gates new recovery on ``_polling_error_task.done()``; if that task ever wedges
+# on a hung await that no local bound covers, the whole gateway goes silently
+# deaf with nothing retrying. The heartbeat loop force-escalates a recovery task
+# that stays in-flight far longer than any healthy ladder attempt could take —
+# stop (_UPDATER_STOP_TIMEOUT) + drain (2x_DRAIN_TIMEOUT) + start
+# (_UPDATER_START_TIMEOUT) + max backoff (60s) is ~135s, so 300s is
+# unambiguously stuck.
+_POLLING_ERROR_TASK_STUCK_TIMEOUT = 300.0
 # A generation is not healthy until the dedicated getUpdates request returns
 # successfully. This exceeds a normal long-poll cycle for healthy idle bots.
 _POLLING_PROGRESS_TIMEOUT = 60.0
@@ -2464,6 +2475,16 @@ class TelegramAdapter(BasePlatformAdapter):
         HEARTBEAT_INTERVAL = 90   # seconds between probes
         PROBE_TIMEOUT = 15        # seconds before declaring the path dead
 
+        # Wedged-recovery watchdog state (#66377). Tracked locally so no
+        # _polling_error_task assignment site needs to stamp a timestamp: the
+        # heartbeat notes when it first observes a given recovery task still
+        # in-flight, and force-escalates if the *same* task object is still
+        # running after _POLLING_ERROR_TASK_STUCK_TIMEOUT. A healthy ladder
+        # attempt completes (task done) or chains to a new task well before
+        # then, so a single long-lived task is unambiguously wedged.
+        stuck_task_ref: Optional[asyncio.Task] = None
+        stuck_task_since = 0.0
+
         while True:
             try:
                 await asyncio.sleep(HEARTBEAT_INTERVAL)
@@ -2471,6 +2492,42 @@ class TelegramAdapter(BasePlatformAdapter):
                     return
                 if self.has_fatal_error:
                     return
+
+                # Independent wedged-recovery watchdog (#66377): if the tracked
+                # recovery task has hung (any await no local bound covers), every
+                # other recovery path is gated behind it and returns early
+                # forever — the gateway stays alive but deaf. Force a
+                # retryable-fatal so the background reconnector rebuilds the
+                # adapter instead of relying on the frozen ladder.
+                recovery_task = self._polling_error_task
+                if recovery_task is not None and not recovery_task.done():
+                    now = time.monotonic()
+                    if recovery_task is not stuck_task_ref:
+                        stuck_task_ref = recovery_task
+                        stuck_task_since = now
+                    elif now - stuck_task_since > _POLLING_ERROR_TASK_STUCK_TIMEOUT:
+                        stuck_for = now - stuck_task_since
+                        logger.error(
+                            "[%s] Telegram reconnect task wedged for %.0fs with no "
+                            "ladder progress; forcing retryable-fatal so the gateway "
+                            "reconnects instead of staying silently deaf.",
+                            self.name, stuck_for,
+                        )
+                        try:
+                            recovery_task.cancel()
+                        except Exception:
+                            pass
+                        self._set_fatal_error(
+                            "telegram_network_error",
+                            "Telegram reconnect task wedged for %.0fs; forcing "
+                            "gateway reconnect." % stuck_for,
+                            retryable=True,
+                        )
+                        await self._notify_fatal_error()
+                        return
+                else:
+                    stuck_task_ref = None
+
                 bot = self._app.bot if self._app else None
                 if bot is None:
                     continue

tests/gateway/test_telegram_network_reconnect.py

@@ -373,6 +373,56 @@ async def test_reconnect_continues_if_drain_hangs(monkeypatch):
     )
 
 
+@pytest.mark.asyncio
+async def test_heartbeat_force_escalates_wedged_recovery_task(monkeypatch):
+    """#66377: the heartbeat is an independent, cause-agnostic watchdog.
+
+    Every recovery path (ladder re-entry, pending-update probe, PTB error
+    callback) gates new recovery on ``_polling_error_task.done()``. If that task
+    wedges on ANY hung await — not just the drain closed by #66492 — the gateway
+    stays alive but deaf with nothing retrying. The heartbeat must detect a
+    recovery task that stays in-flight past ``_POLLING_ERROR_TASK_STUCK_TIMEOUT``
+    and force a retryable-fatal so the background reconnector rebuilds the
+    adapter.
+    """
+    adapter = _make_adapter()
+
+    async def _wedged():
+        await asyncio.Event().wait()  # never completes — simulates the hang
+
+    wedged_task = asyncio.ensure_future(_wedged())
+    adapter._polling_error_task = wedged_task
+
+    mock_bot = MagicMock()
+    mock_bot.get_me = AsyncMock()
+    mock_app = MagicMock()
+    mock_app.bot = mock_bot
+    adapter._app = mock_app
+    adapter._probe_pending_updates = AsyncMock()
+    adapter._notify_fatal_error = AsyncMock()
+
+    # Controllable monotonic clock advanced by each (mocked) heartbeat sleep so
+    # the same wedged task is observed across the stuck threshold deterministically.
+    clock = [1000.0]
+
+    async def _fake_sleep(*_a, **_k):
+        clock[0] += 200.0
+
+    monkeypatch.setattr(tg_adapter.time, "monotonic", lambda: clock[0])
+
+    with patch("asyncio.sleep", new=AsyncMock(side_effect=_fake_sleep)):
+        await asyncio.wait_for(adapter._polling_heartbeat_loop(), timeout=5)
+
+    assert adapter.has_fatal_error, "wedged recovery task must force a fatal escalation"
+    adapter._notify_fatal_error.assert_awaited()
+
+    wedged_task.cancel()
+    try:
+        await wedged_task
+    except asyncio.CancelledError:
+        pass
+
+
 @pytest.mark.asyncio
 async def test_conflict_retry_also_drains_polling_connections():
     """_handle_polling_conflict must also drain the polling pool on retry."""