fix(telegram): retain loop cleanup across bounded stops (#6772)

* fix(telegram): retain loop cleanup across bounded stops

* fix(telegram): release abandoned cleanup generations

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -1042,6 +1042,7 @@ Notes:
 - If `assistant_id` is set to a custom agent name, DeerFlow still routes through `lead_agent` and injects that value as `agent_name`, so the custom agent's SOUL/config takes effect for IM channels.
 - IM channel workers call Gateway's LangGraph-compatible API internally and automatically attach process-local internal auth plus the CSRF cookie/header pair required for thread and run creation.
 - Inbound work is bounded to `inbound_queue_maxsize` pending messages plus `max_concurrency` active workers. When capacity is exhausted, socket/polling providers drop new messages before sending DeerFlow's working acknowledgment and emit a rate-limited warning. Buzz leaves its replay cursor unchanged and reconnects for relay replay; GitHub webhooks return `503`, marking the delivery failed for manual/API redelivery. Shutdown closes admission immediately, keeps channel transports available while accepted messages drain for up to `shutdown_grace_period_seconds`, then cancels and awaits active handlers before closing provider resources; the Gateway's outer timeout can cancel an incomplete shutdown without detaching those resources.
+- Telegram shutdown retains ongoing cleanup across a timeout or cancellation so a retry can finish it. If the Telegram loop has already stopped, shutdown warns that download clients may remain open and releases the abandoned cleanup state; later stops do not repeat that warning.
 - Feishu/Lark now queues rapid follow-up messages per mapped DeerFlow `thread_id` instead of immediately surfacing the generic busy reply, and topic replies keep a per-message card with a compact source-message preview across queued/running/final patches.
 - Streaming IM channels treat backend `error` events like transport failures and follow the same reply and retry path. The channel logs the error type and message for diagnosis.
 

backend/app/channels/AGENTS.md

@@ -1,11 +1,15 @@
 ### IM Channels System (`app/channels/`)
 
-Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk, GitHub) to the DeerFlow agent via Gateway's LangGraph-compatible API.
+IM channels reach DeerFlow through `langgraph-sdk` HTTP; Gateway owns thread lifecycle.
 
 WeChat `_read_outbound_bytes` reads cap+1 off-loop; overflow returns `None` before encryption/upload; caps <=0 are unlimited.
 Cursor writes drain before cancellation; token expiry drains cursor/auth clearing under `_auth_lock`.
 
-**Architecture**: Channels use `langgraph-sdk` HTTP to reach Gateway; Gateway owns thread lifecycle. The SDK client and raw-httpx `/goal` calls send process-local auth plus `_csrf_headers()`; internal auth is not CSRF-exempt.
+Telegram stop retries share live cleanup. If its loop dies, warn once and
+drop only the matching future under `_shutdown_resources_lock`; client closure
+remains unconfirmed.
+
+**Auth**: SDK and raw-httpx `/goal` calls send process-local auth plus `_csrf_headers()`; internal auth requires CSRF.
 
 **Components**:
 - `message_bus.py` - Inbound queue → dispatcher; outbound uses a dispatch-start listener snapshot. Unsubscribe neither cancels nor drains captured callbacks.

backend/app/channels/telegram.py

@@ -8,6 +8,7 @@ import re
 import threading
 import time
 from collections.abc import Coroutine
+from concurrent.futures import Future
 from typing import Any
 
 from app.channels.allowed_users import parse_allowed_users
@@ -149,6 +150,10 @@ class TelegramChannel(Channel):
         # its pool on the loop that performs the download.
         self._download_bot: Any = None
         self._download_bot_lock = asyncio.Lock()
+        # The first stop() owns this Telegram-loop cleanup. A bounded or
+        # cancelled caller leaves it running so a retry can join the same work.
+        self._shutdown_resources_future: Future[None] | None = None
+        self._shutdown_resources_lock = threading.Lock()
         # Tasks submitted from the main dispatcher loop back to PTB's loop.
         # Only the Telegram loop mutates this set.
         self._tg_bridge_tasks: set[asyncio.Task[Any]] = set()
