fix(telegram): recover from silently stalled polling

- Wrap the getUpdates request pool to record completed round trips, since a healthy long poll completes one every ~10s even with no traffic
- Replace the idle keep-alive loop with a supervisor that tears down and rebuilds the application (including its HTTPX pools) when no round trip completes for 120s
- Retry failed startups with exponential backoff so the bot self-heals once the network path recovers
- Give up immediately on InvalidToken since a rejected token is a config error that retries cannot fix
- Tear down the freshly built app when stop() lands mid-startup so the supervisor never leaks a running application
- Bridge python-telegram-bot and httpx stdlib logging into loguru so polling failures are visible in nanobot logs

nanobot/channels/telegram/runtime.py

@@ -25,9 +25,9 @@ from telegram import (
     Update,
     User,
 )
-from telegram.error import BadRequest, NetworkError, TimedOut
+from telegram.error import BadRequest, InvalidToken, NetworkError, TimedOut
 from telegram.ext import Application, CallbackQueryHandler, ContextTypes, MessageHandler, filters
-from telegram.request import HTTPXRequest
+from telegram.request import BaseRequest, HTTPXRequest
 
 from nanobot.bus.events import OutboundMessage
 from nanobot.bus.outbound_events import ProgressEvent
@@ -38,6 +38,7 @@ from nanobot.config.paths import get_media_dir
 from nanobot.config.schema import Base
 from nanobot.security.network import validate_url_target
 from nanobot.utils.helpers import split_message
+from nanobot.utils.logging_bridge import redirect_lib_logging
 
 TELEGRAM_MAX_MESSAGE_LEN = 4000  # Telegram message character limit
 # Telegram's actual API limit is 4096; we split raw markdown at 4000 as a
@@ -53,6 +54,39 @@ TELEGRAM_REPLY_CONTEXT_MAX_LEN = TELEGRAM_MAX_MESSAGE_LEN  # Max length for repl
 TelegramApplication: TypeAlias = Application[Any, Any, Any, Any, Any, Any]
 _T = TypeVar("_T")
 
+# A healthy getUpdates long poll completes every ~10s even with no traffic;
+# PTB retries timeouts silently, so stalls must be detected here.
+POLL_STALE_SECONDS = 120.0
+POLL_WATCH_INTERVAL = 1.0
+RESTART_BACKOFF_INITIAL_SECONDS = 5.0
+RESTART_BACKOFF_MAX_SECONDS = 300.0
+
+
+class _LivenessTrackedRequest(BaseRequest):
+    """Wrap the getUpdates request pool, reporting each completed round trip."""
+
+    __slots__ = ("inner", "_on_round_trip")
+
+    def __init__(self, inner: BaseRequest, on_round_trip: Callable[[], None]) -> None:
+        super().__init__()
+        self.inner = inner
+        self._on_round_trip = on_round_trip
+
+    @property
+    def read_timeout(self) -> float | None:
+        return self.inner.read_timeout
+
+    async def initialize(self) -> None:
+        await self.inner.initialize()
+
+    async def shutdown(self) -> None:
+        await self.inner.shutdown()
+
+    async def do_request(self, *args: Any, **kwargs: Any) -> tuple[int, bytes]:
+        result = await self.inner.do_request(*args, **kwargs)
+        self._on_round_trip()
+        return result
+
 
 def _split_telegram_markdown(content: str, max_len: int) -> list[str]:
     """Split raw Telegram Markdown without leaving fenced code blocks unbalanced."""
@@ -477,6 +511,7 @@ class TelegramChannel(BaseChannel):
         self._inbound_buffers: dict[str, list[_QueuedTelegramUpdate]] = {}
         self._inbound_workers: dict[str, asyncio.Task[None]] = {}
         self._rich_send_disabled: bool = False  # Latch off if Bot API < 10.1
+        self._last_poll_ok: float = 0.0  # monotonic time of last getUpdates round trip
 
     def _require_app(self) -> TelegramApplication:
         if self._app is None:
@@ -516,13 +551,56 @@ class TelegramChannel(BaseChannel):
         return content
 
     async def start(self) -> None:
-        """Start the Telegram bot."""
+        """Start the Telegram bot, rebuilding the app whenever polling stalls."""
         if not self.config.token:
             self.logger.error("bot token not configured")
             return
 
-        self._running = True
+        redirect_lib_logging("telegram")
+        redirect_lib_logging("httpx", level="WARNING")
 
