From 1b88e17e771b63b6b819020e11dede234908c7e5 Mon Sep 17 00:00:00 2001 From: Totoro Date: Mon, 26 Oct 2026 15:39:39 +0800 Subject: [PATCH] fix(channels): report failed attachment delivery (#7532) * fix(channels): report failed attachment delivery * refactor(channels): clarify attachment feedback helpers --------- Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com> Co-authored-by: Willem Jiang --- README.md | 2 + README_zh.md | 2 + backend/app/channels/AGENTS.md | 2 + backend/app/channels/base.py | 70 +++++++- backend/app/channels/wecom.py | 33 ++-- backend/docs/IM_CHANNEL_CONNECTIONS.md | 8 + .../tests/test_channel_file_attachments.py | 167 +++++++++++++++++- 7 files changed, 254 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 54495afd78..6cfa6c6255 100644 --- a/README.md +++ b/README.md @@ -1115,6 +1115,8 @@ cleanup when migrating from legacy metadata credentials. DeerFlow supports receiving tasks from messaging apps. Channels auto-start when configured — no public IP required for any of them. +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. + 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/README_zh.md b/README_zh.md index 9f492b5632..f0027cbb79 100644 --- a/README_zh.md +++ b/README_zh.md @@ -433,6 +433,8 @@ DeerFlow 支持可配置的 MCP Server 和 skills，用来扩展能力。 DeerFlow 支持从即时通讯应用接收任务。只要配置完成，对应渠道会自动启动，而且都不需要公网 IP。 +渠道不支持发送附件时，原回复会直接说明。部分或全部上传失败时，只要平台仍允许回复，DeerFlow 会补一条汇总提示，列出未发出的文件；全部成功时不重复确认。文件继续保留在 DeerFlow 会话中，回复窗口过期只记录失败，不自动切换为主动通知。 + `channels/store.json` 含无效 UTF-8 字节时，采用与 JSON 格式损坏相同的恢复方式：SQL 迁移记录警告并跳过，保留原文件，让渠道可以继续启动；memory 后端的 JSON 存储记录警告并从空状态开始，在下次写入时替换该文件。 DeerFlow 还可以在 workspace UI 里暴露用户自有的 IM 渠道连接。启用 `channel_connections` 后，已登录用户可以从侧边栏 / Settings > Channels 绑定 Telegram、Slack、Discord、Feishu/Lark、DingTalk、WeChat、WeCom、QQ 或 Buzz。它复用现有的 `channels.*` 出站传输，因此不需要公网 IP 或 provider 回调地址。入站 IM 消息会以所连接的 DeerFlow 用户身份运行。设置和安全注意事项参见 [IM Channel Connections](backend/docs/IM_CHANNEL_CONNECTIONS.md)。 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 13d29d6709..f0e491f106 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -15,6 +15,8 @@ Finalize asyncgens; keep primary errors; propagate lone Bot errors. DingTalk retains tasks and logs failures. +Shared attachment feedback uses public `attachment_delivery_text`, preserves routing/cancellation, and never retries or pushes proactively. Built-in channels choose EN/ZH from the first non-empty reply line; an explicit locale overrides it. WeCom keeps frame cleanup and refuses feedback after its reply context expires. + **Components**: - `message_bus.py` - Inbound queue → dispatcher; outbound uses a dispatch-start listener snapshot. Unsubscribe neither cancels nor drains captured callbacks. - `store.py` - `ChannelStore` protocol (async `get_thread_id`/`set_thread_id`/`claim_thread_id`/`remove`/`list_entries`) for `channel_name:chat_id[:topic_id]` → `thread_id`. `SqlChannelStore` (`channel_thread_bindings`, harness `persistence/channel_thread_bindings/`; upsert keeps `created_at`, topic-less `remove` deletes by key prefix) for sqlite/postgres, so replicas share bindings; `JsonChannelStore` (`{base_dir}/channels/store.json`, loaded once per process, file I/O via `to_thread` under `_lock`) only for `database.backend: memory`. `resolve_channel_store` picks; `ChannelService.start()` runs `SqlChannelStore.import_legacy_json` once (file + empty table → `INSERT … ON CONFLICT DO NOTHING`, rename to `store.json.migrated`, a peer's rename tolerated; populated table → file kept). Feishu's sync lark callback (on the SDK's own loop) bridges `_resolve_routing_bounded` to the Gateway loop via `_run_store_call` (tracked future; refuses only a wait from the Gateway loop itself). Store calls run under `FEISHU_STORE_LOOKUP_TIMEOUT_SECONDS` (3s; the SDK loop also services ws keepalives); timeout or store error = *unresolved* (`StoreLookupUnresolved`), never a miss: the message joins a per-chat FIFO drained on the Gateway loop with backoff up to `FEISHU_STORE_RESOLVE_TOTAL_BUDGET_SECONDS` (30s), then is dropped with a warning, never routed to a new conversation; each deferred message holds a bus intake reservation (the `inbound_queue_maxsize` bound) and its 30s budget runs from arrival. `discord.py` reads `channels_dir`. Tests: `test_channel_thread_binding_store.py` (also under `blocking_io/`). diff --git a/backend/app/channels/base.py b/backend/app/channels/base.py index 35f55ccd0b..7c23c5dfd6 100644 --- a/backend/app/channels/base.py +++ b/backend/app/channels/base.py @@ -9,11 +9,13 @@ from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable, Coroutine from concurrent.futures import CancelledError as FutureCancelledError from concurrent.futures import Future -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any, TypeVar from app.channels.commands import extract_connect_code from app.channels.message_bus import ( + HUMAN_INPUT_REQUEST_METADATA_KEY, + PENDING_CLARIFICATION_METADATA_KEY, InboundMessage, InboundMessageType, InboundQueueClosedError, @@ -30,6 +32,26 @@ logger = logging.getLogger(__name__) T = TypeVar("T") +def attachment_delivery_text(msg: OutboundMessage, failed: list[ResolvedAttachment], *, unsupported: bool = False) -> str: + """Format shared feedback without host paths or provider errors. + + Built-in IM outbounds do not populate ``locale``; their notices follow + the first non-empty reply line. An explicit locale remains an override. + """ + locale = msg.metadata.get("locale") + first_line = next((line for line in msg.text.splitlines() if line.strip()), "") + chinese = locale.lower().startswith("zh") if isinstance(locale, str) else any("\u4e00" <= char <= "\u9fff" for char in first_line) + if unsupported: + return "当前渠道不支持发送附件。请到 DeerFlow 会话下载文件。" if chinese else "This channel does not support sending files. Open the DeerFlow conversation to download them." + filenames = "\n".join("- " + " ".join(attachment.filename.split()) for attachment in failed) + sent = len(msg.attachments) - len(failed) + if chinese: + status = f"已发送 {sent}/{len(msg.attachments)} 个附件，以下附件未能发送：" if sent else "附件未能发送：" + return f"{status}\n{filenames}\n文件仍保留在 DeerFlow 会话中，可前往下载。" + status = f"Sent {sent}/{len(msg.attachments)} files. These files could not be sent:" if sent else "These files could not be sent in this chat:" + return f"{status}\n{filenames}\nThe files remain available in the DeerFlow conversation." + + class ChannelUnavailable(Exception): """Local connectivity failure: the channel cannot accept traffic right now. @@ -502,19 +524,49 @@ class Channel(ABC): partial deliveries (files without accompanying text). """ if msg.channel_name == self.name: + # The inherited implementation explicitly means "unsupported". + # Inspect the bound method so custom and instance overrides keep + # working without a second provider-capabilities registry. + unsupported = bool(msg.attachments) and getattr(self.send_file, "__func__", None) is Channel.send_file + reply = msg + if unsupported: + notice = attachment_delivery_text(msg, msg.attachments, unsupported=True) + reply = replace(msg, text=f"{notice}\n\n{msg.text}" if msg.text else notice, attachments=[]) try: - await self.send(msg) + await self.send(reply) except Exception: logger.exception("Failed to send outbound message on channel %s", self.name) return # Do not attempt file uploads when the text message failed - for attachment in msg.attachments: - try: - success = await self.send_file(msg, attachment) - if not success: - logger.warning("[%s] file upload skipped for %s", self.name, attachment.filename) - except Exception: - logger.exception("[%s] failed to upload file %s", self.name, attachment.filename) + if unsupported: + return + failed = await self._send_attachments(msg) + if failed: + await self._send_attachment_feedback(msg, failed) + + async def _send_attachments(self, msg: OutboundMessage) -> list[ResolvedAttachment]: + """Try each upload once; platform-specific retries stay in the adapter.""" + failed: list[ResolvedAttachment] = [] + for attachment in msg.attachments: + try: + if await self.send_file(msg, attachment): + continue + logger.warning("[%s] file upload skipped for %s", self.name, attachment.filename) + except Exception: + logger.exception("[%s] failed to upload file %s", self.name, attachment.filename) + failed.append(attachment) + return failed + + async def _send_attachment_feedback(self, msg: OutboundMessage, failed: list[ResolvedAttachment]) -> None: + """Reply in the same conversation; never resubmit files or push elsewhere.""" + metadata = {key: value for key, value in msg.metadata.items() if key not in {PENDING_CLARIFICATION_METADATA_KEY, HUMAN_INPUT_REQUEST_METADATA_KEY}} + feedback = replace(msg, text=attachment_delivery_text(msg, failed), attachments=[], artifacts=[], metadata=metadata) + try: + await self.send(feedback) + except Exception: + # The platform may no longer accept a reply. Retain files and logs; + # do not turn this into a proactive notification. + logger.exception("[%s] failed to send attachment delivery feedback", self.name) async def receive_file(self, msg: InboundMessage, thread_id: str, *, user_id: str | None = None) -> InboundMessage: """ diff --git a/backend/app/channels/wecom.py b/backend/app/channels/wecom.py index 0ffee1a5da..4eeae3950f 100644 --- a/backend/app/channels/wecom.py +++ b/backend/app/channels/wecom.py @@ -8,7 +8,7 @@ import logging from collections.abc import Awaitable, Callable, Mapping from typing import Any, cast -from app.channels.base import Channel, ChannelUnavailable +from app.channels.base import Channel, ChannelUnavailable, attachment_delivery_text from app.channels.commands import is_known_channel_command from app.channels.connection_identity import attach_connection_identity from app.channels.message_bus import ( @@ -415,25 +415,24 @@ class WeComChannel(Channel): if msg.channel_name != self.name: return - try: - await self.send(msg) - except Exception: - logger.exception("Failed to send outbound message on channel %s", self.name) - if msg.is_final: - self._clear_ws_context(msg.thread_ts) - return - - for attachment in msg.attachments: - try: - success = await self.send_file(msg, attachment) - if not success: - logger.warning("[%s] file upload skipped for %s", self.name, attachment.filename) - except Exception: - logger.exception("[%s] failed to upload file %s", self.name, attachment.filename) - + await super()._on_outbound(msg) if msg.is_final: self._clear_ws_context(msg.thread_ts) + async def _send_attachment_feedback(self, msg: OutboundMessage, failed: list[ResolvedAttachment]) -> None: + if not msg.thread_ts or msg.thread_ts not in self._ws_frames: + # send() without a frame uses the proactive API. A failed upload + # must not silently switch from the original reply to that path. + logger.warning("[WeCom] attachment feedback unavailable: reply context expired") + return + # The final stream is already finished. Reply with text on the same + # inbound frame instead of reopening that stream or pushing elsewhere. + try: + text = _clip_to_byte_limit(attachment_delivery_text(msg, failed), _WECOM_MAX_CONTENT_BYTES) + await self._ws_client.reply(self._ws_frames[msg.thread_ts], {"msgtype": "text", "text": {"content": text}}) + except Exception: + logger.exception("[WeCom] failed to send attachment delivery feedback") + async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool: if not msg.is_final: return True diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 4abadef79f..6149c85fd2 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -183,6 +183,14 @@ graph TB For the special GitHub case (`fire_and_forget=True` channel policy), the manager calls `runs.create()` and returns once the run is `pending` — no outbound reply, because GitHub agents post via the `gh` CLI from inside their sandbox. See [GITHUB_AGENTS.md](GITHUB_AGENTS.md) for the full GitHub flow. +## Attachment delivery feedback + +The shared channel outbound path sends text before uploading attachments. Channels inheriting the default, unsupported `send_file` implementation explain the limitation in that first reply instead of attempting uploads. Supported channels continue trying the other attachments after an individual failure, then send one combined notice listing only unsent files. Successful uploads need no additional confirmation. + +The notice stays in the original conversation, preserves connection/owner routing, and carries no attachments or clarification-card payload. Built-in IM outbounds do not currently populate `metadata.locale`, so their notices choose Chinese or English from the first non-empty response line. This path does not read user language preferences; an explicitly supplied locale still overrides the heuristic. Files stay in the original DeerFlow conversation for download. If the platform no longer accepts the notice, the upload failures and notice failure remain in the logs; no proactive notification, regeneration, or new upload retry is started. + +WeCom reuses this shared processing, sends failure notices as text replies on the original inbound frame after the final stream, and retains its frame/stream cleanup after delivery. Provider-specific upload protocols, reply windows, and send ordering are unchanged. + ## Owner-scoped File Storage `ChannelManager` resolves the storage owner **once** at the top of `_handle_chat` via `_channel_storage_user_id(msg)` and threads that value through the entire file pipeline. The same identity is used as the run `user_id` in `run_context` and as the bucket for memory, uploads, and outputs — so the bucket the agent reads/writes is always the bucket where channel files were staged. diff --git a/backend/tests/test_channel_file_attachments.py b/backend/tests/test_channel_file_attachments.py index 163a7c6fd0..37c462ee8e 100644 --- a/backend/tests/test_channel_file_attachments.py +++ b/backend/tests/test_channel_file_attachments.py @@ -6,13 +6,13 @@ import asyncio import os import stat from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from support.symlinks import symlink_or_skip from app.channels.base import Channel -from app.channels.message_bus import InboundMessage, MessageBus, OutboundMessage, ResolvedAttachment +from app.channels.message_bus import HUMAN_INPUT_REQUEST_METADATA_KEY, PENDING_CLARIFICATION_METADATA_KEY, InboundMessage, MessageBus, OutboundMessage, ResolvedAttachment def _run(coro): @@ -820,6 +820,15 @@ class _DummyChannel(Channel): return True +def _file_reply(tmp_path, *, channel_name="dummy", text="Your files are ready.", metadata=None): + attachments = [] + for name in ("first.csv", "second.csv"): + path = tmp_path / name + path.write_text("value\n1\n") + attachments.append(ResolvedAttachment(f"/mnt/user-data/outputs/{name}", path, name, "text/csv", path.stat().st_size, False)) + return OutboundMessage(channel_name=channel_name, chat_id="chat-1", thread_id="thread-1", thread_ts="source-1", connection_id="connection-1", owner_user_id="owner-1", text=text, attachments=attachments, metadata=metadata or {}) + + class TestBaseChannelOnOutbound: def test_default_receive_file_returns_original_message(self): """The base Channel.receive_file returns the original message unchanged.""" @@ -891,7 +900,8 @@ class TestBaseChannelOnOutbound: assert len(ch.sent_messages) == 1 assert len(ch.sent_files) == 0 - def test_send_file_failure_does_not_block_others(self, tmp_path): + @pytest.mark.parametrize("failure", ["false", "exception"]) + def test_send_file_failure_does_not_block_others(self, tmp_path, failure): """If one attachment upload fails, remaining attachments still get sent.""" bus = MessageBus() ch = _DummyChannel(bus) @@ -904,7 +914,9 @@ class TestBaseChannelOnOutbound: nonlocal call_count call_count += 1 if call_count == 1: - raise RuntimeError("upload failed") + if failure == "exception": + raise RuntimeError("private upload details must not reach the chat") + return False return await original_send_file(msg, att) ch.send_file = flaky_send_file # type: ignore @@ -930,6 +942,96 @@ class TestBaseChannelOnOutbound: # First upload failed, second succeeded assert len(ch.sent_files) == 1 assert ch.sent_files[0][1].filename == "ok.txt" + assert len(ch.sent_messages) == 2 + feedback = ch.sent_messages[-1] + assert "1/2" in feedback.text + assert "fail.txt" in feedback.text and "ok.txt" not in feedback.text + assert "DeerFlow" in feedback.text + assert "private upload details" not in feedback.text + assert feedback.attachments == [] + assert f1.exists() and f2.exists() + + @pytest.mark.parametrize("locale,expected", [("en-US", "could not be sent"), ("zh-CN", "未能发送")]) + def test_all_failed_files_get_one_localized_reply_in_the_same_conversation(self, tmp_path, locale, expected): + channel = _DummyChannel(MessageBus()) + metadata = {"locale": locale, "context_token": "private-routing-token", PENDING_CLARIFICATION_METADATA_KEY: True, HUMAN_INPUT_REQUEST_METADATA_KEY: {"question": "Choose one"}} + msg = _file_reply(tmp_path, metadata=metadata) + + async def refuse(message, attachment): + channel.sent_files.append((message, attachment)) + return False + + channel.send_file = refuse + + async def deliver(): + channel.bus.subscribe_outbound(channel._on_outbound) + await channel.bus.publish_outbound(msg) + + _run(deliver()) + + assert len(channel.sent_messages) == 2 + feedback = channel.sent_messages[-1] + assert expected in feedback.text + assert all(attachment.filename in feedback.text for attachment in msg.attachments) + assert str(tmp_path) not in feedback.text and "private-routing-token" not in feedback.text + assert (feedback.chat_id, feedback.thread_id, feedback.thread_ts, feedback.connection_id, feedback.owner_user_id) == (msg.chat_id, msg.thread_id, msg.thread_ts, msg.connection_id, msg.owner_user_id) + assert not feedback.attachments and not feedback.artifacts + assert feedback.metadata == {"locale": locale, "context_token": "private-routing-token"} + assert msg.metadata == metadata and len(msg.attachments) == 2 + + @pytest.mark.parametrize("text,expected", [("Your files are ready.", "does not support sending files"), ("文件已经生成。", "不支持发送附件")]) + def test_unsupported_channel_explains_in_the_original_reply(self, tmp_path, text, expected): + class TextOnlyChannel(_DummyChannel): + send_file = Channel.send_file + + channel = TextOnlyChannel(MessageBus()) + original_text = f"{text}\n\nCreated Files: first.csv, second.csv" + msg = _file_reply(tmp_path, text=original_text) + _run(channel._on_outbound(msg)) + + assert len(channel.sent_messages) == 1 + assert expected in channel.sent_messages[0].text + assert channel.sent_messages[0].text.index(expected) < channel.sent_messages[0].text.index(text) + assert all(a.filename in channel.sent_messages[0].text for a in msg.attachments) + assert all(channel.sent_messages[0].text.count(a.filename) == 1 for a in msg.attachments) + assert "DeerFlow" in channel.sent_messages[0].text + assert not channel.sent_files + assert msg.text == original_text and len(msg.attachments) == 2 + + def test_expired_feedback_does_not_retry_uploads_or_remove_files(self, tmp_path, caplog): + channel = _DummyChannel(MessageBus()) + msg = _file_reply(tmp_path) + attempts = [] + + async def refuse(message, attachment): + attempts.append(attachment.filename) + return False + + async def send(message): + channel.sent_messages.append(message) + if len(channel.sent_messages) > 1: + raise RuntimeError("reply window expired") + + channel.send_file = refuse + channel.send = send + _run(channel._on_outbound(msg)) + + assert attempts == ["first.csv", "second.csv"] + assert len(channel.sent_messages) == 2 + assert all(a.actual_path.exists() for a in msg.attachments) + assert "feedback" in caplog.text.lower() + + def test_cancellation_during_upload_does_not_send_a_completion_notice(self, tmp_path): + channel = _DummyChannel(MessageBus()) + msg = _file_reply(tmp_path) + + async def cancelled(message, attachment): + raise asyncio.CancelledError + + channel.send_file = cancelled + with pytest.raises(asyncio.CancelledError): + _run(channel._on_outbound(msg)) + assert channel.sent_messages == [msg] def test_send_raises_skips_file_uploads(self, tmp_path): """When send() raises, file uploads are skipped entirely.""" @@ -984,6 +1086,63 @@ class TestBaseChannelOnOutbound: # --------------------------------------------------------------------------- +class TestWecomAttachmentDeliveryFeedback: + def test_lost_reply_context_never_falls_back_to_a_proactive_send(self, tmp_path): + from types import SimpleNamespace + + from app.channels.wecom import WeComChannel + + channel = WeComChannel(MessageBus(), {}) + msg = _file_reply(tmp_path, channel_name="wecom") + channel._ws_frames[msg.thread_ts] = {"body": {"msgid": msg.thread_ts}} + channel._ws_stream_ids[msg.thread_ts] = "stream-1" + channel._ws_client = SimpleNamespace(reply_stream=AsyncMock(), send_message=AsyncMock()) + + async def expired(message, attachment): + channel._clear_ws_context(msg.thread_ts) + return False + + channel.send_file = expired + _run(channel._on_outbound(msg)) + + channel._ws_client.reply_stream.assert_awaited_once() + channel._ws_client.send_message.assert_not_awaited() + assert all(a.actual_path.exists() for a in msg.attachments) + + @pytest.mark.parametrize("feedback_expired", [False, True]) + def test_failed_upload_feedback_precedes_context_cleanup(self, tmp_path, feedback_expired): + from types import SimpleNamespace + + from app.channels.wecom import WeComChannel + + channel = WeComChannel(MessageBus(), {}) + msg = _file_reply(tmp_path, channel_name="wecom") + frame = {"body": {"msgid": msg.thread_ts}} + channel._ws_frames[msg.thread_ts] = frame + channel._ws_stream_ids[msg.thread_ts] = "stream-1" + replies = [] + + async def reply(reply_frame, body): + assert channel._ws_frames[msg.thread_ts] is frame + assert reply_frame is frame + replies.append(body) + if feedback_expired: + raise RuntimeError("reply window expired") + + channel._ws_client = SimpleNamespace(reply_stream=AsyncMock(), reply=AsyncMock(side_effect=reply)) + channel._upload_media_ws = AsyncMock(return_value=None) + _run(channel._on_outbound(msg)) + + channel._ws_client.reply_stream.assert_awaited_once_with(frame, "stream-1", msg.text, True) + assert len(replies) == 1 + assert replies[0]["msgtype"] == "text" + assert "could not be sent" in replies[0]["text"]["content"] + assert channel._upload_media_ws.await_count == 2 + assert msg.thread_ts not in channel._ws_frames + assert msg.thread_ts not in channel._ws_stream_ids + assert all(a.actual_path.exists() for a in msg.attachments) + + class TestManagerArtifactResolution: def test_handle_chat_populates_attachments(self): """Verify _resolve_attachments is importable and works with the manager module."""