fix(telegram): classify PTB heartbeat transport errors

plugins/platforms/telegram/adapter.py

@@ -1175,12 +1175,23 @@ def _looks_like_polling_conflict(error: Exception) -> bool:
 
     @staticmethod
     def _looks_like_network_error(error: Exception) -> bool:
-        """Return True for transient network errors that warrant a reconnect attempt."""
+        """Return True for transient transport failures that warrant reconnect."""
         name = error.__class__.__name__.lower()
+        if name in {"badrequest", "invalidtoken", "forbidden", "retryafter"}:
+            return False
         if name in {"networkerror", "timedout", "connectionerror"}:
             return True
         try:
-            from telegram.error import NetworkError, TimedOut
+            from telegram.error import (
+                BadRequest,
+                Forbidden,
+                InvalidToken,
+                NetworkError,
+                RetryAfter,
+                TimedOut,
+            )
+            if isinstance(error, (BadRequest, InvalidToken, Forbidden, RetryAfter)):
+                return False
             if isinstance(error, (NetworkError, TimedOut)):
                 return True
         except ImportError:
@@ -2204,17 +2215,11 @@ async def _polling_heartbeat_loop(self) -> None:
             except asyncio.CancelledError:
                 return
             except (asyncio.TimeoutError, OSError) as probe_err:
-                logger.warning(
-                    "[%s] Polling heartbeat probe failed (%s); triggering reconnect",
-                    self.name, probe_err,
-                )
-                if self._polling_error_task and not self._polling_error_task.done():
-                    continue   # reconnect already in progress
-                loop = asyncio.get_running_loop()
-                self._polling_error_task = loop.create_task(
-                    self._handle_polling_network_error(probe_err)
-                )
-            except Exception:
+                self._schedule_polling_recovery(probe_err, reason="heartbeat probe")
+            except Exception as probe_err:
+                if self._looks_like_network_error(probe_err):
+                    self._schedule_polling_recovery(probe_err, reason="heartbeat probe")
+                    continue
                 # Non-connectivity errors (e.g. TelegramError 401) are not
                 # CLOSE-WAIT symptoms — let PTB's own handlers surface them.
                 pass
@@ -3192,10 +3197,6 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
             # Start polling — retry initialize() for transient TLS resets.
             # Each attempt is capped by _init_timeout so a single unreachable
             # fallback-IP chain can't block startup indefinitely.
-            try:
-                from telegram.error import NetworkError, TimedOut
-            except ImportError:
-                NetworkError = TimedOut = OSError  # type: ignore[misc,assignment]
             _max_connect = 8
             _init_timeout = _env_float("HERMES_TELEGRAM_INIT_TIMEOUT", 30.0)
             for _attempt in range(_max_connect):
@@ -3230,7 +3231,19 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
                             f"({_init_timeout:.0f}s each). Check network connectivity to api.telegram.org "
                             f"or set HERMES_TELEGRAM_HTTP_CONNECT_TIMEOUT to a lower value."
                         )
