From a2b7f6b7bb724cda37429d0f9fa55157f89e940c Mon Sep 17 00:00:00 2001 From: Hyu Date: Sat, 10 Oct 2026 18:02:42 +0800 Subject: [PATCH] fix: stabilize native streaming and improve bot controls --- src/langbot/libs/wecom_ai_bot_api/api.py | 10 +- .../libs/wecom_ai_bot_api/ws_client.py | 17 ++-- src/langbot/pkg/box/service.py | 4 + .../pkg/pipeline/process/handlers/chat.py | 16 +++- .../pkg/pipeline/process/stream_results.py | 44 +++++++++ .../pkg/platform/adapters/wecom/manifest.yaml | 6 +- .../pkg/platform/adapters/wecombot/adapter.py | 21 ++++- .../platform/adapters/wecombot/manifest.yaml | 41 ++++++--- .../adapters/wecombot/message_converter.py | 7 +- .../platform/adapters/wecombot/stream_text.py | 31 +++++++ .../platform/adapters/wecomcs/manifest.yaml | 6 +- .../box/test_attachment_host_capabilities.py | 23 +++++ .../unit_tests/pipeline/test_chat_handler.py | 51 +++++++++++ .../pipeline/test_stream_results.py | 91 +++++++++++++++++++ .../platform/test_lark_eba_adapter.py | 1 + .../platform/test_native_stream_completion.py | 85 +++++++++++++++++ .../platform/test_wecombot_eba_adapter.py | 6 +- .../test_wecombot_stream_empty_final.py | 58 ++++++++++++ .../platform/test_wecombot_stream_text.py | 30 ++++++ .../platform/test_wecombot_template_card.py | 3 + web/src/app/home/bots/BotDetailContent.tsx | 39 +++++--- .../components/home-sidebar/HomeSidebar.tsx | 4 +- .../monitoring/hooks/useMonitoringData.ts | 4 +- web/src/app/home/monitoring/page.tsx | 8 +- web/src/i18n/locales/en-US.ts | 4 + web/src/i18n/locales/es-ES.ts | 4 + web/src/i18n/locales/ja-JP.ts | 4 + web/src/i18n/locales/ru-RU.ts | 4 + web/src/i18n/locales/th-TH.ts | 4 + web/src/i18n/locales/vi-VN.ts | 4 + web/src/i18n/locales/zh-Hans.ts | 4 + web/src/i18n/locales/zh-Hant.ts | 4 + 32 files changed, 583 insertions(+), 55 deletions(-) create mode 100644 src/langbot/pkg/pipeline/process/stream_results.py create mode 100644 src/langbot/pkg/platform/adapters/wecombot/stream_text.py create mode 100644 tests/unit_tests/box/test_attachment_host_capabilities.py create mode 100644 tests/unit_tests/pipeline/test_stream_results.py create mode 100644 tests/unit_tests/platform/test_native_stream_completion.py create mode 100644 tests/unit_tests/platform/test_wecombot_stream_text.py diff --git a/src/langbot/libs/wecom_ai_bot_api/api.py b/src/langbot/libs/wecom_ai_bot_api/api.py index 785e82f28..d72ffe22b 100644 --- a/src/langbot/libs/wecom_ai_bot_api/api.py +++ b/src/langbot/libs/wecom_ai_bot_api/api.py @@ -2123,7 +2123,7 @@ class WecomBotClient: Args: msg_id: 原始企业微信消息 ID。 - content: 模型产生的片段内容。 + content: 流水线传入的累计正文快照。 is_final: 是否为最终片段。 Returns: @@ -2138,12 +2138,8 @@ class WecomBotClient: return False previous_content = self._stream_last_content.get(msg_id, '') - if previous_content and content.startswith(previous_content): - next_content = content - elif previous_content and not content: - next_content = previous_content - else: - next_content = previous_content + content if previous_content else content + # Pipeline delivery provides a snapshot, not a provider delta. + next_content = content or previous_content if len(next_content) > _MAX_STREAM_CONTENT_CHARS: next_content = next_content[-_MAX_STREAM_CONTENT_CHARS:] diff --git a/src/langbot/libs/wecom_ai_bot_api/ws_client.py b/src/langbot/libs/wecom_ai_bot_api/ws_client.py index 3f4cb0504..d61659950 100644 --- a/src/langbot/libs/wecom_ai_bot_api/ws_client.py +++ b/src/langbot/libs/wecom_ai_bot_api/ws_client.py @@ -11,6 +11,7 @@ Official Node.js SDK: https://github.com/WecomTeam/aibot-node-sdk from __future__ import annotations import asyncio +import hashlib import json import secrets import time @@ -134,6 +135,9 @@ class WecomBotWsClient: self._stream_ids: dict[str, str] = {} # msg_id -> req_id|stream_id # Dedup: skip sending when content hasn't changed self._stream_last_content: dict[str, str] = {} # msg_id -> last content sent + # Preserve completion identity after removing active stream mappings. + # A runner may emit both a final delta and a completed message snapshot. + self._completed_stream_content: dict[str, str] = {} # Stream session info for feedback tracking self._stream_sessions: dict[str, dict] = {} # msg_id -> session info # Feedback tracking: feedback_id -> session info @@ -661,16 +665,15 @@ class WecomBotWsClient: """ key = self._stream_ids.get(msg_id) if not key: + if is_final and self._completed_stream_content.get(msg_id) == hashlib.sha256(content.encode()).hexdigest(): + return True return False req_id, stream_id = key.split('|', 1) try: previous_content = self._stream_last_content.get(msg_id, '') - if previous_content and content.startswith(previous_content): - next_content = content - elif previous_content and not content: - next_content = previous_content - else: - next_content = previous_content + content if previous_content else content + # Pipeline delivery provides complete snapshots, including edits + # that need not start with the previous text (e.g. reasoning tags). + next_content = content or previous_content if len(next_content) > _MAX_STREAM_CONTENT_CHARS: next_content = next_content[-_MAX_STREAM_CONTENT_CHARS:] @@ -720,6 +723,8 @@ class WecomBotWsClient: await self.reply_stream(req_id, stream_id, next_content, finish=is_final, feedback_id=feedback_id) self._stream_last_content[msg_id] = next_content if is_final: + self._completed_stream_content[msg_id] = hashlib.sha256(next_content.encode()).hexdigest() + self._cap_mapping(self._completed_stream_content, _STREAM_CACHE_MAX) self._stream_ids.pop(msg_id, None) self._stream_last_content.pop(msg_id, None) self._stream_sessions.pop(msg_id, None) diff --git a/src/langbot/pkg/box/service.py b/src/langbot/pkg/box/service.py index a18ebd914..c558cca50 100644 --- a/src/langbot/pkg/box/service.py +++ b/src/langbot/pkg/box/service.py @@ -811,6 +811,10 @@ class BoxService: to the sandbox (and vice-versa). It is ``None`` / not a local dir for E2B and remote runtimes, where we must fall back to the exec channel. """ + # secure_fs relies on POSIX openat/O_NOFOLLOW. On Windows use the + # sandbox execution channel rather than misclassifying EACCES as a link. + if os.open not in os.supports_dir_fd or not hasattr(os, 'O_NOFOLLOW'): + return None root = self._tenant_workspace(self._query_execution_context(query)) if not root or not os.path.isdir(root) or os.path.islink(root): return None diff --git a/src/langbot/pkg/pipeline/process/handlers/chat.py b/src/langbot/pkg/pipeline/process/handlers/chat.py index 2cd681fa1..1bc6143b9 100644 --- a/src/langbot/pkg/pipeline/process/handlers/chat.py +++ b/src/langbot/pkg/pipeline/process/handlers/chat.py @@ -6,6 +6,7 @@ import traceback from .. import handler +from ..stream_results import coalesce_stream_results from ... import entities from ... import plugin_diagnostics @@ -132,7 +133,10 @@ class ChatMessageHandler(handler.MessageHandler): # Use AgentRunOrchestrator to run the agent # This replaces direct runner lookup and PluginRunnerWrapper - async for result in self.ap.agent_run_orchestrator.run_from_query(query): + results = self.ap.agent_run_orchestrator.run_from_query(query) + if is_stream: + results = coalesce_stream_results(results) + async for result in results: has_result = True self._check_response_size(result) @@ -149,6 +153,11 @@ class ChatMessageHandler(handler.MessageHandler): ) result.attachments = attachments + if is_stream and result.tool_calls: + # A model round ending in tool calls does not end the + # reply. Keep the native stream open for the answer. + result = result.model_copy(update={'is_final': False}) + result.resp_message_id = str(resp_message_id) # For streaming mode, pop previous response before adding new chunk @@ -192,6 +201,11 @@ class ChatMessageHandler(handler.MessageHandler): if is_stream: yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query) + if result.is_final and not result.tool_calls: + # Native adapters release their card on completion. + # A distinct later message needs a fresh card identity. + is_create_card = False + resp_message_id = uuid.uuid4() # Log final summary after streaming completes if is_stream: diff --git a/src/langbot/pkg/pipeline/process/stream_results.py b/src/langbot/pkg/pipeline/process/stream_results.py new file mode 100644 index 000000000..7e5e889d6 --- /dev/null +++ b/src/langbot/pkg/pipeline/process/stream_results.py @@ -0,0 +1,44 @@ +"""Coalesce a runner's terminal delta and completed snapshot before delivery.""" + +from collections.abc import AsyncIterable, AsyncIterator + +from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk + + +async def coalesce_stream_results( + results: AsyncIterable[Message | MessageChunk], +) -> AsyncIterator[Message | MessageChunk]: + pending: MessageChunk | None = None + snapshot = '' + async for result in results: + if pending is not None: + # Completion snapshots can add attachments and usage to the final + # delta. Deliver that richer result once, before closing the card. + content = pending.all_content if pending.all_content is not None else pending.content + same_message = ( + isinstance(result, Message) + and result.role == pending.role + and result.tool_calls == pending.tool_calls + and (not content or result.content == content) + ) + if not same_message: + yield pending + pending = None + snapshot = '' + + if isinstance(result, MessageChunk) and isinstance(result.content, (str, type(None))): + # Runner output is already cumulative (unlike raw provider deltas). + # Explicit all_content also supports runners forwarding raw deltas. + # Never guess delta semantics from a missing all_content field. + snapshot = result.all_content if result.all_content is not None else (result.content or snapshot) + result = result.model_copy(update={'all_content': snapshot}) + elif isinstance(result, Message): + snapshot = '' + + if isinstance(result, MessageChunk) and result.is_final: + pending = result + else: + yield result + + if pending is not None: + yield pending diff --git a/src/langbot/pkg/platform/adapters/wecom/manifest.yaml b/src/langbot/pkg/platform/adapters/wecom/manifest.yaml index fd63e8348..a5b11284e 100644 --- a/src/langbot/pkg/platform/adapters/wecom/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/wecom/manifest.yaml @@ -8,9 +8,9 @@ metadata: zh_Hans: 企业微信 zh_Hant: 企業微信 description: - en_US: WeCom application message adapter with event-driven orchestration support - zh_Hans: 企业微信内部应用消息适配器，支持事件驱动编排 - zh_Hant: 企業微信內部應用訊息適配器，支援事件驅動編排 + en_US: Serve employees through a custom WeCom app + zh_Hans: 通过企业自建应用，为内部员工提供服务 + zh_Hant: 透過企業自建應用，為內部員工提供服務 icon: wecom.png spec: diff --git a/src/langbot/pkg/platform/adapters/wecombot/adapter.py b/src/langbot/pkg/platform/adapters/wecombot/adapter.py index 38fb5e8df..1bbd7b179 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/adapter.py +++ b/src/langbot/pkg/platform/adapters/wecombot/adapter.py @@ -148,6 +148,16 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo raw = await self.bot.send_message(str(target_id), content) return platform_events.MessageResult(raw={'result': raw}) + async def reply_to_event( + self, *, context: dict, target_type: str, target_id: str, message: platform_message.MessageChain + ) -> platform_events.MessageResult: + """Bind Agent replies to the originating callback and its loading stream.""" + message_id = ((context.get('delivery') or {}).get('reply_target') or {}).get('message_id') + if context.get('event_type') == 'message.received' and message_id: + source = await self.get_message(target_type, target_id, message_id) + return await self.reply_message(source, message) + return await self.send_message(target_type, target_id, message) + async def reply_message( self, message_source: platform_events.MessageEvent, @@ -160,9 +170,14 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo items = await WecomBotMessageConverter.yiri2target(message) content = self._join_text_components(items) raw = None + from .stream_text import format_stream_text + + content = format_stream_text(content) if not self.config.get('enable-webhook', False) and event.get('req_id'): if content: - raw = await self.bot.reply_text(event.get('req_id'), content) + finished = await self.bot.push_stream_chunk(event.message_id, content, is_final=True) + if not finished: + raw = await self.bot.reply_text(event.get('req_id'), content) for item in self._iter_media_components(items): await self._send_media(self.bot, event.get('req_id'), item) else: @@ -181,7 +196,9 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo if not isinstance(event, WecomBotEvent): raise ValueError('WeComBot reply_message_chunk requires a WecomBotEvent source object') items = await WecomBotMessageConverter.yiri2target(message) - content = self._join_text_components(items) + from .stream_text import format_stream_text + + content = format_stream_text(self._join_text_components(items)) success = await self.bot.push_stream_chunk(event.message_id, content, is_final=is_final) if not success and is_final and not self.config.get('enable-webhook', False) and event.get('req_id'): await self.bot.reply_text(event.get('req_id'), content) diff --git a/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml b/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml index 2b722142f..1c1c8213f 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml @@ -8,9 +8,9 @@ metadata: zh_Hans: 企业微信智能机器人 zh_Hant: 企業微信智慧機器人 description: - en_US: WeCom AI Bot adapter with event-driven orchestration support - zh_Hans: 企业微信智能机器人适配器，支持事件驱动编排、长连接和 Webhook 两种接入方式 - zh_Hant: 企業微信智慧機器人適配器，支援事件驅動編排、長連線和 Webhook 兩種接入方式 + en_US: Answer questions and assist teams in WeCom chats and groups + zh_Hans: 在企业微信单聊和群聊中提供智能问答与协作 + zh_Hant: 在企業微信單聊和群聊中提供智慧問答與協作 icon: wecombot.png spec: @@ -27,10 +27,9 @@ spec: zh_Hans: 一键创建机器人 zh_Hant: 一鍵建立機器人 description: - en_US: 'Scan QR code with WeCom to automatically create a bot and fill in BotId and Secret. Note: Robot Name needs to - be filled in manually.' - zh_Hans: 使用企业微信扫码自动创建机器人并填写 BotId 和 Secret。注意：机器人名称需手动填写。 - zh_Hant: 使用企業微信掃碼自動建立機器人並填寫 BotId 和 Secret。注意：機器人名稱需手動填寫。 + en_US: Scan with WeCom to create a bot and fill in BotId and Secret automatically. + zh_Hans: 使用企业微信扫码自动创建机器人并填写 BotId 和 Secret。 + zh_Hant: 使用企業微信掃碼自動建立機器人並填寫 BotId 和 Secret。 type: qr-code-login login_platform: wecombot required: false @@ -60,8 +59,12 @@ spec: en_US: Robot Name zh_Hans: 机器人名称 zh_Hant: 機器人名稱 + description: + en_US: Optional. Only removes matching @name text in group messages; bot mentions are identified by BotId. + zh_Hans: 可选。仅用于清理群消息中的 @名称 文本，机器人提及通过 BotId 识别。 + zh_Hant: 選填。僅用於清理群訊息中的 @名稱 文字，機器人提及透過 BotId 識別。 type: string - required: true + required: false default: "" - name: enable-webhook label: @@ -103,6 +106,10 @@ spec: type: string required: false default: "" + show_if: + field: enable-webhook + operator: neq + value: true - name: Corpid label: en_US: Corpid @@ -115,6 +122,10 @@ spec: type: string required: false default: "" + show_if: + field: enable-webhook + operator: eq + value: true - name: Token label: en_US: Token @@ -127,18 +138,26 @@ spec: type: string required: false default: "" + show_if: + field: enable-webhook + operator: eq + value: true - name: EncodingAESKey label: en_US: EncodingAESKey zh_Hans: 消息加解密密钥 (EncodingAESKey) zh_Hant: 訊息加解密密鑰 (EncodingAESKey) description: - en_US: Required for webhook mode. Optional for WebSocket mode when encrypted files need to be decrypted. - zh_Hans: Webhook 模式必填。WebSocket 模式下如需解密文件则填写。 - zh_Hant: Webhook 模式必填。WebSocket 模式下如需解密檔案則填寫。 + en_US: Required for webhook mode. Long connections use the decryption key included in each message. + zh_Hans: Webhook 模式必填。长连接模式使用消息自带的解密密钥。 + zh_Hant: Webhook 模式必填。長連線模式使用訊息自帶的解密金鑰。 type: string required: false default: "" + show_if: + field: enable-webhook + operator: eq + value: true - name: enable-stream-reply label: en_US: Enable Stream Reply diff --git a/src/langbot/pkg/platform/adapters/wecombot/message_converter.py b/src/langbot/pkg/platform/adapters/wecombot/message_converter.py index fca3e1f5d..95ce52202 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/message_converter.py +++ b/src/langbot/pkg/platform/adapters/wecombot/message_converter.py @@ -1,6 +1,7 @@ from __future__ import annotations import datetime +import re from langbot.libs.wecom_ai_bot_api.wecombotevent import WecomBotEvent import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter @@ -46,12 +47,14 @@ class WecomBotMessageConverter(abstract_platform_adapter.AbstractMessageConverte platform_message.Source(id=event.message_id, time=datetime.datetime.now()), ] if event.type == 'group' and event.ai_bot_id: + # Group callbacks identify the mentioned bot by ID, independently + # of its display name or the optional text-cleanup configuration. components.append(platform_message.At(target=event.ai_bot_id)) if event.content: content = event.content - if bot_name: - content = content.replace(f'@{bot_name}', '').strip() + if event.type == 'group' and bot_name: + content = re.sub(rf'(? str: + if content.strip().lower() in {'<', ')', content, flags=re.IGNORECASE) + thoughts: list[str] = [] + answers: list[str] = [] + thinking = False + for token in tokens: + if re.fullmatch(r'', token, flags=re.IGNORECASE): + thinking = True + elif re.fullmatch(r'', token, flags=re.IGNORECASE): + thinking = False + elif token.strip(): + (thoughts if thinking else answers).append(token.strip()) + if not thoughts: + return '\n\n'.join(answers) + reasoning = '\n\n'.join(thoughts) + answer = '\n\n'.join(answers) + if thinking and not answer: + return f'\n{reasoning}' + return f'\n{reasoning}\n' + (f'\n\n{answer}' if answer else '') diff --git a/src/langbot/pkg/platform/adapters/wecomcs/manifest.yaml b/src/langbot/pkg/platform/adapters/wecomcs/manifest.yaml index 0ab00eaac..3c8821675 100644 --- a/src/langbot/pkg/platform/adapters/wecomcs/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/wecomcs/manifest.yaml @@ -8,9 +8,9 @@ metadata: zh_Hans: 企业微信客服 zh_Hant: 企業微信客服 description: - en_US: WeCom Customer Service adapter with event-driven orchestration support - zh_Hans: 企业微信客服适配器，支持事件驱动编排，通过统一 Webhook 接收客服会话消息 - zh_Hant: 企業微信客服適配器，支援事件驅動編排，透過統一 Webhook 接收客服會話訊息 + en_US: Support customers on WeChat through WeChat Customer Service + zh_Hans: 通过微信客服，为微信用户提供咨询与售后服务 + zh_Hant: 透過微信客服，為微信使用者提供諮詢與售後服務 icon: wecom.png spec: diff --git a/tests/unit_tests/box/test_attachment_host_capabilities.py b/tests/unit_tests/box/test_attachment_host_capabilities.py new file mode 100644 index 000000000..48acd2862 --- /dev/null +++ b/tests/unit_tests/box/test_attachment_host_capabilities.py @@ -0,0 +1,23 @@ +from unittest.mock import AsyncMock, Mock + +import pytest + +from langbot.pkg.box.service import BoxService +from langbot_plugin.box.client import BoxRuntimeClient +from tests.unit_tests.box.test_box_service import make_app, make_query + + +@pytest.mark.asyncio +async def test_host_without_openat_uses_sandbox_attachment_channel(monkeypatch, tmp_path): + monkeypatch.setattr('os.supports_dir_fd', set()) + service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient)) + service._available = True + service._tenant_workspace = Mock(return_value=str(tmp_path)) + service._read_outbox_host = Mock(side_effect=AssertionError('POSIX host reader must not run')) + service._read_outbox_via_exec = AsyncMock(return_value=[]) + service._clear_outbox = AsyncMock() + query = make_query() + assert service._host_query_dir(service.OUTBOX_SUBDIR, query) is None + assert await service.collect_outbound_attachments(query) == [] + service._read_outbox_via_exec.assert_awaited_once_with(query) + service._clear_outbox.assert_awaited_once_with(query, None) diff --git a/tests/unit_tests/pipeline/test_chat_handler.py b/tests/unit_tests/pipeline/test_chat_handler.py index 439c41491..ce4be91af 100644 --- a/tests/unit_tests/pipeline/test_chat_handler.py +++ b/tests/unit_tests/pipeline/test_chat_handler.py @@ -263,6 +263,57 @@ class TestChatMessageHandlerReal: class TestChatHandlerStreaming: """Tests for streaming behavior.""" + @pytest.mark.asyncio + @pytest.mark.parametrize('with_tool_call', [False, True]) + async def test_terminal_delta_and_snapshot_close_card_once( + self, fake_app, mock_event_ctx, set_runner, with_tool_call + ): + from tests.factories import text_query + from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk + from langbot_plugin.api.entities.builtin.platform.message import File, MessageChain + + fake_app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx) + query = text_query('stream test') + query.adapter = Mock() + query.adapter.is_stream_output_supported = AsyncMock(return_value=True) + query.adapter.create_message_card = AsyncMock() + attachments = MessageChain([File(url='https://example.com/file.txt', name='file.txt')]) + + class StreamRunner: + def __init__(self, app, config): + pass + + async def run(self, query): + if with_tool_call: + yield MessageChunk( + role='assistant', + is_final=True, + tool_calls=[ + {'id': 'call-1', 'type': 'function', 'function': {'name': 'search', 'arguments': '{}'}} + ], + ) + yield MessageChunk(role='assistant', content='hello') + yield MessageChunk(role='assistant', content='hello world', is_final=True) + completed = Message(role='assistant', content='hello world') + completed.attachments = attachments + yield completed + + set_runner(StreamRunner) + handler = get_chat_handler().ChatMessageHandler(fake_app) + handler._ensure_conversation_for_history = AsyncMock() + delivered = [] + async for stage_result in handler.handle(query): + assert stage_result.result_type == get_entities().ResultType.CONTINUE + delivered.append(query.resp_messages[-1]) + assert [result.content for result in delivered if result.content] == ['hello', 'hello world'] + if with_tool_call: + assert delivered[0].tool_calls + assert not delivered[0].is_final + assert sum(result.is_final for result in delivered) == 1 + assert delivered[-1].attachments is attachments + assert len({result.resp_message_id for result in delivered}) == 1 + query.adapter.create_message_card.assert_awaited_once() + @pytest.mark.asyncio async def test_streaming_chunks_collected(self, fake_app, mock_event_ctx, set_runner): """Streaming produces multiple results.""" diff --git a/tests/unit_tests/pipeline/test_stream_results.py b/tests/unit_tests/pipeline/test_stream_results.py new file mode 100644 index 000000000..3643ec2d1 --- /dev/null +++ b/tests/unit_tests/pipeline/test_stream_results.py @@ -0,0 +1,91 @@ +import pytest + +from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk +from langbot_plugin.api.entities.builtin.platform.message import File, MessageChain +from langbot.pkg.pipeline.process.stream_results import coalesce_stream_results + + +async def source(*items): + for item in items: + yield item + + +@pytest.mark.asyncio +async def test_terminal_snapshot_is_delivered_once_with_attachments(): + text = 'reasoninganswer' + completed = Message(role='assistant', content=text) + completed.attachments = MessageChain([File(url='https://example.com/report.txt', name='report.txt')]) + results = [ + item + async for item in coalesce_stream_results( + source( + MessageChunk(role='assistant', content='answer', all_content=text, is_final=True), + completed, + ) + ) + ] + assert results == [completed] + assert results[0].attachments is completed.attachments + + +@pytest.mark.asyncio +@pytest.mark.parametrize('with_snapshot', [False, True]) +async def test_progressive_chunks_are_not_buffered(with_snapshot): + first = MessageChunk(role='assistant', content='hel', all_content='hel') + final = MessageChunk(role='assistant', content='lo', all_content='hello', is_final=True) + completed = Message(role='assistant', content='hello') + stream = coalesce_stream_results(source(first, final, *([completed] if with_snapshot else []))) + assert await anext(stream) == first + assert [item async for item in stream] == [completed if with_snapshot else final] + + +@pytest.mark.asyncio +async def test_completed_only_and_distinct_messages_are_preserved(): + final = MessageChunk(role='assistant', content='first', is_final=True) + other = Message(role='assistant', content='second') + results = [item async for item in coalesce_stream_results(source(final, other))] + assert [item.content for item in results] == ['first', 'second'] + assert [item async for item in coalesce_stream_results(source(other))] == [other] + + +@pytest.mark.asyncio +async def test_blank_terminal_delta_uses_completed_snapshot(): + completed = Message(role='assistant', content='answer') + results = [ + item + async for item in coalesce_stream_results( + source( + MessageChunk(role='assistant', is_final=True), + completed, + ) + ) + ] + assert results == [completed] + + +@pytest.mark.asyncio +async def test_failures_propagate_without_successful_final(): + async def failing(): + yield MessageChunk(role='assistant', content='partial') + raise RuntimeError('provider failed') + + stream = coalesce_stream_results(failing()) + assert (await anext(stream)).content == 'partial' + with pytest.raises(RuntimeError, match='provider failed'): + await anext(stream) + + +@pytest.mark.asyncio +async def test_cumulative_stream_ends_with_last_snapshot_even_when_final_is_blank(): + results = [ + item + async for item in coalesce_stream_results( + source( + MessageChunk(role='assistant', content='hel'), + MessageChunk(role='assistant', content='hello'), + MessageChunk(role='assistant', is_final=True), + ) + ) + ] + assert [item.all_content for item in results] == ['hel', 'hello', 'hello'] + assert results[-1].is_final diff --git a/tests/unit_tests/platform/test_lark_eba_adapter.py b/tests/unit_tests/platform/test_lark_eba_adapter.py index 827a2f4f6..9590fbed8 100644 --- a/tests/unit_tests/platform/test_lark_eba_adapter.py +++ b/tests/unit_tests/platform/test_lark_eba_adapter.py @@ -301,6 +301,7 @@ async def test_lark_event_converter_maps_group_and_private_message(): @pytest.mark.asyncio async def test_lark_adapter_dispatches_and_caches_message_event(): adapter = make_adapter() + adapter._get_bot_open_id = AsyncMock(return_value='bot-open-id') calls: list[platform_events.Event] = [] async def listener(event, adapter): diff --git a/tests/unit_tests/platform/test_native_stream_completion.py b/tests/unit_tests/platform/test_native_stream_completion.py new file mode 100644 index 000000000..9495169f8 --- /dev/null +++ b/tests/unit_tests/platform/test_native_stream_completion.py @@ -0,0 +1,85 @@ +"""Exercise terminal runner snapshots against native adapter delivery methods.""" + +import datetime +import time +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import telegram + +from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk +from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain +from langbot.pkg.pipeline.process.stream_results import coalesce_stream_results + + +@pytest.mark.asyncio +@pytest.mark.parametrize('platform', ['lark', 'dingtalk', 'telegram', 'discord', 'qqofficial', 'wecombot']) +async def test_native_stream_receives_one_terminal_snapshot(platform): + module = __import__(f'tests.unit_tests.platform.test_{platform}_eba_adapter', fromlist=['make_adapter']) + adapter = module.make_adapter() + object.__setattr__(adapter, 'reply_message', AsyncMock()) + source = SimpleNamespace() + if platform == 'lark': + adapter.card_id_dict['response'] = 'card' + adapter.card_sequence_dict['card'] = 0 + adapter._replace_streaming_card = AsyncMock() + source.source_platform_object = None + elif platform == 'dingtalk': + adapter.card_instance_id_dict['response'] = ('card', 'id') + elif platform == 'telegram': + source.source_platform_object = telegram.Update( + 1, + message=telegram.Message( + 1, + datetime.datetime.now(datetime.timezone.utc), + telegram.Chat(1, 'private'), + text='hi', + ), + ) + adapter.bot = SimpleNamespace(send_message_draft=AsyncMock(), send_message=AsyncMock()) + adapter.msg_stream_id['response'] = ('private', 1, False) + elif platform == 'discord': + adapter._stream_buffer['response'] = { + 'channel': SimpleNamespace(send=AsyncMock(return_value=SimpleNamespace(content='hello', edit=AsyncMock()))), + 'sent_message': None, + 'last_content': '', + 'chunk_count': 0, + 'updated_at': time.time(), + } + channel = adapter._stream_buffer['response']['channel'] + elif platform == 'qqofficial': + source = await module.QQOfficialEventConverter().target2yiri(module.qq_event()) + await adapter.create_message_card('response', source) + else: + source = await module.WecomBotEventConverter().target2yiri(module.wecombot_event()) + + async def runner(): + yield MessageChunk(role='assistant', content='hello', all_content='hello', is_final=True) + yield Message(role='assistant', content='hello') + + count = 0 + async for result in coalesce_stream_results(runner()): + chunk = MessageChunk(role='assistant', content=result.content, resp_message_id='response', is_final=True) + await adapter.reply_message_chunk(source, chunk, MessageChain([Plain(text=result.content)]), is_final=True) + count += 1 + assert count == 1 + adapter.reply_message.assert_not_awaited() + if platform == 'lark': + adapter._replace_streaming_card.assert_awaited_once_with('card', 'hello') + assert not adapter.card_id_dict + elif platform == 'dingtalk': + adapter.bot.send_card_message.assert_awaited_once_with('card', 'id', 'hello', True) + assert not adapter.card_instance_id_dict + elif platform == 'telegram': + adapter.bot.send_message.assert_awaited_once() + assert not adapter.msg_stream_id + elif platform == 'discord': + channel.send.assert_awaited_once_with('hello') + assert not adapter._stream_buffer + elif platform == 'qqofficial': + assert len([call for call in adapter.bot.sent if call[0] == 'stream']) == 1 + assert not adapter._stream_ctx + else: + adapter.bot.push_stream_chunk.assert_awaited_once_with('msg-1', 'hello', is_final=True) + adapter.bot.reply_text.assert_not_awaited() diff --git a/tests/unit_tests/platform/test_wecombot_eba_adapter.py b/tests/unit_tests/platform/test_wecombot_eba_adapter.py index 1952505da..12d3386e3 100644 --- a/tests/unit_tests/platform/test_wecombot_eba_adapter.py +++ b/tests/unit_tests/platform/test_wecombot_eba_adapter.py @@ -241,7 +241,7 @@ async def test_wecombot_event_converter_maps_private_and_group_messages_to_eba() assert private_event.adapter_name == 'wecombot-omni' assert private_event.chat_type == platform_entities.ChatType.PRIVATE assert private_event.chat_id == 'user-1' - assert str(private_event.message_chain) == 'hello' + assert str(private_event.message_chain) == '@EBA Bot hello' assert isinstance(group_event, platform_events.MessageReceivedEvent) assert group_event.chat_type == platform_entities.ChatType.GROUP @@ -302,7 +302,9 @@ async def test_wecombot_send_reply_feedback_and_platform_api_use_underlying_clie source_event = await WecomBotEventConverter().target2yiri(wecombot_event()) await adapter.reply_message(source_event, message) - adapter.bot.reply_text.assert_awaited_once_with('req-1', 'hello') + adapter.bot.push_stream_chunk.assert_awaited_once_with('msg-1', 'hello', is_final=True) + adapter.bot.reply_text.assert_not_awaited() + adapter.bot.push_stream_chunk.reset_mock() await adapter.reply_message_chunk(source_event, None, message, is_final=True) adapter.bot.push_stream_chunk.assert_awaited_once_with('msg-1', 'hello', is_final=True) diff --git a/tests/unit_tests/platform/test_wecombot_stream_empty_final.py b/tests/unit_tests/platform/test_wecombot_stream_empty_final.py index e21d163b6..8f18fa866 100644 --- a/tests/unit_tests/platform/test_wecombot_stream_empty_final.py +++ b/tests/unit_tests/platform/test_wecombot_stream_empty_final.py @@ -117,3 +117,61 @@ async def test_real_final_chunk_is_sent_and_closes_stream(): assert frames[-1]['stream']['content'] == 'the answer' assert frames[-1]['stream']['finish'] is True assert not client.stream_still_open() + + +@pytest.mark.asyncio +async def test_duplicate_final_is_acknowledged_without_sending_again(): + client = RecordingClient() + client.seed_stream() + assert await client.push_stream_chunk('msg-1', 'answer', is_final=True) + assert await client.push_stream_chunk('msg-1', 'answer', is_final=True) + assert len(client.stream_frames()) == 1 + assert not await client.push_stream_chunk('different-message', 'answer', is_final=True) + assert not await client.push_stream_chunk('msg-1', 'different answer', is_final=True) + + +@pytest.mark.asyncio +async def test_runner_cumulative_snapshots_reach_wire_once_without_concatenation(): + from langbot.pkg.pipeline.process.stream_results import coalesce_stream_results + from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk + from tests.unit_tests.platform.test_wecombot_eba_adapter import ( + make_adapter, + wecombot_event, + WecomBotEventConverter, + ) + + client = RecordingClient() + client.seed_stream() + adapter = make_adapter() + adapter.bot = client + event = await WecomBotEventConverter().target2yiri(wecombot_event()) + snapshots = ['test', 'test\nOK', 'test\nOK!'] + + async def runner(): + # LocalAgent's StreamingModelCaller sets content to the accumulated + # text, leaves all_content unset, then emits message.completed. + for index, text in enumerate(snapshots): + yield MessageChunk(role='assistant', content=text, msg_sequence=index + 1, is_final=index == 2) + yield Message(role='assistant', content=snapshots[-1]) + + async for result in coalesce_stream_results(runner()): + final = isinstance(result, Message) or result.is_final + if isinstance(result, MessageChunk): + result = result.model_copy(update={'content': result.all_content}) + await adapter.reply_message_chunk(event, result, result.get_content_platform_message_chain(), is_final=final) + + frames = client.stream_frames() + from langbot.pkg.platform.adapters.wecombot.stream_text import format_stream_text + + assert [frame['stream']['content'] for frame in frames] == [format_stream_text(text) for text in snapshots] + assert [frame['stream']['finish'] for frame in frames] == [False, False, True] + assert len(client.replies) == 3 # No extra plain-text reply after completion. + + +@pytest.mark.asyncio +async def test_revised_snapshot_replaces_previous_text(): + client = RecordingClient() + client.seed_stream() + await client.push_stream_chunk('msg-1', 'test\nanswer') + await client.push_stream_chunk('msg-1', 'answer', is_final=True) + assert client.stream_frames()[-1]['stream']['content'] == 'answer' diff --git a/tests/unit_tests/platform/test_wecombot_stream_text.py b/tests/unit_tests/platform/test_wecombot_stream_text.py new file mode 100644 index 000000000..fe19d7bab --- /dev/null +++ b/tests/unit_tests/platform/test_wecombot_stream_text.py @@ -0,0 +1,30 @@ +import pytest + +from langbot.pkg.platform.adapters.wecombot.stream_text import format_stream_text + + +def test_multiple_rounds_have_one_thinking_section_and_separated_answers(): + raw = 'oneChecking.twoMore checks.threeDone.' + result = format_stream_text(raw) + assert result == '\none\n\ntwo\n\nthree\n\n\nChecking.\n\nMore checks.\n\nDone.' + assert format_stream_text(result) == result + + +@pytest.mark.parametrize('suffix', ['<', 'oneChecking.' + suffix) + assert result.endswith('Checking.') + + +def test_later_incomplete_thinking_remains_inside_single_section(): + result = format_stream_text('oneChecking.two') + assert result == '\none\n\ntwo\n\n\nChecking.' + + +def test_plain_markdown_is_unchanged(): + text = ' hello\n\n```python\na < b\n```\n' + assert format_stream_text(text) == text + + +def test_initial_partial_thinking_tag_waits_for_next_snapshot(): + assert format_stream_text('test你好', is_final=False) assert await client.push_stream_chunk('msg-1', '你好', is_final=True) assert sent == [ ('req-1', 'stream-1', '你', False), ('req-1', 'stream-1', '你好', False), + ('req-1', 'stream-1', 'test你好', False), ('req-1', 'stream-1', '你好', True), ] @@ -385,6 +387,7 @@ async def test_webhook_stream_queues_cumulative_snapshots_for_followups(): assert await client.push_stream_chunk('msg-1', '你', is_final=False) assert await client.push_stream_chunk('msg-1', '你好', is_final=False) + assert await client.push_stream_chunk('msg-1', 'test你好', is_final=False) assert await client.push_stream_chunk('msg-1', '你好', is_final=True) assert session.queue.qsize() == 1 diff --git a/web/src/app/home/bots/BotDetailContent.tsx b/web/src/app/home/bots/BotDetailContent.tsx index 78b380bf4..bcce0e9da 100644 --- a/web/src/app/home/bots/BotDetailContent.tsx +++ b/web/src/app/home/bots/BotDetailContent.tsx @@ -4,7 +4,7 @@ import { useNavigate } from 'react-router-dom'; import { Tabs, TabsList, TabsTrigger, TabsContent } from '@/components/ui/tabs'; import { Button } from '@/components/ui/button'; import { Switch } from '@/components/ui/switch'; -import { Label } from '@/components/ui/label'; +import { Badge } from '@/components/ui/badge'; import { Dialog, DialogContent, @@ -22,7 +22,7 @@ import type { BotSessionMonitorHandle } from '@/app/home/bots/components/bot-ses import { httpClient } from '@/app/infra/http/HttpClient'; import { useSidebarData } from '@/app/home/components/home-sidebar/SidebarDataContext'; import { useTranslation } from 'react-i18next'; -import { Settings, FileText, Users, RefreshCw, Trash2 } from 'lucide-react'; +import { Settings, FileText, Users, RefreshCw, Trash2, Loader2 } from 'lucide-react'; import { cn } from '@/lib/utils'; import { toast } from 'sonner'; import { showBotError } from './bot-error'; @@ -72,6 +72,7 @@ export default function BotDetailContent({ id }: { id: string }) { // Enable state managed here so the header switch works const [botEnabled, setBotEnabled] = useState(true); + const [isTogglingEnable, setIsTogglingEnable] = useState(false); const [enableLoaded, setEnableLoaded] = useState(false); // Fetch bot enable state @@ -91,20 +92,25 @@ export default function BotDetailContent({ id }: { id: string }) { const handleEnableToggle = useCallback( async (checked: boolean) => { + if (isTogglingEnable) return; const prev = botEnabled; + setIsTogglingEnable(true); setBotEnabled(checked); try { await httpClient.updateBot(id, { enable: checked }); setBot((current) => current ? { ...current, enable: checked } : current, ); + toast.success(t(checked ? 'bots.enableConfirmed' : 'bots.disableConfirmed')); refreshBots(); } catch (error) { setBotEnabled(prev); showBotError(error, t('bots.setBotEnableError'), t); + } finally { + setIsTogglingEnable(false); } }, - [id, botEnabled, refreshBots, t], + [id, botEnabled, isTogglingEnable, refreshBots, t], ); function handleFormSubmit() { @@ -210,18 +216,27 @@ export default function BotDetailContent({ id }: { id: string }) { {enableLoaded && (
+ + {isTogglingEnable + ? -
)} @@ -237,11 +252,11 @@ export default function BotDetailContent({ id }: { id: string }) {