fix(telegram): keep lifecycle state consistent during polling recovery

- Fail outbound sends while the app is rebuilding instead of returning quietly, so ChannelManager retries and logs the failure instead of counting the message as delivered
- Close the bot's HTTPX pools during teardown: Application.shutdown() skips them when initialize() never finished, leaking one pool per startup retry
- Propagate terminal startup errors (rejected token, bad proxy, bound webhook port) and clear the running flag instead of retrying forever while the channel still reports itself healthy
- Restrict startup retries to NetworkError/TimedOut, the exceptions HTTPXRequest wraps every httpx failure into
- Scrub the rejected-token failure so PTB's token-bearing message never reaches the log

nanobot/channels/telegram/runtime.py

@@ -60,6 +60,9 @@ POLL_STALE_SECONDS = 120.0
 POLL_WATCH_INTERVAL = 1.0
 RESTART_BACKOFF_INITIAL_SECONDS = 5.0
 RESTART_BACKOFF_MAX_SECONDS = 300.0
+# How long a send waits out a rebuild; short because ChannelManager dispatches
+# every channel from one serial loop.
+APP_RESTART_SEND_WAIT_SECONDS = 2.0
 
 
 class _LivenessTrackedRequest(BaseRequest):
@@ -512,6 +515,7 @@ class TelegramChannel(BaseChannel):
         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
+        self._app_ready = asyncio.Event()  # cleared while the app is being rebuilt
 
     def _require_app(self) -> TelegramApplication:
         if self._app is None:
@@ -564,16 +568,23 @@ class TelegramChannel(BaseChannel):
         while self._running:
             try:
                 await self._start_app()
-            except InvalidToken as e:
+            except InvalidToken:
-                # A rejected token is a config error, not a transient network
+                # A config error, not a blip: fail the channel. The scrubbed
-                # failure — retrying forever would only spam the log.
+                # re-raise keeps PTB's token-bearing message out of the log.
                 await self._teardown_app()
-                self.logger.error("bot token rejected: {}", self._format_telegram_error(e))
+                self._running = False
-                return
+                self.logger.error("bot token rejected by Telegram")
+                raise RuntimeError("Telegram bot token was rejected by the server") from None
             except Exception as e:
                 await self._teardown_app()
                 if not self._running:
                     break
+                if not self._is_transient_startup_error(e):
+                    # Never heals on its own: fail instead of retrying forever
+                    # while ChannelManager keeps reporting the channel running.
+                    self._running = False
+                    self.logger.error("startup failed: {}", self._format_telegram_error(e))
+                    raise
                 self.logger.error(
                     "startup failed: {}; retrying in {:.0f}s",
                     self._format_telegram_error(e),
@@ -706,6 +717,35 @@ class TelegramChannel(BaseChannel):
                 error_callback=self._on_polling_error,
             )
 
+        self._app_ready.set()
+
+    @staticmethod
+    def _is_transient_startup_error(exc: Exception) -> bool:
+        """Report whether a startup failure is worth retrying.
+
+        HTTPXRequest wraps every httpx failure into NetworkError/TimedOut, so
+        anything else is terminal: a bad proxy raises ValueError, an already
+        bound webhook port raises OSError.
+        """
+        return isinstance(exc, NetworkError | TimedOut | asyncio.TimeoutError)
+
+    async def _wait_for_app(self) -> TelegramApplication | None:
+        """Return the live app, briefly waiting out an in-flight rebuild.
+
+        Returning quietly while ``start()`` rebuilds would let the manager count
+        the message as delivered, so raise once the wait runs out. None means the
+        channel is stopped: nothing left to deliver.
+        """
+        if self._app is not None:
+            return self._app
+        if not self._running:
+            return None
+        with suppress(asyncio.TimeoutError):
+            await asyncio.wait_for(self._app_ready.wait(), APP_RESTART_SEND_WAIT_SECONDS)
+        if self._app is None:
+            raise RuntimeError("Telegram application is restarting; message not delivered")
+        return self._app
+
     def _note_poll_ok(self) -> None:
         # HTTP error statuses count too: the watchdog detects transport stalls,
         # not logical failures.
@@ -729,6 +769,7 @@ class TelegramChannel(BaseChannel):
     async def _teardown_app(self) -> None:
         """Shut down the application, tolerating partially started state."""
         app, self._app = self._app, None
+        self._app_ready.clear()
         if not app:
             return
         for step in (cast(Any, app.updater).stop, app.stop, app.shutdown):
