From bf4aa82f09939ba364f43385ddefa7b3619099d7 Mon Sep 17 00:00:00 2001 From: ZephyrWang <152375907+Wang-Zephyr@users.noreply.github.com> Date: Wed, 11 Nov 2026 08:03:40 +0800 Subject: [PATCH] fix(telegram): deliver rejected photos as documents (#8056) * fix(telegram): deliver rejected photos as documents * fix(telegram): log rejected-photo document fallback --- README.md | 2 + backend/app/channels/AGENTS.md | 2 +- backend/app/channels/telegram.py | 13 +- .../test_channel_outbound_files.py | 11 +- backend/tests/test_telegram_photo_fallback.py | 260 ++++++++++++++++++ 5 files changed, 284 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_telegram_photo_fallback.py diff --git a/README.md b/README.md index adf1ff999c..e8a097d218 100644 --- a/README.md +++ b/README.md @@ -1311,6 +1311,8 @@ DeerFlow supports receiving tasks from messaging apps. Channels auto-start when Generated-file replies explain when a channel cannot send attachments. If some or all uploads fail, DeerFlow sends one combined notice naming the unsent files when the platform still accepts replies. Successful uploads add no duplicate confirmation. Files remain in the DeerFlow conversation; an expired reply window is logged without switching to proactive messaging. +Telegram retries photos rejected for invalid dimensions or image processing once as documents, preserving the original file and reply target. + Cancelling a channel restart discards its pending configuration reload, preserving newer runtime settings applied afterward. Invalid UTF-8 in `channels/store.json` follows the same recovery policy as malformed JSON: diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index baa68aae92..22c21a1470 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -36,7 +36,7 @@ Shared attachment feedback uses public `attachment_delivery_text`, preserves rou **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. + - Telegram inbound: largest photo/document metadata; keep captions, cap downloads at 20,000,000 bytes before/after, hide file URLs. Private bytes cross only via `message_bus.INBOUND_FILE_CONTENT_KEY`, consumed before persisting metadata. Outbound: only photo BadRequest PHOTO_INVALID_DIMENSIONS / IMAGE_PROCESS_FAILED falls back once to document; same bytes/name/reply. Other errors/cancellation unchanged; track success only. - Feishu/Lark inbound image/file downloads read at most 20,000,001 bytes and reject anything above 20,000,000 bytes before persistence or sandbox sync. Oversize, unsafe-path, and path-resolution failures rewrite only that attachment placeholder to `Failed to obtain the [type]` so later attachments in the same message can still load. - WeChat iLink inbound image/file downloads stream with an in-flight cap derived from `channels.wechat.max_inbound_image_bytes` / `max_inbound_file_bytes` (20 MB / 50 MB defaults) that aborts before the full read, and the payload-supplied `media.full_url` must pass a scheme + dot-boundary host-suffix allowlist (`channels.wechat.allowed_media_hosts`, defaulting to the `qq.com` family plus the configured `cdn_base_url` host) before anything is fetched. The limits are PLAINTEXT limits while the stream measures CIPHERTEXT, so the in-flight cap is the PKCS#7-padded size of exactly-limit plaintext (`WechatChannel._stream_cap_for`); the exact post-decryption check stays as the authority. There is deliberately **no URL re-fetch fallback**: `_read_wechat_inbound_file` (manager) only reads the locally staged `path` — the channel always stages one before publishing — so nothing fetches a WeChat URL outside the channel-side gate. WeCom media URLs (frame-supplied `url`, no per-channel size config) get the same style of host gate — `qq.com` suffixes plus the COS shape `ww-aibot-img-.cos..myqcloud.com` WeCom serves signed links from, where the numeric suffix must be one of the verified WeCom-owned Tencent Cloud APPIDs (`_WECOM_MEDIA_COS_APPIDS`, currently only the `1258476243` observed in Tencent's published callback examples) because bucket names are user-chosen and any Tencent Cloud account can register a lookalike `ww-aibot-img-*` bucket; other accounts go through operator suffixes from `channels.wecom.allowed_media_hosts` resolved from the live channel at read time — and the manager-level `MAX_INBOUND_URL_FILE_BYTES` (50 MB) streaming cap. Both URL downloads and `_download_cdn_bytes` request `Accept-Encoding: identity`, refuse a response with any residual `Content-Encoding` before reading, and iterate `aiter_raw()`: httpx's transparent decoding would allocate the full decompressed body before the byte cap sees a chunk. No URL-based inbound-media log line may contain any part of the URL (signed links carry credentials in path and query): failure labels use the attachment filename or host only (`manager._inbound_file_label`), and reader failures log a sanitized exception summary — class name plus HTTP status (`manager._reader_error_summary`, `wechat._media_download_error_summary`), never the URL-bearing message or traceback. The rule extends to the HTTP client libraries' own records: httpx emits `HTTP Request: GET ` at INFO via the bare `httpx` logger, and urllib3 — enumerated against the installed source; every remaining emitter logs host:port only (connection establishment/reset) or one absolute URL (the header-parse warning) that the generic pattern rewrites — renders a request target in five shapes, each needing a dedicated redaction pattern because the generic absolute-URL pass cannot see a target with no scheme: `Redirecting -> ` at INFO via `urllib3.poolmanager` and DEBUG via `urllib3.connectionpool` (either slot may be an origin-form relative Location, which pool/connection callers pass or the Location header carries; the shape keeps the `^Redirecting ` prefix anchor because `-> /path` arrows also appear in non-URL logs — sandbox mount mappings — and must pass through untouched, while both slots consume through end-of-line: the raw Location header may carry interior spaces, and the recursive urlopen frame can pass it on as the next request target. A slot is kept verbatim only when the generic pass consumes it whole (`logging_config._url_pass_consumes_slot` asks that pass itself, so the kept set cannot drift from its `host`/`rest` stop conditions) — every other RFC 3986 relative-reference form (slash-less paths, query-only, fragments, network-path), non-hierarchical scheme, space-carrying slot, quote-at-boundary slot (`https://h/a')b?sig=…`, where `rest` stops at the closing mark), and empty-host absolute (`https:///path?sig=…`, which `host` never matches) collapses, so neither `download?sign=…` nor `other page?sig=…` leaks its signed query), the per-request `scheme://host:port "METHOD /path?query HTTP/x.x"` DEBUG line, `Retry: ` DEBUG and `Retrying (…) after connection broken by '': ` **WARNING** — above the INFO root, so reachable at the default log level — via `urllib3.connectionpool`, and `Incremented Retry for (url='')` DEBUG via `urllib3.util.retry` (origin-form on the request path, absolute on the redirect path). The generic pass's `rest` treats a quote as a closing mark only at a boundary — followed by whitespace, a closing parenthesis, or end of string — so urllib3's `Incremented Retry for (url='…')` keeps its `')` scaffolding while a quote embedded in a URL stays consumed and redacted. Logger filters only see records emitted through that exact logger — child loggers neither inherit them nor trigger them on propagation — so `UrlRedactionFilter` is attached to the `httpx` logger AND to every root handler by `configure_logging` (handler-level filters see propagated records), which also keeps Telegram's token-bearing Bot API paths (`api.telegram.org/bot/...`, emitted by python-telegram-bot's httpx client) out of the logs. 2. `ChannelManager._dispatch_loop()` consumes from queue diff --git a/backend/app/channels/telegram.py b/backend/app/channels/telegram.py index 7d67806147..59da84a67a 100644 --- a/backend/app/channels/telegram.py +++ b/backend/app/channels/telegram.py @@ -598,13 +598,24 @@ class TelegramChannel(Channel): try: input_file = await asyncio.to_thread(_load_telegram_input_file, attachment.actual_path, attachment.filename) + from telegram.error import BadRequest + if attachment.is_image and attachment.size <= 10 * 1024 * 1024: kwargs: dict[str, Any] = {"chat_id": chat_id, "photo": input_file} if reply_to: kwargs["reply_to_message_id"] = reply_to if chat_id > 0: kwargs["allow_sending_without_reply"] = True - sent = await bot.send_photo(**kwargs) + try: + sent = await bot.send_photo(**kwargs) + except BadRequest as exc: + if str(exc).casefold() not in {"photo_invalid_dimensions", "image_process_failed"}: + raise + logger.info("[Telegram] photo rejected (%s); delivering as document: %s", exc, attachment.filename) + # A rejected photo can still be delivered unchanged as a file. + # Keep the reply target and reuse the already loaded bytes. + kwargs["document"] = kwargs.pop("photo") + sent = await bot.send_document(**kwargs) else: kwargs = {"chat_id": chat_id, "document": input_file} if reply_to: diff --git a/backend/tests/blocking_io/test_channel_outbound_files.py b/backend/tests/blocking_io/test_channel_outbound_files.py index ffdc19cc07..b86dd663d1 100644 --- a/backend/tests/blocking_io/test_channel_outbound_files.py +++ b/backend/tests/blocking_io/test_channel_outbound_files.py @@ -17,6 +17,7 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest +from telegram.error import BadRequest from app.channels.feishu import FeishuChannel from app.channels.message_bus import MessageBus, OutboundMessage, ResolvedAttachment @@ -133,22 +134,27 @@ async def test_feishu_outbound_uploads_do_not_block_event_loop(tmp_path: Path, m assert file_worker_thread != event_loop_thread -@pytest.mark.parametrize("is_image", [False, True]) -async def test_telegram_outbound_upload_does_not_block_event_loop(tmp_path: Path, is_image: bool) -> None: +@pytest.mark.parametrize("is_image,rejected_photo", [(False, False), (True, False), (True, True)]) +async def test_telegram_outbound_upload_does_not_block_event_loop(tmp_path: Path, is_image: bool, rejected_photo: bool) -> None: path = tmp_path / ("chart.png" if is_image else "report.bin") payload = b"telegram-payload" await asyncio.to_thread(path.write_bytes, payload) sent_file = None + calls = [] class _Bot: async def send_document(self, **kwargs): nonlocal sent_file + calls.append("document") sent_file = kwargs["document"] return SimpleNamespace(message_id=7) async def send_photo(self, **kwargs): nonlocal sent_file + calls.append("photo") + if rejected_photo: + raise BadRequest("PHOTO_INVALID_DIMENSIONS") sent_file = kwargs["photo"] if hasattr(sent_file, "read"): sent_file.read() @@ -160,6 +166,7 @@ async def test_telegram_outbound_upload_does_not_block_event_loop(tmp_path: Path assert await channel.send_file(_outbound("telegram"), _attachment(path, is_image=is_image)) assert sent_file.input_file_content == payload assert sent_file.filename == path.name + assert calls == (["photo", "document"] if rejected_photo else ["photo"] if is_image else ["document"]) async def test_wecom_outbound_upload_does_not_block_event_loop(tmp_path: Path) -> None: diff --git a/backend/tests/test_telegram_photo_fallback.py b/backend/tests/test_telegram_photo_fallback.py new file mode 100644 index 0000000000..5f28014ea5 --- /dev/null +++ b/backend/tests/test_telegram_photo_fallback.py @@ -0,0 +1,260 @@ +"""Telegram photo rejection recovery through real SDK request serialization. + +The offline transport models documented photo geometry limits. It is not a +live Telegram reproduction, and these tests never contact the platform. +""" + +from __future__ import annotations + +import asyncio +import json +import struct +import zlib +from dataclasses import replace +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import pytest_asyncio +from telegram import Bot +from telegram.error import NetworkError, TimedOut +from telegram.request import BaseRequest + +from app.channels.message_bus import MessageBus, OutboundMessage, ResolvedAttachment +from app.channels.telegram import TelegramChannel + + +def _png(width: int, height: int) -> bytes: + def chunk(kind: bytes, data: bytes) -> bytes: + return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data)) + + rows = (b"\0" + bytes((width + 7) // 8)) * height + header = struct.pack(">IIBBBBB", width, height, 1, 0, 0, 0, 0) + return b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", header) + chunk(b"IDAT", zlib.compress(rows)) + chunk(b"IEND", b"") + + +class _PhotoRequest(BaseRequest): + def __init__(self): + self.calls = [] + self.errors = {} + self.blocked_endpoint = None + self.started = asyncio.Event() + self.release = asyncio.Event() + + @property + def read_timeout(self): + return 5 + + async def initialize(self): + pass + + async def shutdown(self): + pass + + async def do_request(self, url, method, request_data=None, **kwargs): + endpoint = url.rsplit("/", 1)[-1] + params = request_data.parameters if request_data else {} + if endpoint == "getMe": + result = {"id": 1, "is_bot": True, "first_name": "Offline", "username": "offline_bot"} + else: + uploads = request_data.multipart_data + self.calls.append((endpoint, params, uploads)) + if endpoint == self.blocked_endpoint: + self.started.set() + await self.release.wait() + error = self.errors.get(endpoint) + if isinstance(error, Exception): + raise error + if error: + status, description = error + return status, json.dumps({"ok": False, "error_code": status, "description": description, "parameters": {"retry_after": 30} if status == 429 else {}}).encode() + if endpoint == "sendPhoto": + data = next(iter(uploads.values()))[1] + width, height = struct.unpack(">II", data[16:24]) + if width + height > 10000 or max(width, height) > 20 * min(width, height): + return 400, b'{"ok": false, "error_code": 400, "description": "Bad Request: PHOTO_INVALID_DIMENSIONS"}' + result = {"message_id": 100 + len(self.calls), "date": 0, "chat": {"id": params["chat_id"], "type": "private" if params["chat_id"] > 0 else "supergroup"}} + return 200, json.dumps({"ok": True, "result": result}).encode() + + +@pytest_asyncio.fixture +async def channel(): + request = _PhotoRequest() + async with Bot("123:offline-test-only", request=request, get_updates_request=request) as bot: + instance = TelegramChannel(MessageBus(), {"bot_token": "123:offline-test-only"}) + instance._application = SimpleNamespace(bot=bot) + instance._remember_reply_thread = AsyncMock(return_value=True) + yield instance, request + + +def _attachment(tmp_path, width=64, height=64): + path = tmp_path / "original chart.png" + data = _png(width, height) + path.write_bytes(data) + return ResolvedAttachment("/mnt/user-data/outputs/" + path.name, path, path.name, "image/png", len(data), True) + + +def _message(chat_id="12345"): + return OutboundMessage(channel_name="telegram", chat_id=chat_id, thread_id="thread", text="Complete answer") + + +def _endpoints(request): + return [call[0] for call in request.calls] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("chat_id", ["12345", "-12345"]) +@pytest.mark.parametrize("dimensions", [(100, 3000), (6000, 5000)]) +async def test_rejected_photo_delivers_original_document_and_preserves_reply(channel, tmp_path, chat_id, dimensions): + instance, request = channel + attachment = _attachment(tmp_path, *dimensions) + original = attachment.actual_path.read_bytes() + instance._last_bot_message[chat_id] = 42 + message = _message(chat_id) + + assert await instance.send_file(message, attachment) + + assert _endpoints(request) == ["sendPhoto", "sendDocument"] + photo, document = request.calls + assert photo[2]["photo"] == document[2]["document"] + assert document[2]["document"][:2] == (attachment.filename, original) + assert photo[1]["chat_id"] == document[1]["chat_id"] == int(chat_id) + assert photo[1]["reply_parameters"] == document[1]["reply_parameters"] + reply = document[1]["reply_parameters"] + assert reply["message_id"] == 42 + assert bool(reply.get("allow_sending_without_reply")) is (int(chat_id) > 0) + assert instance._last_bot_message[chat_id] == 102 + instance._remember_reply_thread.assert_awaited_once_with(message, 102) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("description", ["Bad Request: IMAGE_PROCESS_FAILED", "Bad Request: image_process_failed", "Bad Request: photo_invalid_dimensions"]) +async def test_explicit_media_rejections_fall_back_once(channel, tmp_path, description): + instance, request = channel + request.errors["sendPhoto"] = (400, description) + assert await instance.send_file(_message(), _attachment(tmp_path)) + assert _endpoints(request) == ["sendPhoto", "sendDocument"] + assert "reply_parameters" not in request.calls[-1][1] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_image,size,endpoint", [(True, 10 * 1024 * 1024, "sendPhoto"), (True, 10 * 1024 * 1024 + 1, "sendDocument"), (False, 72, "sendDocument"), (False, 50 * 1024 * 1024, "sendDocument")]) +async def test_existing_photo_and_document_selection(channel, tmp_path, is_image, size, endpoint): + instance, request = channel + # Metadata sizes exercise the existing routing boundaries without large files. + attachment = replace(_attachment(tmp_path), is_image=is_image, size=size) + assert await instance.send_file(_message(), attachment) + assert _endpoints(request) == [endpoint] + instance._remember_reply_thread.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + (400, "Bad Request: message to be replied not found"), + (400, "Bad Request: unknown photo failure"), + (400, "Bad Request: PHOTO_INVALID_DIMENSIONS extra detail"), + (403, "Forbidden: bot was blocked by the user"), + (403, "PHOTO_INVALID_DIMENSIONS"), + (429, "Too Many Requests"), + NetworkError("PHOTO_INVALID_DIMENSIONS"), + TimedOut(), + ], +) +async def test_unrelated_or_uncertain_photo_errors_do_not_resend(channel, tmp_path, error): + instance, request = channel + request.errors["sendPhoto"] = error + instance._last_bot_message["12345"] = 42 + assert not await instance.send_file(_message(), _attachment(tmp_path)) + assert _endpoints(request) == ["sendPhoto"] + assert instance._last_bot_message["12345"] == 42 + instance._remember_reply_thread.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error", [(400, "Bad Request: PHOTO_INVALID_DIMENSIONS"), (403, "Forbidden: bot was blocked by the user"), (429, "Too Many Requests"), NetworkError("offline connection failed")]) +async def test_document_fallback_failure_is_not_retried_or_registered(channel, tmp_path, error): + instance, request = channel + request.errors["sendDocument"] = error + instance._last_bot_message["12345"] = 42 + assert not await instance.send_file(_message(), _attachment(tmp_path, 100, 3000)) + assert _endpoints(request) == ["sendPhoto", "sendDocument"] + assert instance._last_bot_message["12345"] == 42 + instance._remember_reply_thread.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("endpoint", ["sendPhoto", "sendDocument"]) +async def test_cancellation_during_send_propagates_without_registration(channel, tmp_path, endpoint): + instance, request = channel + request.blocked_endpoint = endpoint + attachment = _attachment(tmp_path, 100, 3000) + instance._last_bot_message["12345"] = 42 + task = asyncio.create_task(instance.send_file(_message(), attachment)) + try: + await asyncio.wait_for(request.started.wait(), timeout=5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + assert _endpoints(request) == (["sendPhoto"] if endpoint == "sendPhoto" else ["sendPhoto", "sendDocument"]) + assert instance._last_bot_message["12345"] == 42 + instance._remember_reply_thread.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_missing_file_and_oversized_attachment_do_not_send(channel, tmp_path): + instance, request = channel + attachment = _attachment(tmp_path) + assert not await instance.send_file(_message(), replace(attachment, size=50 * 1024 * 1024 + 1)) + attachment.actual_path.unlink() + assert not await instance.send_file(_message(), attachment) + assert not request.calls + instance._remember_reply_thread.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_mixed_attachment_batch_continues_without_resending_success(channel, tmp_path): + instance, request = channel + wide = _attachment(tmp_path, 100, 3000) + ordinary_path = tmp_path / "notes.txt" + ordinary_path.write_bytes(b"notes") + ordinary = ResolvedAttachment("/mnt/user-data/outputs/notes.txt", ordinary_path, "notes.txt", "text/plain", 5, False) + message = replace(_message(), attachments=[wide, ordinary]) + assert await instance._send_attachments(message) == [] + assert _endpoints(request) == ["sendPhoto", "sendDocument", "sendDocument"] + assert request.calls[-1][2]["document"][:2] == ("notes.txt", b"notes") + assert request.calls[-1][1]["reply_parameters"]["message_id"] == 102 + assert instance._remember_reply_thread.await_count == 2 + + +@pytest.mark.asyncio +async def test_fallback_binds_delivered_group_document_to_original_thread(channel, tmp_path): + instance, request = channel + instance._remember_reply_thread = TelegramChannel._remember_reply_thread.__get__(instance) + store = SimpleNamespace(claim_thread_id=AsyncMock()) + instance.config["channel_store"] = store + instance._last_bot_message["-12345"] = 42 + message = replace(_message("-12345"), thread_ts="17") + + assert await instance.send_file(message, _attachment(tmp_path, 100, 3000)) + + assert _endpoints(request) == ["sendPhoto", "sendDocument"] + assert store.claim_thread_id.await_count == 2 + claims = store.claim_thread_id.await_args_list + assert {entry.args for entry in claims} == {("telegram", "-12345", "thread")} + assert {entry.kwargs["topic_id"] for entry in claims} == {"17", "102"} + + +@pytest.mark.asyncio +async def test_unavailable_bot_and_invalid_chat_do_not_send(channel, tmp_path): + instance, request = channel + attachment = _attachment(tmp_path) + assert not await instance.send_file(_message("not-a-chat-id"), attachment) + instance._application = None + assert not await instance.send_file(_message(), attachment) + assert not request.calls + instance._remember_reply_thread.assert_not_awaited()