fix(channels): clean up interrupted Telegram initialization (#6810)

* fix(channels): clean up interrupted Telegram initialization

* fix(channels): preserve Telegram shutdown failures and finalize generators

---------

Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com>

README.md

@@ -1049,7 +1049,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.
+- Telegram shutdown retains ongoing cleanup across a timeout or cancellation so a retry can finish it. Stopping during startup also cancels initialization and closes the partially initialized bot's HTTP clients before retiring its worker loop. 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.
 - Slack stop waits up to five seconds for an in-flight Socket Mode connection and its cleanup. If cleanup is still pending, the previous instance stays tracked and a restart can be retried after it finishes. SDK cleanup failures report their original error and remain retryable.
 - 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

@@ -5,7 +5,9 @@ IM channels reach DeerFlow through `langgraph-sdk` HTTP; Gateway owns thread lif
 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`.
 
-Telegram stop retries share live cleanup. If its loop dies, warn once and drop only the matching future under `_shutdown_resources_lock`; client closure stays unconfirmed.
+Telegram retries share bootstrap/Bot cleanup. Dead loops warn once; clear
+matching futures under `_shutdown_resources_lock` (closure unconfirmed).
+Finalize asyncgens; keep primary errors; propagate lone Bot errors.
 
 **Auth**: SDK and raw-httpx `/goal` calls send process-local auth plus `_csrf_headers()`; internal auth requires CSRF.
 
@@ -15,7 +17,7 @@ Telegram stop retries share live cleanup. If its loop dies, warn once and drop o
 - `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`
   Streaming `error` frames use the same failure path as transport exceptions. Publish the final outbound before releasing the inbound dedupe key, so provider redelivery can retry without overtaking the terminal reply. Log the frame's error name and message at WARNING with the thread ID; keep the synthetic error generic so payload details do not change reply selection.
   **What may be published from the stream is an allowlist, not a denylist** (`_accumulate_stream_text`): only assistant message types — LangChain serializes `AIMessage.type` as `"ai"` and `AIMessageChunk.type` as `"AIMessageChunk"`, plus the OpenAI-style `"assistant"` spelling for foreign runtimes — become displayable text. The previous rule rejected only payloads whose `type` contained `"tool"` and therefore published everything else, which leaked DeerFlow's hidden model context to every streaming IM channel: `DynamicContextMiddleware` injects recalled memory as a hidden `HumanMessage` (`type == "human"`) and rewrites the user's own turn into a new `HumanMessage`, `DurableContextMiddleware` injects a hidden `<durable_context_data>` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay (a `<memory>` fact block; a verbatim echo of the user's message as the reply). Matching is by prefix (`ai` / `assistant`), never substring, because ordinary words contain `"ai"` (`chain`, `domain`). The message type is resolved by `_stream_payload_type`, which handles both the `model_dump()` shape DeerFlow's own gateway emits and LangChain's `to_json()` constructor shape (whose top-level `type` is the literal `"constructor"`, with the class name at the tail of the `id` path). A bare `str` payload is no longer accepted at all: it carries no type information, so it cannot be attributed to the assistant, and nothing in DeerFlow produces one (`runtime/serialization.py::serialize_messages_tuple` always emits `[message_dict, metadata]`).
-- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`: every run and channel shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack cleanup awaits connect then closes off-loop; a 5s deadline/cancellation retains it for retry. SDK close errors propagate. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): the scheduled-task outbox pushes markdown to a bound identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running). `capabilities.py` gates enqueue/UI; unsupported legacy rows fail terminally (`test_channel_capabilities.py`). Await ACK; `ChannelUnavailable` parks transport failures, rejections count. Group targets use owner-bound connection IDs and a separate group hook; see [group delivery](../../docs/IM_CHANNEL_CONNECTIONS.md#feishu-group-notifications). **WeCom/Feishu push**; Feishu uses owner-scoped p2p routing. WeCom uses `WSClient.send_message`; its SDK raises `RuntimeError` for a dead socket and a rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding (owner-scoped `resolve_connections`, before liveness): a gone target is `mark_failed(terminal=True)`; a lookup failure parks. Terminal rows ignore later status writes. Scheduler side: `_on_finalization` enqueues each notice in the outcome's transaction, so recovery notifies once (unique key; idle ends use `idle:`); one message per occurrence (stop > pause > finish > run outcome); plain trials and interrupts stay silent; text (`scheduler/notification_text.py`) is localized, without IDs or links. Lifespan starts `NotificationDeliveryWorker` after the channel service (detaching enqueue if it cannot) and stops it first.
+- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`: the shared Gateway loop also serves `POST /api/channels/{name}/restart`; Slack cleanup awaits connect then closes off-loop; a 5s deadline/cancellation retains it for retry. SDK close errors propagate. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): the scheduled-task outbox pushes markdown to a bound identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running). `capabilities.py` gates enqueue/UI; unsupported legacy rows fail terminally (`test_channel_capabilities.py`). Await ACK; `ChannelUnavailable` parks transport failures, rejections count. Group targets use owner-bound connection IDs and a separate group hook; see [group delivery](../../docs/IM_CHANNEL_CONNECTIONS.md#feishu-group-notifications). **WeCom/Feishu push**; Feishu uses owner-scoped p2p routing. WeCom uses `WSClient.send_message`; its SDK raises `RuntimeError` for a dead socket and a rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding (owner-scoped `resolve_connections`, before liveness): a gone target is `mark_failed(terminal=True)`; a lookup failure parks. Terminal rows ignore later status writes. Scheduler side: `_on_finalization` enqueues each notice in the outcome's transaction, so recovery notifies once (unique key; idle ends use `idle:`); one message per occurrence (stop > pause > finish > run outcome); plain trials and interrupts stay silent; text (`scheduler/notification_text.py`) is localized, without IDs or links. Lifespan starts `NotificationDeliveryWorker` after the channel service (detaching enqueue if it cannot) and stops it first.
 - `wecom.py` - WeCom WebSocket channel with proactive `send_notification`; `_ws_transport_up` maps disconnects to `ChannelUnavailable` (no retry used). `capabilities.py` - SDK-free `CHANNEL_CAPABILITIES`.
 - `service.py` - Drain manager work before transports; retain timed-out instances for retry. Feishu/DingTalk `stop()` stop the SDK thread's private loop via `_request_*_shutdown` (Feishu disarms lark's `_auto_reconnect` first, else its reconnect branch eats the stop; DingTalk runs its own copy of the SDK connect loop), then join; a live join or Slack cleanup raises `ChannelStopTimeout`: retain, defer replacement/restart, log WARNING without traceback; wedged instances report not running. Reload returns snapshots; restart owns cached fallback/publication under the channel lock. `_stopping` fences starts/publication (including after pre-start I/O) until shutdown succeeds. Lock manager startup through stale cleanup; capture the shutdown generation before locking to fence queued starts. `stop()` stays unlocked to fence in-flight startup.
 - `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages` — the Rich path only fires when `rich_messages` is on **and** the text contains a rich construct (fenced code, emphasis, a table, a task list, `<details>`, block math, or a link), so structured command/error replies stay plain text and their newlines and `<placeholder>` tokens survive; `discord.py` runs `_on_message` on discord.py's private loop, so connection-repository work (identity lookup with the reservation commit, `/connect` binding) goes to the Gateway loop through `_submit_threadsafe_coroutine` — the SQLAlchemy engine is bound there — and bind replies return through `_run_on_discord_loop`; typing registers before that hand-off settles, and a failed hand-off skips the ack and stops the indicator it started unless another message still relies on it; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; each tick sends the indicator with discord.py 2.x's `await channel.typing()` (first failed tick logs at WARNING, later ones at DEBUG); because `stop()` runs on the main loop while typing tasks belong to `_discord_loop`, normal cross-thread cancellation, awaiting, and map cleanup are scheduled there with a bounded wait, while `_run_client()` drains tasks in its `finally` block before an exception/disconnect can make that loop unusable; an already-stopped foreign loop must never have its tasks awaited from the main loop; to that end every outbound cross-loop call (`send` / `send_file` / `_get_channel_or_thread`) goes through `_run_on_discord_loop`, which bounds the await (`DISCORD_OUTBOUND_TIMEOUT_SECONDS` 30s; `DISCORD_UPLOAD_TIMEOUT_SECONDS` 120s for file uploads: slow uplink plus 429 retry-after) and fails fast with a `RuntimeError` when the client loop is missing or not running — a dead client becomes a logged send failure instead of a permanently hung `ChannelManager` worker — and `is_running` reports client-thread aliveness (like `feishu.py`) so `ensure_channel_ready` can restart the channel after `_run_client()` exits on a fatal error; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`)

