From c78aa0bad5dc96c8080b1d16def868f1ab039b0c Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 18 Jul 2026 19:55:49 +0530 Subject: [PATCH] refactor(gateway): dedupe detached-task consumer + reconnect backoff policy /simplify-code findings on the #66222 salvage: - consume_detached_task_result moves to agent/async_utils.py (shared home); gateway/run.py and the Discord adapter both had near-identical copies of the same pattern (a third lives in the telegram adapter). One canonical implementation, both new callsites import it. - Reconnect backoff formula min(30 * 2^(n-1), 300) was copied verbatim at 3 sites in run.py (primary watcher x2, secondary-profile reconnect), the third hardcoding the cap. Hoisted to module-level _reconnect_backoff() with a single _RECONNECT_BACKOFF_CAP so a future tune can't silently miss one path. Behavior-preserving: 70 gateway teardown/liveness/reconnect tests green. --- agent/async_utils.py | 16 ++++++++++++++ gateway/run.py | 32 ++++++++++++++-------------- plugins/platforms/discord/adapter.py | 10 ++++----- 3 files changed, 36 insertions(+), 22 deletions(-) diff --git a/agent/async_utils.py b/agent/async_utils.py index d268e1a3a84a..07442b63c543 100644 --- a/agent/async_utils.py +++ b/agent/async_utils.py @@ -66,3 +66,19 @@ def safe_schedule_threadsafe( coro.close() log.log(log_level, "%s: %s", log_message, exc) return None + + +def consume_detached_task_result(task: "asyncio.Future[Any]") -> None: + """Retrieve a detached task's result without surfacing cancellation. + + Used as an ``add_done_callback`` on tasks that were cancelled and + detached (e.g. an adapter close path that swallows ``CancelledError`` + past its teardown deadline). Observing ``task.exception()`` prevents + "exception was never retrieved" noise on the event loop; cancellation + and any terminal error are deliberately swallowed — the task's owner + already gave up on it. + """ + try: + task.exception() + except (asyncio.CancelledError, Exception): + pass diff --git a/gateway/run.py b/gateway/run.py index fb3fe77f6bd3..2247b2a63ca8 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -53,7 +53,7 @@ from typing import Awaitable, Callable, Dict, Optional, Any, List, Union # gateway is a long-running daemon, so its boot cost matters less than # preserving the established test-patch surface. from agent.account_usage import fetch_account_usage, render_account_usage_lines -from agent.async_utils import safe_schedule_threadsafe +from agent.async_utils import consume_detached_task_result, safe_schedule_threadsafe from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX from agent.i18n import t from hermes_cli.config import cfg_get @@ -2950,6 +2950,16 @@ async def _dispose_unused_adapter(adapter: "BasePlatformAdapter | None") -> None ) +# Max seconds between platform reconnect retries (primary watcher and +# secondary-profile reconnects share this policy — tune in one place). +_RECONNECT_BACKOFF_CAP = 300 + + +def _reconnect_backoff(attempt: int) -> int: + """Exponential reconnect backoff: 30s, 60s, 120s, ... capped at 5 min.""" + return min(30 * (2 ** (attempt - 1)), _RECONNECT_BACKOFF_CAP) + + class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, GatewaySlashCommandsMixin): """ Main gateway controller. @@ -3540,14 +3550,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if mode in {"voice_only", "all"} and key.startswith(prefix) ) - @staticmethod - def _consume_detached_adapter_cleanup_result(task: asyncio.Future[Any]) -> None: - """Retrieve a detached cleanup task result without surfacing cancellation.""" - try: - task.exception() - except (asyncio.CancelledError, Exception): - pass - async def _await_adapter_cleanup_with_timeout( self, awaitable: Awaitable[Any], timeout: float ) -> bool: @@ -3567,14 +3569,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew done, _pending = await asyncio.wait({task}, timeout=timeout) except asyncio.CancelledError: task.cancel() - task.add_done_callback(self._consume_detached_adapter_cleanup_result) + task.add_done_callback(consume_detached_task_result) raise if task in done: await task return True task.cancel() - task.add_done_callback(self._consume_detached_adapter_cleanup_result) + task.add_done_callback(consume_detached_task_result) return False async def _safe_adapter_disconnect(self, adapter, platform) -> None: @@ -8309,8 +8311,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew automatically — auto-pausing a recovered platform was the cause of bots silently staying dead after a transient DNS failure. """ - _BACKOFF_CAP = 300 # 5 minutes max between retries - await asyncio.sleep(10) # initial delay — let startup finish while self._running: if not self._failed_platforms: @@ -8441,7 +8441,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew error_code=adapter.fatal_error_code, error_message=adapter.fatal_error_message or "failed to reconnect", ) - backoff = min(30 * (2 ** (attempt - 1)), _BACKOFF_CAP) + backoff = _reconnect_backoff(attempt) info["attempts"] = attempt info["next_retry"] = time.monotonic() + backoff logger.info( @@ -8479,7 +8479,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew error_code=None, error_message=str(e), ) - backoff = min(30 * (2 ** (attempt - 1)), _BACKOFF_CAP) + backoff = _reconnect_backoff(attempt) info["attempts"] = attempt info["next_retry"] = time.monotonic() + backoff logger.warning( @@ -9325,7 +9325,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if not self._running: return attempts += 1 - backoff = min(30 * (2 ** (attempts - 1)), 300) + backoff = _reconnect_backoff(attempts) logger.info( "Secondary %s reconnect retry in %ds (profile: %s)", platform.value, diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 9356d2068f79..b373ef4b8872 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -26,6 +26,10 @@ from collections import defaultdict from contextlib import suppress from typing import Callable, Dict, List, Optional, Any, Tuple +from agent.async_utils import ( + consume_detached_task_result as _consume_background_task_result, +) + logger = logging.getLogger(__name__) @@ -153,12 +157,6 @@ def _abort_discord_websocket_transport(websocket: Any) -> bool: return True -def _consume_background_task_result(task: asyncio.Task) -> None: - """Retrieve a detached cleanup task result without surfacing cancellation.""" - with suppress(asyncio.CancelledError, Exception): - task.exception() - - async def _wait_for_ready_or_bot_exit( ready_event: asyncio.Event, bot_task: asyncio.Task,