+        self._running = True
+        backoff = RESTART_BACKOFF_INITIAL_SECONDS
+        while self._running:
+            try:
+                await self._start_app()
+            except InvalidToken as e:
+                # A rejected token is a config error, not a transient network
+                # failure — retrying forever would only spam the log.
+                await self._teardown_app()
+                self.logger.error("bot token rejected: {}", self._format_telegram_error(e))
+                return
+            except Exception as e:
+                await self._teardown_app()
+                if not self._running:
+                    break
+                self.logger.error(
+                    "startup failed: {}; retrying in {:.0f}s",
+                    self._format_telegram_error(e),
+                    backoff,
+                )
+                await self._idle(backoff)
+                backoff = min(backoff * 2, RESTART_BACKOFF_MAX_SECONDS)
+                continue
+
+            backoff = RESTART_BACKOFF_INITIAL_SECONDS
+            if not self._running:
+                # stop() ran while _start_app() was mid-flight and tore down the
+                # previous (possibly None) app; this one would leak otherwise.
+                await self._teardown_app()
+                break
+            stalled = await self._watch_polling()
+            if not stalled or not self._running:
+                break
+            self.logger.warning(
+                "polling stalled: no getUpdates round trip for {:.0f}s; "
+                "rebuilding connection pools and restarting",
+                time.monotonic() - self._last_poll_ok,
+            )
+            await self._teardown_app()
+
+    async def _start_app(self) -> None:
+        """Build, initialize and start the Telegram application."""
         proxy = self.config.proxy or None
 
         # Separate pools so long-polling (getUpdates) never starves outbound sends.
@@ -544,7 +622,7 @@ class TelegramChannel(BaseChannel):
             Application.builder()
             .token(self.config.token)
             .request(api_request)
-            .get_updates_request(poll_request)
+            .get_updates_request(_LivenessTrackedRequest(poll_request, self._note_poll_ok))
         )
         self._app = builder.build()
         self._app.add_error_handler(self._on_error)
@@ -621,16 +699,43 @@ class TelegramChannel(BaseChannel):
                 max_connections=self.config.webhook_max_connections,
             )
         else:
-            # Start polling (this runs until stopped)
+            self._last_poll_ok = time.monotonic()
             await cast(Any, self._app.updater).start_polling(
                 allowed_updates=allowed_updates,
                 drop_pending_updates=False,  # Process pending messages on startup
                 error_callback=self._on_polling_error,
             )
 
-        # Keep running until stopped
+    def _note_poll_ok(self) -> None:
+        # HTTP error statuses count too: the watchdog detects transport stalls,
+        # not logical failures.
+        self._last_poll_ok = time.monotonic()
+
+    async def _watch_polling(self) -> bool:
+        """Idle until stop(); in polling mode, return True when getUpdates goes stale."""
+        watch = self.config.mode != "webhook"
         while self._running:
-            await asyncio.sleep(1)
+            await asyncio.sleep(POLL_WATCH_INTERVAL)
+            if watch and time.monotonic() - self._last_poll_ok > POLL_STALE_SECONDS:
+                return True
+        return False
+
+    async def _idle(self, seconds: float) -> None:
+        """Sleep in short steps so stop() stays responsive."""
+        deadline = time.monotonic() + seconds
+        while self._running and time.monotonic() < deadline:
+            await asyncio.sleep(POLL_WATCH_INTERVAL)
+
+    async def _teardown_app(self) -> None:
+        """Shut down the application, tolerating partially started state."""
+        app, self._app = self._app, None
+        if not app:
+            return
+        for step in (cast(Any, app.updater).stop, app.stop, app.shutdown):
+            try:
+                await step()
+            except Exception as e:
+                self.logger.debug("teardown step failed: {}", e)
 
     async def stop(self) -> None:
         """Stop the Telegram bot."""
@@ -652,10 +757,7 @@ class TelegramChannel(BaseChannel):
 
         if self._app:
             self.logger.info("Stopping bot...")
-            await cast(Any, self._app.updater).stop()
+            await self._teardown_app()
-            await self._app.stop()
-            await self._app.shutdown()
-            self._app = None
 
     @staticmethod
     def _get_media_type(path: str) -> str:

nanobot/channels/telegram/tests/test_telegram_channel.py

@@ -337,7 +337,7 @@ async def test_start_creates_separate_pools_with_proxy(monkeypatch) -> None:
     assert api_req.kwargs["connection_pool_size"] == 32
     assert poll_req.kwargs["connection_pool_size"] == 4
     assert builder.request_value is api_req
-    assert builder.get_updates_request_value is poll_req
+    assert builder.get_updates_request_value.inner is poll_req
     assert callable(app.updater.start_polling_kwargs["error_callback"])
     assert any(cmd.command == "status" for cmd in app.bot.commands)
     assert any(cmd.command == "history" for cmd in app.bot.commands)
@@ -378,6 +378,161 @@ async def test_start_respects_custom_pool_config(monkeypatch) -> None:
     assert poll_req.kwargs["pool_timeout"] == 10.0
 
 