backend/app/channels/telegram.py

@@ -142,6 +142,8 @@ class TelegramChannel(Channel):
         self._thread: threading.Thread | None = None
         self._tg_loop: asyncio.AbstractEventLoop | None = None
         self._main_loop: asyncio.AbstractEventLoop | None = None
+        self._bootstrap_task: asyncio.Task[None] | None = None
+        self._polling_cleanup_started = False
         # Dedicated Bot for inbound file downloads. The Application's Bot is
         # built on the manager/main loop (see start()), so its httpx
         # connection pool is bound to that loop; opening a fresh download
@@ -171,6 +173,13 @@ class TelegramChannel(Channel):
     async def start(self) -> None:
         if self._running:
             return
+        if self._thread is not None and self._thread.is_alive():
+            raise ChannelStopTimeout("Telegram polling worker has not finished shutting down")
+        with self._shutdown_resources_lock:
+            if self._shutdown_resources_future is not None and not self._shutdown_resources_future.done():
+                raise ChannelStopTimeout("Telegram resource cleanup has not finished")
+            self._shutdown_resources_future = None
+        self._polling_cleanup_started = False
 
         try:
             from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters
@@ -238,12 +247,8 @@ class TelegramChannel(Channel):
         # 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:
+            cleanup_future = self._request_telegram_cleanup(telegram_loop)
+        if cleanup_future is not None and not telegram_loop_running and not cleanup_future.done():
             # 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:
@@ -268,7 +273,7 @@ class TelegramChannel(Channel):
 
         if telegram_loop_running:
             try:
-                telegram_loop.call_soon_threadsafe(telegram_loop.stop)
+                telegram_loop.call_soon_threadsafe(self._stop_telegram_loop, telegram_loop)
             except RuntimeError:
                 pass
         if worker_thread is not None and worker_thread.is_alive():
@@ -291,6 +296,19 @@ class TelegramChannel(Channel):
                 self._shutdown_resources_future = None
         logger.info("Telegram channel stopped")
 
+    def _request_telegram_cleanup(self, loop: asyncio.AbstractEventLoop) -> Future[None]:
+        """Share one owner-loop cleanup across stop callers and startup failure."""
+        with self._shutdown_resources_lock:
+            if self._shutdown_resources_future is None:
+                self._shutdown_resources_future = asyncio.run_coroutine_threadsafe(self._shutdown_telegram_resources(), loop)
+            return self._shutdown_resources_future
+
+    def _stop_telegram_loop(self, loop: asyncio.AbstractEventLoop) -> None:
+        # A stop retry must not interrupt run_until_complete while the polling
+        # thread is joining its cleanup after bootstrap cancellation/failure.
+        if not self._polling_cleanup_started:
+            loop.stop()
+
     async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
         if not self._application:
             return
@@ -731,21 +749,46 @@ class TelegramChannel(Channel):
                 logger.warning("[Telegram] %d inbound file download task(s) did not cancel promptly", len(still_pending))
 
     async def _shutdown_telegram_resources(self) -> None:
-        """Cancel in-flight PTB bridge work and close the download Bot.
+        """Drain startup and PTB work, then close both Bots on their owner loop.
 
-        Runs on the Telegram loop (scheduled from ``stop()``) and must finish
-        before that loop is stopped, so the download Bot's HTTPX clients are
-        shut down on the loop that owns them.
+        The application may not have reached running or even initialized: PTB
+        Application.shutdown() then skips Bot.shutdown(), despite initialized
+        HTTP clients from an interrupted/failed getMe request.
         """
+        bootstrap = self._bootstrap_task
+        if bootstrap is not None and bootstrap is not asyncio.current_task():
+            if not bootstrap.done():
+                bootstrap.cancel()
+            await asyncio.gather(bootstrap, return_exceptions=True)
         await self._cancel_telegram_bridge_tasks()
         await self._shutdown_download_bot()
+        application = self._application
+        if application is not None:
+            application_cleanup_failed = False
+            try:
+                if application.updater.running:
+                    await application.updater.stop()
+                if application.running:
+                    await application.stop()
+                await application.shutdown()
+            except BaseException:
+                application_cleanup_failed = True
+                raise
+            finally:
+                # Idempotent after full shutdown, necessary after partial init.
+                try:
+                    await application.bot.shutdown()
+                except Exception:
+                    if not application_cleanup_failed:
+                        raise
+                    logger.warning("[Telegram] failed to shut down the application bot", exc_info=True)
 
     async def _shutdown_download_bot(self) -> None:
         """Close the download Bot's HTTPX clients on the Telegram loop.
 
         python-telegram-bot 22.7 exposes no ``Bot.session``; the HTTPX clients
