fix(telegram): bound polling drain so wedged pool close can't stall reconnect ladder (#66377)

_drain_polling_connections() awaited polling_req.shutdown() and
.initialize() without a timeout. When the getUpdates httpx connection is
wedged on a stale CLOSE-WAIT socket, that close can block forever, hanging
_handle_polling_network_error (the tracked _polling_error_task). The task
never completes, so every escalation path — _schedule_polling_recovery,
_probe_pending_updates, the heartbeat verifier — stays gated behind its
in-flight guard, the ladder freezes mid-way, _set_fatal_error is never
reached, and Restart=always never fires: the gateway is alive but silently
dead.

Wrap both drain awaits in asyncio.wait_for with a new module-level
_DRAIN_TIMEOUT (15.0s, matching _UPDATER_STOP_TIMEOUT), mirroring the
existing bounded stop()/start_polling() sites. On timeout the drain logs and
continues, so the handler task completes and the ladder always advances
toward the fatal-restart escalation.

Adds test_reconnect_continues_if_drain_hangs, which wedges the drain and
asserts the handler still reaches start_polling within a hard bound.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

plugins/platforms/telegram/adapter.py

@@ -543,6 +543,14 @@ _UPDATER_STOP_TIMEOUT = 15.0
 # reconnect ladder from stalling indefinitely and allows the heartbeat loop to
 # trigger its own recovery path. Refs: NousResearch/hermes-agent#59614
 _UPDATER_START_TIMEOUT = 30.0
+# shutdown()/initialize() on the getUpdates httpx request close and rebuild the
+# connection pool. When a connection is wedged on a stale CLOSE-WAIT socket that
+# close can block forever, hanging _drain_polling_connections() and freezing the
+# whole reconnect ladder (the tracked _polling_error_task never completes, so
+# every escalation path stays gated behind its in-flight guard). Bound the drain
+# so the ladder always advances toward the fatal-restart escalation. Matches
+# _UPDATER_STOP_TIMEOUT. Refs: NousResearch/hermes-agent#66377
+_DRAIN_TIMEOUT = 15.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
@@ -1987,20 +1995,22 @@ class TelegramAdapter(BasePlatformAdapter):
         except Exception:
             return
         try:
-            await polling_req.shutdown()
+            # Bounded: a wedged CLOSE-WAIT socket can make this close hang
+            # forever and freeze the reconnect ladder (#66377).
+            await asyncio.wait_for(polling_req.shutdown(), timeout=_DRAIN_TIMEOUT)
         except Exception:
             logger.debug(
-                "[%s] Polling request shutdown failed (non-fatal)",
+                "[%s] Polling request shutdown failed/timed out (non-fatal)",
                 self.name, exc_info=True,
             )
         try:
-            await polling_req.initialize()
+            await asyncio.wait_for(polling_req.initialize(), timeout=_DRAIN_TIMEOUT)
             logger.debug(
                 "[%s] Polling request pool drained before reconnect", self.name
             )
         except Exception:
             logger.debug(
-                "[%s] Polling request re-initialize failed (non-fatal)",
+                "[%s] Polling request re-initialize failed/timed out (non-fatal)",
                 self.name, exc_info=True,
             )
 

tests/gateway/test_telegram_network_reconnect.py

@@ -325,6 +325,54 @@ async def test_initialize_still_runs_when_shutdown_fails():
     mock_app.updater.start_polling.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_reconnect_continues_if_drain_hangs(monkeypatch):
+    """If the polling request drain HANGS (wedged httpx pool close on a
+    CLOSE-WAIT socket), the reconnect ladder must still advance rather than
+    freezing the tracked _polling_error_task forever.
+
+    Regression test for #66377: an unbounded ``shutdown()`` /
+    ``initialize()`` in ``_drain_polling_connections`` leaves the handler
+    task pending, which gates every escalation path and silently kills the
+    gateway. The drain awaits are bounded by ``_DRAIN_TIMEOUT``, so the
+    handler must complete and reach ``start_polling`` within a hard bound.
+    """
+    adapter = _make_adapter()
+    adapter._polling_network_error_count = 1
+
+    mock_app, mock_polling_req = _make_mock_app()
+
+    async def _hang(*args, **kwargs):
+        await asyncio.Event().wait()  # never returns
+
+    # Both drain awaits wedge indefinitely.
+    mock_polling_req.shutdown = AsyncMock(side_effect=_hang)
+    mock_polling_req.initialize = AsyncMock(side_effect=_hang)
+    adapter._app = mock_app
+
+    # Keep the drain timeout tiny so the test stays fast; the real default
+    # is generous enough not to truncate healthy closes.
+    monkeypatch.setattr(tg_adapter, "_DRAIN_TIMEOUT", 0.01, raising=False)
+
+    with patch("asyncio.sleep", new_callable=AsyncMock):
+        # Hard outer bound: on unfixed code the drain hangs forever and this
+        # trips; with the fix the inner wait_for releases well before it.
+        await asyncio.wait_for(
+            adapter._handle_polling_network_error(Exception("Timed out")),
+            timeout=5,
+        )
+
+    # Ladder advanced past the wedged drain despite it never returning.
+    mock_app.updater.start_polling.assert_called_once()
+    assert adapter._polling_network_error_count == 2
+    # The tracked task must not be stuck pending — otherwise every
+    # escalation path stays gated behind an in-flight guard.
+    assert (
+        adapter._polling_error_task is None
+        or adapter._polling_error_task.done()
+    )
+
+
 @pytest.mark.asyncio
 async def test_conflict_retry_also_drains_polling_connections():
     """_handle_polling_conflict must also drain the polling pool on retry."""