fix(feishu): delete truncated stream previews on fallback

Feishu had no delete_message, so a failed finalize-edit plus fallback
send left the truncated edit bubble next to the full final. Implement
the SDK delete and thread the fallback send to the originating message.

(cherry picked from commit c61add84ad40b1bc39288405a0a05b4f621e00fc)

gateway/stream_consumer_fallback.py

@@ -123,7 +123,8 @@ class StreamFallbackMixin:
         sent_any_chunk = False
         for chunk in chunks:
             result = await self._send_with_flood_retry(
-                content=chunk, retry_log="Flood control on fallback send, retrying in %.1fs")
+                content=chunk, reply_to=self._initial_reply_to_id,
+                retry_log="Flood control on fallback send, retrying in %.1fs")
             if not result or not result.success:
                 # Partial continuation landed: do NOT set _final_response_sent (the
                 # gateway must still deliver the full answer); _already_sent only

plugins/platforms/feishu/adapter.py

@@ -64,7 +64,8 @@ _LARK_SDK_IMPORTS = (
     ("lark_oapi.api.im.v1", (
         "CreateFileRequest", "CreateFileRequestBody", "CreateImageRequest", "CreateImageRequestBody",
         "CreateMessageRequest", "CreateMessageRequestBody", "GetChatRequest", "GetMessageRequest",
-        "GetMessageResourceRequest", "P2ImMessageMessageReadV1", "ReplyMessageRequest", "ReplyMessageRequestBody",
+        "DeleteMessageRequest", "GetMessageResourceRequest", "P2ImMessageMessageReadV1",
+        "ReplyMessageRequest", "ReplyMessageRequestBody",
         "UpdateMessageRequest", "UpdateMessageRequestBody",
     )),
     ("lark_oapi.core", ("AccessTokenType", "HttpMethod")),
@@ -1723,6 +1724,23 @@ class FeishuAdapter(BasePlatformAdapter):
             logger.error("[Feishu] Failed to edit message %s: %s", message_id, exc, exc_info=True)
             return SendResult(success=False, error=str(exc))
 
+    async def delete_message(self, chat_id: str, message_id: str) -> bool:
+        """Delete a bot-posted message so stream-consumer fallback/fresh-final can
+        remove a truncated edit bubble instead of leaving it next to the full send.
+
+        Feishu has no ``delete_message`` today, so ``_delete_previews`` no-ops and a
+        failed finalize-edit + fallback send (#103068) keeps both bubbles.
+        """
+        if not self._client or not message_id:
+            return False
+        try:
+            request = self._build_delete_message_request(message_id)
+            response = await self._run_blocking(self._client.im.v1.message.delete, request)
+            return self._response_succeeded(response)
+        except Exception:
+            logger.debug("[Feishu] Failed to delete message %s", message_id, exc_info=True)
+            return False
+
     # Template attrs for the shared _format_exec_approval core. The card
     # header carries the title, so the text core starts at the code fence.
     _EA_HEADER = ""
@@ -3973,6 +3991,10 @@ class FeishuAdapter(BasePlatformAdapter):
     def _build_update_message_request(message_id: str, request_body: Any) -> Any:
         return _sdk_build(UpdateMessageRequest, message_id=message_id, request_body=request_body)
 
+    @staticmethod
+    def _build_delete_message_request(message_id: str) -> Any:
+        return _sdk_build(DeleteMessageRequest, message_id=message_id)
+
     @staticmethod
     def _build_create_message_body(*, receive_id: str, msg_type: str, content: str, uuid_value: str) -> Any:
         return _sdk_build(

tests/gateway/test_feishu.py

@@ -273,6 +273,42 @@ class TestFeishuAdapterMessaging(unittest.TestCase):
         )
 
 
+@patch.dict(os.environ, {}, clear=True)
+class TestDeleteMessage(unittest.TestCase):
+    def test_delete_message_calls_im_message_delete(self):
+        from gateway.config import PlatformConfig
+        from plugins.platforms.feishu.adapter import FeishuAdapter
+
+        captured = {"ids": []}
+
+        class _MessageAPI:
+            def delete(self, request):
+                captured["ids"].append(getattr(request, "message_id", None))
+                return SimpleNamespace(success=lambda: True)
+
+        adapter = FeishuAdapter(PlatformConfig())
+        adapter._client = SimpleNamespace(
+            im=SimpleNamespace(v1=SimpleNamespace(message=_MessageAPI()))
+        )
+
+        async def _direct(func, *args, **kwargs):
+            return func(*args, **kwargs)
+
+        with patch.object(adapter, "_run_blocking", side_effect=_direct):
+            ok = asyncio.run(adapter.delete_message("oc_chat", "om_preview"))
+
+        self.assertTrue(ok)
+        self.assertEqual(captured["ids"], ["om_preview"])
+
+    def test_delete_message_returns_false_when_disconnected(self):
+        from gateway.config import PlatformConfig
+        from plugins.platforms.feishu.adapter import FeishuAdapter
+
+        adapter = FeishuAdapter(PlatformConfig())
+        adapter._client = None
+        self.assertFalse(asyncio.run(adapter.delete_message("oc_chat", "om_preview")))
+
+
 class TestAdapterModule(unittest.TestCase):
     def test_load_settings_uses_sdk_defaults_for_invalid_ws_reconnect_values(self):
         from plugins.platforms.feishu.adapter import FeishuAdapter

tests/gateway/test_telegram_final_delivery.py

@@ -149,6 +149,28 @@ async def test_complete_preview_survives_long_flood_fallback_failure(monkeypatch
     assert consumer.delivered_final_matches("Final answer") is True
 
 
+@pytest.mark.asyncio
+async def test_full_fallback_resend_threads_and_deletes_truncated_preview():
+    """When the visible prefix is not a prefix of the final (Feishu post vs
+    cleaned text), fallback resends the whole answer. Thread it to the user
+    message and delete the truncated edit bubble (#103068)."""
+    adapter = _adapter()
+    adapter.send.return_value = SendResult(success=True, message_id="full-1")
+
+    consumer = GatewayStreamConsumer(
+        adapter, "chat-1", initial_reply_to_id="user-1",
+    )
+    consumer._message_id = "preview-1"
+    consumer._last_sent_text = "truncated snapshot"
+    consumer._already_sent = True
+    consumer._fallback_final_send = True
+
+    await consumer._send_fallback_final("full completed answer that does not start with the snapshot")
+
+    assert adapter.send.await_args.kwargs["reply_to"] == "user-1"
+    adapter.delete_message.assert_awaited()
+
+
 @pytest.mark.asyncio
 async def test_telegram_long_flood_result_keeps_retry_after():
     """The real adapter contract preserves the server delay for consumers."""