@@ -224,44 +229,66 @@ class TelegramChannel(Channel):
         deadline = shutdown_loop.time() + TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS
         telegram_loop = self._tg_loop
         worker_thread = self._thread
+        telegram_loop_running = telegram_loop is not None and telegram_loop.is_running()
 
-        try:
-            if telegram_loop and telegram_loop.is_running():
-                drain_future = asyncio.run_coroutine_threadsafe(self._shutdown_telegram_resources(), telegram_loop)
-                try:
-                    remaining = max(0.0, deadline - shutdown_loop.time())
-                    await asyncio.wait_for(asyncio.wrap_future(drain_future), timeout=remaining)
-                except asyncio.CancelledError:
-                    drain_future.cancel()
-                    raise
-                except TimeoutError:
-                    drain_future.cancel()
-                    logger.warning("[Telegram] timed out shutting down Telegram loop resources during shutdown")
-                except Exception as exc:
-                    logger.warning("[Telegram] failed to shut down Telegram loop resources during shutdown: %s", type(exc).__name__)
-        finally:
-            if telegram_loop and telegram_loop.is_running():
-                try:
-                    telegram_loop.call_soon_threadsafe(telegram_loop.stop)
-                except RuntimeError:
-                    pass
+        # Bridge tasks and the download Bot's HTTPX clients may only be released
+        # on the Telegram loop, so that loop has to outlive the cleanup. Exactly
+        # one cleanup generation is scheduled per channel: a bounded or
+        # cancelled caller leaves it running, and a retried stop() joins the
+        # same work instead of scheduling a second shutdown next to it.
+        cleanup_future = self._shutdown_resources_future
+        if cleanup_future is None and telegram_loop_running:
+            with self._shutdown_resources_lock:
+                cleanup_future = self._shutdown_resources_future
+                if cleanup_future is None:
+                    cleanup_future = asyncio.run_coroutine_threadsafe(self._shutdown_telegram_resources(), telegram_loop)
+                    self._shutdown_resources_future = cleanup_future
+        if cleanup_future is not None and not telegram_loop_running:
+            # Its owner loop is gone, so nothing can finish this cleanup.
+            logger.warning("[Telegram] Telegram loop stopped before its resource cleanup finished; the download bot's HTTPX clients may not have been closed")
+            with self._shutdown_resources_lock:
+                if self._shutdown_resources_future is cleanup_future:
+                    self._shutdown_resources_future = None
+            cleanup_future = None
+        if cleanup_future is not None:
             try:
