From 2b4a04fb71ce367b631b7df6a9f649049dd2f514 Mon Sep 17 00:00:00 2001
From: chengyongru <chengyongru.ai@gmail.com>
Date: Fri, 7 Aug 2026 17:30:08 +0800
Subject: [PATCH] fix(telegram): serialize application teardown

---
 nanobot/channels/telegram/runtime.py          | 32 ++++++++-------
 .../telegram/tests/test_telegram_channel.py   | 40 +++++++++++++++++++
 2 files changed, 58 insertions(+), 14 deletions(-)

diff --git a/nanobot/channels/telegram/runtime.py b/nanobot/channels/telegram/runtime.py
index 72cc3daec..e475b1afe 100644
--- a/nanobot/channels/telegram/runtime.py
+++ b/nanobot/channels/telegram/runtime.py
@@ -516,6 +516,7 @@ class TelegramChannel(BaseChannel):
         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
+        self._teardown_lock = asyncio.Lock()
 
     def _require_app(self) -> TelegramApplication:
         if self._app is None:
@@ -768,21 +769,22 @@ 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):
+        async with self._teardown_lock:
+            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):
+                try:
+                    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 step()
+                await app.bot.shutdown()
             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)
+                self.logger.debug("bot shutdown failed: {}", e)
 
     async def stop(self) -> None:
         """Stop the Telegram bot."""
@@ -804,7 +806,9 @@ class TelegramChannel(BaseChannel):
 
         if self._app:
             self.logger.info("Stopping bot...")
-            await self._teardown_app()
+        # Join an in-flight supervisor teardown before ChannelManager cancels
+        # start(), otherwise cancellation can strand the old HTTPX pools.
+        await self._teardown_app()
 
     @staticmethod
     def _get_media_type(path: str) -> str:
diff --git a/nanobot/channels/telegram/tests/test_telegram_channel.py b/nanobot/channels/telegram/tests/test_telegram_channel.py
index 11ed4f692..bf5ded907 100644
--- a/nanobot/channels/telegram/tests/test_telegram_channel.py
+++ b/nanobot/channels/telegram/tests/test_telegram_channel.py
@@ -563,6 +563,46 @@ 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_stop_waits_for_inflight_watchdog_teardown(monkeypatch) -> None:
+    """Manager cancellation after stop() must not interrupt an active teardown."""
+    _FakeHTTPXRequest.clear()
+    channel = TelegramChannel(
+        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
+        MessageBus(),
+    )
+    teardown_started = asyncio.Event()
+    finish_teardown = asyncio.Event()
+    app = _FakeApp(lambda: None)
+
+    async def slow_updater_stop() -> None:
+        teardown_started.set()
+        await finish_teardown.wait()
+
+    app.updater.stop = slow_updater_stop
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest)
+    monkeypatch.setattr(
+        "nanobot.channels.telegram.runtime.Application",
+        SimpleNamespace(builder=lambda: _FakeBuilder(app)),
+    )
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_STALE_SECONDS", -1.0)
+    monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_WATCH_INTERVAL", 0.0)
+
+    start_task = asyncio.create_task(channel.start())
+    await teardown_started.wait()
+    assert channel._app is None
+
+    stop_task = asyncio.create_task(channel.stop())
+    await asyncio.sleep(0)
+    assert not stop_task.done()
+
+    finish_teardown.set()
+    await stop_task
+    await start_task
+
+    assert app.bot.shutdown_calls == 1
+
+
 @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."""