fix(telegram): retain loop cleanup across bounded stops

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 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 @@ def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
         # 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 @@ async def stop(self) -> None:
         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()
+
+        # 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:
+                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__)
 
-        try:
-            if telegram_loop and telegram_loop.is_running():
-                drain_future = asyncio.run_coroutine_threadsafe(self._shutdown_telegram_resources(), telegram_loop)
+        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:
-                    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
+                    await asyncio.wait_for(
+                        asyncio.to_thread(worker_thread.join, remaining),
+                        timeout=remaining,
+                    )
                 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
-            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
+                    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: