fix(feishu): keep streaming replies in existing topics

Made-with: Cursor

nanobot/channels/feishu.py

@@ -1174,6 +1174,17 @@ class FeishuChannel(BaseChannel):
         """Return whether a group reply should create a Feishu thread/topic."""
         return metadata.get("chat_type", "group") == "group" and self.config.reply_to_message
 
+    def _thread_reply_target(self, metadata: dict[str, Any]) -> str | None:
+        """Return the message_id that should receive a Reply API response."""
+        if metadata.get("chat_type", "group") != "group":
+            return None
+        message_id = metadata.get("message_id")
+        if not message_id:
+            return None
+        if metadata.get("thread_id") or self.config.reply_to_message:
+            return message_id
+        return None
+
     def _send_message_sync(
         self, receive_id_type: str, receive_id: str, msg_type: str, content: str
     ) -> str | None:
@@ -1415,15 +1426,14 @@ class FeishuChannel(BaseChannel):
                     {"config": {"wide_screen_mode": True}, "elements": chunk},
                     ensure_ascii=False,
                 )
-                # Fallback: only create a group thread/topic when reply-to-message
+                # Fallback replies stay in existing topics, but only create a
-                # is enabled. Otherwise use the plain create-message API.
+                # new topic when reply-to-message is enabled.
-                _f_msg = meta.get("message_id")
+                fallback_msg_id = self._thread_reply_target(meta)
-                fallback_msg_id = _f_msg if self._should_use_reply_in_thread(meta) else None
                 if fallback_msg_id:
                     await loop.run_in_executor(
                         None, lambda: self._reply_message_sync(
                             fallback_msg_id, "interactive", card,
-                            reply_in_thread=True,
+                            reply_in_thread=self._should_use_reply_in_thread(meta),
                         ),
                     )
                 else:
@@ -1443,10 +1453,10 @@ class FeishuChannel(BaseChannel):
 
         now = time.monotonic()
         if buf.card_id is None:
-            # Send the streaming card as a group thread/topic reply only when
+            # Use the Reply API for existing topics, and only create new topics
-            # reply-to-message is enabled.
+            # when reply-to-message is enabled.
             use_reply_in_thread = self._should_use_reply_in_thread(meta)
-            reply_msg_id = meta.get("message_id") if use_reply_in_thread else None
+            reply_msg_id = self._thread_reply_target(meta)
             card_id = await loop.run_in_executor(
                 None,
                 lambda: self._create_streaming_card_sync(
@@ -1497,20 +1507,20 @@ class FeishuChannel(BaseChannel):
                     )
                     return
                 # No active streaming card — send as a regular interactive card
-                # with the same 🔧 prefix style. Only create a group thread/topic
+                # with the same 🔧 prefix style. Existing topics stay threaded;
-                # when reply-to-message is enabled.
+                # new topics are created only when reply-to-message is enabled.
                 card = json.dumps(
                     {"config": {"wide_screen_mode": True}, "elements": [
                         {"tag": "markdown", "content": self._format_tool_hint_delta(hint)},
                     ]},
                     ensure_ascii=False,
                 )
-                _th_msg_id = msg.metadata.get("message_id")
+                _th_msg_id = self._thread_reply_target(msg.metadata)
-                if _th_msg_id and self._should_use_reply_in_thread(msg.metadata):
+                if _th_msg_id:
                     await loop.run_in_executor(
                         None, lambda: self._reply_message_sync(
                             _th_msg_id, "interactive", card,
-                            reply_in_thread=True,
+                            reply_in_thread=self._should_use_reply_in_thread(msg.metadata),
                         ),
                     )
                 else:

tests/channels/test_feishu_streaming.py

@@ -165,6 +165,26 @@ class TestSendDelta:
         ch._client.im.v1.message.create.assert_called_once()
         ch._client.im.v1.message.reply.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_group_delta_keeps_existing_topic_when_reply_disabled(self):
+        ch = _make_channel(reply_to_message=False)
+        ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_new")
+        reply_resp = MagicMock()
+        reply_resp.success.return_value = True
+        ch._client.im.v1.message.reply.return_value = reply_resp
+        ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
+
+        await ch.send_delta(
+            "oc_chat1",
+            "Hello ",
+            metadata={"message_id": "om_001", "chat_type": "group", "thread_id": "ot_001"},
+        )
+
+        ch._client.im.v1.message.reply.assert_called_once()
+        ch._client.im.v1.message.create.assert_not_called()
+        request = ch._client.im.v1.message.reply.call_args[0][0]
+        assert request.request_body.reply_in_thread is not True
+
     @pytest.mark.asyncio
     async def test_group_delta_replies_in_thread_when_reply_enabled(self):
         ch = _make_channel(reply_to_message=True)
@@ -258,6 +278,32 @@ class TestSendDelta:
         ch._client.im.v1.message.create.assert_called_once()
         ch._client.im.v1.message.reply.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_stream_end_fallback_keeps_existing_topic_when_reply_disabled(self):
+        ch = _make_channel(reply_to_message=False)
+        ch._stream_bufs["om_001"] = _FeishuStreamBuf(
+            text="Fallback content", card_id=None, sequence=0, last_edit=0.0,
+        )
+        reply_resp = MagicMock()
+        reply_resp.success.return_value = True
+        ch._client.im.v1.message.reply.return_value = reply_resp
+
+        await ch.send_delta(
+            "oc_chat1",
+            "",
+            metadata={
+                "_stream_end": True,
+                "message_id": "om_001",
+                "chat_type": "group",
+                "thread_id": "ot_001",
+            },
+        )
+
+        ch._client.im.v1.message.reply.assert_called_once()
+        ch._client.im.v1.message.create.assert_not_called()
+        request = ch._client.im.v1.message.reply.call_args[0][0]
+        assert request.request_body.reply_in_thread is not True
+
     @pytest.mark.asyncio
     async def test_stream_end_fallback_group_replies_when_reply_enabled(self):
         ch = _make_channel(reply_to_message=True)
@@ -406,6 +452,30 @@ class TestToolHintInlineStreaming:
         ch._client.im.v1.message.create.assert_called_once()
         ch._client.im.v1.message.reply.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_tool_hint_keeps_existing_topic_when_reply_disabled(self):
+        ch = _make_channel(reply_to_message=False)
+        reply_resp = MagicMock()
+        reply_resp.success.return_value = True
+        ch._client.im.v1.message.reply.return_value = reply_resp
+
+        msg = OutboundMessage(
+            channel="feishu", chat_id="oc_chat1",
+            content='read_file("path")',
+            metadata={
+                "_tool_hint": True,
+                "message_id": "om_001",
+                "chat_type": "group",
+                "thread_id": "ot_001",
+            },
+        )
+        await ch.send(msg)
+
+        ch._client.im.v1.message.reply.assert_called_once()
+        ch._client.im.v1.message.create.assert_not_called()
+        request = ch._client.im.v1.message.reply.call_args[0][0]
+        assert request.request_body.reply_in_thread is not True
+
     @pytest.mark.asyncio
     async def test_tool_hint_group_replies_when_reply_enabled(self):
         ch = _make_channel(reply_to_message=True)