fix(telegram): receive inbound attachments (#4392)

* fix(telegram): receive inbound attachments

* refactor(telegram): tighten inbound attachment handoff

---------

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

README.md

@@ -502,6 +502,7 @@ DINGTALK_CLIENT_SECRET=your_client_secret
 
 1. Chat with [@BotFather](https://t.me/BotFather), send `/newbot`, and copy the HTTP API token.
 2. Set `TELEGRAM_BOT_TOKEN` in `.env` and enable the channel in `config.yaml`.
+3. The bot accepts inbound text, photos, and documents (with or without captions). Hosted Bot API downloads are limited to 20 MB per attachment.
 
 **Slack Setup**
 

README_zh.md

@@ -421,6 +421,7 @@ DINGTALK_CLIENT_SECRET=your_client_secret
 
 1. 打开 [@BotFather](https://t.me/BotFather)，发送 `/newbot`，复制生成的 HTTP API token。
 2. 在 `.env` 中设置 `TELEGRAM_BOT_TOKEN`，并在 `config.yaml` 里启用该渠道。
+3. 机器人支持接收入站文本、图片和文档（可带说明文字，也可不带）；托管版 Bot API 的单个附件下载上限为 20 MB。
 
 **Slack 配置**
 

backend/AGENTS.md

@@ -666,7 +666,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
   A swallowed streaming failure publishes its final outbound before releasing the inbound dedupe key, so a provider redelivery can retry without overtaking the terminal reply.
 - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle)
 - `service.py` - Manages lifecycle of all configured channels from `config.yaml`
-- `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` registers the "Working on it..." placeholder as the stream target and edits it in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured)
+- `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, and edits the "Working on it..." stream target in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured)
 - `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR
 - `app/gateway/routers/channel_connections.py` - Browser-facing user connection and disconnect APIs
 - `deerflow.persistence.channel_connections` - SQL-backed user-owned connection, optional credential, connect state, and conversation store
@@ -674,6 +674,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 **Message Flow**:
 1. External platform -> Channel impl -> `MessageBus.publish_inbound()`
    - For GitHub, the webhook router verifies the delivery then calls `fanout_event(bus, ...)`; matching agent bindings publish one `InboundMessage` each instead of a long-polling channel worker.
+   - Telegram photo/document updates use the largest photo size or document metadata, preserve `message.caption`, enforce the hosted Bot API's 20,000,000-byte download ceiling before and after download, and never expose the token-bearing Bot API file URL. Downloaded bytes cross the adapter/manager boundary only through `message_bus.INBOUND_FILE_CONTENT_KEY`; the manager consumes that transient field before persisting safe upload metadata.
 2. `ChannelManager._dispatch_loop()` consumes from queue
 3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id`. The Gateway accepts `channel_user_id` only from an internally authenticated channel caller's top-level `body.context`, clears it from both free-form `body.config` sections, and writes it into runtime context only (never `configurable`, which is checkpointed). `bash_tool` exposes it to sandbox commands as the fixed env var `DEERFLOW_CHANNEL_USER_ID` — via a shell-quoted command-string prefix, NOT the `execute_command(env=...)` channel, which is reserved for request-scoped secrets and would switch `AioSandbox` onto the `bash.exec` path (image >= 1.9.3, fresh session per call). Per-call injection keeps group-chat identity correct (one thread/sandbox, many senders) **without depending on the AIO shell's session semantics**: every IM-channel command carries an explicit `export VAR=<id>; ` (valid id) or `unset VAR; ` (empty / non-str / over the 256-char cap). The AIO no-env path reuses a persistent shell session (the reason for the class lock, #1433), so a bare command could otherwise resolve a stale id an earlier sender exported; the `unset` closes the window the length/type guard would open (a dropped id would inherit the previous sender's value). Non-IM runs (no `channel_user_id` in context) are left untouched. Not injected on the Windows local sandbox (its PowerShell/cmd.exe fallback has no `export`/`unset`). Propagates across `task` delegation: `task_tool` captures the dispatching turn's id and the subagent executor forwards it into the subagent's runtime context, same as the guardrail attribution fields. The runtime-context value is authorization-grade at the Gateway/guardrail boundary, but the exported shell variable remains informational because any bash command can overwrite its own environment; skills must not treat the shell variable itself as authenticated identity. Tests: `tests/test_gateway_services.py`, `tests/test_channel_user_id_env.py`
 4. For chat: look up/create thread through Gateway's LangGraph-compatible API

backend/app/channels/manager.py

@@ -21,6 +21,7 @@ from langgraph_sdk.errors import ConflictError
 from app.channels import feishu_run_policy as _feishu_run_policy  # noqa: F401
 from app.channels.commands import KNOWN_CHANNEL_COMMANDS
 from app.channels.message_bus import (
+    INBOUND_FILE_CONTENT_KEY,
     PENDING_CLARIFICATION_METADATA_KEY,
     InboundMessage,
     InboundMessageType,
@@ -833,15 +834,21 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
             ftype = f.get("type") if isinstance(f.get("type"), str) else "file"
             filename = f.get("filename") if isinstance(f.get("filename"), str) else ""
 
-            try:
-                data = await file_reader(f, client)
-            except Exception:
-                logger.exception(
-                    "[Manager] failed to read inbound file: channel=%s, file=%s",
-                    msg.channel_name,
-                    f.get("url") or filename or idx,
-                )
-                continue
+            inline_content = f.pop(INBOUND_FILE_CONTENT_KEY, None)
+            if isinstance(inline_content, bytes):
+                data = inline_content
+            elif isinstance(inline_content, (bytearray, memoryview)):
+                data = bytes(inline_content)
+            else:
+                try:
+                    data = await file_reader(f, client)
+                except Exception:
+                    logger.exception(
+                        "[Manager] failed to read inbound file: channel=%s, file=%s",
+                        msg.channel_name,
+                        f.get("url") or filename or idx,
+                    )
+                    continue
 
             if data is None:
                 logger.warning(

backend/app/channels/message_bus.py

@@ -15,6 +15,9 @@ logger = logging.getLogger(__name__)
 
 PENDING_CLARIFICATION_METADATA_KEY = "pending_clarification"
 RESOLVED_FROM_PENDING_CLARIFICATION_METADATA_KEY = "resolved_from_pending_clarification"
+# Adapter-owned bytes may use this transient key while crossing the channel
+# boundary. ChannelManager consumes and removes it before persisting metadata.
+INBOUND_FILE_CONTENT_KEY = "_content"
 
 
 # ---------------------------------------------------------------------------

backend/app/channels/telegram.py

@@ -6,15 +6,29 @@ import asyncio
 import logging
 import threading
 import time
+from collections.abc import Coroutine
 from typing import Any
 
 from app.channels.base import Channel
 from app.channels.connection_identity import attach_connection_identity
-from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment
+from app.channels.message_bus import (
+    INBOUND_FILE_CONTENT_KEY,
+    InboundMessage,
+    InboundMessageType,
+    MessageBus,
+    OutboundMessage,
+    ResolvedAttachment,
+)
+from deerflow.uploads.manager import is_upload_staging_file, normalize_filename
 
 logger = logging.getLogger(__name__)
 
 TELEGRAM_MAX_MESSAGE_LENGTH = 4096
+# Telegram's hosted Bot API documents this as 20 MB (decimal bytes).
+TELEGRAM_MAX_INBOUND_FILE_BYTES = 20_000_000
+# Keep Telegram cleanup inside the Gateway's five-second shutdown-hook budget.
+TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS = 4.0
+TELEGRAM_BRIDGE_DRAIN_TIMEOUT_SECONDS = 1.0
 STREAM_EDIT_MIN_INTERVAL_SECONDS = 1.0
 # Groups (negative chat_id) are capped at 20 messages/minute by Telegram,
 # so stream edits there must pace well below the private-chat 1 msg/s guideline.
@@ -41,6 +55,9 @@ class TelegramChannel(Channel):
         self._thread: threading.Thread | None = None
         self._tg_loop: asyncio.AbstractEventLoop | None = None
         self._main_loop: asyncio.AbstractEventLoop | None = None
+        # 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()
         self._allowed_users: set[int] = set()
         for uid in config.get("allowed_users", []):
             try:
@@ -96,6 +113,10 @@ class TelegramChannel(Channel):
         # General message handler
         app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self._on_text))
 
+        # Telegram keeps attachment captions separate from message.text, and
+        # photo/document updates do not match filters.TEXT.
+        app.add_handler(MessageHandler(filters.PHOTO | filters.Document.ALL, self._on_text))
+
         self._application = app
 
         # Run polling in a dedicated thread with its own event loop
@@ -106,12 +127,47 @@ class TelegramChannel(Channel):
     async def stop(self) -> None:
         self._running = False
         self.bus.unsubscribe_outbound(self._on_outbound)
-        if self._tg_loop and self._tg_loop.is_running():
-            self._tg_loop.call_soon_threadsafe(self._tg_loop.stop)
-        if self._thread:
-            self._thread.join(timeout=10)
-            self._thread = None
-        self._application = None
+        shutdown_loop = asyncio.get_running_loop()
+        deadline = shutdown_loop.time() + TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS
+        telegram_loop = self._tg_loop
+        worker_thread = self._thread
+
+        try:
+            if telegram_loop and telegram_loop.is_running():
+                drain_future = asyncio.run_coroutine_threadsafe(self._cancel_telegram_bridge_tasks(), 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 cancelling inbound file downloads during shutdown")
+                except Exception as exc:
+                    logger.warning("[Telegram] failed to cancel inbound file downloads 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():
+                    logger.warning("[Telegram] polling thread is still exiting after bounded shutdown")
+                self._thread = None
+                self._application = None
         logger.info("Telegram channel stopped")
 
     async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
@@ -293,8 +349,137 @@ class TelegramChannel(Channel):
             logger.exception("[Telegram] failed to send file: %s", attachment.filename)
             return False
 
+    async def receive_file(self, msg: InboundMessage, thread_id: str, *, user_id: str | None = None) -> InboundMessage:
+        """Download inbound Telegram attachments for the shared upload pipeline.
+
+        The Bot API download URL contains the bot token, so this adapter never
+        exposes that URL to ``InboundMessage`` or logs. Instead it hands the
+        downloaded bytes to ``ChannelManager`` through a short-lived private
+        field that the manager consumes before persisting safe upload metadata.
+        """
+        # Owner/thread scoping is intentionally applied by ChannelManager when
+        # it persists these bytes through _ingest_inbound_files().
+        del thread_id, user_id
+        if not msg.files:
+            return msg
+
+        bot = self._application.bot if self._application is not None else None
+        materialized: list[dict[str, Any]] = []
+        unavailable: list[str] = []
+
+        for file_info in msg.files:
+            if not isinstance(file_info, dict):
+                continue
+
+            filename = self._safe_inbound_filename(file_info.get("filename"), "attachment")
+            file_id = file_info.get("file_id") if isinstance(file_info.get("file_id"), str) else ""
+            declared_size = self._file_size(file_info.get("size"))
+            if declared_size is not None and declared_size > TELEGRAM_MAX_INBOUND_FILE_BYTES:
+                logger.warning("[Telegram] inbound file exceeds 20 MB download limit, skipping: %s", filename)
+                unavailable.append(f"{filename} (exceeds the 20 MB download limit)")
+                continue
+
+            if bot is None or not file_id:
+                logger.error("[Telegram] cannot download inbound file: %s", filename)
+                unavailable.append(f"{filename} (download unavailable)")
+                continue
+
+            try:
+                resolved_size, content = await self._run_on_telegram_loop(self._download_inbound_file(bot, file_id))
+            except Exception as exc:
+                # Exception strings from HTTP clients can contain request URLs.
+                # Log only the class name so a Bot API token can never leak.
+                logger.error("[Telegram] failed to download inbound file %s: %s", filename, type(exc).__name__)
+                unavailable.append(f"{filename} (download failed)")
+                continue
+
+            if content is None:
+                logger.warning("[Telegram] resolved inbound file exceeds 20 MB download limit, skipping: %s", filename)
+                unavailable.append(f"{filename} (exceeds the 20 MB download limit)")
+                continue
+
+            if len(content) > TELEGRAM_MAX_INBOUND_FILE_BYTES:
+                logger.warning("[Telegram] downloaded inbound file exceeds 20 MB limit, skipping: %s", filename)
+                unavailable.append(f"{filename} (exceeds the 20 MB download limit)")
+                continue
+
+            materialized.append(
+                {
+                    "type": "image" if file_info.get("type") == "image" else "file",
+                    "filename": filename,
+                    "mime_type": file_info.get("mime_type") if isinstance(file_info.get("mime_type"), str) else "application/octet-stream",
+                    "size": len(content),
+                    INBOUND_FILE_CONTENT_KEY: content,
+                }
+            )
+
+        msg.files = materialized
+        if unavailable:
+            notice = f"[Telegram attachment unavailable: {', '.join(unavailable)}]"
+            msg.text = f"{msg.text}\n\n{notice}" if msg.text else notice
+        return msg
+
     # -- helpers -----------------------------------------------------------
 
+    async def _download_inbound_file(self, bot: Any, file_id: str) -> tuple[int | None, bytearray | None]:
+        """Fetch one file entirely on the event loop that owns PTB's HTTP client."""
+        telegram_file = await bot.get_file(file_id)
+        resolved_size = self._file_size(getattr(telegram_file, "file_size", None))
+        if resolved_size is not None and resolved_size > TELEGRAM_MAX_INBOUND_FILE_BYTES:
+            return resolved_size, None
+        return resolved_size, await telegram_file.download_as_bytearray()
+
+    async def _track_telegram_bridge_task(self, coroutine: Coroutine[Any, Any, Any]) -> Any:
+        """Keep a main-loop submission visible to Telegram-loop shutdown."""
+        task = asyncio.current_task()
+        if task is None:
+            coroutine.close()
+            raise RuntimeError("Telegram bridge task is unavailable")
+        self._tg_bridge_tasks.add(task)
+        try:
+            return await coroutine
+        finally:
+            self._tg_bridge_tasks.discard(task)
+
+    async def _cancel_telegram_bridge_tasks(self) -> None:
+        """Cancel and drain cross-loop PTB work before its event loop exits."""
+        current = asyncio.current_task()
+        pending = [task for task in self._tg_bridge_tasks if task is not current and not task.done()]
+        for task in pending:
+            task.cancel()
+        if pending:
+            done, still_pending = await asyncio.wait(pending, timeout=TELEGRAM_BRIDGE_DRAIN_TIMEOUT_SECONDS)
+            if done:
+                await asyncio.gather(*done, return_exceptions=True)
+            if still_pending:
+                logger.warning("[Telegram] %d inbound file download task(s) did not cancel promptly", len(still_pending))
+
+    async def _run_on_telegram_loop(self, coroutine: Coroutine[Any, Any, Any]) -> Any:
+        """Await a PTB coroutine without using its HTTP client across event loops."""
+        telegram_loop = self._tg_loop
+        current_loop = asyncio.get_running_loop()
+        if telegram_loop is None or telegram_loop is current_loop:
+            return await coroutine
+
+        if not telegram_loop.is_running():
+            coroutine.close()
+            raise RuntimeError("Telegram event loop is not running")
+
+        tracked_coroutine = self._track_telegram_bridge_task(coroutine)
+        try:
+            future = asyncio.run_coroutine_threadsafe(tracked_coroutine, telegram_loop)
+        except BaseException:
+            # A concurrent shutdown can close the loop between the running
+            # check and scheduling. Close the unscheduled coroutine explicitly.
+            tracked_coroutine.close()
+            coroutine.close()
+            raise
+        try:
+            return await asyncio.wrap_future(future)
+        except asyncio.CancelledError:
+            future.cancel()
+            raise
+
     @staticmethod
     def _stream_key(chat_id: str, thread_ts: str | None) -> str:
         return f"{chat_id}:{thread_ts or ''}"
@@ -303,8 +488,83 @@ class TelegramChannel(Channel):
     def _parse_message_id(value: str | None) -> int | None:
         try:
             return int(value) if value else None
+        except (OverflowError, TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _file_size(value: Any) -> int | None:
+        if isinstance(value, bool):
+            return None
+        try:
+            size = int(value)
         except (TypeError, ValueError):
             return None
+        return size if size >= 0 else None
+
+    @staticmethod
+    def _safe_inbound_filename(value: Any, fallback: str) -> str:
+        if not isinstance(value, str):
+            return fallback
+        try:
+            candidate = normalize_filename(value.strip())
+        except (UnicodeError, ValueError):
+            return fallback
+        if is_upload_staging_file(candidate) or any(ord(char) < 32 for char in candidate):
+            return fallback
+        return candidate
+
+    @classmethod
+    def _extract_inbound_files(cls, message: Any) -> list[dict[str, Any]]:
+        files: list[dict[str, Any]] = []
+        message_id = str(getattr(message, "message_id", "message"))
+
+        # Materialize the PTB sequence once. Test doubles and partially shaped
+        # update objects can expose a truthy but empty iterable here.
+        photo_sizes = tuple(getattr(message, "photo", None) or ())
+        if photo_sizes:
+            photo = max(
+                photo_sizes,
+                key=lambda item: (
+                    (cls._file_size(getattr(item, "width", None)) or 0) * (cls._file_size(getattr(item, "height", None)) or 0),
+                    cls._file_size(getattr(item, "file_size", None)) or 0,
+                ),
+            )
+            file_id = getattr(photo, "file_id", None)
+            if isinstance(file_id, str) and file_id:
+                file_unique_id = getattr(photo, "file_unique_id", None)
+                files.append(
+                    {
+                        "type": "image",
+                        "file_id": file_id,
+                        "file_unique_id": file_unique_id if isinstance(file_unique_id, str) else None,
+                        "filename": f"telegram-photo-{message_id}.jpg",
+                        "mime_type": "image/jpeg",
+                        "size": cls._file_size(getattr(photo, "file_size", None)),
+                    }
+                )
+
+        document = getattr(message, "document", None)
+        document_id = getattr(document, "file_id", None) if document is not None else None
+        if isinstance(document_id, str) and document_id:
+            file_unique_id = getattr(document, "file_unique_id", None)
+            mime_type = getattr(document, "mime_type", None)
+            if not isinstance(mime_type, str) or not mime_type:
+                mime_type = "application/octet-stream"
+            # Avoid the lazy system MIME database lookup on the Telegram event
+            # loop. The original name is preferred; an opaque extension is safe.
+            fallback = f"telegram-document-{message_id}.bin"
+            files.append(
+                {
+                    "type": "file",
+                    "file_id": document_id,
+                    "file_unique_id": file_unique_id if isinstance(file_unique_id, str) else None,
+                    "filename": cls._safe_inbound_filename(getattr(document, "file_name", None), fallback),
+                    "mime_type": mime_type,
+                    "size": cls._file_size(getattr(document, "file_size", None)),
+                }
+            )
+
+        return files
 
     def _register_stream_message(self, key: str, *, message_id: int, last_text: str, last_edit_at: float) -> None:
         self._stream_messages.pop(key, None)
@@ -358,15 +618,18 @@ class TelegramChannel(Channel):
 
     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)
         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(self._application.initialize())
-            self._tg_loop.run_until_complete(self._application.start())
-            self._tg_loop.run_until_complete(self._application.updater.start_polling())
+            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()
         except Exception:
             if self._running:
@@ -374,10 +637,11 @@ class TelegramChannel(Channel):
         finally:
             # Graceful shutdown
             try:
-                if self._application.updater.running:
-                    self._tg_loop.run_until_complete(self._application.updater.stop())
-                self._tg_loop.run_until_complete(self._application.stop())
-                self._tg_loop.run_until_complete(self._application.shutdown())
+                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())
             except Exception:
                 logger.exception("Error during Telegram shutdown")
 
@@ -517,12 +781,19 @@ class TelegramChannel(Channel):
             logger.warning("[Telegram] Main loop not running. Cannot publish inbound message.")
 
     async def _on_text(self, update, context) -> None:
-        """Handle regular text messages."""
+        """Handle regular text, photo, and document messages."""
         if not self._check_user(update.effective_user.id):
             return
 
-        text = self._strip_bot_username_from_leading_command(update.message.text.strip(), self._get_bot_username(context))
-        if not text:
+        message = update.message
+        message_text = getattr(message, "text", None)
+        caption = getattr(message, "caption", None)
+        raw_text = message_text if isinstance(message_text, str) else caption if isinstance(caption, str) else ""
+        text = raw_text.strip()
+        if text:
+            text = self._strip_bot_username_from_leading_command(text, self._get_bot_username(context))
+        files = self._extract_inbound_files(message)
+        if not text and not files:
             return
 
         chat_id = str(update.effective_chat.id)
@@ -549,6 +820,7 @@ class TelegramChannel(Channel):
             text=text,
             msg_type=InboundMessageType.CHAT,
             thread_ts=msg_id,
+            files=files,
             metadata={"message_id": msg_id},
         )
         inbound.topic_id = topic_id

backend/tests/test_channel_file_attachments.py

@@ -255,6 +255,33 @@ class TestResolveAttachments:
 
 
 class TestInboundFileIngestion:
+    def test_consumes_inline_channel_bytes_without_exposing_them_downstream(self, tmp_path):
+        from app.channels import manager
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        msg = InboundMessage(
+            channel_name="telegram",
+            chat_id="chat-1",
+            user_id="user-1",
+            text="see attachment",
+            files=[{"type": "file", "filename": "report.pdf", "_content": b"pdf bytes"}],
+        )
+
+        with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir):
+            result = _run(manager._ingest_inbound_files("thread-1", msg))
+
+        assert result == [
+            {
+                "filename": "report.pdf",
+                "size": len(b"pdf bytes"),
+                "path": "/mnt/user-data/uploads/report.pdf",
+                "is_image": False,
+            }
+        ]
+        assert (uploads_dir / "report.pdf").read_bytes() == b"pdf bytes"
+        assert "_content" not in msg.files[0]
+
     def test_rejects_preexisting_symlink_destination(self, tmp_path):
         from app.channels import manager
 

backend/tests/test_channels.py

@@ -6,6 +6,7 @@ import asyncio
 import json
 import logging
 import tempfile
+import threading
 from concurrent.futures import Future
 from pathlib import Path
 from types import SimpleNamespace
@@ -7805,6 +7806,9 @@ class TestTelegramSendRetry:
             def __and__(self, other):
                 return FakeFilter(f"{self.expr}&{other.expr}")
 
+            def __or__(self, other):
+                return FakeFilter(f"{self.expr}|{other.expr}")
+
             def __invert__(self):
                 return FakeFilter(f"~{self.expr}")
 
@@ -7836,7 +7840,12 @@ class TestTelegramSendRetry:
         telegram_ext_mod.ApplicationBuilder = FakeApplicationBuilder
         telegram_ext_mod.CommandHandler = fake_command_handler
         telegram_ext_mod.MessageHandler = fake_message_handler
-        telegram_ext_mod.filters = SimpleNamespace(TEXT=FakeFilter("TEXT"), COMMAND=FakeFilter("COMMAND"))
+        telegram_ext_mod.filters = SimpleNamespace(
+            TEXT=FakeFilter("TEXT"),
+            COMMAND=FakeFilter("COMMAND"),
+            PHOTO=FakeFilter("PHOTO"),
+            Document=SimpleNamespace(ALL=FakeFilter("DOCUMENT")),
+        )
         telegram_mod.ext = telegram_ext_mod
         monkeypatch.setitem(sys.modules, "telegram", telegram_mod)
         monkeypatch.setitem(sys.modules, "telegram.ext", telegram_ext_mod)
@@ -7852,6 +7861,9 @@ class TestTelegramSendRetry:
             def join(self, timeout=None):
                 return None
 
+            def is_alive(self):
+                return False
+
         monkeypatch.setattr("app.channels.telegram.threading.Thread", FakeThread)
 
         async def go():
@@ -7866,6 +7878,7 @@ class TestTelegramSendRetry:
                 assert "start" in registered_commands
                 message_filters = {handler.filter_expr.expr for handler in fake_app.handlers if handler.kind == "message"}
                 assert {"TEXT&COMMAND", "TEXT&~COMMAND"} <= message_filters
+                assert "PHOTO|DOCUMENT" in message_filters
             finally:
                 await ch.stop()
 
@@ -7958,13 +7971,25 @@ class TestFeishuSendRetry:
 # ---------------------------------------------------------------------------
 
 
-def _make_telegram_update(chat_type: str, message_id: int, *, reply_to_message_id: int | None = None, text: str = "hello"):
+def _make_telegram_update(
+    chat_type: str,
+    message_id: int,
+    *,
+    reply_to_message_id: int | None = None,
+    text: str | None = "hello",
+    caption: str | None = None,
+    photo: list[SimpleNamespace] | None = None,
+    document: SimpleNamespace | None = None,
+):
     """Build a minimal mock telegram Update for testing _on_text / _cmd_generic."""
     update = MagicMock()
     update.effective_chat.type = chat_type
     update.effective_chat.id = 100
     update.effective_user.id = 42
     update.message.text = text
+    update.message.caption = caption
+    update.message.photo = photo or []
+    update.message.document = document
     update.message.message_id = message_id
     if reply_to_message_id is not None:
         reply_msg = MagicMock()
@@ -7975,8 +8000,8 @@ def _make_telegram_update(chat_type: str, message_id: int, *, reply_to_message_i
     return update
 
 
-class TestTelegramPrivateChatThread:
-    """Verify that private chats use topic_id=None (single thread per chat)."""
+class TestTelegramInboundMessages:
+    """Verify Telegram inbound normalization and conversation thread context."""
 
     def test_private_chat_no_reply_uses_none_topic(self):
         from app.channels.telegram import TelegramChannel
@@ -7984,7 +8009,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("private", message_id=10)
             await ch._on_text(update, None)
@@ -7994,13 +8019,519 @@ class TestTelegramPrivateChatThread:
 
         _run(go())
 
+    def test_photo_caption_uses_largest_size_and_preserves_thread_context(self):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            ch._main_loop = asyncio.get_running_loop()
+            small = SimpleNamespace(file_id="photo-small", file_unique_id="unique-small", file_size=10, width=90, height=90)
+            large = SimpleNamespace(file_id="photo-large", file_unique_id="unique-large", file_size=200, width=800, height=600)
+            update = _make_telegram_update(
+                "group",
+                message_id=40,
+                reply_to_message_id=15,
+                text=None,
+                caption="  Describe this image  ",
+                # Do not rely on Telegram returning PhotoSize objects in order.
+                photo=[large, small],
+            )
+
+            await ch._on_text(update, None)
+
+            msg = await asyncio.wait_for(bus.get_inbound(), timeout=2)
+            assert msg.text == "Describe this image"
+            assert msg.topic_id == "15"
+            assert msg.thread_ts == "40"
+            assert len(msg.files) == 1
+            assert msg.files[0] == {
+                "type": "image",
+                "file_id": "photo-large",
+                "file_unique_id": "unique-large",
+                "filename": "telegram-photo-40.jpg",
+                "mime_type": "image/jpeg",
+                "size": 200,
+            }
+
+        _run(go())
+
+    def test_document_without_caption_still_publishes_an_inbound_turn(self):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            ch._main_loop = asyncio.get_running_loop()
+            document = SimpleNamespace(
+                file_id="document-id",
+                file_unique_id="document-unique",
+                file_name="../report.pdf",
+                mime_type="application/pdf",
+                file_size=1234,
+            )
+            update = _make_telegram_update("private", message_id=41, text=None, document=document)
+
+            await ch._on_text(update, None)
+
+            msg = await asyncio.wait_for(bus.get_inbound(), timeout=2)
+            assert msg.text == ""
+            assert msg.topic_id is None
+            assert msg.files == [
+                {
+                    "type": "file",
+                    "file_id": "document-id",
+                    "file_unique_id": "document-unique",
+                    "filename": "report.pdf",
+                    "mime_type": "application/pdf",
+                    "size": 1234,
+                }
+            ]
+
+        _run(go())
+
+    def test_photo_without_caption_still_publishes_an_inbound_turn(self):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            ch._main_loop = asyncio.get_running_loop()
+            photo = SimpleNamespace(file_id="photo-id", file_unique_id="photo-unique", file_size=25)
+            update = _make_telegram_update("private", message_id=42, text=None, photo=[photo])
+
+            await ch._on_text(update, None)
+
+            msg = await asyncio.wait_for(bus.get_inbound(), timeout=2)
+            assert msg.text == ""
+            assert msg.files[0]["filename"] == "telegram-photo-42.jpg"
+            assert msg.files[0]["file_id"] == "photo-id"
+
+        _run(go())
+
+    def test_document_caption_is_preserved_and_missing_filename_gets_safe_fallback(self):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            ch._main_loop = asyncio.get_running_loop()
+            document = SimpleNamespace(
+                file_id="document-id",
+                file_unique_id="document-unique",
+                file_name=None,
+                mime_type="text/plain",
+                file_size=12,
+            )
+            update = _make_telegram_update("private", message_id=43, text=None, caption="  Review this  ", document=document)
+
+            await ch._on_text(update, None)
+
+            msg = await asyncio.wait_for(bus.get_inbound(), timeout=2)
+            assert msg.text == "Review this"
+            assert msg.files[0]["filename"] == "telegram-document-43.bin"
+
+        _run(go())
+
+    def test_document_staging_filename_gets_visible_safe_fallback(self):
+        from app.channels.telegram import TelegramChannel
+
+        document = SimpleNamespace(
+            file_id="document-id",
+            file_unique_id="document-unique",
+            file_name=".upload-hidden.part",
+            mime_type="application/pdf",
+            file_size=12,
+        )
+        update = _make_telegram_update("private", message_id=44, text=None, document=document)
+
+        files = TelegramChannel._extract_inbound_files(update.message)
+
+        assert files[0]["filename"] == "telegram-document-44.bin"
+
+    def test_receive_file_downloads_bytes_without_retaining_telegram_file_id(self):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            downloaded = bytearray(b"data")
+            telegram_file = SimpleNamespace(file_size=4, download_as_bytearray=AsyncMock(return_value=downloaded))
+            bot = SimpleNamespace(get_file=AsyncMock(return_value=telegram_file))
+            ch._application = SimpleNamespace(bot=bot)
+            msg = InboundMessage(
+                channel_name="telegram",
+                chat_id="100",
+                user_id="42",
+                text="caption",
+                files=[
+                    {
+                        "type": "file",
+                        "file_id": "document-id",
+                        "file_unique_id": "document-unique",
+                        "filename": "report.pdf",
+                        "mime_type": "application/pdf",
+                        "size": 4,
+                    }
+                ],
+            )
+
+            result = await ch.receive_file(msg, "thread-1")
+
+            bot.get_file.assert_awaited_once_with("document-id")
+            telegram_file.download_as_bytearray.assert_awaited_once_with()
+            assert result.text == "caption"
+            assert result.files[0]["_content"] is downloaded
+            assert result.files == [
+                {
+                    "type": "file",
+                    "filename": "report.pdf",
+                    "mime_type": "application/pdf",
+                    "size": 4,
+                    "_content": b"data",
+                }
+            ]
+
+        _run(go())
+
+    def test_receive_file_marshals_ptb_download_to_telegram_event_loop(self):
+        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()
+
+            def run_telegram_loop():
+                asyncio.set_event_loop(telegram_loop)
+                telegram_loop.call_soon(loop_started.set_result, None)
+                telegram_loop.run_forever()
+
+            loop_thread = threading.Thread(target=run_telegram_loop, daemon=True)
+            loop_thread.start()
+            try:
+                loop_started.result(timeout=2)
+
+                class LoopBoundFile:
+                    file_size = 4
+
+                    async def download_as_bytearray(self):
+                        assert asyncio.get_running_loop() is telegram_loop
+                        return bytearray(b"data")
+
+                class LoopBoundBot:
+                    async def get_file(self, file_id):
+                        assert asyncio.get_running_loop() is telegram_loop
+                        assert file_id == "document-id"
+                        return LoopBoundFile()
+
+                ch._tg_loop = telegram_loop
+                ch._application = SimpleNamespace(bot=LoopBoundBot())
+                msg = InboundMessage(
+                    channel_name="telegram",
+                    chat_id="100",
+                    user_id="42",
+                    text="caption",
+                    files=[{"type": "file", "file_id": "document-id", "filename": "report.pdf", "size": 4}],
+                )
+                result = await ch.receive_file(msg, "thread-1")
+            finally:
+                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()
+
+            assert result.files[0]["_content"] == b"data"
+
+        _run(go())
+
+    def test_stop_cancels_and_drains_an_inflight_receive_download(self):
+        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()
+            download_started = threading.Event()
+            download_cancelled = threading.Event()
+
+            def run_telegram_loop():
+                asyncio.set_event_loop(telegram_loop)
+                telegram_loop.call_soon(loop_started.set_result, None)
+                telegram_loop.run_forever()
+
+            class SlowTelegramFile:
+                file_size = 4
+
+                async def download_as_bytearray(self):
+                    download_started.set()
+                    try:
+                        await asyncio.Event().wait()
+                    finally:
+                        download_cancelled.set()
+
+            class LoopBoundBot:
+                async def get_file(self, file_id):
+                    assert asyncio.get_running_loop() is telegram_loop
+                    assert file_id == "document-id"
+                    return SlowTelegramFile()
+
+            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=LoopBoundBot())
+                msg = InboundMessage(
+                    channel_name="telegram",
+                    chat_id="100",
+                    user_id="42",
+                    text="caption",
+                    files=[{"type": "file", "file_id": "document-id", "filename": "report.pdf", "size": 4}],
+                )
+                receive_task = asyncio.create_task(ch.receive_file(msg, "thread-1"))
+                assert await asyncio.to_thread(download_started.wait, 2)
+
+                await ch.stop()
+
+                with pytest.raises(asyncio.CancelledError):
+                    await receive_task
+                assert download_cancelled.is_set()
+                assert not ch._tg_bridge_tasks
+            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_cancelled_stop_still_stops_thread_and_clears_state(self):
+        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()
+            drain_started = threading.Event()
+
+            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()
+
+            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._cancel_telegram_bridge_tasks = slow_drain
+
+                stop_task = asyncio.create_task(ch.stop())
+                assert await asyncio.to_thread(drain_started.wait, 2)
+                stop_task.cancel()
+
+                with pytest.raises(asyncio.CancelledError):
+                    await stop_task
+
+                assert not loop_thread.is_alive()
+                assert ch._thread is None
+                assert ch._application is None
+            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_receive_file_does_not_reuse_ptb_client_after_telegram_loop_stops(self):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            stopped_loop = asyncio.new_event_loop()
+            bot = SimpleNamespace(get_file=AsyncMock())
+            ch._tg_loop = stopped_loop
+            ch._application = SimpleNamespace(bot=bot)
+            msg = InboundMessage(
+                channel_name="telegram",
+                chat_id="100",
+                user_id="42",
+                text="caption",
+                files=[{"type": "file", "file_id": "document-id", "filename": "report.pdf", "size": 4}],
+            )
+
+            try:
+                result = await ch.receive_file(msg, "thread-1")
+            finally:
+                stopped_loop.close()
+
+            bot.get_file.assert_not_awaited()
+            assert result.files == []
+            assert result.text.startswith("caption")
+            assert "report.pdf" in result.text
+
+        _run(go())
+
+    def test_receive_file_rejects_declared_oversize_before_download(self):
+        from app.channels.telegram import TELEGRAM_MAX_INBOUND_FILE_BYTES, TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            bot = SimpleNamespace(get_file=AsyncMock())
+            ch._application = SimpleNamespace(bot=bot)
+            msg = InboundMessage(
+                channel_name="telegram",
+                chat_id="100",
+                user_id="42",
+                text="",
+                files=[
+                    {
+                        "type": "file",
+                        "file_id": "too-large",
+                        "filename": "archive.zip",
+                        "mime_type": "application/zip",
+                        "size": TELEGRAM_MAX_INBOUND_FILE_BYTES + 1,
+                    }
+                ],
+            )
+
+            result = await ch.receive_file(msg, "thread-1")
+
+            bot.get_file.assert_not_called()
+            assert result.files == []
+            assert "archive.zip" in result.text
+            assert "20 MB" in result.text
+
+        _run(go())
+
+    def test_receive_file_rejects_download_larger_than_reported(self, monkeypatch):
+        from app.channels import telegram
+
+        async def go():
+            monkeypatch.setattr(telegram, "TELEGRAM_MAX_INBOUND_FILE_BYTES", 3)
+            bus = MessageBus()
+            ch = telegram.TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            telegram_file = SimpleNamespace(file_size=2, download_as_bytearray=AsyncMock(return_value=bytearray(b"four")))
+            ch._application = SimpleNamespace(bot=SimpleNamespace(get_file=AsyncMock(return_value=telegram_file)))
+            msg = InboundMessage(
+                channel_name="telegram",
+                chat_id="100",
+                user_id="42",
+                text="caption",
+                files=[{"type": "image", "file_id": "photo-id", "filename": "photo.jpg", "mime_type": "image/jpeg", "size": 2}],
+            )
+
+            result = await ch.receive_file(msg, "thread-1")
+
+            assert result.files == []
+            assert result.text.startswith("caption")
+            assert "photo.jpg" in result.text
+
+        _run(go())
+
+    def test_receive_file_rejects_resolved_oversize_before_downloading(self, monkeypatch):
+        from app.channels import telegram
+
+        async def go():
+            monkeypatch.setattr(telegram, "TELEGRAM_MAX_INBOUND_FILE_BYTES", 3)
+            bus = MessageBus()
+            ch = telegram.TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            download = AsyncMock(return_value=bytearray(b"four"))
+            telegram_file = SimpleNamespace(file_size=4, download_as_bytearray=download)
+            ch._application = SimpleNamespace(bot=SimpleNamespace(get_file=AsyncMock(return_value=telegram_file)))
+            msg = InboundMessage(
+                channel_name="telegram",
+                chat_id="100",
+                user_id="42",
+                text="",
+                files=[{"type": "file", "file_id": "file-id", "filename": "large.bin", "size": 2}],
+            )
+
+            result = await ch.receive_file(msg, "thread-1")
+
+            download.assert_not_awaited()
+            assert result.files == []
+            assert "large.bin" in result.text
+
+        _run(go())
+
+    def test_receive_file_accepts_exact_download_limit(self, monkeypatch):
+        from app.channels import telegram
+
+        async def go():
+            monkeypatch.setattr(telegram, "TELEGRAM_MAX_INBOUND_FILE_BYTES", 4)
+            bus = MessageBus()
+            ch = telegram.TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            telegram_file = SimpleNamespace(file_size=4, download_as_bytearray=AsyncMock(return_value=bytearray(b"four")))
+            ch._application = SimpleNamespace(bot=SimpleNamespace(get_file=AsyncMock(return_value=telegram_file)))
+            msg = InboundMessage(
+                channel_name="telegram",
+                chat_id="100",
+                user_id="42",
+                text="",
+                files=[{"type": "file", "file_id": "file-id", "filename": "exact.bin", "size": 4}],
+            )
+
+            result = await ch.receive_file(msg, "thread-1")
+
+            assert result.files[0]["_content"] == b"four"
+            assert result.files[0]["size"] == 4
+
+        _run(go())
+
+    def test_receive_file_download_failure_keeps_caption_and_drops_attachment(self, caplog):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+            ch._application = SimpleNamespace(bot=SimpleNamespace(get_file=AsyncMock(side_effect=RuntimeError("GET https://api.telegram.org/bottest-token/getFile failed"))))
+            msg = InboundMessage(
+                channel_name="telegram",
+                chat_id="100",
+                user_id="42",
+                text="Summarize this",
+                files=[{"type": "file", "file_id": "document-id", "filename": "report.pdf", "mime_type": "application/pdf", "size": 10}],
+            )
+
+            with caplog.at_level(logging.ERROR):
+                result = await ch.receive_file(msg, "thread-1")
+
+            assert result.files == []
+            assert result.text.startswith("Summarize this")
+            assert "report.pdf" in result.text
+            assert "test-token" not in caplog.text
+
+        _run(go())
+
     def test_private_chat_slash_skill_text_routes_as_chat(self):
         from app.channels.telegram import TelegramChannel
 
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("private", message_id=12, text="/data-analysis analyze uploads/foo.csv")
             await ch._on_text(update, None)
@@ -8018,7 +8549,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update(
                 "group",
@@ -8041,7 +8572,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("private", message_id=11, reply_to_message_id=5)
             await ch._on_text(update, None)
@@ -8057,7 +8588,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("group", message_id=20)
             await ch._on_text(update, None)
@@ -8073,7 +8604,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("group", message_id=21, reply_to_message_id=15)
             await ch._on_text(update, None)
@@ -8089,7 +8620,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("supergroup", message_id=25)
             await ch._on_text(update, None)
@@ -8105,7 +8636,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("private", message_id=30, text="/new")
             await ch._cmd_generic(update, None)
@@ -8122,7 +8653,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("group", message_id=31, text="/status")
             await ch._cmd_generic(update, None)
@@ -8139,7 +8670,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("group", message_id=32, reply_to_message_id=20, text="/status")
             await ch._cmd_generic(update, None)
@@ -8156,7 +8687,7 @@ class TestTelegramPrivateChatThread:
         async def go():
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             update = _make_telegram_update("group", message_id=33, text="/status@DeerFlowBot")
             context = SimpleNamespace(bot=SimpleNamespace(username="DeerFlowBot"))
@@ -8180,7 +8711,7 @@ class TestTelegramProcessingOrder:
             bus = MessageBus()
             ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
 
-            ch._main_loop = asyncio.get_event_loop()
+            ch._main_loop = asyncio.get_running_loop()
 
             order = []