fix(feishu): keep progress messages in topic threads

gateway/platforms/feishu.py

@@ -3907,7 +3907,8 @@ async def _send_raw_message(
         reply_to: Optional[str],
         metadata: Optional[Dict[str, Any]],
     ) -> Any:
-        reply_in_thread = bool((metadata or {}).get("thread_id"))
+        thread_id = (metadata or {}).get("thread_id")
+        reply_in_thread = bool(thread_id)
         if reply_to:
             body = self._build_reply_message_body(
                 content=payload,
@@ -3918,13 +3919,15 @@ async def _send_raw_message(
             request = self._build_reply_message_request(reply_to, body)
             return await asyncio.to_thread(self._client.im.v1.message.reply, request)
 
+        receive_id = str(thread_id) if thread_id else chat_id
+        receive_id_type = "thread_id" if thread_id else "chat_id"
         body = self._build_create_message_body(
-            receive_id=chat_id,
+            receive_id=receive_id,
             msg_type=msg_type,
             content=payload,
             uuid_value=str(uuid.uuid4()),
         )
-        request = self._build_create_message_request("chat_id", body)
+        request = self._build_create_message_request(receive_id_type, body)
         return await asyncio.to_thread(self._client.im.v1.message.create, request)
 
     @staticmethod

gateway/run.py

@@ -10316,6 +10316,11 @@ def _run_still_current() -> bool:
                         chat_id=source.chat_id,
                         config=_consumer_cfg,
                         metadata=_thread_metadata,
+                        initial_reply_to_id=(
+                            event_message_id
+                            if source.platform == Platform.FEISHU and source.thread_id
+                            else None
+                        ),
                     )
             except Exception as _sc_err:
                 logger.debug("Proxy: could not set up stream consumer: %s", _sc_err)
@@ -11063,6 +11068,11 @@ def run_sync():
                             chat_id=source.chat_id,
                             config=_consumer_cfg,
                             metadata={"thread_id": _progress_thread_id} if _progress_thread_id else None,
+                            initial_reply_to_id=(
+                                event_message_id
+                                if source.platform == Platform.FEISHU and source.thread_id
+                                else None
+                            ),
                             on_new_message=(
                                 (lambda: progress_queue.put(("__reset__",)))
                                 if progress_queue is not None

gateway/stream_consumer.py

@@ -92,11 +92,16 @@ def __init__(
         config: Optional[StreamConsumerConfig] = None,
         metadata: Optional[dict] = None,
         on_new_message: Optional[callable] = None,
+        initial_reply_to_id: Optional[str] = None,
     ):
         self.adapter = adapter
         self.chat_id = chat_id
         self.cfg = config or StreamConsumerConfig()
         self.metadata = metadata
+        # Message id that triggered this stream.  The first platform send must
+        # reply to it so topic/thread-aware adapters do not create a top-level
+        # message before an editable message id exists.
+        self._initial_reply_to_id = initial_reply_to_id
         # Fired whenever a fresh content bubble is created on the platform
         # (first-send of a new message, commentary, overflow chunk, or
         # fallback continuation). The gateway uses this to linearize the
@@ -541,7 +546,7 @@ async def _send_new_chunk(self, text: str, reply_to_id: Optional[str]) -> Option
             result = await self.adapter.send(
                 chat_id=self.chat_id,
                 content=text,
-                reply_to=reply_to_id,
+                reply_to=reply_to_id or self._initial_reply_to_id,
                 metadata=meta,
             )
             if result.success and result.message_id:
@@ -983,6 +988,7 @@ async def _send_or_edit(self, text: str, *, finalize: bool = False) -> bool:
                 result = await self.adapter.send(
                     chat_id=self.chat_id,
                     content=text,
+                    reply_to=self._initial_reply_to_id,
                     metadata=self.metadata,
                 )
                 if result.success:

tests/gateway/test_feishu.py

@@ -1855,6 +1855,48 @@ async def _direct(func, *args, **kwargs):
         self.assertEqual(result.message_id, "om_reply")
         self.assertTrue(captured["request"].request_body.reply_in_thread)
 
+    @patch.dict(os.environ, {}, clear=True)
+    def test_send_uses_thread_id_receive_type_when_no_reply_to(self):
+        from gateway.config import PlatformConfig
+        from gateway.platforms.feishu import FeishuAdapter
+
+        adapter = FeishuAdapter(PlatformConfig())
+        captured = {}
+
+        class _MessageAPI:
+            def create(self, request):
+                captured["request"] = request
+                return SimpleNamespace(
+                    success=lambda: True,
+                    data=SimpleNamespace(message_id="om_thread_msg"),
+                )
+
+        adapter._client = SimpleNamespace(
+            im=SimpleNamespace(
+                v1=SimpleNamespace(
+                    message=_MessageAPI(),
+                )
+            )
+        )
+
+        async def _direct(func, *args, **kwargs):
+            return func(*args, **kwargs)
+
+        with patch("gateway.platforms.feishu.asyncio.to_thread", side_effect=_direct):
+            result = asyncio.run(
+                adapter.send(
+                    chat_id="oc_chat",
+                    content="tool progress",
+                    reply_to=None,
+                    metadata={"thread_id": "omt-thread"},
+                )
+            )
+
+        self.assertTrue(result.success)
+        self.assertEqual(result.message_id, "om_thread_msg")
+        self.assertEqual(captured["request"].receive_id_type, "thread_id")
+        self.assertEqual(captured["request"].request_body.receive_id, "omt-thread")
+
     @patch.dict(os.environ, {}, clear=True)
     def test_send_retries_transient_failure(self):
         from gateway.config import PlatformConfig

tests/gateway/test_stream_consumer.py

@@ -189,6 +189,56 @@ async def test_first_send_strips_media(self):
         assert "MEDIA:" not in sent_text
         assert "Here is your image" in sent_text
 
+    @pytest.mark.asyncio
+    async def test_first_send_uses_initial_reply_to_id(self):
+        """Initial streaming send replies to the inbound message/thread root."""
+        adapter = MagicMock()
+        send_result = SimpleNamespace(success=True, message_id="msg_1")
+        adapter.send = AsyncMock(return_value=send_result)
+        adapter.MAX_MESSAGE_LENGTH = 4096
+
+        consumer = GatewayStreamConsumer(
+            adapter,
+            "chat_123",
+            initial_reply_to_id="om_inbound",
+        )
+        await consumer._send_or_edit("Starting streamed response")
+
+        adapter.send.assert_called_once()
+        assert adapter.send.call_args[1]["reply_to"] == "om_inbound"
+
+    @pytest.mark.asyncio
+    async def test_first_send_without_initial_reply_to_stays_unanchored(self):
+        """Default streaming behavior should not force reply anchoring."""
+        adapter = MagicMock()
+        send_result = SimpleNamespace(success=True, message_id="msg_1")
+        adapter.send = AsyncMock(return_value=send_result)
+        adapter.MAX_MESSAGE_LENGTH = 4096
+
+        consumer = GatewayStreamConsumer(adapter, "chat_123")
+        await consumer._send_or_edit("Starting streamed response")
+
+        adapter.send.assert_called_once()
+        assert adapter.send.call_args[1].get("reply_to") is None
+
+    @pytest.mark.asyncio
+    async def test_first_overflow_chunk_uses_initial_reply_to_id(self):
+        """Overflow first chunks should also preserve the inbound thread."""
+        adapter = MagicMock()
+        send_result = SimpleNamespace(success=True, message_id="msg_1")
+        adapter.send = AsyncMock(return_value=send_result)
+        adapter.MAX_MESSAGE_LENGTH = 4096
+
+        consumer = GatewayStreamConsumer(
+            adapter,
+            "chat_123",
+            initial_reply_to_id="om_inbound",
+        )
+        await consumer._send_new_chunk("chunk text", None)
+
+        adapter.send.assert_called_once()
+        assert adapter.send.call_args[1]["reply_to"] == "om_inbound"
+
     @pytest.mark.asyncio
     async def test_edit_strips_media(self):
         """Edit call removes MEDIA: tags from visible text."""