-        are closed by ``Bot.shutdown()``, which is a no-op if the bot was never
-        initialized. Must run on the loop the bot was initialized on.
+        are closed by ``Bot.shutdown()``, which is a no-op if request resources
+        were never initialized or were already shut down. Run on their owner loop.
         """
         bot = self._download_bot
         if bot is None:
@@ -939,34 +982,53 @@ class TelegramChannel(Channel):
         except Exception:
             logger.exception("[Telegram] failed to send running reply in chat=%s", chat_id)
 
+    async def _bootstrap_telegram(self, application: Any) -> None:
+        if self._running:
+            await application.initialize()
+        if self._running:
+            await application.start()
+        if self._running:
+            await application.updater.start_polling()
+
     def _run_polling(self) -> None:
         """Run telegram polling in a dedicated thread."""
         application = self._application
         if application is None:
             return
-        self._tg_loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(self._tg_loop)
+        loop = asyncio.new_event_loop()
+        self._tg_loop = loop
+        asyncio.set_event_loop(loop)
         try:
             # Cannot use run_polling() because it calls add_signal_handler(),
             # which only works in the main thread.  Instead, manually
             # initialize the application and start the updater.
-            self._tg_loop.run_until_complete(application.initialize())
-            self._tg_loop.run_until_complete(application.start())
-            self._tg_loop.run_until_complete(application.updater.start_polling())
-            self._tg_loop.run_forever()
+            self._bootstrap_task = loop.create_task(self._bootstrap_telegram(application))
+            loop.run_until_complete(self._bootstrap_task)
+            if self._running:
+                loop.run_forever()
+        except asyncio.CancelledError:
+            # stop() cancelled and awaits bootstrap on this same loop.
+            pass
         except Exception:
             if self._running:
                 logger.exception("Telegram polling error")
         finally:
-            # Graceful shutdown
+            self._running = False
+            self._polling_cleanup_started = True
             try:
-                self._tg_loop.run_until_complete(self._cancel_telegram_bridge_tasks())
-                if application.updater.running:
-                    self._tg_loop.run_until_complete(application.updater.stop())
-                self._tg_loop.run_until_complete(application.stop())
-                self._tg_loop.run_until_complete(application.shutdown())
+                cleanup = self._request_telegram_cleanup(loop)
+                loop.run_until_complete(asyncio.wrap_future(cleanup, loop=loop))
             except Exception:
                 logger.exception("Error during Telegram shutdown")
+            finally:
+                try:
+                    loop.run_until_complete(loop.shutdown_asyncgens())
+                except Exception:
+                    logger.exception("Error shutting down Telegram async generators")
+                finally:
+                    self._bootstrap_task = None
+                    asyncio.set_event_loop(None)
+                    loop.close()
 
     def _check_user(self, user_id: int) -> bool:
         if self._allowed_users is None:

backend/tests/blocking_io/test_channel_stop_offloop.py

@@ -29,6 +29,7 @@ from app.channels.discord import DiscordChannel
 from app.channels.feishu import FeishuChannel
 from app.channels.message_bus import MessageBus
 from app.channels.slack import SlackChannel
+from app.channels.telegram import TelegramChannel
 
 pytestmark = pytest.mark.asyncio
 
@@ -88,7 +89,7 @@ async def test_slack_stop_keeps_connect_cleanup_off_loop(monkeypatch) -> None:
         await channel.stop()
 
 
-@pytest.mark.parametrize("channel_cls", [FeishuChannel, DingTalkChannel, DiscordChannel])
+@pytest.mark.parametrize("channel_cls", [FeishuChannel, DingTalkChannel, DiscordChannel, TelegramChannel])
 async def test_stop_joins_sdk_thread_off_loop(channel_cls) -> None:
     release = threading.Event()
     sdk_thread = threading.Thread(target=release.wait, daemon=True)

backend/tests/test_channels.py

@@ -11440,7 +11440,7 @@ class TestTelegramInboundMessages:
                 ch._tg_loop = telegram_loop
                 ch._thread = loop_thread
                 ch._running = True
-                ch._application = SimpleNamespace(bot=SimpleNamespace())
+                ch._application = SimpleNamespace(bot=SimpleNamespace(shutdown=AsyncMock()), updater=SimpleNamespace(running=False), running=False, shutdown=AsyncMock())
                 ch._download_bot = DownloadBot()
 
                 stop_task = asyncio.create_task(ch.stop())
@@ -11512,7 +11512,7 @@ class TestTelegramInboundMessages:
                 ch._tg_loop = telegram_loop
                 ch._thread = loop_thread
                 ch._running = True
-                ch._application = SimpleNamespace(bot=SimpleNamespace())
+                ch._application = SimpleNamespace(bot=SimpleNamespace(shutdown=AsyncMock()), updater=SimpleNamespace(running=False), running=False, shutdown=AsyncMock())
                 ch._download_bot = DownloadBot()
 
                 first = asyncio.create_task(ch.stop())
@@ -11577,7 +11577,7 @@ class TestTelegramInboundMessages:
                 ch._tg_loop = telegram_loop
                 ch._thread = loop_thread
                 ch._running = True
-                ch._application = SimpleNamespace(bot=SimpleNamespace())
+                ch._application = SimpleNamespace(bot=SimpleNamespace(shutdown=AsyncMock()), updater=SimpleNamespace(running=False), running=False, shutdown=AsyncMock())
                 ch._download_bot = DownloadBot()
                 monkeypatch.setattr(telegram_module, "TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS", 0.01)
 

backend/tests/test_telegram_polling_shutdown.py

@@ -0,0 +1,311 @@
+"""Real PTB HTTP connections must close even when startup never completes."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import threading
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.channels.base import ChannelStopTimeout
+from app.channels.message_bus import MessageBus
+from app.channels.telegram import TelegramChannel
+
+
+async def _until(predicate, timeout=5):
+    async with asyncio.timeout(timeout):
+        while not predicate():
+            await asyncio.sleep(0.01)
+
+
+class _BotAPI:
+    """Small HTTP/1.1 Bot API fixture with observable client EOF."""
+
+    def __init__(self, mode):
+        self.mode = mode
+        self.requests = []
+        self.connections = []
+        self.release = asyncio.Event()
+        self.handlers = set()
+
+    async def handle(self, reader, writer):
+        self.handlers.add(asyncio.current_task())
+        connection = {"writer": writer, "eof": asyncio.Event()}
+        self.connections.append(connection)
+        try:
+            while line := await reader.readline():
+                method = line.decode().split()[1].rsplit("/", 1)[-1].lower()
+                headers = {}
+                while (line := await reader.readline()) not in (b"\r\n", b"\n", b""):
+                    key, value = line.decode().split(":", 1)
+                    headers[key.lower()] = value.strip()
+                await reader.readexactly(int(headers.get("content-length", "0")))
+                self.requests.append(method)
+                if method == "getme" and self.mode == "blocked":
+                    gate = asyncio.create_task(self.release.wait())
+                    peer = asyncio.create_task(reader.read(1))
+                    try:
+                        done, _ = await asyncio.wait((gate, peer), return_when=asyncio.FIRST_COMPLETED)
+                        if peer in done and not peer.result():
+                            connection["eof"].set()
+                            return
+                    finally:
+                        gate.cancel()
+                        peer.cancel()
+                        await asyncio.gather(gate, peer, return_exceptions=True)
+                if method == "getme":
+                    result = {"id": 12345, "is_bot": True, "first_name": "Offline", "username": "offline_test_bot"}
+                elif method == "getupdates":
+                    await asyncio.sleep(0.01)
+                    result = []
+                else:
+                    assert method == "deletewebhook", method
+                    result = True
+                rejected = method == "getme" and self.mode == "rejected"
+                payload = {"ok": False, "error_code": 401, "description": "Unauthorized"} if rejected else {"ok": True, "result": result}
+                data = json.dumps(payload).encode()
+                status = "401 Unauthorized" if rejected else "200 OK"
+                writer.write(f"HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {len(data)}\r\n\r\n".encode() + data)
+                await writer.drain()
+            connection["eof"].set()
+        except (ConnectionError, asyncio.IncompleteReadError):
+            connection["eof"].set()
+        finally:
+            writer.close()
+            await writer.wait_closed()
+            self.handlers.discard(asyncio.current_task())
+
+
+def _cleanup_failed_assertion(loop, application):
+    """Tests inspect resources before this fallback; RED must not leak them."""
+    if loop.is_closed():
+        return
+    asyncio.set_event_loop(loop)
+
+    async def close():
+        pending = [task for task in asyncio.all_tasks(loop) if task is not asyncio.current_task() and not task.done()]
+        for task in pending:
+            task.cancel()
+        if pending:
+            await asyncio.gather(*pending, return_exceptions=True)
+        await application.bot.shutdown()
+
+    try:
+        loop.run_until_complete(close())
+        loop.run_until_complete(loop.shutdown_asyncgens())
+    finally:
+        loop.close()
+        asyncio.set_event_loop(None)
+
+
+@asynccontextmanager
+async def _channel(monkeypatch, *, mode="normal", shutdown_gate=None):
+    ext = pytest.importorskip("telegram.ext")
+    from telegram.request import HTTPXRequest
+
+    class Request(HTTPXRequest):
+        async def shutdown(self):
+            if shutdown_gate is not None and not self._client.is_closed:
+                shutdown_gate["calls"] += 1
+                shutdown_gate["started"].set()
+                while not shutdown_gate["release"].is_set():
+                    await asyncio.sleep(0.01)
+            await super().shutdown()
+
+    api = _BotAPI(mode)
+    server = await asyncio.start_server(api.handle, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    original = ext.ApplicationBuilder
+
+    def builder():
+        return original().base_url(f"http://127.0.0.1:{port}/bot").request(Request(httpx_kwargs={"trust_env": False})).get_updates_request(Request(httpx_kwargs={"trust_env": False}))
+
+    monkeypatch.setattr(ext, "ApplicationBuilder", builder)
+    channel = TelegramChannel(MessageBus(), {"bot_token": "12345:offline-test-token"})
+    owned = []
+
+    async def start():
+        before = len(api.requests)
+        await channel.start()
+        application, thread = channel._application, channel._thread
+        await _until(lambda: len(api.requests) > before)
+        owned.append((channel._tg_loop, application, thread))
+        return application, thread, channel._tg_loop
+
+    try:
+        yield channel, api, start
+    finally:
+        api.release.set()
+        if shutdown_gate is not None:
+            shutdown_gate["release"].set()
+        try:
+            await asyncio.wait_for(channel.stop(), 6)
+        except (ChannelStopTimeout, TimeoutError):
+            pass
+        for loop, application, thread in owned:
+            if loop.is_running():
+                loop.call_soon_threadsafe(loop.stop)
+            await asyncio.to_thread(thread.join, 3)
+            assert not thread.is_alive(), "test must not leave a Telegram worker"
+            await asyncio.to_thread(_cleanup_failed_assertion, loop, application)
+        server.close()
+        await server.wait_closed()
+        for connection in api.connections:
+            connection["writer"].close()
+        pending = list(api.handlers)
+        for task in pending:
+            task.cancel()
+        if pending:
+            await asyncio.gather(*pending, return_exceptions=True)
+
+
+async def _assert_closed(api, application, thread, loop):
+    assert not thread.is_alive()
+    assert all(request._client.is_closed for request in application.bot._request)
+    await _until(lambda: all(connection["eof"].is_set() for connection in api.connections))
+    assert not asyncio.all_tasks(loop)
+    assert loop.is_closed()
+
+
+@pytest.mark.asyncio
+async def test_stop_during_real_ptb_initialize_closes_requests_and_cancels_startup(monkeypatch):
+    async with _channel(monkeypatch, mode="blocked") as (channel, api, start):
+        application, thread, loop = await start()
+        assert not application.running
+        assert not application._initialized
+        await channel.stop()
+        await _assert_closed(api, application, thread, loop)
+        assert "deletewebhook" not in api.requests
+        assert "getupdates" not in api.requests
+
+
+@pytest.mark.asyncio
+async def test_failed_real_ptb_initialize_closes_partial_bot_without_explicit_stop(monkeypatch):
+    async with _channel(monkeypatch, mode="rejected") as (channel, api, start):
+        application, thread, loop = await start()
+        await _until(lambda: not thread.is_alive())
+        await _assert_closed(api, application, thread, loop)
+        assert channel.is_running is False
+        await channel.stop()
+
+
+@pytest.mark.asyncio
+async def test_real_ptb_can_restart_after_complete_stop(monkeypatch):
+    async with _channel(monkeypatch) as (channel, api, start):
+        for _ in range(2):
+            application, thread, loop = await start()
+            await _until(lambda: application.running and application.updater.running)
+            await channel.stop()
+            await _assert_closed(api, application, thread, loop)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("interruption", ["cancel", "timeout"])
+async def test_stop_interruption_retains_partial_application_cleanup_for_retry(monkeypatch, interruption):
+    from app.channels import telegram as telegram_module
+
+    gate = {"started": threading.Event(), "release": threading.Event(), "calls": 0}
+    if interruption == "timeout":
+        monkeypatch.setattr(telegram_module, "TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS", 0.05)
+    async with _channel(monkeypatch, mode="blocked", shutdown_gate=gate) as (channel, api, start):
+        application, thread, loop = await start()
+        stop = asyncio.create_task(channel.stop())
+        try:
+            await _until(gate["started"].is_set)
+            cleanup = channel._shutdown_resources_future
+            if interruption == "cancel":
+                stop.cancel()
+                with pytest.raises(asyncio.CancelledError):
+                    await stop
+            else:
+                with pytest.raises(ChannelStopTimeout):
+                    await stop
+            assert channel._thread is thread
+            assert thread.is_alive() and loop.is_running()
+            assert cleanup is not None and not cleanup.cancelled()
+            assert channel._shutdown_resources_future is cleanup
+            gate["release"].set()
+            monkeypatch.setattr(telegram_module, "TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS", 4)
+            await channel.stop()
+            await _assert_closed(api, application, thread, loop)
+            assert gate["calls"] == 2
+        finally:
+            gate["release"].set()
+            if not stop.done():
+                stop.cancel()
+            await asyncio.gather(stop, return_exceptions=True)
+
+
+def test_polling_finalizes_async_generators_when_resource_cleanup_fails(monkeypatch, caplog):
+    channel = TelegramChannel(MessageBus(), {"bot_token": "12345:offline-test-token"})
+    primary = ValueError("updater shutdown failed")
+    channel._application = SimpleNamespace(
+        updater=SimpleNamespace(running=True, stop=AsyncMock(side_effect=primary)),
+        running=False,
+        shutdown=AsyncMock(),
+        bot=SimpleNamespace(shutdown=AsyncMock()),
+    )
+    finalized = threading.Event()
+    generators = []
+
+    async def resource():
+        try:
+            yield
+        finally:
+            finalized.set()
+
+    async def bootstrap(application):
+        generator = resource()
+        generators.append(generator)
+        await anext(generator)
+        channel._running = False
+
+    monkeypatch.setattr(channel, "_bootstrap_telegram", bootstrap)
+    channel._running = True
+    thread = threading.Thread(target=channel._run_polling)
+    thread.start()
+    try:
+        thread.join(timeout=5)
+        assert not thread.is_alive()
+        assert finalized.is_set()
+        assert channel._tg_loop.is_closed()
+        assert any(record.exc_info and record.exc_info[1] is primary for record in caplog.records)
+    finally:
+        if thread.is_alive():
+            channel._tg_loop.call_soon_threadsafe(channel._tg_loop.stop)
+            thread.join(timeout=5)
+        assert not thread.is_alive(), "test must not leave a Telegram worker"
+        # RED must not leak the generator; assertions above precede fallback cleanup.
+        for generator in generators:
+            asyncio.run(generator.aclose())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("primary_kind", ["exception", "cancelled", None])
+async def test_application_bot_shutdown_preserves_primary_error_and_reports_its_own(primary_kind, caplog):
+    channel = TelegramChannel(MessageBus(), {"bot_token": "12345:offline-test-token"})
+    primary = None
+    if primary_kind == "exception":
+        primary = ValueError("updater shutdown failed")
+    elif primary_kind == "cancelled":
+        primary = asyncio.CancelledError("application cleanup cancelled")
+    bot_error = RuntimeError("application bot shutdown failed")
+    channel._application = SimpleNamespace(
+        updater=SimpleNamespace(running=True, stop=AsyncMock(side_effect=primary)),
+        running=False,
+        shutdown=AsyncMock(),
+        bot=SimpleNamespace(shutdown=AsyncMock(side_effect=bot_error)),
+    )
+
+    expected = primary if primary is not None else bot_error
+    with pytest.raises(type(expected)) as raised:
+        await channel._shutdown_telegram_resources()
+
+    assert raised.value is expected
+    channel._application.bot.shutdown.assert_awaited_once()
+    if primary is not None:
+        assert any("failed to shut down the application bot" in record.message and record.exc_info and record.exc_info[1] is bot_error for record in caplog.records)