diff --git a/astrbot/core/config/default.py b/astrbot/core/config/default.py index eac4bafc01..d61b4adab4 100644 --- a/astrbot/core/config/default.py +++ b/astrbot/core/config/default.py @@ -457,6 +457,10 @@ "telegram_command_auto_refresh": True, "telegram_command_register_interval": 300, "telegram_polling_restart_delay": 5.0, + "telegram_connect_timeout": 15.0, + "telegram_polling_watchdog_interval": 15.0, + "telegram_polling_watchdog_failure_threshold": 3, + "telegram_polling_watchdog_pending_update_threshold": 2, }, "Discord": { "id": "discord", @@ -776,6 +780,26 @@ "type": "float", "hint": "当轮询意外结束尝试自动重启时的延迟时间，理论上越短恢复越快，但过短（<0.1s）可能导致死循环针对 API 服务器的请求阻断。单位为秒。默认为 5s。", }, + "telegram_connect_timeout": { + "description": "Telegram 连接超时", + "type": "float", + "hint": "建立 Telegram Bot API 连接时的超时时间，单位为秒。默认为 15s。", + }, + "telegram_polling_watchdog_interval": { + "description": "Telegram 轮询看门狗检查间隔", + "type": "float", + "hint": "检查 Telegram 轮询是否仍有进展的间隔时间，单位为秒。最小为 1s，默认为 15s。", + }, + "telegram_polling_watchdog_failure_threshold": { + "description": "Telegram 轮询看门狗失败阈值", + "type": "int", + "hint": "看门狗连续请求失败多少次后重建轮询客户端。最小为 1，默认为 3。", + }, + "telegram_polling_watchdog_pending_update_threshold": { + "description": "Telegram 待处理更新停滞阈值", + "type": "int", + "hint": "看门狗连续多少次发现待处理更新没有投递进展后重建轮询客户端。最小为 1，默认为 2。", + }, "id": { "description": "机器人名称", "type": "string", diff --git a/astrbot/core/platform/sources/telegram/tg_adapter.py b/astrbot/core/platform/sources/telegram/tg_adapter.py index d8efd7f5d8..b69319c1e4 100644 --- a/astrbot/core/platform/sources/telegram/tg_adapter.py +++ b/astrbot/core/platform/sources/telegram/tg_adapter.py @@ -1,9 +1,12 @@ +from __future__ import annotations + import asyncio import os import re import sys import uuid from contextlib import suppress +from math import isfinite from typing import cast from apscheduler.events import EVENT_JOB_ERROR @@ -11,7 +14,7 @@ from telegram import BotCommand, Update from telegram.constants import ChatType from telegram.error import Forbidden, InvalidToken, NetworkError -from telegram.ext import ApplicationBuilder, ContextTypes, ExtBot, filters +from telegram.ext import Application, ApplicationBuilder, ContextTypes, ExtBot, filters from telegram.ext import MessageHandler as TelegramMessageHandler import astrbot.api.message_components as Comp @@ -42,6 +45,50 @@ from typing_extensions import override +def _get_bounded_config_number( + config: dict, + key: str, + default: int | float, + minimum: int | float, + value_type: type[int] | type[float], +) -> int | float: + """Read and validate a bounded numeric adapter setting. + + Args: + config: Telegram adapter configuration. + key: Configuration key to read. + default: Value used when the configured value is invalid. + minimum: Smallest accepted value. + value_type: Numeric type used to parse the configured value. + + Returns: + Parsed value, clamped to the configured minimum. + """ + raw_value = config.get(key, default) + try: + value = value_type(raw_value) + if not isfinite(value): + raise ValueError + except (TypeError, ValueError, OverflowError): + logger.warning( + "Invalid %r value %r in config, falling back to default %s", + key, + raw_value, + default, + ) + value = default + + if value < minimum: + logger.warning( + "Configured %r value %s is too small; enforcing minimum %s", + key, + value, + minimum, + ) + value = minimum + return value + + @register_platform_adapter("telegram", "telegram 适配器") class TelegramPlatformAdapter(Platform): def __init__( @@ -95,28 +142,76 @@ def __init__( self._polling_recovery_requested = asyncio.Event() self._consecutive_polling_failures = 0 self._last_polling_failure_at = 0.0 - raw_delay = self.config.get("telegram_polling_restart_delay", 5.0) - try: - delay = float(raw_delay) - except (TypeError, ValueError): - logger.warning( - "Invalid 'telegram_polling_restart_delay' value %r in config, " - "falling back to default 5.0s", - raw_delay, - ) - delay = 5.0 - - if delay < 0.1: - logger.warning( - "Configured 'telegram_polling_restart_delay' (%s) is too small; " - "enforcing minimum of 0.1s to avoid tight restart loops", - delay, - ) - delay = 0.1 - self._polling_restart_delay = delay + self._polling_restart_delay = cast( + float, + _get_bounded_config_number( + self.config, + "telegram_polling_restart_delay", + 5.0, + 0.1, + float, + ), + ) self._polling_recovery_threshold = 3 self._polling_failure_window = 60.0 + self._telegram_connect_timeout = cast( + float, + _get_bounded_config_number( + self.config, + "telegram_connect_timeout", + 15.0, + 0.1, + float, + ), + ) + self._polling_watchdog_interval = cast( + float, + _get_bounded_config_number( + self.config, + "telegram_polling_watchdog_interval", + 15.0, + 1.0, + float, + ), + ) + self._polling_watchdog_failure_threshold = cast( + int, + _get_bounded_config_number( + self.config, + "telegram_polling_watchdog_failure_threshold", + 3, + 1, + int, + ), + ) + self._polling_pending_update_threshold = cast( + int, + _get_bounded_config_number( + self.config, + "telegram_polling_watchdog_pending_update_threshold", + 2, + 1, + int, + ), + ) + # Stopping an updater has a separate deadline so tuning network + # connections cannot make adapter termination excessively slow. + self._polling_shutdown_timeout = 15.0 + self._polling_watchdog_failures = 0 + self._polling_pending_update_checks = 0 + self._received_message_count = 0 + self._last_watchdog_message_count = 0 + self._last_pending_update_count: int | None = None + self._next_polling_watchdog_at = 0.0 + self._replace_outbound_application_on_recovery = False self._application_started = False + # Lifecycle invariants: + # - application owns the active polling updater. + # - when _outbound_application exists, client is its bot. + # - polling recovery may preserve an initialized outbound application, + # so application and _outbound_application can intentionally differ. + self._outbound_application: Application | None = None + self._outbound_application_initialized = False self._build_application() # Media group handling @@ -129,24 +224,71 @@ def __init__( "telegram_media_group_max_wait", 10.0 ) # max seconds - hard cap to prevent indefinite delay + def _set_outbound_application( + self, + application: Application | None, + *, + initialized: bool, + ) -> None: + """Update outbound application ownership as one lifecycle transition. + + Args: + application: Application whose bot handles outbound requests, or + None after its resources have been released. + initialized: Whether the application completed initialization. + + Raises: + ValueError: If None is marked as initialized. + """ + if application is None and initialized: + raise ValueError("An absent outbound application cannot be initialized") + self._outbound_application = application + self._outbound_application_initialized = initialized + if application is not None: + self.client = application.bot + + def _request_polling_recovery( + self, + replace_outbound_application: bool = False, + ) -> None: + """Request polling recovery and optionally escalate its scope. + + Args: + replace_outbound_application: Whether recovery must also replace + the client used for outbound Bot API requests. + """ + if replace_outbound_application: + self._replace_outbound_application_on_recovery = True + self._polling_recovery_requested.set() + def _build_application(self) -> None: - self.application = ( - ApplicationBuilder() - .token(self.config["telegram_token"]) - .base_url(self.base_url) - .base_file_url(self.file_base_url) - .build() - ) + builder = ApplicationBuilder() + builder.token(self.config["telegram_token"]) + builder.base_url(self.base_url) + builder.base_file_url(self.file_base_url) + builder.connect_timeout(self._telegram_connect_timeout) + builder.get_updates_connect_timeout(self._telegram_connect_timeout) + self.application = builder.build() message_handler = TelegramMessageHandler( filters=filters.ALL, callback=self.message_handler, ) self.application.add_handler(message_handler) - self.client = self.application.bot - logger.debug(f"Telegram base url: {self.client.base_url}") + # Keep a healthy outbound client alive when only getUpdates needs recovery. + if self._outbound_application is None: + self._set_outbound_application(self.application, initialized=False) + self._polling_watchdog_failures = 0 + self._polling_pending_update_checks = 0 + self._last_watchdog_message_count = self._received_message_count + self._last_pending_update_count = None + self._next_polling_watchdog_at = 0.0 + self._replace_outbound_application_on_recovery = False + logger.debug(f"Telegram base url: {self.application.bot.base_url}") async def _start_application(self) -> None: await self.application.initialize() + if self.application is self._outbound_application: + self._set_outbound_application(self.application, initialized=True) await self.application.start() if self.enable_command_register: @@ -158,41 +300,129 @@ async def _shutdown_application( self, *, delete_commands: bool, + preserve_outbound_application: bool = False, ) -> None: + """Stop the current Telegram polling application. + + Args: + delete_commands: Whether to remove registered Telegram commands. + preserve_outbound_application: Whether to keep the outbound Bot API + application initialized while replacing only the polling client. + """ self._application_started = False + application = self.application + is_outbound_application = application is self._outbound_application - updater = self.application.updater + updater = application.updater if updater is not None: with suppress(Exception): - await updater.stop() + await asyncio.wait_for( + updater.stop(), + timeout=self._polling_shutdown_timeout, + ) if delete_commands and self.enable_command_register: with suppress(Exception): await self.client.delete_my_commands() with suppress(Exception): - await self.application.stop() + await application.stop() + + if ( + preserve_outbound_application + and is_outbound_application + and self._outbound_application_initialized + ): + return - shutdown = getattr(self.application, "shutdown", None) + shutdown = getattr(application, "shutdown", None) if shutdown is not None: with suppress(Exception): await shutdown() + if is_outbound_application: + self._set_outbound_application(None, initialized=False) async def _recreate_application(self) -> None: if self._terminating: self._polling_recovery_requested.clear() return + replace_outbound_application = ( + self._replace_outbound_application_on_recovery + and self.application is self._outbound_application + ) logger.warning( - "Telegram polling hit repeated network errors; rebuilding the " - "Telegram application and HTTP client.", + "Telegram polling recovery requested; rebuilding %s.", + "the polling and outbound clients" + if replace_outbound_application + else "the polling client", + ) + await self._shutdown_application( + delete_commands=False, + preserve_outbound_application=not replace_outbound_application, ) - await self._shutdown_application(delete_commands=False) self._build_application() self._consecutive_polling_failures = 0 self._last_polling_failure_at = 0.0 self._polling_recovery_requested.clear() + async def _check_polling_health(self) -> None: + """Request recovery when polling has no observable delivery progress.""" + try: + webhook_info = await self.application.bot.get_webhook_info() + except NetworkError as e: + self._polling_watchdog_failures += 1 + self._polling_pending_update_checks = 0 + self._last_watchdog_message_count = self._received_message_count + self._last_pending_update_count = None + logger.warning( + "Telegram polling watchdog request failed (%s/%s): %s", + self._polling_watchdog_failures, + self._polling_watchdog_failure_threshold, + e, + ) + if ( + self._polling_watchdog_failures + >= self._polling_watchdog_failure_threshold + ): + # A failing regular Bot API pool must not remain the outbound client. + self._request_polling_recovery( + replace_outbound_application=( + self.application is self._outbound_application + ) + ) + return + + self._polling_watchdog_failures = 0 + pending_update_count = webhook_info.pending_update_count or 0 + received_messages = ( + self._received_message_count != self._last_watchdog_message_count + ) + pending_updates_decreased = ( + self._last_pending_update_count is not None + and pending_update_count < self._last_pending_update_count + ) + self._last_watchdog_message_count = self._received_message_count + self._last_pending_update_count = pending_update_count + + if pending_update_count == 0 or received_messages or pending_updates_decreased: + self._polling_pending_update_checks = 0 + return + + self._polling_pending_update_checks += 1 + logger.warning( + "Telegram polling watchdog found %s pending updates without delivery " + "progress (%s/%s).", + pending_update_count, + self._polling_pending_update_checks, + self._polling_pending_update_threshold, + ) + if ( + self._polling_pending_update_checks + >= self._polling_pending_update_threshold + ): + self._request_polling_recovery() + def _start_command_scheduler(self) -> None: if not self.enable_command_refresh or not self.enable_command_register: return @@ -249,7 +479,20 @@ async def run(self) -> None: logger.info("Starting Telegram polling...") await updater.start_polling(error_callback=self._on_polling_error) logger.info("Telegram Platform Adapter is running.") + self._next_polling_watchdog_at = ( + self._loop.time() + self._polling_watchdog_interval + ) while updater.running and not self._terminating: # noqa: ASYNC110 + if not self._polling_recovery_requested.is_set(): + now = self._loop.time() + if now >= self._next_polling_watchdog_at: + self._next_polling_watchdog_at = ( + now + self._polling_watchdog_interval + ) + # getWebhookInfo uses the regular Bot API pool, which + # is separate from the getUpdates long-polling pool. + await self._check_polling_health() + if self._polling_recovery_requested.is_set(): await self._recreate_application() break @@ -263,7 +506,9 @@ async def run(self) -> None: continue if not self._terminating: - logger.info("Telegram polling restarted with a fresh client.") + logger.info( + "Telegram polling restarted with a fresh polling client." + ) continue except asyncio.CancelledError: raise @@ -279,7 +524,10 @@ async def run(self) -> None: f"Retrying in {self._polling_restart_delay}s.", ) with suppress(Exception): - await self._shutdown_application(delete_commands=False) + await self._shutdown_application( + delete_commands=False, + preserve_outbound_application=True, + ) self._build_application() if not self._terminating: @@ -307,14 +555,14 @@ def _on_polling_error(self, error: Exception) -> None: logger.warning( "Telegram polling encountered %s network failures within %.1fs; " - "scheduling client rebuild.", + "scheduling polling client rebuild.", self._consecutive_polling_failures, self._polling_failure_window, ) if self._loop.is_closed(): return try: - self._loop.call_soon_threadsafe(self._polling_recovery_requested.set) + self._loop.call_soon_threadsafe(self._request_polling_recovery) except RuntimeError: return @@ -424,6 +672,7 @@ async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non async def message_handler( self, update: Update, context: ContextTypes.DEFAULT_TYPE ) -> None: + self._received_message_count += 1 logger.debug(f"Telegram message: {update.message}") # Handle media group messages @@ -773,6 +1022,11 @@ async def terminate(self) -> None: self.scheduler.shutdown() self._polling_recovery_requested.set() await self._shutdown_application(delete_commands=True) + outbound_application = self._outbound_application + if outbound_application is not None: + with suppress(Exception): + await outbound_application.shutdown() + self._set_outbound_application(None, initialized=False) logger.info("Telegram adapter has been closed.") except Exception as e: diff --git a/dashboard/src/i18n/locales/en-US/features/config-metadata.json b/dashboard/src/i18n/locales/en-US/features/config-metadata.json index fdd430c63e..e7012e1a5a 100644 --- a/dashboard/src/i18n/locales/en-US/features/config-metadata.json +++ b/dashboard/src/i18n/locales/en-US/features/config-metadata.json @@ -600,6 +600,22 @@ "description": "Telegram Polling Restart Delay", "hint": "Waiting time in seconds when the polling loop needs to restart after unexpected exits. Defaults to 5s." }, + "telegram_connect_timeout": { + "description": "Telegram Connection Timeout", + "hint": "Timeout in seconds for establishing Telegram Bot API connections. Defaults to 15s." + }, + "telegram_polling_watchdog_interval": { + "description": "Telegram Polling Watchdog Interval", + "hint": "Interval in seconds between polling progress checks. The minimum is 1s and the default is 15s." + }, + "telegram_polling_watchdog_failure_threshold": { + "description": "Telegram Polling Watchdog Failure Threshold", + "hint": "Number of consecutive watchdog request failures before rebuilding the polling client. The minimum is 1 and the default is 3." + }, + "telegram_polling_watchdog_pending_update_threshold": { + "description": "Telegram Pending Update Stall Threshold", + "hint": "Number of consecutive checks with pending updates and no delivery progress before rebuilding the polling client. The minimum is 1 and the default is 2." + }, "telegram_token": { "description": "Bot Token", "hint": "If you are in mainland China, set a proxy or change api_base in Other Settings." diff --git a/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json b/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json index e25cb8e0fb..cbcc3a02f9 100644 --- a/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json +++ b/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json @@ -602,6 +602,22 @@ "description": "Telegram 轮询重启延迟", "hint": "当轮询意外结束尝试自动重启时的延迟时间，单位为秒。默认为 5s。" }, + "telegram_connect_timeout": { + "description": "Telegram 连接超时", + "hint": "建立 Telegram Bot API 连接时的超时时间，单位为秒。默认为 15s。" + }, + "telegram_polling_watchdog_interval": { + "description": "Telegram 轮询看门狗检查间隔", + "hint": "检查 Telegram 轮询是否仍有进展的间隔时间，单位为秒。最小为 1s，默认为 15s。" + }, + "telegram_polling_watchdog_failure_threshold": { + "description": "Telegram 轮询看门狗失败阈值", + "hint": "看门狗连续请求失败多少次后重建轮询客户端。最小为 1，默认为 3。" + }, + "telegram_polling_watchdog_pending_update_threshold": { + "description": "Telegram 待处理更新停滞阈值", + "hint": "看门狗连续多少次发现待处理更新没有投递进展后重建轮询客户端。最小为 1，默认为 2。" + }, "telegram_token": { "description": "Bot Token", "hint": "如果你的网络环境为中国大陆，请在 `其他配置` 处设置代理或更改 api_base。" diff --git a/tests/fixtures/helpers.py b/tests/fixtures/helpers.py index 68d5e342a4..c6ddbf24a9 100644 --- a/tests/fixtures/helpers.py +++ b/tests/fixtures/helpers.py @@ -49,6 +49,11 @@ def make_platform_config(platform_type: str, **kwargs) -> dict: "telegram_command_register": True, "telegram_command_auto_refresh": True, "telegram_command_register_interval": 300, + "telegram_polling_restart_delay": 5.0, + "telegram_connect_timeout": 15.0, + "telegram_polling_watchdog_interval": 15.0, + "telegram_polling_watchdog_failure_threshold": 3, + "telegram_polling_watchdog_pending_update_threshold": 2, "telegram_media_group_timeout": 2.5, "telegram_media_group_max_wait": 10.0, "start_message": "Welcome to AstrBot!", diff --git a/tests/fixtures/mocks/telegram.py b/tests/fixtures/mocks/telegram.py index e5e31f178c..0cff98cc1d 100644 --- a/tests/fixtures/mocks/telegram.py +++ b/tests/fixtures/mocks/telegram.py @@ -129,6 +129,9 @@ def create_bot(): bot.set_message_reaction = AsyncMock() bot.edit_message_text = AsyncMock() bot.send_message_draft = AsyncMock() + webhook_info = MagicMock() + webhook_info.pending_update_count = 0 + bot.get_webhook_info = AsyncMock(return_value=webhook_info) return bot @staticmethod diff --git a/tests/test_telegram_adapter.py b/tests/test_telegram_adapter.py index 17fe60f111..6d471c840b 100644 --- a/tests/test_telegram_adapter.py +++ b/tests/test_telegram_adapter.py @@ -320,6 +320,347 @@ async def test_telegram_polling_error_requests_rebuild_after_threshold(): assert adapter._polling_recovery_requested.is_set() +def test_telegram_polling_watchdog_uses_configured_settings(): + TelegramPlatformAdapter = _load_telegram_adapter() + module_globals = TelegramPlatformAdapter.__init__.__globals__ + builder = MagicMock() + builder.build.return_value = MockTelegramBuilder.create_application() + + with patch.dict( + module_globals, + { + "ApplicationBuilder": MagicMock(return_value=builder), + "AsyncIOScheduler": MagicMock( + return_value=MockTelegramBuilder.create_scheduler() + ), + }, + ): + adapter = TelegramPlatformAdapter( + make_platform_config( + "telegram", + telegram_connect_timeout=7.5, + telegram_polling_watchdog_interval=20.0, + telegram_polling_watchdog_failure_threshold=4, + telegram_polling_watchdog_pending_update_threshold=5, + ), + {}, + asyncio.Queue(), + ) + + builder.connect_timeout.assert_called_once_with(7.5) + builder.get_updates_connect_timeout.assert_called_once_with(7.5) + assert adapter._polling_watchdog_interval == 20.0 + assert adapter._polling_watchdog_failure_threshold == 4 + assert adapter._polling_pending_update_threshold == 5 + assert adapter._polling_shutdown_timeout == 15.0 + + +def test_telegram_polling_watchdog_validates_configured_settings(): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config( + "telegram", + telegram_polling_restart_delay=float("inf"), + telegram_connect_timeout="invalid", + telegram_polling_watchdog_interval=0, + telegram_polling_watchdog_failure_threshold=0, + telegram_polling_watchdog_pending_update_threshold="invalid", + ), + {}, + asyncio.Queue(), + ) + + assert adapter._polling_restart_delay == 5.0 + assert adapter._telegram_connect_timeout == 15.0 + assert adapter._polling_watchdog_interval == 1.0 + assert adapter._polling_watchdog_failure_threshold == 1 + assert adapter._polling_pending_update_threshold == 2 + + +@pytest.mark.asyncio +async def test_telegram_polling_watchdog_tolerates_transient_network_error(): + TelegramPlatformAdapter = _load_telegram_adapter() + module_globals = TelegramPlatformAdapter.__init__.__globals__ + application = MockTelegramBuilder.create_application() + webhook_info = MagicMock() + webhook_info.pending_update_count = 0 + application.bot.get_webhook_info.side_effect = [ + MockTelegramNetworkError("proxy switching"), + webhook_info, + ] + builder = MagicMock() + builder.build.return_value = application + + with patch.dict( + module_globals, + { + "ApplicationBuilder": MagicMock(return_value=builder), + "AsyncIOScheduler": MagicMock( + return_value=MockTelegramBuilder.create_scheduler() + ), + }, + ): + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + await adapter._check_polling_health() + await adapter._check_polling_health() + + builder.connect_timeout.assert_called_once_with(15.0) + builder.get_updates_connect_timeout.assert_called_once_with(15.0) + assert adapter._polling_watchdog_failures == 0 + assert not adapter._polling_recovery_requested.is_set() + + +@pytest.mark.asyncio +async def test_telegram_polling_watchdog_replaces_failed_outbound_client(): + TelegramPlatformAdapter = _load_telegram_adapter() + module_globals = TelegramPlatformAdapter.__init__.__globals__ + app_one = MockTelegramBuilder.create_application() + app_one.bot.get_webhook_info.side_effect = MockTelegramNetworkError( + "proxy unavailable" + ) + app_two = MockTelegramBuilder.create_application() + builder = MagicMock() + builder.build.side_effect = [app_one, app_two] + + with patch.dict( + module_globals, + { + "ApplicationBuilder": MagicMock(return_value=builder), + "AsyncIOScheduler": MagicMock( + return_value=MockTelegramBuilder.create_scheduler() + ), + }, + ): + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + await adapter._start_application() + for _ in range(adapter._polling_watchdog_failure_threshold): + await adapter._check_polling_health() + await adapter._recreate_application() + + assert app_one.bot.get_webhook_info.await_count == 3 + app_one.stop.assert_awaited_once() + app_one.shutdown.assert_awaited_once() + assert adapter.client is app_two.bot + assert adapter._outbound_application is app_two + assert adapter._outbound_application_initialized is False + + +@pytest.mark.asyncio +async def test_telegram_polling_watchdog_ignores_pending_updates_with_progress(): + TelegramPlatformAdapter = _load_telegram_adapter() + module_globals = TelegramPlatformAdapter.__init__.__globals__ + application = MockTelegramBuilder.create_application() + pending_counts = [3, 2, 2, 2] + application.bot.get_webhook_info.side_effect = [ + MagicMock(pending_update_count=count) for count in pending_counts + ] + builder = MagicMock() + builder.build.return_value = application + + with patch.dict( + module_globals, + { + "ApplicationBuilder": MagicMock(return_value=builder), + "AsyncIOScheduler": MagicMock( + return_value=MockTelegramBuilder.create_scheduler() + ), + }, + ): + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + await adapter._check_polling_health() + await adapter._check_polling_health() + adapter._received_message_count += 1 + await adapter._check_polling_health() + await adapter._check_polling_health() + + assert adapter._polling_pending_update_checks == 1 + assert not adapter._polling_recovery_requested.is_set() + + +@pytest.mark.asyncio +async def test_telegram_polling_watchdog_preserves_sends_and_cleans_up(): + TelegramPlatformAdapter = _load_telegram_adapter() + module_globals = TelegramPlatformAdapter.__init__.__globals__ + app_one = MockTelegramBuilder.create_application() + webhook_info = MagicMock() + webhook_info.pending_update_count = 1 + app_one.bot.get_webhook_info.return_value = webhook_info + app_two = MockTelegramBuilder.create_application() + builder = MagicMock() + builder.build.side_effect = [app_one, app_two] + + with patch.dict( + module_globals, + { + "ApplicationBuilder": MagicMock(return_value=builder), + "AsyncIOScheduler": MagicMock( + return_value=MockTelegramBuilder.create_scheduler() + ), + }, + ): + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + await adapter._start_application() + for _ in range(adapter._polling_pending_update_threshold): + await adapter._check_polling_health() + await adapter._recreate_application() + + assert adapter.client is app_one.bot + app_one.shutdown.assert_not_awaited() + await adapter.client.send_message(chat_id=123, text="still available") + app_one.bot.send_message.assert_awaited_once_with( + chat_id=123, + text="still available", + ) + + await adapter.terminate() + + app_one.shutdown.assert_awaited_once() + app_two.shutdown.assert_awaited_once() + assert adapter._outbound_application is None + assert adapter._outbound_application_initialized is False + + +@pytest.mark.asyncio +async def test_telegram_run_rebuilds_polling_when_watchdog_detects_stall(): + TelegramPlatformAdapter = _load_telegram_adapter() + module_globals = TelegramPlatformAdapter.__init__.__globals__ + app_one = MockTelegramBuilder.create_application() + app_one.updater.running = True + webhook_info = MagicMock() + webhook_info.pending_update_count = 1 + app_one.bot.get_webhook_info.return_value = webhook_info + app_two = MockTelegramBuilder.create_application() + app_two.updater.running = True + builder = MagicMock() + builder.build.side_effect = [app_one, app_two] + adapter = None + + async def second_start_polling(*args, **kwargs): + assert adapter is not None + adapter._terminating = True + + app_two.updater.start_polling.side_effect = second_start_polling + + with patch.dict( + module_globals, + { + "ApplicationBuilder": MagicMock(return_value=builder), + "AsyncIOScheduler": MagicMock( + return_value=MockTelegramBuilder.create_scheduler() + ), + }, + ): + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + adapter._polling_watchdog_interval = 0.01 + await adapter.run() + + assert app_one.bot.get_webhook_info.await_count == 2 + assert builder.build.call_count == 2 + app_one.shutdown.assert_not_awaited() + app_two.initialize.assert_awaited_once() + app_two.start.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_telegram_initialization_failure_replaces_uninitialized_client(): + TelegramPlatformAdapter = _load_telegram_adapter() + module_globals = TelegramPlatformAdapter.__init__.__globals__ + app_one = MockTelegramBuilder.create_application() + app_one.initialize.side_effect = MockTelegramNetworkError("proxy switching") + app_two = MockTelegramBuilder.create_application() + builder = MagicMock() + builder.build.side_effect = [app_one, app_two] + adapter = None + + async def second_start_polling(*args, **kwargs): + assert adapter is not None + adapter._terminating = True + + app_two.updater.start_polling.side_effect = second_start_polling + + with patch.dict( + module_globals, + { + "ApplicationBuilder": MagicMock(return_value=builder), + "AsyncIOScheduler": MagicMock( + return_value=MockTelegramBuilder.create_scheduler() + ), + }, + ): + adapter = TelegramPlatformAdapter( + make_platform_config( + "telegram", + telegram_polling_restart_delay=0.1, + ), + {}, + asyncio.Queue(), + ) + await adapter.run() + + app_one.shutdown.assert_awaited_once() + assert adapter.client is app_two.bot + assert adapter._outbound_application is app_two + assert adapter._outbound_application_initialized + + +@pytest.mark.asyncio +async def test_telegram_shutdown_bounds_stalled_updater_stop(): + TelegramPlatformAdapter = _load_telegram_adapter() + module_globals = TelegramPlatformAdapter.__init__.__globals__ + application = MockTelegramBuilder.create_application() + builder = MagicMock() + builder.build.return_value = application + + async def stalled_stop(): + await asyncio.Event().wait() + + application.updater.stop.side_effect = stalled_stop + + with patch.dict( + module_globals, + { + "ApplicationBuilder": MagicMock(return_value=builder), + "AsyncIOScheduler": MagicMock( + return_value=MockTelegramBuilder.create_scheduler() + ), + }, + ): + adapter = TelegramPlatformAdapter( + make_platform_config( + "telegram", + telegram_connect_timeout=120.0, + ), + {}, + asyncio.Queue(), + ) + adapter._polling_shutdown_timeout = 0.01 + await adapter._shutdown_application(delete_commands=False) + + builder.connect_timeout.assert_called_once_with(120.0) + application.stop.assert_awaited_once() + application.shutdown.assert_awaited_once() + + @pytest.mark.asyncio async def test_telegram_run_rebuilds_application_after_repeated_polling_errors(): TelegramPlatformAdapter = _load_telegram_adapter() @@ -379,9 +720,10 @@ async def second_start_polling(*args, **kwargs): app_one.updater.stop.assert_awaited() app_one.bot.delete_my_commands.assert_not_awaited() app_one.stop.assert_awaited() - app_one.shutdown.assert_awaited() + app_one.shutdown.assert_not_awaited() app_two.initialize.assert_awaited() app_two.start.assert_awaited() + assert adapter.client is app_one.bot @pytest.mark.asyncio