-                if worker_thread is not None and worker_thread.is_alive():
-                    remaining = max(0.0, deadline - shutdown_loop.time())
-                    if remaining:
-                        try:
-                            await asyncio.wait_for(
-                                asyncio.to_thread(worker_thread.join, remaining),
-                                timeout=remaining,
-                            )
-                        except TimeoutError:
-                            logger.warning("[Telegram] polling thread did not stop within the shutdown budget")
-            finally:
-                if worker_thread is not None and worker_thread.is_alive():
-                    raise ChannelStopTimeout("Telegram polling thread is still running after stop timeout")
-                if self._thread is worker_thread:
-                    self._thread = None
-                self._application = None
+                remaining = max(0.0, deadline - shutdown_loop.time())
+                # shield(): the deadline and caller cancellation must interrupt
+                # only this wait, never the cleanup the Telegram loop owns.
+                await asyncio.wait_for(asyncio.shield(asyncio.wrap_future(cleanup_future)), timeout=remaining)
+            except asyncio.CancelledError:
+                # Retain the loop and its cleanup: ChannelService keeps this
+                # instance, and its next stop() converges on the same generation.
+                raise
+            except TimeoutError:
+                logger.warning("[Telegram] timed out shutting down Telegram loop resources during shutdown")
+                raise ChannelStopTimeout("Telegram loop cleanup is still running after stop timeout") from None
+            except Exception as exc:
+                logger.warning("[Telegram] failed to shut down Telegram loop resources during shutdown: %s", type(exc).__name__)
+
+        if telegram_loop_running:
+            try:
+                telegram_loop.call_soon_threadsafe(telegram_loop.stop)
+            except RuntimeError:
+                pass
+        if worker_thread is not None and worker_thread.is_alive():
+            remaining = max(0.0, deadline - shutdown_loop.time())
+            if remaining:
+                try:
+                    await asyncio.wait_for(
+                        asyncio.to_thread(worker_thread.join, remaining),
+                        timeout=remaining,
+                    )
+                except TimeoutError:
+                    logger.warning("[Telegram] polling thread did not stop within the shutdown budget")
+        if worker_thread is not None and worker_thread.is_alive():
+            raise ChannelStopTimeout("Telegram polling thread is still running after stop timeout")
+        if self._thread is worker_thread:
+            self._thread = None
+        self._application = None
+        with self._shutdown_resources_lock:
+            if self._shutdown_resources_future is cleanup_future:
+                self._shutdown_resources_future = None
         logger.info("Telegram channel stopped")
 
     async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:

backend/tests/test_channels.py

@@ -11354,7 +11354,60 @@ class TestTelegramInboundMessages:
 
         _run(go())
 
-    def test_cancelled_stop_still_stops_thread_and_clears_state(self):
+    @pytest.mark.parametrize("owner_loop", ["missing", "stopped", "closed"])
+    def test_stop_releases_abandoned_cleanup_generation(self, caplog, owner_loop):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            ch = TelegramChannel(bus=MessageBus(), config={"bot_token": "test-token"})
+            telegram_loop = asyncio.new_event_loop() if owner_loop != "missing" else None
+            if owner_loop == "closed":
+                telegram_loop.close()
+            cleanup_future: Future[None] = Future()
+            ch._tg_loop = telegram_loop
+            ch._shutdown_resources_future = cleanup_future
+            ch._download_bot = SimpleNamespace(shutdown=AsyncMock())
+            try:
+                with caplog.at_level(logging.WARNING, logger="app.channels.telegram"):
+                    await ch.stop()
+                    await ch.stop()
+
+                assert ch._shutdown_resources_future is None
+                assert caplog.text.count("Telegram loop stopped before its resource cleanup finished") == 1
+                # Dropping an abandoned generation is not evidence that its
+                # loop-owned clients were closed, nor a reason to cancel it.
+                ch._download_bot.shutdown.assert_not_awaited()
+                assert not cleanup_future.done()
+            finally:
+                if telegram_loop is not None and not telegram_loop.is_closed():
+                    telegram_loop.close()
+
+        _run(go())
+
+    def test_stop_abandons_only_the_matching_cleanup_generation(self, monkeypatch):
+        from app.channels import telegram as telegram_module
+
+        async def go():
+            ch = telegram_module.TelegramChannel(bus=MessageBus(), config={"bot_token": "test-token"})
+            abandoned: Future[None] = Future()
+            replacement: Future[None] = Future()
+            ch._shutdown_resources_future = abandoned
+
+            def replace_generation_before_release(*args, **kwargs):
+                with ch._shutdown_resources_lock:
+                    ch._shutdown_resources_future = replacement
+
+            monkeypatch.setattr(telegram_module.logger, "warning", replace_generation_before_release)
+            await ch.stop()
+
+            assert ch._shutdown_resources_future is replacement
+            assert not abandoned.done()
+            assert not replacement.done()
+
+        _run(go())
+
+    def test_cancelled_stop_keeps_owned_cleanup_and_retry_converges(self):
+        """A cancelled stop must not cancel the cleanup the Telegram loop owns."""
         from app.channels.telegram import TelegramChannel
 
         async def go():
