diff --git a/gateway/platforms/feishu.py b/gateway/platforms/feishu.py index 718f01e9954d..393abd5ac713 100644 --- a/gateway/platforms/feishu.py +++ b/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 diff --git a/gateway/run.py b/gateway/run.py index 19dc5eae74a8..1d5cb84c4207 100644 --- a/gateway/run.py +++ b/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 diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index c0ab907100ef..213427bc61c9 100644 --- a/gateway/stream_consumer.py +++ b/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: diff --git a/tests/gateway/test_feishu.py b/tests/gateway/test_feishu.py index f21b7dcef829..68e5d229b638 100644 --- a/tests/gateway/test_feishu.py +++ b/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 diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py index 6878ddcab4dd..5024ecda7805 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/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."""