+@pytest.mark.asyncio
+async def test_stalled_polling_triggers_pool_rebuild(monkeypatch) -> None:
+    """When no getUpdates round trip completes for too long, the app is rebuilt."""
+    _FakeHTTPXRequest.clear()
+    config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"])
+    bus = MessageBus()
+    channel = TelegramChannel(config, bus)
+
+    apps: list[_FakeApp] = []
+
+    def on_start_polling() -> None:
+        if len(apps) >= 2:
+            channel._running = False
+
+    def make_builder():
+        app = _FakeApp(on_start_polling)
+        apps.append(app)
+        return _FakeBuilder(app)
+
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest)
+    monkeypatch.setattr(
+        "nanobot.channels.telegram.runtime.Application",
+        SimpleNamespace(builder=make_builder),
+    )
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_STALE_SECONDS", -1.0)
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_WATCH_INTERVAL", 0.0)
+
+    await channel.start()
+
+    assert len(apps) == 2
+    assert apps[0].updater.start_polling_kwargs is not None
+    assert apps[1].updater.start_polling_kwargs is not None
+    # 2 fresh pools per app
+    assert len(_FakeHTTPXRequest.instances) == 4
+
+
+@pytest.mark.asyncio
+async def test_startup_failure_retries_with_backoff(monkeypatch) -> None:
+    """Transient startup failures back off and retry until the app comes up."""
+    from telegram.error import NetworkError
+
+    _FakeHTTPXRequest.clear()
+    config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"])
+    bus = MessageBus()
+    channel = TelegramChannel(config, bus)
+
+    apps: list[_FakeApp] = []
+
+    def make_builder():
+        app = _FakeApp(lambda: setattr(channel, "_running", False))
+        if len(apps) < 2:
+            async def _fail() -> None:
+                raise NetworkError("connect failed")
+
+            app.initialize = _fail
+        apps.append(app)
+        return _FakeBuilder(app)
+
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest)
+    monkeypatch.setattr(
+        "nanobot.channels.telegram.runtime.Application",
+        SimpleNamespace(builder=make_builder),
+    )
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_WATCH_INTERVAL", 0.0)
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.RESTART_BACKOFF_INITIAL_SECONDS", 0.0)
+
+    await channel.start()
+
+    assert len(apps) == 3
+    assert apps[0].updater.start_polling_kwargs is None
+    assert apps[1].updater.start_polling_kwargs is None
+    assert apps[2].updater.start_polling_kwargs is not None
+
+
+@pytest.mark.asyncio
+async def test_invalid_token_stops_without_retry(monkeypatch) -> None:
+    """A rejected token is a config error: log it and give up instead of retrying."""
+    from telegram.error import InvalidToken
+
+    _FakeHTTPXRequest.clear()
+    config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"])
+    bus = MessageBus()
+    channel = TelegramChannel(config, bus)
+
+    apps: list[_FakeApp] = []
+
+    def make_builder():
+        app = _FakeApp(lambda: None)
+
+        async def _reject() -> None:
+            raise InvalidToken("token rejected by Telegram")
+
+        app.initialize = _reject
+        apps.append(app)
+        return _FakeBuilder(app)
+
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest)
+    monkeypatch.setattr(
+        "nanobot.channels.telegram.runtime.Application",
+        SimpleNamespace(builder=make_builder),
+    )
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.RESTART_BACKOFF_INITIAL_SECONDS", 0.0)
+
+    await channel.start()
+
+    assert len(apps) == 1
+    assert channel._app is None
+
+
+@pytest.mark.asyncio
+async def test_stop_during_startup_does_not_leak_app(monkeypatch) -> None:
+    """stop() landing while _start_app() is mid-flight must not leave the app running."""
+    _FakeHTTPXRequest.clear()
+    config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"])
+    bus = MessageBus()
+    channel = TelegramChannel(config, bus)
+
+    # Simulate stop() winning the race just before start_polling returns.
+    app = _FakeApp(lambda: setattr(channel, "_running", False))
+    builder = _FakeBuilder(app)
+
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest)
+    monkeypatch.setattr(
+        "nanobot.channels.telegram.runtime.Application",
+        SimpleNamespace(builder=lambda: builder),
+    )
+
+    await channel.start()
+
+    assert channel._app is None  # torn down, not leaked
+
+
+@pytest.mark.asyncio
+async def test_liveness_tracked_request_stamps_on_round_trip() -> None:
+    from nanobot.channels.telegram.runtime import _LivenessTrackedRequest
+
+    stamps: list[int] = []
+
+    class _Inner:
+        read_timeout = 5.0
+
+        async def initialize(self) -> None:
+            pass
+
+        async def shutdown(self) -> None:
+            pass
+
+        async def do_request(self, *args, **kwargs):
+            return 200, b"{}"
+
+    wrapped = _LivenessTrackedRequest(_Inner(), lambda: stamps.append(1))
+    assert await wrapped.do_request(url="https://example.org", method="POST") == (200, b"{}")
+    assert stamps == [1]
+
+
 def test_webhook_config_requires_https_url_and_secret() -> None:
     with pytest.raises(ValueError, match="webhook_url is required"):
         TelegramConfig(enabled=True, token="123:abc", mode="webhook")