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 <willem.jiang@gmail.com>

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

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

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 <name>` 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/<name>/chats/<thread_id>`; `lead_agent` omits it and stays on the ordinary chat route. 

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.

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