From 499fbe11423d1fdac0c78990e3c7fb0635b1276b Mon Sep 17 00:00:00 2001 From: ooiuuii Date: Tue, 13 Oct 2026 19:32:30 -0700 Subject: [PATCH] fix(telegram): allow deleted reply targets in private chats (#6791) * fix(telegram): allow deleted reply targets in private chats * fix(telegram): preserve previews after reply target deletion --- README.md | 1 + backend/app/channels/AGENTS.md | 2 +- backend/app/channels/telegram.py | 20 +- backend/tests/test_channels.py | 1 + backend/tests/test_telegram_deleted_reply.py | 198 +++++++++++++++++++ 5 files changed, 216 insertions(+), 6 deletions(-) create mode 100644 backend/tests/test_telegram_deleted_reply.py diff --git a/README.md b/README.md index 4933d0745b..4d72b438dd 100644 --- a/README.md +++ b/README.md @@ -1057,6 +1057,7 @@ Notes: - Inbound work is bounded to `inbound_queue_maxsize` pending messages plus `max_concurrency` active workers. When capacity is exhausted, socket/polling providers drop new messages before sending DeerFlow's working acknowledgment and emit a rate-limited warning. Buzz leaves its replay cursor unchanged and reconnects for relay replay; GitHub webhooks return `503`, marking the delivery failed for manual/API redelivery. Shutdown closes admission immediately, keeps channel transports available while accepted messages drain for up to `shutdown_grace_period_seconds`, then cancels and awaits active handlers before closing provider resources; the Gateway's outer timeout can cancel an incomplete shutdown without detaching those resources. - Telegram shutdown retains ongoing cleanup across a timeout or cancellation so a retry can finish it. Stopping during startup also cancels initialization and closes the partially initialized bot's HTTP clients before retiring its worker loop. If the Telegram loop has already stopped, shutdown warns that download clients may remain open and releases the abandoned cleanup state; later stops do not repeat that warning. - Slack stop waits up to five seconds for an in-flight Socket Mode connection and its cleanup. If cleanup is still pending, the previous instance stays tracked and a restart can be retried after it finishes. SDK cleanup failures report their original error and remain retryable. +- In Telegram private chats, text, attachment and streaming replies (including "Working on it...") can still be delivered if their referenced user or bot message was deleted, such as after clearing chat history during a run. Group reply-target requirements remain unchanged to preserve forum routing. - Feishu/Lark now queues rapid follow-up messages per mapped DeerFlow `thread_id` instead of immediately surfacing the generic busy reply, and topic replies keep a per-message card with a compact source-message preview across queued/running/final patches. - Streaming IM channels treat backend `error` events like transport failures and follow the same reply and retry path. The channel logs the error type and message for diagnosis. diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 90e79aedcb..d49f9b9af9 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -41,7 +41,7 @@ Finalize asyncgens; keep primary errors; propagate lone Bot errors. 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound 6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound; the agent posts its own reply via sandbox `gh`. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub); with `buffer_followups_on_busy=True` (GitHub's default) the triggering message is captured into a per-thread buffer so a concurrent comment is not silently dropped (see "Follow-up buffering while busy"). 7. Feishu patches one card per source (`config.update_multi=true`), with topic previews and queued → running → final states. Opt-in cards show context/question; bounded tokens preserve request ID/source. Originator/option checks consume once; consumption fails closed across workers (`feishu_card_interaction.py`). Card replies keep bound aliases in fields/metadata; ordinary replies preserve `/new`. Accepted intake retires only that card's buttons/text hint before terminal PATCH; rejected intake keeps text fallback. -8. Telegram streaming: the "Working on it..." placeholder message is registered as the stream target; non-final updates `editMessageText` it in place (channel-side throttle: 1s in private chats, 3s in groups due to Telegram's 20 msg/min group cap; 4096-char truncation; rate-limited updates dropped); the final update performs the last edit and splits >4096 texts into follow-up messages +8. Telegram edits the "Working on it..." stream target: 1s private/3s group throttle (20 msg/min), 4096-char truncation, rate-limited updates dropped; final edits split overflow. Text/preview/photo/document reply sends use `allow_sending_without_reply` only for `chat_id > 0`: missing private targets degrade atomically; group/forum routing and transport retries stay unchanged. 9. DingTalk AI Cards (`card_template_id`): `runs.stream()` → create card with initial text → `PUT /v1.0/card/streaming` → finalize (`is_final=True`). Creation/streaming failure uses `sampleMarkdown`; tables preserve escaped pipes and empty edge cells. 10. For commands (`/new`, `/status`, `/models`, `/model`, `/memory`, `/goal`, `/agent`, `/help`): handle locally or query Gateway API. `/agent list` reads the effective owner's Custom Agents; `/agent use ` validates in the same owner bucket, creates a new thread, and persists `channel_agent_name` in its metadata. A Custom Agent writes canonical `metadata.agent_name` so thread-search routes Web continuation to `/workspace/agents//chats/`; `lead_agent` omits it and stays on the ordinary chat route. `/model ` persists `channel_model_name` in the *current* thread's metadata (in place, no fresh thread); `/model` reports the effective model, `/model default` clears it (stale pins self-clear). diff --git a/backend/app/channels/telegram.py b/backend/app/channels/telegram.py index 3f65406175..7aa0a6648c 100644 --- a/backend/app/channels/telegram.py +++ b/backend/app/channels/telegram.py @@ -370,6 +370,8 @@ class TelegramChannel(Channel): send_kwargs: dict[str, Any] = {"chat_id": chat_id, "text": display} if reply_to: send_kwargs["reply_to_message_id"] = reply_to + if chat_id > 0: + send_kwargs["allow_sending_without_reply"] = True if state is None: try: @@ -505,6 +507,10 @@ class TelegramChannel(Channel): reply_to = self._last_bot_message.get(chat_key) if reply_to: kwargs["reply_to_message_id"] = reply_to + # A deleted private-chat reply must not block delivery. Keep group + # replies strict: dropping their target could change forum routing. + if chat_id > 0: + kwargs["allow_sending_without_reply"] = True bot = self._application.bot @@ -543,11 +549,15 @@ class TelegramChannel(Channel): 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) else: kwargs = {"chat_id": chat_id, "document": 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_document(**kwargs) self._last_bot_message[msg.chat_id] = sent.message_id @@ -967,11 +977,11 @@ class TelegramChannel(Channel): return try: bot = self._application.bot - sent = await bot.send_message( - chat_id=int(chat_id), - text="Working on it...", - reply_to_message_id=reply_to_message_id, - ) + numeric_chat_id = int(chat_id) + kwargs: dict[str, Any] = {"chat_id": numeric_chat_id, "text": "Working on it...", "reply_to_message_id": reply_to_message_id} + if numeric_chat_id > 0: + kwargs["allow_sending_without_reply"] = True + sent = await bot.send_message(**kwargs) self._register_stream_message( self._stream_key(chat_id, str(reply_to_message_id)), message_id=sent.message_id, diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 6561e7580d..67cbf3e996 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -12701,6 +12701,7 @@ class TestTelegramStreaming: chat_id=12345, text="Working on it...", reply_to_message_id=42, + allow_sending_without_reply=True, ) _run(go()) diff --git a/backend/tests/test_telegram_deleted_reply.py b/backend/tests/test_telegram_deleted_reply.py new file mode 100644 index 0000000000..693c687b6f --- /dev/null +++ b/backend/tests/test_telegram_deleted_reply.py @@ -0,0 +1,198 @@ +"""Offline Bot API simulations exercising real Telegram SDK serialization.""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest +import pytest_asyncio +from telegram import Bot +from telegram.error import BadRequest, NetworkError +from telegram.request import BaseRequest + +from app.channels import telegram as telegram_module +from app.channels.message_bus import MessageBus, OutboundMessage, ResolvedAttachment +from app.channels.telegram import TelegramChannel + + +class _BotApiRequest(BaseRequest): + @property + def read_timeout(self): + return 5 + + def __init__(self): + self.messages = {42} # The inbound user message. + self.calls = [] + self.next_id = 100 + + 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] + parameters = request_data.parameters if request_data else {} + self.calls.append((endpoint, parameters)) + if endpoint == "getMe": + result = {"id": 1, "is_bot": True, "first_name": "Test", "username": "test_bot"} + else: + reply = parameters.get("reply_parameters", {}) + if reply.get("message_id") not in self.messages and reply and not reply.get("allow_sending_without_reply"): + return 400, json.dumps({"ok": False, "error_code": 400, "description": "Bad Request: message to be replied not found"}).encode() + if endpoint == "editMessageText" and parameters["message_id"] not in self.messages: + return 400, json.dumps({"ok": False, "error_code": 400, "description": "Bad Request: message to edit not found"}).encode() + message_id = parameters.get("message_id", self.next_id) + self.next_id += 1 + self.messages.add(message_id) + result = { + "message_id": message_id, + "date": 0, + "chat": {"id": parameters["chat_id"], "type": "private" if parameters["chat_id"] > 0 else "supergroup"}, + "text": parameters.get("text", ""), + } + return 200, json.dumps({"ok": True, "result": result}).encode() + + +@pytest_asyncio.fixture +async def telegram_channel(): + request = _BotApiRequest() + async with Bot("123:test-token", request=request, get_updates_request=request) as bot: + channel = TelegramChannel(MessageBus(), {"bot_token": "123:test-token"}) + channel._application = SimpleNamespace(bot=bot) + yield channel, request + + +def _reply(chat_id="12345", **kwargs): + return OutboundMessage(channel_name="telegram", chat_id=chat_id, thread_id="thread", text="Complete answer", **kwargs) + + +@pytest.mark.asyncio +async def test_private_final_reply_survives_chat_history_cleared_during_run(telegram_channel): + channel, request = telegram_channel + await channel.send(_reply()) + old_reply = channel._last_bot_message["12345"] + await channel._send_running_reply("12345", 42) + assert channel._stream_messages["12345:42"]["message_id"] in request.messages + + # Clearing history removes both the prior reply and the in-flight preview. + request.messages.clear() + before = len(request.calls) + await channel.send(_reply(thread_ts="42")) + + assert [endpoint for endpoint, _ in request.calls[before:]] == ["editMessageText", "sendMessage"] + assert request.calls[-1][1]["reply_parameters"]["message_id"] == old_reply + assert channel._last_bot_message["12345"] in request.messages + assert "12345:42" not in channel._stream_messages + + +@pytest.mark.asyncio +@pytest.mark.parametrize("chat_id", ["12345", "-12345"]) +@pytest.mark.parametrize("stage", ["running", "initial_stream", "replacement_stream"]) +async def test_deleted_inbound_target_is_optional_only_for_private_previews(telegram_channel, monkeypatch, chat_id, stage): + channel, request = telegram_channel + key = f"{chat_id}:42" + monkeypatch.setattr(telegram_module, "_monotonic", lambda: 10.0) + if stage == "replacement_stream": + await channel.send(_reply(chat_id, thread_ts="42", is_final=False)) + assert channel._stream_messages[key]["message_id"] in request.messages + monkeypatch.setattr(telegram_module, "_monotonic", lambda: 20.0) + + # Deletion removes the inbound target and, when present, the preview too. + request.messages.clear() + before = len(request.calls) + if stage == "running": + await channel._send_running_reply(chat_id, 42) + else: + await channel.send(OutboundMessage(channel_name="telegram", chat_id=chat_id, thread_id="thread", thread_ts="42", text="Updated preview", is_final=False)) + + expected_calls = ["editMessageText", "sendMessage"] if stage == "replacement_stream" else ["sendMessage"] + assert [endpoint for endpoint, _ in request.calls[before:]] == expected_calls + reply = request.calls[-1][1]["reply_parameters"] + assert reply["message_id"] == 42 + state = channel._stream_messages.get(key) + if int(chat_id) > 0: + assert reply["allow_sending_without_reply"] is True + assert state is not None and state["message_id"] in request.messages + assert state["last_text"] == ("Working on it..." if stage == "running" else "Updated preview") + else: + assert not reply.get("allow_sending_without_reply") + assert state is None or state["message_id"] not in request.messages + + +@pytest.mark.asyncio +@pytest.mark.parametrize("chat_id", ["12345", "-12345"]) +@pytest.mark.parametrize("stage", ["running", "initial_stream"]) +async def test_existing_inbound_target_remains_attached_to_preview(telegram_channel, chat_id, stage): + channel, request = telegram_channel + if stage == "running": + await channel._send_running_reply(chat_id, 42) + else: + await channel.send(_reply(chat_id, thread_ts="42", is_final=False)) + assert request.calls[-1][1]["reply_parameters"]["message_id"] == 42 + assert channel._stream_messages[f"{chat_id}:42"]["message_id"] in request.messages + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_image", [False, True]) +@pytest.mark.parametrize("chat_id", ["12345", "-12345"]) +async def test_deleted_file_reply_target_is_optional_only_in_private_chat(telegram_channel, tmp_path, is_image, chat_id): + channel, request = telegram_channel + await channel.send(_reply(chat_id)) + request.messages.clear() + file_path = tmp_path / "output.bin" + file_path.write_bytes(b"offline attachment") + attachment = ResolvedAttachment("/mnt/user-data/outputs/output.bin", file_path, file_path.name, "image/png" if is_image else "application/octet-stream", file_path.stat().st_size, is_image) + before = len(request.calls) + + assert await channel.send_file(_reply(chat_id), attachment) is (int(chat_id) > 0) + + assert [endpoint for endpoint, _ in request.calls[before:]] == ["sendPhoto" if is_image else "sendDocument"] + assert (channel._last_bot_message[chat_id] in request.messages) is (int(chat_id) > 0) + if int(chat_id) < 0: + assert not request.calls[-1][1]["reply_parameters"].get("allow_sending_without_reply") + + +@pytest.mark.asyncio +async def test_existing_private_reply_keeps_its_reply_target(telegram_channel): + channel, request = telegram_channel + await channel.send(_reply()) + previous_id = channel._last_bot_message["12345"] + await channel.send(_reply()) + assert request.calls[-1][1]["reply_parameters"]["message_id"] == previous_id + + +@pytest.mark.asyncio +async def test_group_missing_reply_is_not_silently_rerouted(telegram_channel): + channel, request = telegram_channel + await channel.send(_reply("-12345")) + previous_id = channel._last_bot_message["-12345"] + request.messages.clear() + before = len(request.calls) + + with pytest.raises(BadRequest, match="(?i)message to be replied not found"): + await channel.send(_reply("-12345"), _max_retries=1) + + assert len(request.calls) == before + 1 + assert request.calls[-1][1]["reply_parameters"] == {"message_id": previous_id} + + +@pytest.mark.asyncio +async def test_network_failure_does_not_add_a_replyless_resend(telegram_channel, monkeypatch): + channel, request = telegram_channel + await channel.send(_reply()) + previous_id = channel._last_bot_message["12345"] + calls = [] + + async def fail_request(url, method, request_data=None, **kwargs): + calls.append(request_data.parameters) + raise NetworkError("Ambiguous transport failure") + + monkeypatch.setattr(request, "do_request", fail_request) + with pytest.raises(NetworkError, match="Ambiguous transport failure"): + await channel.send(_reply(), _max_retries=1) + assert len(calls) == 1 + assert calls[0]["reply_parameters"]["message_id"] == previous_id + assert channel._last_bot_message["12345"] == previous_id