mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-26 17:38:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(telegram): require initial polling readiness

Browse source 
Use wall deadlines for deleteWebhook and start_polling, then fail cold startup unless getUpdates proves progress. This lets the gateway discard partial PTB state and retry with a fresh adapter.\n\nRefs #67498
This commit is contained in:
mannnrachman 2026-07-22 17:11:47 +08:00 • committed by Teknium
parent 1cfc3425af
commit 83e30e371f
3 changed files with 98 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -545,6 +545,10 @@ _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
# Initial connect is not healthy until the dedicated getUpdates request completes
# one successful round trip. Unlike reconnect, initial bootstrap must fail closed
# so GatewayRunner disposes the partial adapter and retries with a fresh PTB app.
_INITIAL_POLLING_PROGRESS_TIMEOUT = 60.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
@ -2129,6 +2133,7 @@ class TelegramAdapter(BasePlatformAdapter):
*,
drop_pending_updates: bool,
error_callback,
abandon_app_on_timeout: bool = False,
) -> None:
"""Start one generation and verify real getUpdates progress."""
if getattr(self, "_polling_teardown_started", False):
@ -2151,13 +2156,22 @@ class TelegramAdapter(BasePlatformAdapter):
context_token = _POLLING_GENERATION_CONTEXT.set(generation)
try:
await asyncio.wait_for(
# asyncio.wait_for can wait forever for cancellation to escape
# httpcore/AnyIO shielded scopes (#58236/#67498). Reuse the
# proven wall-deadline helper and abandon the partial updater;
# caller recovery will dispose/rebuild the whole adapter.
await _await_with_thread_deadline(
app.updater.start_polling(
allowed_updates=Update.ALL_TYPES,
drop_pending_updates=drop_pending_updates,
error_callback=_generation_error_callback,
),
timeout=_UPDATER_START_TIMEOUT,
on_abandon=(
(lambda app=app: _shutdown_abandoned_app(app))
if abandon_app_on_timeout
else None
),
)
finally:
_POLLING_GENERATION_CONTEXT.reset(context_token)
@ -2264,12 +2278,14 @@ class TelegramAdapter(BasePlatformAdapter):
self._background_tasks.add(self._polling_error_task)
self._polling_error_task.add_done_callback(self._background_tasks.discard)
async def _delete_webhook_best_effort(self) -> bool:
"""Clear any stale webhook, but never fail polling on a network error.
async def _delete_webhook_best_effort(
self, *, require_success: bool = False
) -> bool:
"""Clear stale webhook, optionally failing closed on initial connect.
Returns True when the webhook was cleared (or there was nothing to do)
and False when a transient network error was swallowed so bootstrap can
continue to polling; the reconnect ladder recovers from there.
Reconnect can recover a transient error in background. Cold startup uses
``require_success`` so GatewayRunner disposes the partial adapter and
retries with a fresh PTB Application instead of publishing degraded state.
"""
if not self._bot:
return False
@ -2277,10 +2293,19 @@ class TelegramAdapter(BasePlatformAdapter):
if not callable(delete_webhook):
return True
try:
await delete_webhook(drop_pending_updates=False)
# Same shielded-cancellation class as initialize/start_polling:
# never let a wedged duplicate deleteWebhook pin initial connect.
await _await_with_thread_deadline(
delete_webhook(drop_pending_updates=False),
timeout=_UPDATER_START_TIMEOUT,
)
return True
except Exception as err:
if self._looks_like_network_error(err):
if require_success:
raise OSError(
"Telegram deleteWebhook did not complete during initial connect"
) from err
logger.warning(
"[%s] deleteWebhook failed with a recoverable network error; "
"continuing to polling so getUpdates/retry can recover: %s",
@ -2290,12 +2315,19 @@ class TelegramAdapter(BasePlatformAdapter):
return False
raise
async def _start_polling_resilient(self, *, drop_pending_updates: bool, error_callback) -> bool:
"""Start PTB polling; on a transient bootstrap failure, recover in background.
async def _start_polling_resilient(
self,
*,
drop_pending_updates: bool,
error_callback,
require_progress: bool = False,
) -> bool:
"""Start PTB polling and optionally require real getUpdates readiness.
Returns True when polling started, False when a transient conflict or
network error was scheduled for background recovery instead of raising
(keeping the gateway process alive).
Reconnects may recover in background. Initial connect sets
``require_progress`` so a bootstrap failure or missing first successful
getUpdates response raises; GatewayRunner then disposes this partial
adapter and retries with a fresh PTB Application.
"""
if getattr(self, "_polling_teardown_started", False):
return False
@ -2312,13 +2344,30 @@ class TelegramAdapter(BasePlatformAdapter):
self._app,
drop_pending_updates=drop_pending_updates,
error_callback=error_callback,
abandon_app_on_timeout=require_progress,
)
if require_progress:
try:
await _await_with_thread_deadline(
self._polling_progress_event.wait(),
timeout=_INITIAL_POLLING_PROGRESS_TIMEOUT,
)
except asyncio.TimeoutError as exc:
raise OSError(
"Telegram getUpdates made no progress during initial connect"
) from exc
if not self._polling_progress_event.is_set():
raise OSError(
"Telegram getUpdates did not become ready during initial connect"
)
return True
except _PollingLifecycleAbort:
return False
except Exception as err:
if getattr(self, "_polling_teardown_started", False):
return False
if require_progress:
raise
if self._looks_like_polling_conflict(err):
logger.warning(
"[%s] Telegram polling bootstrap conflict; gateway stays alive "
@ -3815,7 +3864,9 @@ class TelegramAdapter(BasePlatformAdapter):
# updates. Best-effort: a transient Bot API network error here
# must not fail gateway startup — degrade to background polling
# recovery instead.
await self._delete_webhook_best_effort()
await self._delete_webhook_best_effort(
require_success=not is_reconnect
)
loop = asyncio.get_running_loop()
@ -3853,6 +3904,7 @@ class TelegramAdapter(BasePlatformAdapter):
# sent while the bot was offline are delivered (#46621).
drop_pending_updates=not is_reconnect,
error_callback=_polling_error_callback,
require_progress=not is_reconnect,
)
if not polling_started:
logger.warning(

9
tests/gateway/test_telegram_polling_progress.py
View file

@ -138,13 +138,20 @@ async def test_polling_disconnect_webhook_reconnect_heals_webhook_send_path(monk
adapter = _make_adapter()
polling_app = _lifecycle_app()
webhook_app = _lifecycle_app()
async def start_polling_with_progress(**_kwargs):
adapter._record_polling_progress(adapter._polling_generation)
polling_app.updater.start_polling = AsyncMock(
side_effect=start_polling_with_progress
)
_configure_lifecycle_connect(monkeypatch, adapter, [polling_app, webhook_app])
monkeypatch.delenv("TELEGRAM_WEBHOOK_URL", raising=False)
monkeypatch.delenv("TELEGRAM_WEBHOOK_SECRET", raising=False)
assert await adapter.connect() is True
assert adapter._webhook_mode is False
assert adapter._send_path_degraded is True
assert adapter._send_path_degraded is False
await adapter.disconnect()
monkeypatch.setenv("TELEGRAM_WEBHOOK_URL", "https://example.test/telegram")

35
tests/gateway/test_telegram_start_polling_timeout.py
View file

@ -8,9 +8,10 @@ httpx connection pool — it neither returns nor raises. Unbounded, that wedges:
2. the heartbeat loop (sees the recovery task as alive-but-wedged and skips),
3. the fatal-error escalation (never reached).
The fix wraps every ``start_polling()`` await in ``asyncio.wait_for`` with
``_UPDATER_START_TIMEOUT`` so a hung call raises and feeds the existing retry
ladder. These tests patch the timeout down to keep the suite fast.
The fix wraps every ``start_polling()`` await in the wall-deadline helper with
``_UPDATER_START_TIMEOUT`` so a cancellation-shielded hung call still raises and
feeds the existing retry ladder. These tests patch the timeout down to keep the
suite fast.
"""
import asyncio
import sys
@ -138,20 +139,34 @@ async def test_start_polling_success_path_unaffected(monkeypatch):
def test_every_start_polling_call_site_is_time_bounded():
"""Bug-class contract: every `updater.start_polling(` await in the adapter
must be wrapped in asyncio.wait_for. A new unbounded call site reintroduces
the #59614 wedge."""
"""Every updater.start_polling call must use the wall-deadline helper."""
import inspect
import re
src = inspect.getsource(tg_adapter)
# Find each start_polling( call and check an enclosing wait_for within the
# preceding 6 lines (the wrapper always sits directly above).
lines = src.splitlines()
unbounded = []
for i, line in enumerate(lines):
if re.search(r"updater\.start_polling\(", line) and "def " not in line:
window = "\n".join(lines[max(0, i - 6):i + 1])
if "wait_for" not in window:
window = "\n".join(lines[max(0, i - 8):i + 1])
if "_await_with_thread_deadline" not in window:
unbounded.append((i + 1, line.strip()))
assert not unbounded, f"unbounded start_polling() call sites: {unbounded}"
@pytest.mark.asyncio
async def test_initial_connect_requires_get_updates_progress(monkeypatch):
"""Cold connect must not claim success before real getUpdates I/O."""
monkeypatch.setattr(tg_adapter, "_INITIAL_POLLING_PROGRESS_TIMEOUT", 0.05)
a = _bare_adapter()
app = MagicMock()
app.updater = AsyncMock()
app.updater.start_polling = AsyncMock(return_value=None)
a._app = app
with pytest.raises(OSError, match="getUpdates made no progress"):
await a._start_polling_resilient(
drop_pending_updates=True,
error_callback=None,
require_progress=True,
)