@@ -11362,16 +11415,23 @@ class TestTelegramInboundMessages:
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
             telegram_loop = asyncio.new_event_loop()
             loop_started: Future[None] = Future()
-            drain_started = threading.Event()
+            cleanup_started = threading.Event()
+            cleanup_finished = threading.Event()
+            release_cleanup = threading.Event()
+            shutdown_calls = 0
 
             def run_telegram_loop():
                 asyncio.set_event_loop(telegram_loop)
                 telegram_loop.call_soon(loop_started.set_result, None)
                 telegram_loop.run_forever()
 
-            async def slow_drain():
-                drain_started.set()
-                await asyncio.Event().wait()
+            class DownloadBot:
+                async def shutdown(self):
+                    nonlocal shutdown_calls
+                    shutdown_calls += 1
+                    cleanup_started.set()
+                    await asyncio.to_thread(release_cleanup.wait)
+                    cleanup_finished.set()
 
             loop_thread = threading.Thread(target=run_telegram_loop, daemon=True)
             loop_thread.start()
@@ -11381,18 +11441,173 @@ class TestTelegramInboundMessages:
                 ch._thread = loop_thread
                 ch._running = True
                 ch._application = SimpleNamespace(bot=SimpleNamespace())
-                ch._cancel_telegram_bridge_tasks = slow_drain
+                ch._download_bot = DownloadBot()
 
                 stop_task = asyncio.create_task(ch.stop())
-                assert await asyncio.to_thread(drain_started.wait, 2)
+                assert await asyncio.to_thread(cleanup_started.wait, 2)
                 stop_task.cancel()
 
                 with pytest.raises(asyncio.CancelledError):
                     await stop_task
 
-                assert not loop_thread.is_alive()
+                # Cancellation reached the caller only: the loop and the cleanup
+                # it owns stay with the channel so a retry can finish them.
+                assert not cleanup_finished.is_set()
+                assert telegram_loop.is_running()
+                assert loop_thread.is_alive()
+                assert ch._thread is loop_thread
+                assert ch._application is not None
+                assert ch._download_bot is not None
+
+                release_cleanup.set()
+                assert await asyncio.to_thread(cleanup_finished.wait, 2)
+                await ch.stop()
+
+                assert shutdown_calls == 1
+                assert ch._download_bot is None
                 assert ch._thread is None
                 assert ch._application is None
