From 137c721d2b14f4d3de17bf67898cfcd4ca271423 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Tue, 27 Oct 2026 11:56:31 +0900 Subject: [PATCH] fix(channels): write each Feishu thread alias once per streamed reply (#7593) * fix(channels): write each Feishu thread alias once per streamed reply Every outbound to a Feishu running card re-sent set_thread_id for all of the reply's aliases (source message, card, topic, root, parent and Feishu thread ids), although only updated_at changed after the first chunk. That was about five store writes per streamed chunk: a full store.json rewrite under the store lock each, or a database upsert. Remember the aliases written for each in-flight reply and skip them on non-final chunks. A failed write is not remembered, so the next chunk retries it, and the final outbound still writes every alias, so the store ends up as before. * docs(changelog): reference #7593 for the Feishu streamed thread alias writes --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 9 ++++ CHANGELOG_zh.md | 6 +++ backend/app/channels/AGENTS.md | 2 +- backend/app/channels/feishu.py | 21 ++++++++ backend/tests/test_channels.py | 98 ++++++++++++++++++++++++++++++++++ 5 files changed, 135 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7650a8c602..5f6cbbab70 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1402,6 +1402,14 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**. ### Fixed +- **channels:** A streamed Feishu reply no longer rewrites the same thread + bindings on every chunk. Each outbound to the running card stored all of + the reply's aliases again (about five `set_thread_id` writes per chunk: a + full `store.json` rewrite under the store lock each, or a database upsert), + although nothing but `updated_at` changed. Non-final chunks now store each + alias once per reply, a failed write is retried on the next chunk, and the + final outbound still writes them all. ([#7593]) + - **runtime:** Loop-detection warnings now ask the agent to present the files it already created in `/mnt/user-data/outputs` before its final answer, so a warned run no longer fails its delivery check with finished files left @@ -13187,3 +13195,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7585]: https://github.com/bytedance/deer-flow/pull/7585 [#7588]: https://github.com/bytedance/deer-flow/pull/7588 [#7591]: https://github.com/bytedance/deer-flow/pull/7591 +[#7593]: https://github.com/bytedance/deer-flow/pull/7593 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 6cb3b16868..8ce3376e8c 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -1116,6 +1116,11 @@ ### 修复 +- **渠道：** 飞书流式回复不再在每个分块上重写相同的线程绑定。此前每次向运行中卡片发送消息都会 + 重新保存该回复的全部别名（每个分块约五次 `set_thread_id`：每次都是在存储锁下完整重写 + `store.json`，或一次数据库 upsert），而实际只有 `updated_at` 变化。现在非最终分块对每个 + 别名每次回复只写一次，写入失败会在下一个分块重试，最终消息仍会写入全部别名。([#7593]) + - **运行时：** 循环检测警告现在会让 agent 在作答前，先交付已经写入 `/mnt/user-data/outputs` 的文件，被警告的运行不会再因为做好的文件没交付而判为失败。 运行达到 `recursion_limit` 时，现在以 `stop_reason: recursion_capped` 结束，并给出 @@ -10519,3 +10524,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7580]: https://github.com/bytedance/deer-flow/pull/7580 [#7588]: https://github.com/bytedance/deer-flow/pull/7588 [#7591]: https://github.com/bytedance/deer-flow/pull/7591 +[#7593]: https://github.com/bytedance/deer-flow/pull/7593 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index f0e491f106..6ecd64dfc5 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -46,7 +46,7 @@ Shared attachment feedback uses public `attachment_delivery_text`, preserves rou 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`) 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. +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. Non-final chunks store each thread alias once per reply; the final rewrites all. 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. diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index 937a6dc994..fa9f10ef95 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -186,6 +186,9 @@ class FeishuChannel(Channel): self._background_tasks: set[asyncio.Task] = set() self._running_card_ids: dict[str, str] = {} self._running_card_tasks: dict[str, asyncio.Task] = {} + # Thread aliases (chat_id, thread_id, topic_id, user_id) already written + # for each in-flight reply, keyed by source message id. + self._remembered_thread_aliases: dict[str, set[tuple[str, str, str, str]]] = {} self._pending_clarifications: dict[tuple[str, str], list[dict[str, Any]]] = {} # Interactive clarification cards (button answers). Off by default: the # feature needs a one-time console subscription to card.action.trigger. @@ -1111,6 +1114,7 @@ class FeishuChannel(Channel): if msg.is_final: self._running_card_ids.pop(source_message_id, None) + self._remembered_thread_aliases.pop(source_message_id, None) await self._add_reaction(source_message_id, "DONE") return @@ -1119,6 +1123,14 @@ class FeishuChannel(Channel): # -- internal ---------------------------------------------------------- async def _remember_thread_mapping(self, msg: OutboundMessage, *topic_ids: str | None) -> None: + """Map the reply's message, card and topic ids to its DeerFlow thread. + + Every streamed chunk of a reply carries the same aliases, and each write + is a full ``store.json`` rewrite or a database upsert, so a non-final + chunk only writes the aliases this reply has not stored yet. The final + outbound writes all of them again, leaving the store as it was when every + chunk wrote them. + """ store = self.config.get("channel_store") if store is None or not msg.thread_id: return @@ -1134,12 +1146,18 @@ class FeishuChannel(Channel): if isinstance(raw_user_id, str): user_id = raw_user_id + source_message_id = self._non_empty_str(msg.thread_ts) + remembered = self._remembered_thread_aliases.setdefault(source_message_id, set()) if source_message_id and not msg.is_final else set() + seen: set[str] = set() for topic_id in [*topic_ids, *metadata_topic_ids]: topic_id = self._non_empty_str(topic_id) if not topic_id or topic_id in seen: continue seen.add(topic_id) + alias = (msg.chat_id, msg.thread_id, topic_id, user_id) + if alias in remembered: + continue try: await store.set_thread_id( self.name, @@ -1149,7 +1167,10 @@ class FeishuChannel(Channel): user_id=user_id, ) except Exception: + # Not remembered, so the next chunk retries it. logger.exception("[Feishu] failed to remember thread mapping for topic_id=%s", topic_id) + else: + remembered.add(alias) def _run_store_call(self, make_coroutine: Callable[[], Coroutine[Any, Any, _T]], *, name: str, msg_id: Any) -> _T: """Run an async ``channel_store`` call from the synchronous lark callback. diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index ecca2e7104..fd1226d3f8 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -8359,6 +8359,104 @@ class TestFeishuChannel: _run(go()) + @staticmethod + def _streaming_alias_channel(tmp_path, *, fail_once: str | None = None): + """A Feishu channel with a running card and a store that records every alias write.""" + from lark_oapi.api.im.v1 import ( + CreateMessageReactionRequest, + CreateMessageReactionRequestBody, + Emoji, + PatchMessageRequest, + PatchMessageRequestBody, + ReplyMessageRequest, + ReplyMessageRequestBody, + ) + + from app.channels.feishu import FeishuChannel + + class RecordingStore(JsonChannelStore): + def __init__(self, path): + super().__init__(path) + self.writes: list[str] = [] + self.fail_once = fail_once + + async def set_thread_id(self, channel_name, chat_id, thread_id, *, topic_id=None, user_id=""): + self.writes.append(topic_id) + if topic_id == self.fail_once: + self.fail_once = None + raise RuntimeError("offline store failure") + await super().set_thread_id(channel_name, chat_id, thread_id, topic_id=topic_id, user_id=user_id) + + store = RecordingStore(tmp_path / "store.json") + channel = FeishuChannel(MessageBus(), config={"channel_store": store}) + channel._api_client = MagicMock() + channel._ReplyMessageRequest = ReplyMessageRequest + channel._ReplyMessageRequestBody = ReplyMessageRequestBody + channel._PatchMessageRequest = PatchMessageRequest + channel._PatchMessageRequestBody = PatchMessageRequestBody + channel._CreateMessageReactionRequest = CreateMessageReactionRequest + channel._CreateMessageReactionRequestBody = CreateMessageReactionRequestBody + channel._Emoji = Emoji + reply_response = MagicMock() + reply_response.data.message_id = "om-running-card" + channel._api_client.im.v1.message.reply = MagicMock(return_value=reply_response) + channel._api_client.im.v1.message.patch = MagicMock() + channel._api_client.im.v1.message_reaction.create = MagicMock() + return channel, store + + @staticmethod + def _streamed_chunk(text: str, *, is_final: bool) -> OutboundMessage: + return OutboundMessage( + channel_name="feishu", + chat_id="chat-1", + thread_id="thread-1", + text=text, + is_final=is_final, + thread_ts="om-source-msg", + metadata={"user_id": "ou-user", "message_id": "om-source-msg", "topic_id": "om-root", "root_id": "om-root", "parent_id": "om-parent", "thread_id": "omt-thread"}, + ) + + def test_streaming_writes_each_thread_alias_once_per_reply(self, tmp_path): + aliases = ["om-source-msg", "om-running-card", "om-root", "om-parent", "omt-thread"] + + async def go(): + channel, store = self._streaming_alias_channel(tmp_path) + await channel._send_running_reply("om-source-msg") + + await channel.send(self._streamed_chunk("He", is_final=False)) + assert store.writes == aliases + + for text in ("Hell", "Hello", "Hello wo"): + await channel.send(self._streamed_chunk(text, is_final=False)) + assert store.writes == aliases + assert channel._api_client.im.v1.message.patch.call_count == 4 + + # An alias dropped mid-stream is restored: the final outbound writes + # every alias again, as each chunk used to. + await store.remove("feishu", "chat-1", topic_id="om-parent") + await channel.send(self._streamed_chunk("Hello world", is_final=True)) + assert store.writes == aliases * 2 + for topic_id in aliases: + assert await store.get_thread_id("feishu", "chat-1", topic_id=topic_id) == "thread-1" + assert "om-source-msg" not in channel._remembered_thread_aliases + + _run(go()) + + def test_streaming_retries_a_failed_thread_alias_on_the_next_chunk(self, tmp_path): + async def go(): + channel, store = self._streaming_alias_channel(tmp_path, fail_once="om-root") + await channel._send_running_reply("om-source-msg") + + await channel.send(self._streamed_chunk("He", is_final=False)) + assert await store.get_thread_id("feishu", "chat-1", topic_id="om-root") is None + + store.writes.clear() + await channel.send(self._streamed_chunk("Hello", is_final=False)) + assert store.writes == ["om-root"] + assert await store.get_thread_id("feishu", "chat-1", topic_id="om-root") == "thread-1" + + _run(go()) + class TestFeishuSendFileSuccessChecks: """``send_file`` uploads via ``_upload_image``/``_upload_file`` (which already