diff --git a/README.md b/README.md index 27e3363c3..732977912 100644 --- a/README.md +++ b/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** diff --git a/README_zh.md b/README_zh.md index af030c5a3..ddd57ca88 100644 --- a/README_zh.md +++ b/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 配置** diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 5d3044df0..1bb3a785a 100644 --- a/backend/AGENTS.md +++ b/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=; ` (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 diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index a50c99c68..30a9a18fe 100644 --- a/backend/app/channels/manager.py +++ b/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( diff --git a/backend/app/channels/message_bus.py b/backend/app/channels/message_bus.py index e04f51006..25fc3f1d8 100644 --- a/backend/app/channels/message_bus.py +++ b/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" # --------------------------------------------------------------------------- diff --git a/backend/app/channels/telegram.py b/backend/app/channels/telegram.py index 14ee28a4f..eca123f26 100644 --- a/backend/app/channels/telegram.py +++ b/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 diff --git a/backend/tests/test_channel_file_attachments.py b/backend/tests/test_channel_file_attachments.py index aa2e9b004..0b8e27496 100644 --- a/backend/tests/test_channel_file_attachments.py +++ b/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 diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index b3352a370..4058403d2 100644 --- a/backend/tests/test_channels.py +++ b/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 = []