+                assert not loop_thread.is_alive()
+            finally:
+                if telegram_loop.is_running():
+                    telegram_loop.call_soon_threadsafe(telegram_loop.stop)
+                await asyncio.to_thread(loop_thread.join, 2)
+                if loop_thread.is_alive():
+                    pytest.fail("Telegram test event loop did not stop")
+                telegram_loop.close()
+
+        _run(go())
+
+    def test_concurrent_stops_share_one_telegram_cleanup_generation(self):
+        """Concurrent stops must not schedule a second Telegram-loop cleanup."""
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            telegram_loop = asyncio.new_event_loop()
+            loop_started: Future[None] = Future()
+            cleanup_started = threading.Event()
+            cleanup_finished = threading.Event()
+            release_cleanup = threading.Event()
+            shutdown_calls = 0
+
+            def run_telegram_loop():
+                asyncio.set_event_loop(telegram_loop)
+                telegram_loop.call_soon(loop_started.set_result, None)
+                telegram_loop.run_forever()
+
+            class DownloadBot:
+                async def shutdown(self):
+                    nonlocal shutdown_calls
+                    shutdown_calls += 1
+                    cleanup_started.set()
+                    await asyncio.to_thread(release_cleanup.wait)
+                    cleanup_finished.set()
+
+            loop_thread = threading.Thread(target=run_telegram_loop, daemon=True)
+            loop_thread.start()
+            try:
+                loop_started.result(timeout=2)
+                ch._tg_loop = telegram_loop
+                ch._thread = loop_thread
+                ch._running = True
+                ch._application = SimpleNamespace(bot=SimpleNamespace())
+                ch._download_bot = DownloadBot()
+
+                first = asyncio.create_task(ch.stop())
+                assert await asyncio.to_thread(cleanup_started.wait, 2)
+                second = asyncio.create_task(ch.stop())
+                await asyncio.sleep(0.05)
+
+                assert not first.done()
+                assert not second.done()
+
+                release_cleanup.set()
+                await asyncio.gather(first, second)
+
+                assert shutdown_calls == 1
+                assert await asyncio.to_thread(cleanup_finished.wait, 2)
+                assert ch._download_bot is None
+                assert ch._thread is None
+                assert ch._application is None
+                assert not loop_thread.is_alive()
+            finally:
+                if telegram_loop.is_running():
+                    telegram_loop.call_soon_threadsafe(telegram_loop.stop)
+                await asyncio.to_thread(loop_thread.join, 2)
+                if loop_thread.is_alive():
+                    pytest.fail("Telegram test event loop did not stop")
+                telegram_loop.close()
+
+        _run(go())
+
+    def test_shutdown_timeout_keeps_download_bot_cleanup_running(self, monkeypatch):
+        """The Gateway deadline must not cancel Telegram-loop-owned Bot cleanup."""
+        from app.channels import telegram as telegram_module
+        from app.channels.base import ChannelStopTimeout
+
+        async def go():
+            bus = MessageBus()
+            ch = telegram_module.TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            telegram_loop = asyncio.new_event_loop()
+            loop_started: Future[None] = Future()
+            cleanup_started = threading.Event()
+            cleanup_finished = threading.Event()
+            release_cleanup = threading.Event()
+            shutdown_calls = 0
+
+            def run_telegram_loop():
+                asyncio.set_event_loop(telegram_loop)
+                telegram_loop.call_soon(loop_started.set_result, None)
+                telegram_loop.run_forever()
+
+            class DownloadBot:
+                async def shutdown(self):
+                    nonlocal shutdown_calls
+                    shutdown_calls += 1
+                    cleanup_started.set()
+                    await asyncio.to_thread(release_cleanup.wait)
+                    cleanup_finished.set()
+
+            loop_thread = threading.Thread(target=run_telegram_loop, daemon=True)
+            loop_thread.start()
+            try:
+                loop_started.result(timeout=2)
+                ch._tg_loop = telegram_loop
+                ch._thread = loop_thread
+                ch._running = True
+                ch._application = SimpleNamespace(bot=SimpleNamespace())
+                ch._download_bot = DownloadBot()
+                monkeypatch.setattr(telegram_module, "TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS", 0.01)
+
+                with pytest.raises(ChannelStopTimeout, match="Telegram loop cleanup is still running"):
+                    await ch.stop()
+
+                # The deadline interrupted the wait, not the cleanup: the loop
+                # still owns the unfinished download Bot shutdown.
+                assert await asyncio.to_thread(cleanup_started.wait, 2)
+                assert not cleanup_finished.is_set()
+                assert telegram_loop.is_running()
+                assert loop_thread.is_alive()
+                assert ch._thread is loop_thread
+                assert ch._application is not None
+                assert ch._download_bot is not None
+
+                release_cleanup.set()
+                assert await asyncio.to_thread(cleanup_finished.wait, 2)
+
+                # A retried stop joins that same cleanup and only then tears the
+                # channel down.
+                monkeypatch.setattr(telegram_module, "TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS", 5.0)
+                await ch.stop()
+
+                assert shutdown_calls == 1
+                assert ch._download_bot is None
+                assert ch._thread is None
+                assert ch._application is None
+                assert not telegram_loop.is_running()
+                assert not loop_thread.is_alive()
             finally:
                 if telegram_loop.is_running():
                     telegram_loop.call_soon_threadsafe(telegram_loop.stop)