@@ -736,6 +777,12 @@ class TelegramChannel(BaseChannel):
                 await step()
             except Exception as e:
                 self.logger.debug("teardown step failed: {}", e)
+        # Application.shutdown() skips the HTTPX pools unless initialize()
+        # finished, so a failed startup leaks one per retry. This is idempotent.
+        try:
+            await app.bot.shutdown()
+        except Exception as e:
+            self.logger.debug("bot shutdown failed: {}", e)
 
     async def stop(self) -> None:
         """Stop the Telegram bot."""
@@ -848,7 +895,8 @@ class TelegramChannel(BaseChannel):
 
     async def send(self, msg: OutboundMessage) -> None:
         """Send a message through Telegram."""
-        if not self._app:
+        app = await self._wait_for_app()
+        if app is None:
             self.logger.warning("bot not running")
             return
 
@@ -887,11 +935,11 @@ class TelegramChannel(BaseChannel):
             try:
                 media_type = self._get_media_type(media_path)
                 sender = {
-                    "photo": self._app.bot.send_photo,
+                    "photo": app.bot.send_photo,
-                    "video": self._app.bot.send_video,
+                    "video": app.bot.send_video,
-                    "voice": self._app.bot.send_voice,
+                    "voice": app.bot.send_voice,
-                    "audio": self._app.bot.send_audio,
+                    "audio": app.bot.send_audio,
-                }.get(media_type, self._app.bot.send_document)
+                }.get(media_type, app.bot.send_document)
                 param = {
                     "photo": "photo",
                     "video": "video",
@@ -931,7 +979,7 @@ class TelegramChannel(BaseChannel):
             except Exception:
                 filename = media_path.rsplit("/", 1)[-1]
                 self.logger.exception("Failed to send media {}", media_path)
-                await self._app.bot.send_message(
+                await app.bot.send_message(
                     chat_id=chat_id,
                     text=f"[Failed to send: {filename}]",
                     reply_parameters=reply_params,
@@ -1059,7 +1107,8 @@ class TelegramChannel(BaseChannel):
         merge_next: bool = False,
     ) -> None:
         """Progressive message editing: send on first delta, edit on subsequent ones."""
-        if not self._app:
+        app = await self._wait_for_app()
+        if app is None:
             return
         meta = metadata or {}
         int_chat_id = int(chat_id)
@@ -1098,7 +1147,7 @@ class TelegramChannel(BaseChannel):
                     # Delete the streaming preview message
                     try:
                         await self._call_with_retry(
-                            self._app.bot.delete_message,
+                            app.bot.delete_message,
                             chat_id=int_chat_id, message_id=buf.message_id,
                         )
                     except Exception:
@@ -1112,7 +1161,7 @@ class TelegramChannel(BaseChannel):
             extra_html_chunks = html_chunks[1:]
             try:
                 await self._call_with_retry(
-                    self._app.bot.edit_message_text,
+                    app.bot.edit_message_text,
                     chat_id=int_chat_id, message_id=buf.message_id,
                     text=primary_html, parse_mode="HTML",
                 )
@@ -1129,7 +1178,7 @@ class TelegramChannel(BaseChannel):
                 primary_plain = split_message(raw_text, TELEGRAM_MAX_MESSAGE_LEN)[0] if len(raw_text) > TELEGRAM_MAX_MESSAGE_LEN else raw_text
                 try:
                     await self._call_with_retry(
-                        self._app.bot.edit_message_text,
+                        app.bot.edit_message_text,
                         chat_id=int_chat_id, message_id=buf.message_id,
                         text=primary_plain,
                     )
@@ -1142,7 +1191,7 @@ class TelegramChannel(BaseChannel):
             for extra_html_chunk in extra_html_chunks:
                 try:
                     await self._call_with_retry(
-                        self._app.bot.send_message,
+                        app.bot.send_message,
                         chat_id=int_chat_id, text=extra_html_chunk,
                         parse_mode="HTML",
                         **thread_kwargs,
@@ -1172,7 +1221,7 @@ class TelegramChannel(BaseChannel):
             preview = _strip_md_block(buf.text)
             try:
                 sent = await self._call_with_retry(
-                    self._app.bot.send_message,
+                    app.bot.send_message,
                     chat_id=int_chat_id, text=preview,
                     **stream_thread_kwargs,
                 )
@@ -1189,7 +1238,7 @@ class TelegramChannel(BaseChannel):
             preview = _strip_md_block(buf.text)
             try:
                 await self._call_with_retry(
-                    self._app.bot.edit_message_text,
+                    app.bot.edit_message_text,
                     chat_id=int_chat_id, message_id=buf.message_id,
                     text=preview,
                 )

nanobot/channels/telegram/tests/test_telegram_channel.py

@@ -61,6 +61,10 @@ class _FakeBot:
         self.sent_messages: list[dict] = []
         self.sent_media: list[dict] = []
         self.get_me_calls = 0
+        self.shutdown_calls = 0
+
+    async def shutdown(self) -> None:
+        self.shutdown_calls += 1
 
     async def get_me(self):
         self.get_me_calls += 1
@@ -450,11 +454,49 @@ async def test_startup_failure_retries_with_backoff(monkeypatch) -> None:
     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
+    # Pools must be closed via the bot: app.shutdown() skips them here.
+    assert apps[0].bot.shutdown_calls == 1
+    assert apps[1].bot.shutdown_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_terminal_startup_error_is_not_retried(monkeypatch) -> None:
+    """Config errors (bad proxy, bound webhook port) must fail the channel."""
+    _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 _fail() -> None:
+            raise ValueError("Unknown scheme for proxy URL")
+
+        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.RESTART_BACKOFF_INITIAL_SECONDS", 0.0)
+
+    with pytest.raises(ValueError, match="proxy URL"):
+        await channel.start()
+
+    assert len(apps) == 1  # no retry loop
+    assert channel._app is None
+    assert channel.is_running is False
 
 
 @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."""
+    """A rejected token is a config error: fail the channel instead of retrying."""
     from telegram.error import InvalidToken
 
     _FakeHTTPXRequest.clear()
@@ -481,10 +523,13 @@ async def test_invalid_token_stops_without_retry(monkeypatch) -> None:
     )
     monkeypatch.setattr("nanobot.channels.telegram.runtime.RESTART_BACKOFF_INITIAL_SECONDS", 0.0)
 
-    await channel.start()
+    with pytest.raises(RuntimeError) as excinfo:
+        await channel.start()
 
     assert len(apps) == 1
     assert channel._app is None
+    assert channel.is_running is False
+    assert "123:abc" not in str(excinfo.value)  # token must not reach the log
 
 
 @pytest.mark.asyncio
@@ -510,6 +555,52 @@ async def test_stop_during_startup_does_not_leak_app(monkeypatch) -> None:
     assert channel._app is None  # torn down, not leaked
 
 
+@pytest.mark.asyncio
+async def test_send_during_rebuild_fails_instead_of_dropping(monkeypatch) -> None:
+    """A send that cannot reach Telegram must raise so the manager can retry."""
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.APP_RESTART_SEND_WAIT_SECONDS", 0.0)
+    config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"])
+    channel = TelegramChannel(config, MessageBus())
+
+    # Mid-rebuild: still running, but no app to send through.
+    channel._running = True
+    channel._app = None
+
+    msg = OutboundMessage(channel="telegram", chat_id="123", content="hello")
+    with pytest.raises(RuntimeError, match="restarting"):
+        await channel.send(msg)
+    with pytest.raises(RuntimeError, match="restarting"):
+        await channel.send_delta("123", "hello", stream_id="s1")
+
+    # Stopped: nothing to deliver, so stay quiet.
+    channel._running = False
+    await channel.send(msg)
+    await channel.send_delta("123", "hello", stream_id="s1")
+
+
+@pytest.mark.asyncio
+async def test_send_waits_for_rebuild_to_finish(monkeypatch) -> None:
+    """A fast rebuild is waited out rather than surfaced as a delivery failure."""
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.APP_RESTART_SEND_WAIT_SECONDS", 5.0)
+    config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"])
+    channel = TelegramChannel(config, MessageBus())
+
+    app = _FakeApp(lambda: None)
+    channel._running = True
+    channel._app = None
+
+    async def _finish_rebuild() -> None:
+        await asyncio.sleep(0)
+        channel._app = app
+        channel._app_ready.set()
+
+    rebuild = asyncio.create_task(_finish_rebuild())
+    await channel.send(OutboundMessage(channel="telegram", chat_id="123", content="hello"))
+    await rebuild
+
+    assert [m["text"] for m in app.bot.sent_messages] == ["hello"]
+
+
 @pytest.mark.asyncio
 async def test_liveness_tracked_request_stamps_on_round_trip() -> None:
     from nanobot.channels.telegram.runtime import _LivenessTrackedRequest