mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge pull request #67241 from NousResearch/bb/telegram-reconnect-watchdog

Browse source 
fix(telegram): cause-agnostic wedged-recovery watchdog + bounded drain so the reconnect ladder can't freeze silently (#66377, supersedes #66492)
This commit is contained in:
brooklyn! 2026-07-18 21:48:19 -04:00 • committed by GitHub
commit 7235592ad2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 171 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

2
contributors/emails/mkoduri73@gmail.com Normal file
View file

@ -0,0 +1,2 @@
MaheshBhushan
# PR #66492 salvage (#66377)

75
plugins/platforms/telegram/adapter.py
View file

@ -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
@ -543,6 +544,24 @@ _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
# 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
@ -1987,20 +2006,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,
)
@ -2454,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)
@ -2461,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

98
tests/gateway/test_telegram_network_reconnect.py
View file

@ -325,6 +325,104 @@ 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_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."""