fix(wecombot): stop empty bubbles and premature stream close on blank final chunks (#2561)

* fix(pipeline): stop emitting empty final assistant messages

An empty final streaming chunk was appended as an empty message chain when no sandbox outbox attachment was present, so platforms received an empty reply. Only append the chain when attachments were actually collected, and gate the "Call ..." tool notice behind output.misc.track-function-calls so it no longer becomes the final chunk of a stream.

* fix(wecombot): do not close stream on a blank final snapshot

A blank final snapshot closed the WeCom stream with an empty bubble and pushed the real answer into a separate reply_text message. Skip blank final snapshots and keep the session open so the following non-blank chunk can finalize it. Covered by new regression tests.

src/langbot/libs/wecom_ai_bot_api/ws_client.py

@@ -687,6 +687,18 @@ class WecomBotWsClient:
                 if not _re.sub(r'[\s​‌‍]', '', next_content):
                     return True
 
+            # A blank *final* snapshot would close the stream with an empty
+            # bubble and strand the real answer in a separate reply_text
+            # message. Keep the session open so a following non-blank chunk can
+            # finalize it. The non-final branch above already skips blank
+            # snapshots; final snapshots with earlier content are non-blank
+            # here because ``next_content`` falls back to the previous content.
+            if is_final:
+                import re as _re
+
+                if not _re.sub(r'[\s\u200b\u200c\u200d\ufeff]', '', next_content):
+                    return True
+
             # Generate feedback_id for final chunk
             feedback_id = ''
             if is_final:

src/langbot/pkg/pipeline/wrapper/wrapper.py

@@ -161,26 +161,35 @@ class ResponseWrapper(stage.PipelineStage):
                     elif (
                         isinstance(result, provider_message.MessageChunk) and result.is_final and not result.tool_calls
                     ):
-                        # Final streaming chunk with no text content but
-                        # possibly carrying sandbox outbox attachments.
+                        # Final streaming chunk with no text content. It may still
+                        # carry sandbox outbox attachments; deliver them when present.
+                        # Otherwise emit nothing: appending an empty message chain
+                        # would be sent to the platform as an empty bubble and, on
+                        # streaming adapters (e.g. WeCom), would also close the
+                        # stream so the real answer arrives as a separate message.
                         reply_chain = platform_message.MessageChain([])
                         await self._append_outbound_attachments(query, reply_chain)
-                        query.resp_message_chain.append(reply_chain)
-                        yield entities.StageProcessResult(
-                            result_type=entities.ResultType.CONTINUE,
-                            new_query=query,
-                        )
+                        if len(reply_chain) > 0:
+                            query.resp_message_chain.append(reply_chain)
+                            yield entities.StageProcessResult(
+                                result_type=entities.ResultType.CONTINUE,
+                                new_query=query,
+                            )
 
                     if result.tool_calls is not None and len(result.tool_calls) > 0:  # 有函数调用
                         function_names = [tc.function.name for tc in result.tool_calls]
 
                         reply_text = f'Call {".".join(function_names)}...'
 
-                        query.resp_message_chain.append(
-                            platform_message.MessageChain([platform_message.Plain(text=reply_text)])
-                        )
-
+                        # Only surface the tool-call notice when the pipeline option
+                        # is enabled. Emitting it unconditionally would become the
+                        # final streaming chunk, closing the stream early and pushing
+                        # the real answer into a separate message.
                         if query.pipeline_config['output']['misc']['track-function-calls']:
+                            query.resp_message_chain.append(
+                                platform_message.MessageChain([platform_message.Plain(text=reply_text)])
+                            )
+
                             event = events.NormalMessageResponded(
                                 launcher_type=query.launcher_type.value,
                                 launcher_id=query.launcher_id,

tests/unit_tests/platform/test_wecombot_stream_empty_final.py

@@ -0,0 +1,119 @@
+"""Regression tests for WeCom AI Bot streaming chunks with blank content.
+
+A blank *final* snapshot used to be forwarded to the client as-is. On WeCom the
+stream frame replaces the displayed bubble, so this closed the stream with an
+empty bubble and stranded the model's real answer in a separate ``reply_text``
+message (observed as "empty message, then the real reply").
+
+These tests lock in the fixed behaviour:
+
+* blank snapshots are never sent, final or not;
+* a blank final snapshot does not close the stream session, so a following
+  non-blank chunk can still finalize it;
+* a blank final snapshot with earlier content still finalizes with that
+  content (the previous-content fallback keeps it non-blank).
+"""
+
+import pytest
+
+import langbot.pkg.core.app  # noqa: F401
+from langbot.libs.wecom_ai_bot_api.ws_client import WecomBotWsClient
+
+
+class Logger:
+    def __init__(self):
+        self.warnings = []
+        self.errors = []
+
+    async def warning(self, message):
+        self.warnings.append(message)
+
+    async def error(self, message):
+        self.errors.append(message)
+
+    async def info(self, message):
+        return None
+
+
+class RecordingClient(WecomBotWsClient):
+    """A client that records reply frames instead of touching a socket."""
+
+    def __init__(self):
+        super().__init__(bot_id='bot', secret='secret', logger=Logger())
+        self.replies = []
+
+    async def _send_reply(self, req_id, body, cmd='aibot_respond_msg'):
+        self.replies.append((cmd, req_id, body))
+        return {'errcode': 0}
+
+    def seed_stream(self, msg_id='msg-1'):
+        self._stream_ids[msg_id] = 'req-1|stream-1'
+        self._stream_sessions[msg_id] = {
+            'req_id': 'req-1',
+            'stream_id': 'stream-1',
+            'msg_id': msg_id,
+            'user_id': 'user-1',
+            'chat_id': 'chat-1',
+            'chat_type': 'single',
+        }
+
+    def stream_still_open(self, msg_id='msg-1'):
+        return msg_id in self._stream_ids
+
+    def stream_frames(self):
+        return [body for _cmd, _req, body in self.replies if body.get('msgtype') == 'stream']
+
+
+@pytest.mark.asyncio
+async def test_blank_final_chunk_does_not_close_stream():
+    client = RecordingClient()
+    client.seed_stream()
+
+    ok = await client.push_stream_chunk('msg-1', '', is_final=True)
+
+    assert ok is True
+    assert client.stream_frames() == []
+    assert client.stream_still_open()
+
+
+@pytest.mark.asyncio
+async def test_blank_non_final_chunk_is_skipped():
+    client = RecordingClient()
+    client.seed_stream()
+
+    ok = await client.push_stream_chunk('msg-1', '   \u200b', is_final=False)
+
+    assert ok is True
+    assert client.stream_frames() == []
+    assert client.stream_still_open()
+
+
+@pytest.mark.asyncio
+async def test_blank_final_after_content_finalizes_with_previous_content():
+    client = RecordingClient()
+    client.seed_stream()
+
+    await client.push_stream_chunk('msg-1', 'hello', is_final=False)
+    ok = await client.push_stream_chunk('msg-1', '', is_final=True)
+
+    frames = client.stream_frames()
+    assert ok is True
+    assert len(frames) == 2
+    assert frames[-1]['stream']['content'] == 'hello'
+    assert frames[-1]['stream']['finish'] is True
+    assert not client.stream_still_open()
+
+
+@pytest.mark.asyncio
+async def test_real_final_chunk_is_sent_and_closes_stream():
+    client = RecordingClient()
+    client.seed_stream()
+
+    ok = await client.push_stream_chunk('msg-1', 'the answer', is_final=True)
+
+    frames = client.stream_frames()
+    assert ok is True
+    assert len(frames) == 1
+    assert frames[-1]['stream']['content'] == 'the answer'
+    assert frames[-1]['stream']['finish'] is True
+    assert not client.stream_still_open()