-                except (NetworkError, TimedOut, OSError) as init_err:
+                except OSError as init_err:
+                    if _attempt < _max_connect - 1:
+                        wait = min(2 ** _attempt, 15)
+                        logger.warning(
+                            "[%s] Connect attempt %d/%d failed: %s — retrying in %ds",
+                            self.name, _attempt + 1, _max_connect, init_err, wait,
+                        )
+                        await asyncio.sleep(wait)
+                    else:
+                        raise
+                except Exception as init_err:
+                    if not self._looks_like_network_error(init_err):
+                        raise
                     if _attempt < _max_connect - 1:
                         wait = min(2 ** _attempt, 15)
                         logger.warning(

tests/gateway/conftest.py

@@ -69,15 +69,21 @@ def _ensure_telegram_mock() -> None:
     mod.constants.ChatType.SUPERGROUP = "supergroup"
     mod.constants.ChatType.CHANNEL = "channel"
 
-    # Real exception classes so ``except (NetworkError, ...)`` clauses
-    # in production code don't blow up with TypeError.
-    mod.error.NetworkError = type("NetworkError", (OSError,), {})
-    mod.error.TimedOut = type("TimedOut", (OSError,), {})
-    mod.error.BadRequest = type("BadRequest", (Exception,), {})
-    mod.error.Forbidden = type("Forbidden", (Exception,), {})
-    mod.error.InvalidToken = type("InvalidToken", (Exception,), {})
-    mod.error.RetryAfter = type("RetryAfter", (Exception,), {"retry_after": 1})
-    mod.error.Conflict = type("Conflict", (Exception,), {})
+    # Mirror PTB's exception hierarchy: BadRequest is a semantic API error,
+    # but inherits from NetworkError in python-telegram-bot 22.x.
+    mod.error.TelegramError = type("TelegramError", (Exception,), {})
+    mod.error.NetworkError = type("NetworkError", (mod.error.TelegramError,), {})
+    mod.error.TimedOut = type("TimedOut", (mod.error.NetworkError,), {})
+    mod.error.BadRequest = type("BadRequest", (mod.error.NetworkError,), {})
+    mod.error.Forbidden = type("Forbidden", (mod.error.TelegramError,), {})
+    mod.error.InvalidToken = type("InvalidToken", (mod.error.TelegramError,), {})
+
+    class RetryAfter(mod.error.TelegramError):
+        def __init__(self, retry_after=1):
+            self.retry_after = retry_after
+
+    mod.error.RetryAfter = RetryAfter
+    mod.error.Conflict = type("Conflict", (mod.error.TelegramError,), {})
 
     # Update.ALL_TYPES used in start_polling()
     mod.Update.ALL_TYPES = []

tests/gateway/test_telegram_network_reconnect.py

@@ -6,7 +6,9 @@
 rather than silently leaving polling dead.
 """
 
+import ast
 import asyncio
+from pathlib import Path
 import sys
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -678,6 +680,123 @@ async def telegram_error_wait_for(coro, timeout):
     adapter._handle_polling_network_error.assert_not_awaited()
 
 
+async def _heartbeat_exception_case(exc, *, pending_probe=False):
+    adapter = _make_adapter()
+    reconnect_handler = AsyncMock()
+    adapter._handle_polling_network_error = reconnect_handler  # type: ignore[method-assign]
+    mock_app = MagicMock()
+    mock_app.updater.running = True
+    if pending_probe:
+        mock_app.bot.get_me = AsyncMock(return_value=MagicMock())
+        mock_app.bot.get_webhook_info = AsyncMock(side_effect=exc)
+    else:
+        mock_app.bot.get_me = AsyncMock(side_effect=exc)
+    adapter._app = mock_app
+
+    sleep_calls = 0
+
+    async def fast_sleep(_seconds):
+        nonlocal sleep_calls
+        sleep_calls += 1
+        if sleep_calls >= 2:
+            raise asyncio.CancelledError()
+
+    with patch("asyncio.sleep", side_effect=fast_sleep):
+        await adapter._polling_heartbeat_loop()
+    await asyncio.sleep(0)
+    return adapter
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("pending_probe", [False, True])
+async def test_heartbeat_routes_ptb_transport_errors_to_reconnect(pending_probe):
+    from telegram.error import NetworkError, TimedOut
+
+    for exc in (NetworkError("network"), TimedOut("timeout")):
+        adapter = await _heartbeat_exception_case(exc, pending_probe=pending_probe)
+        reconnect_handler = adapter._handle_polling_network_error
+        assert isinstance(reconnect_handler, AsyncMock)
+        reconnect_handler.assert_awaited_once_with(exc)
+        assert adapter._polling_error_task is not None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("pending_probe", [False, True])
+async def test_heartbeat_ignores_ptb_semantic_errors(pending_probe):
+    from telegram.error import BadRequest, Forbidden, InvalidToken, RetryAfter
+
+    for exc in (
+        BadRequest("bad request"),
+        Forbidden("forbidden"),
+        InvalidToken("invalid token"),
+        RetryAfter(1),
+    ):
+        adapter = await _heartbeat_exception_case(exc, pending_probe=pending_probe)
+        reconnect_handler = adapter._handle_polling_network_error
+        assert isinstance(reconnect_handler, AsyncMock)
+        reconnect_handler.assert_not_awaited()
+        assert adapter._polling_error_task is None
+
+
+@pytest.mark.parametrize(
+    ("error_name", "expected"),
+    [
+        ("NetworkError", True),
+        ("TimedOut", True),
+        ("BadRequest", False),
+        ("Forbidden", False),
+        ("InvalidToken", False),
+        ("RetryAfter", False),
+    ],
+)
+def test_network_error_classifier_matches_ptb_semantics(error_name, expected):
+    import telegram.error as telegram_error
+
+    error_type = getattr(telegram_error, error_name)
+    error = error_type(1) if error_name == "RetryAfter" else error_type(error_name)
+    assert TelegramAdapter._looks_like_network_error(error) is expected
+
+
+def _calls_shared_network_classifier(node):
+    return any(
+        isinstance(child, ast.Call)
+        and isinstance(child.func, ast.Attribute)
+        and child.func.attr == "_looks_like_network_error"
+        for child in ast.walk(node)
+    )
+
+
+def test_polling_error_callback_uses_shared_network_classifier():
+    source = Path(TelegramAdapter.connect.__code__.co_filename).read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    callbacks = [
+        node
+        for node in ast.walk(tree)
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        and node.name == "_polling_error_callback"
+    ]
+    assert len(callbacks) == 1
+    assert _calls_shared_network_classifier(callbacks[0])
+
+
+def test_connect_initialize_retry_uses_shared_network_classifier():
+    source = Path(TelegramAdapter.connect.__code__.co_filename).read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    connect = next(
+        node
+        for node in ast.walk(tree)
+        if isinstance(node, ast.AsyncFunctionDef) and node.name == "connect"
+    )
+    exception_handlers = [
+        node
+        for node in ast.walk(connect)
+        if isinstance(node, ast.ExceptHandler)
+        and isinstance(node.type, ast.Name)
+        and node.type.id == "Exception"
+    ]
+    assert any(_calls_shared_network_classifier(handler) for handler in exception_handlers)
+
+
 @pytest.mark.asyncio
 async def test_heartbeat_loop_exits_on_fatal_error():
     """A fatal error short-circuits the loop before probing get_me()."""