diff --git a/pyproject.toml b/pyproject.toml index 9e7353795..83fd41166 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -232,4 +232,4 @@ line-ending = "auto" [tool.uv.sources] # Development contract: update to the matching SDK release before publishing. -langbot-plugin = { git = "https://github.com/langbot-app/langbot-plugin-sdk", rev = "f82b3ce935f9a33afee389fb39fe8dc29a45b615" } +langbot-plugin = { git = "https://github.com/langbot-app/langbot-plugin-sdk", rev = "ca0671b81db5b2ed937d01158a3982de3fd8500f" } diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index a4da83fe6..8ce82816f 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -8,6 +8,9 @@ import typing from langbot_plugin.api.entities.builtin.provider import message as provider_message from langbot_plugin.api.entities.builtin.pipeline import query as pipeline_query +from langbot_plugin.entities.io.actions.enums import PluginToRuntimeAction + +from .reply_stream import ReplyStreamSession from ...core import app from ...api.http.context import ExecutionContext from ...pipeline.pool import get_query_execution_context @@ -155,6 +158,16 @@ class AgentRunOrchestrator: state_context = build_state_context(event, binding, descriptor) run_id = context['run_id'] + context['context']['available_apis']['reply_stream'] = hasattr(PluginToRuntimeAction, 'REPLY_STREAM') and any( + tool.get('tool_name') == 'event_reply' and tool.get('tool_type') == 'platform' + for tool in resources.get('tools', []) + ) + reply_streams = ReplyStreamSession( + event, + adapter=(adapter_context or {}).get('_delivery_adapter') or getattr(execution_query, 'adapter', None), + source=(adapter_context or {}).get('_platform_event') + or getattr((adapter_context or {}).get('_query'), 'message_event', None), + ) available_apis = context.get('context', {}).get('available_apis') run_authorization = { 'runner_id': descriptor.id, @@ -208,6 +221,7 @@ class AgentRunOrchestrator: state_context=state_context, execution_query=execution_query, platform_context=freeze_platform_context(event), + reply_streams=reply_streams, ) event_log_id = await self.journal.write_event_log( @@ -358,6 +372,7 @@ class AgentRunOrchestrator: raise finally: session = await self._session_registry.unregister(run_id) + await reply_streams.close() pending_steering = session.get('steering_queue', []) if session else [] if pending_steering: try: diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index aee86b287..c1d0f2f1b 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -54,8 +54,10 @@ PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = ( 'write', {'zh_Hans': '回复当前会话', 'en_US': 'Reply to current conversation'}, { - 'zh_Hans': '向触发当前事件的会话发送文本消息。目标由 LangBot 固定,Agent 无法改写。', - 'en_US': 'Send text to the conversation that triggered this run. LangBot fixes the target.', + 'zh_Hans': '向触发当前事件的会话发送回复或任务进度。目标由 LangBot 固定,Agent 无法改写。' + '需要发送消息时请调用此工具,Agent 的输出文本不会自动发送。', + 'en_US': 'Send a reply or task progress to the conversation that triggered this run. LangBot fixes the target. ' + 'Call this tool to send a message; Agent output text is not sent automatically.', }, _object_schema({'text': {**_TEXT, 'description': 'Reply text'}}, ['text']), ('message.*', 'friend.*', 'group.*', 'feedback.*'), @@ -209,8 +211,9 @@ PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = ( 'write', {'zh_Hans': '发送消息', 'en_US': 'Send message'}, { - 'zh_Hans': '使用当前机器人向指定用户或群组发送文本消息。', - 'en_US': 'Send text to a specified person or group using the current bot.', + 'zh_Hans': '使用当前机器人向指定用户或群组发送回复或任务进度。Agent 的输出文本不会自动发送。', + 'en_US': 'Send a reply or task progress to a specified person or group using the current bot. ' + 'Agent output text is not sent automatically.', }, _object_schema( {'target_type': _TARGET_TYPE, 'target_id': _ID, 'text': _TEXT}, ['target_type', 'target_id', 'text'] diff --git a/src/langbot/pkg/agent/runner/reply_stream.py b/src/langbot/pkg/agent/runner/reply_stream.py new file mode 100644 index 000000000..aa2f3500f --- /dev/null +++ b/src/langbot/pkg/agent/runner/reply_stream.py @@ -0,0 +1,162 @@ +"""Host-owned explicit replies using the existing adapter streaming lifecycle.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from typing import Literal +from uuid import UUID, uuid4 + +import pydantic +from langbot_plugin.api.entities.builtin.platform import events, message +from langbot_plugin.api.entities.builtin.provider import message as provider_message + + +class ReplyStreamRequest(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra='forbid') + + stream_id: UUID + operation: Literal['update', 'finish', 'abort'] + text: str = pydantic.Field(max_length=200_000, strict=True) + + +@dataclass +class _Reply: + delivery_id: str = field(default_factory=lambda: str(uuid4())) + text: str = '' + started: bool = False + native: bool = False + final_attempted: bool = False + sequence: int = 0 + status: str = 'open' + result: dict | None = None + + +class ReplyStreamSession: + """Ephemeral delivery state owned by one authorized run, never by plugin input.""" + + def __init__(self, event, adapter=None, source=None): + self.adapter = adapter + self.source = source.to_legacy_event() if isinstance(source, events.MessageReceivedEvent) else source + if not isinstance(self.source, events.MessageEvent): + self.source = None + self.target = dict(event.delivery.reply_target or {}) + self.mock = ( + event.delivery.surface == 'webui' and (event.delivery.platform_capabilities or {}).get('debug_mock') is True + ) + self.mock_error = ( + ((event.delivery.platform_capabilities or {}).get('mock_options') or {}) + .get('errors', {}) + .get('event_reply') + ) + self._replies: dict[str, _Reply] = {} + self._lock = asyncio.Lock() + self._closed = False + self._active: set[asyncio.Task] = set() + + async def apply(self, request: ReplyStreamRequest) -> dict: + task = asyncio.create_task(self._apply(request)) + self._active.add(task) + try: + return await task + finally: + self._active.discard(task) + + async def _apply(self, request: ReplyStreamRequest) -> dict: + async with self._lock: + if self._closed: + raise ValueError('Reply stream run has ended') + key = str(request.stream_id) + reply = self._replies.get(key) + if reply is None: + if len(self._replies) >= 16: + raise ValueError('A run may create at most 16 reply streams') + reply = self._replies[key] = _Reply() + if reply.status != 'open': + if request.operation == 'update' or reply.status == 'failed': + raise ValueError('Reply stream is closed') + return reply.result + try: + if request.operation != 'abort': + reply.text = request.text + if self.mock_error: + raise ValueError(str(self.mock_error)) + if not self.mock and reply.text and not reply.started and request.operation != 'abort': + if self.adapter is None: + raise ValueError('This run has no platform delivery adapter') + # Set before I/O: an uncertain create must never be retried as another send. + reply.started = True + reply.native = self.source is not None and await self.adapter.is_stream_output_supported() + if reply.native: + await self.adapter.create_message_card(reply.delivery_id, self.source) + if reply.started and reply.native: + await self._update_native(key, reply, final=request.operation != 'update') + elif request.operation == 'finish' and reply.text and not self.mock: + if self.source is not None: + await self.adapter.reply_message( + message_source=self.source, message=self._message(reply.text), quote_origin=False + ) + else: + target_type, target_id = self.target.get('target_type'), self.target.get('target_id') + if not target_type or not target_id: + raise ValueError('This event has no reply target') + await self.adapter.send_message(str(target_type), str(target_id), self._message(reply.text)) + if request.operation != 'update': + reply.status = 'completed' if request.operation == 'finish' else 'cancelled' + reply.result = { + 'stream_id': key, + 'status': reply.status, + 'delivery': 'simulated' if self.mock else 'streaming' if reply.native else 'buffered', + 'mock': self.mock, + **({'text': reply.text} if request.operation == 'finish' else {}), + } + return reply.result + except BaseException: + reply.status = 'failed' + raise + + @staticmethod + def _message(text): + return message.MessageChain([message.Plain(text=text)]) + + async def _update_native(self, key, reply, *, final): + if final: + reply.final_attempted = True + reply.sequence += 1 + chunk = provider_message.MessageChunk( + role='assistant', + content=reply.text, + all_content=reply.text, + resp_message_id=reply.delivery_id, + msg_sequence=reply.sequence, + is_final=final, + ) + await self.adapter.reply_message_chunk( + message_source=self.source, + bot_message=chunk, + message=self._message(reply.text), + quote_origin=False, + is_final=final, + ) + + async def close(self): + """Close visible streams on cancellation; never send a buffered partial reply.""" + self._closed = True + pending_requests = list(self._active) + for task in pending_requests: + task.cancel() + if pending_requests: + await asyncio.gather(*pending_requests, return_exceptions=True) + async with self._lock: + pending = [ + self._update_native(key, reply, final=True) + for key, reply in self._replies.items() + if reply.started and reply.native and not reply.final_attempted + ] + try: + if pending: + await asyncio.wait_for(asyncio.gather(*pending, return_exceptions=True), timeout=5) + except TimeoutError: + pass + finally: + self._replies.clear() diff --git a/src/langbot/pkg/agent/runner/session_registry.py b/src/langbot/pkg/agent/runner/session_registry.py index eb59ed824..ba135e0ed 100644 --- a/src/langbot/pkg/agent/runner/session_registry.py +++ b/src/langbot/pkg/agent/runner/session_registry.py @@ -77,6 +77,7 @@ class AgentRunSession(typing.TypedDict): plugin_identity: str # author/name authorization: RunAuthorizationSnapshot status: AgentRunSessionStatus + reply_streams: typing.Any steering_queue: list[SteeringQueueItem] @@ -115,6 +116,7 @@ class AgentRunSessionRegistry: state_context: dict[str, typing.Any] | None = None, execution_query: pipeline_query.Query | None = None, platform_context: dict[str, typing.Any] | None = None, + reply_streams: typing.Any = None, ) -> None: """Register a new agent run session. @@ -167,6 +169,7 @@ class AgentRunSessionRegistry: 'runner_id': runner_id, 'query_id': query_id, 'execution_query': execution_query, + 'reply_streams': reply_streams, 'plugin_identity': plugin_identity, 'authorization': authorization, 'status': { diff --git a/src/langbot/pkg/api/http/controller/groups/platform/bots.py b/src/langbot/pkg/api/http/controller/groups/platform/bots.py index 7f0df72fb..ad00d7888 100644 --- a/src/langbot/pkg/api/http/controller/groups/platform/bots.py +++ b/src/langbot/pkg/api/http/controller/groups/platform/bots.py @@ -3,11 +3,20 @@ from sqlalchemy.exc import IntegrityError from ....authz import Permission, has_permission from ....context import RequestContext +from ....service.bot_errors import BotApplyError, bot_error_message from ... import group @group.group_class('bots', '/api/v1/platform/bots') class BotsRouterGroup(group.RouterGroup): + def _apply_error_response(self, exc: BotApplyError): + request_id = self.request_id() + logger = getattr(self.ap, 'logger', self.quart_app.logger) + logger.error(f'Bot configuration apply failed request_id={request_id} bot_uuid={exc.bot_uuid}', exc_info=True) + return quart.jsonify( + code='bot_apply_failed', msg=str(exc), data={'uuid': exc.bot_uuid}, request_id=request_id + ), 400 + async def initialize(self) -> None: @self.route( '', @@ -34,7 +43,14 @@ class BotsRouterGroup(group.RouterGroup): ) async def _(request_context: RequestContext) -> str: json_data = await quart.request.json - bot_uuid = await self.ap.bot_service.create_bot(request_context, json_data) + if not isinstance(json_data, dict): + return self.http_status(400, 'invalid_bot_config', 'Bot configuration must be an object') + try: + bot_uuid = await self.ap.bot_service.create_bot(request_context, json_data) + except BotApplyError as exc: + return self._apply_error_response(exc) + except ValueError as exc: + return self.http_status(400, 'invalid_bot_config', bot_error_message(exc, json_data)) return self.success(data={'uuid': bot_uuid}) @self.route( @@ -63,7 +79,14 @@ class BotsRouterGroup(group.RouterGroup): async def _(bot_uuid: str, request_context: RequestContext) -> str: if quart.request.method == 'PUT': json_data = await quart.request.json - await self.ap.bot_service.update_bot(request_context, bot_uuid, json_data) + if not isinstance(json_data, dict): + return self.http_status(400, 'invalid_bot_config', 'Bot configuration must be an object') + try: + await self.ap.bot_service.update_bot(request_context, bot_uuid, json_data) + except BotApplyError as exc: + return self._apply_error_response(exc) + except ValueError as exc: + return self.http_status(400, 'invalid_bot_config', bot_error_message(exc, json_data)) else: await self.ap.bot_service.delete_bot(request_context, bot_uuid) return self.success() diff --git a/src/langbot/pkg/api/http/service/bot.py b/src/langbot/pkg/api/http/service/bot.py index 05f4a35ca..2190b7159 100644 --- a/src/langbot/pkg/api/http/service/bot.py +++ b/src/langbot/pkg/api/http/service/bot.py @@ -11,6 +11,7 @@ from ....entity.persistence import agent as persistence_agent from ....entity.persistence import bot as persistence_bot from ....entity.persistence import pipeline as persistence_pipeline from ....workspace.errors import WorkspaceNotFoundError +from .bot_errors import BotApplyError, bot_error_message from .tenant import TenantContext, require_workspace_uuid, scope_statement from ....utils import httpclient from ....platform.sources import http_bot_signing @@ -664,7 +665,10 @@ class BotService: bot = await self.get_bot(context, bot_data['uuid'], include_secret=True) - await self.ap.platform_mgr.load_bot(context, bot) + try: + await self.ap.platform_mgr.load_bot(context, bot) + except Exception as exc: + raise BotApplyError(bot_error_message(exc, bot), bot['uuid']) from exc return bot_data['uuid'] @@ -694,17 +698,17 @@ class BotService: runtime_bot.bot_entity.description = update_data['description'] return - await self.ap.platform_mgr.remove_bot(context, bot_uuid) - - # select from db + # Persisted configuration is distinct from applying it to the running adapter. bot = await self.get_bot(context, bot_uuid, include_secret=True) + try: + await self.ap.platform_mgr.remove_bot(context, bot_uuid) + runtime_bot = await self.ap.platform_mgr.load_bot(context, bot) + if runtime_bot.enable: + await runtime_bot.run() + except Exception as exc: + raise BotApplyError(bot_error_message(exc, bot), bot['uuid']) from exc - runtime_bot = await self.ap.platform_mgr.load_bot(context, bot) - - if runtime_bot.enable: - await runtime_bot.run() - - # update all conversation that use this bot + # Reset conversations using this bot after its configuration is applied. for session in self.ap.sess_mgr.session_list: if ( session.using_conversation is not None diff --git a/src/langbot/pkg/api/http/service/bot_errors.py b/src/langbot/pkg/api/http/service/bot_errors.py new file mode 100644 index 000000000..a73c03d54 --- /dev/null +++ b/src/langbot/pkg/api/http/service/bot_errors.py @@ -0,0 +1,43 @@ +"""User-facing bot configuration errors without credentials or validation inputs.""" + +import json + +import pydantic + +from .secrets import redact_secrets + + +class BotApplyError(Exception): + """Configuration was persisted, but the runtime could not apply it.""" + + def __init__(self, message: str, bot_uuid: str | None = None): + super().__init__(message) + self.bot_uuid = bot_uuid + + +def bot_error_message(error: Exception, configuration: dict) -> str: + if isinstance(error, pydantic.ValidationError): + text = '; '.join( + f'{".".join(map(str, item["loc"]))}: {item["msg"]}' + for item in error.errors(include_input=False, include_context=False, include_url=False) + ) + else: + text = str(error).strip() or type(error).__name__ + + replacements = [] + + def collect(original, masked): + if isinstance(original, dict) and isinstance(masked, dict): + for key, value in original.items(): + collect(value, masked.get(key)) + elif isinstance(original, (list, tuple)) and isinstance(masked, (list, tuple)): + for value, replacement in zip(original, masked): + collect(value, replacement) + elif isinstance(original, str) and original and original != masked: + for representation in {original, repr(original)[1:-1], json.dumps(original, ensure_ascii=False)[1:-1]}: + replacements.append((representation, str(masked))) + + collect(configuration, redact_secrets(configuration)) + for original, masked in sorted(replacements, key=lambda item: len(item[0]), reverse=True): + text = text.replace(original, masked) + return text[:2000] diff --git a/src/langbot/pkg/platform/adapters/dingtalk/manifest.yaml b/src/langbot/pkg/platform/adapters/dingtalk/manifest.yaml index 6485b846f..8e74f21d1 100644 --- a/src/langbot/pkg/platform/adapters/dingtalk/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/dingtalk/manifest.yaml @@ -103,9 +103,9 @@ spec: zh_Hans: 启用钉钉卡片流式回复模式 zh_Hant: 啟用釘釘卡片串流回覆模式 description: - en_US: If enabled, the bot will use DingTalk card streaming replies. - zh_Hans: 如果启用,将使用钉钉卡片流式方式来回复内容 - zh_Hant: 如果啟用,將使用釘釘卡片串流方式來回覆內容 + en_US: Used for streaming replies from pipelines or plugins. If enabled, the bot will use DingTalk card streaming replies. + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,将使用钉钉卡片流式方式来回复内容 + zh_Hant: 用於流水線和插件主動發起的串流回覆。如果啟用,將使用釘釘卡片串流方式來回覆內容 type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/adapters/kook/manifest.yaml b/src/langbot/pkg/platform/adapters/kook/manifest.yaml index 7a1abbe50..56f6e5b85 100644 --- a/src/langbot/pkg/platform/adapters/kook/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/kook/manifest.yaml @@ -38,6 +38,10 @@ spec: en_US: Enable stream reply zh_Hans: 启用流式回复 zh_Hant: 啟用串流回覆 + description: + en_US: Used for streaming replies from pipelines or plugins. + zh_Hans: 用于流水线和插件主动发起的流式回复。 + zh_Hant: 用於流水線和插件主動發起的串流回覆。 type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/adapters/lark/manifest.yaml b/src/langbot/pkg/platform/adapters/lark/manifest.yaml index cfdea896a..5d80781ca 100644 --- a/src/langbot/pkg/platform/adapters/lark/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/lark/manifest.yaml @@ -174,10 +174,10 @@ spec: zh_Hant: 啟用飛書串流回覆模式 ja_JP: ストリーミング返信モードを有効化 description: - en_US: If enabled, replies are rendered through an updating Lark card. - zh_Hans: 如果启用,将使用可更新的飞书卡片进行流式回复。 - zh_Hant: 如果啟用,將使用可更新的飛書卡片進行串流回覆。 - ja_JP: 有効にすると、更新可能な Lark カードでストリーミング返信します。 + en_US: Used for streaming replies from pipelines or plugins. If enabled, replies are rendered through an updating Lark card. + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,将使用可更新的飞书卡片进行流式回复。 + zh_Hant: 用於流水線和插件主動發起的串流回覆。如果啟用,將使用可更新的飛書卡片進行串流回覆。 + ja_JP: パイプラインやプラグインのストリーミング返信に適用されます。有効にすると、更新可能な Lark カードでストリーミング返信します。 type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/adapters/qqofficial/manifest.yaml b/src/langbot/pkg/platform/adapters/qqofficial/manifest.yaml index 53ec224d0..bad341ea5 100644 --- a/src/langbot/pkg/platform/adapters/qqofficial/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/qqofficial/manifest.yaml @@ -100,9 +100,9 @@ spec: zh_Hans: 启用流式回复模式 zh_Hant: 啟用串流回覆模式 description: - en_US: If enabled, the adapter uses QQ Official streaming replies for C2C private messages. - zh_Hans: 启用后,适配器会对 C2C 私聊使用 QQ 官方流式回复。 - zh_Hant: 啟用後,適配器會對 C2C 私聊使用 QQ 官方串流回覆。 + en_US: Used for streaming replies from pipelines or plugins. If enabled, the adapter uses QQ Official streaming replies for C2C private messages. + zh_Hans: 用于流水线和插件主动发起的流式回复。启用后,适配器会对 C2C 私聊使用 QQ 官方流式回复。 + zh_Hant: 用於流水線和插件主動發起的串流回覆。啟用後,適配器會對 C2C 私聊使用 QQ 官方串流回覆。 type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/adapters/telegram/manifest.yaml b/src/langbot/pkg/platform/adapters/telegram/manifest.yaml index 4d48d4af6..c54902a97 100644 --- a/src/langbot/pkg/platform/adapters/telegram/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/telegram/manifest.yaml @@ -39,8 +39,8 @@ spec: en_US: Enable Stream Reply Mode zh_Hans: 启用电报流式回复模式 description: - en_US: If enabled, the bot will use the stream of telegram reply mode - zh_Hans: 如果启用,将使用电报流式方式来回复内容 + en_US: Used for streaming replies from pipelines or plugins. If enabled, the bot will use the stream of telegram reply mode + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,将使用电报流式方式来回复内容 type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml b/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml index e2dba9fad..2b722142f 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml @@ -145,9 +145,9 @@ spec: zh_Hans: 启用流式回复 zh_Hant: 啟用串流回覆 description: - en_US: If enabled, the bot will use WeComBot streaming replies. - zh_Hans: 如果启用,机器人将使用企业微信智能机器人流式回复。 - zh_Hant: 如果啟用,機器人將使用企業微信智慧機器人串流回覆。 + en_US: Used for streaming replies from pipelines or plugins. If enabled, the bot will use WeComBot streaming replies. + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,机器人将使用企业微信智能机器人流式回复。 + zh_Hant: 用於流水線和插件主動發起的串流回覆。如果啟用,機器人將使用企業微信智慧機器人串流回覆。 type: boolean required: false default: true diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index 0f1591b48..fd4366ee7 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -38,7 +38,6 @@ from .logger import EventLogger from .adapter_names import canonical_adapter_name import langbot_plugin.api.entities.builtin.provider.session as provider_session -import langbot_plugin.api.entities.builtin.provider.message as provider_message import langbot_plugin.api.entities.events as plugin_events import langbot_plugin.api.entities.builtin.platform.events as platform_events import langbot_plugin.api.entities.builtin.platform.entities as platform_entities @@ -826,7 +825,7 @@ class RuntimeBot: state_policy=StatePolicy(state_scopes=['conversation', 'actor', 'subject', 'runner']), delivery_policy=DeliveryPolicy( enable_streaming=False, - enable_reply=True, + enable_reply=False, enable_interactions=agent.get('kind') != 'event_processor', ), agent_id=agent.get('uuid'), @@ -834,65 +833,6 @@ class RuntimeBot: processor_id=agent.get('uuid'), ) - @staticmethod - def _provider_content_to_text(content: typing.Any) -> str: - if content is None: - return '' - if isinstance(content, str): - return content - if isinstance(content, list): - parts: list[str] = [] - for item in content: - item_data = item.model_dump(mode='json') if hasattr(item, 'model_dump') else item - if isinstance(item_data, dict): - if item_data.get('type') == 'text' and item_data.get('text') is not None: - parts.append(str(item_data.get('text'))) - elif item_data.get('text') is not None: - parts.append(str(item_data.get('text'))) - elif item_data is not None: - parts.append(str(item_data)) - return ''.join(parts) - return str(content) - - @classmethod - def _provider_output_to_text(cls, result: provider_message.Message | provider_message.MessageChunk) -> str: - if getattr(result, 'all_content', None): - return str(getattr(result, 'all_content')) - return cls._provider_content_to_text(getattr(result, 'content', None)) - - async def _deliver_agent_outputs( - self, - envelope: AgentEventEnvelope, - outputs: list[provider_message.Message | provider_message.MessageChunk], - adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter | None = None, - ) -> None: - if not outputs or not envelope.delivery.reply_target: - return - - reply_target = envelope.delivery.reply_target - target_type = reply_target.get('target_type') - target_id = reply_target.get('target_id') - if not target_type or not target_id: - return - - final_text = '' - for output in outputs: - output_text = self._provider_output_to_text(output) - if isinstance(output, provider_message.Message): - final_text = output_text or final_text - elif output_text: - final_text = output_text - - if not final_text: - return - - delivery_adapter = adapter or self.adapter - await delivery_adapter.send_message( - str(target_type), - str(target_id), - platform_message.MessageChain([platform_message.Plain(text=final_text)]), - ) - async def _handle_platform_event( self, event: platform_events.EBAEvent, @@ -1053,17 +993,18 @@ class RuntimeBot: envelope = self._eba_event_to_agent_envelope(event, adapter) if target_type == 'event_processor': envelope.data = event.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'}) - outputs: list[provider_message.Message | provider_message.MessageChunk] = [] try: - async for output in self.ap.agent_run_orchestrator.run( + async for _ in self.ap.agent_run_orchestrator.run( envelope, binding, adapter_context={ '_delivery_adapter': adapter, + '_platform_event': event, '_execution_context': self.execution_context, }, ): - outputs.append(output) + # Results are journaled by the orchestrator; platform sends require explicit actions. + pass except Exception: return await self._record_event_route_trace( event_type=event_type, @@ -1076,21 +1017,6 @@ class RuntimeBot: reason='Agent runner failed', text=f'Failed to run Agent for EBA event {event_type}: {traceback.format_exc()}', ) - - try: - await self._deliver_agent_outputs(envelope, outputs, adapter=adapter) - except Exception: - return await self._record_event_route_trace( - event_type=event_type, - status='failed', - level='error', - binding=event_binding, - target_type=target_type, - target_uuid=target_uuid, - failure_code='delivery_failed', - reason='Agent output delivery failed', - text=f'Failed to deliver Agent output for EBA event {event_type}: {traceback.format_exc()}', - ) return await self._record_event_route_trace( event_type=event_type, status='delivered', @@ -1458,14 +1384,13 @@ class RuntimeBot: data={'interaction': submission}, ) - outputs: list[provider_message.Message | provider_message.MessageChunk] = [] - async for output in self.ap.agent_run_orchestrator.run( + async for _ in self.ap.agent_run_orchestrator.run( envelope, binding, adapter_context={'_delivery_adapter': adapter}, ): - outputs.append(output) - await self._deliver_agent_outputs(envelope, outputs, adapter=adapter) + # Resuming an interaction follows the same explicit-action delivery policy. + pass async def _dispatch_eba_message_to_pipeline( self, diff --git a/src/langbot/pkg/platform/sources/dingtalk.yaml b/src/langbot/pkg/platform/sources/dingtalk.yaml index 7ced7de90..f8a7e6395 100644 --- a/src/langbot/pkg/platform/sources/dingtalk.yaml +++ b/src/langbot/pkg/platform/sources/dingtalk.yaml @@ -82,9 +82,9 @@ spec: zh_Hans: 启用钉钉卡片流式回复模式 zh_Hant: 啟用釘釘卡片串流回覆模式 description: - en_US: If enabled, the bot will use the stream of lark reply mode - zh_Hans: 如果启用,将使用钉钉卡片流式方式来回复内容 - zh_Hant: 如果啟用,將使用釘釘卡片串流方式來回覆內容 + en_US: Used for streaming replies from pipelines or plugins. If enabled, the bot will use the stream of lark reply mode + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,将使用钉钉卡片流式方式来回复内容 + zh_Hant: 用於流水線和插件主動發起的串流回覆。如果啟用,將使用釘釘卡片串流方式來回覆內容 type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/sources/lark.yaml b/src/langbot/pkg/platform/sources/lark.yaml index 6c3a59636..57c5e757f 100644 --- a/src/langbot/pkg/platform/sources/lark.yaml +++ b/src/langbot/pkg/platform/sources/lark.yaml @@ -178,10 +178,10 @@ spec: zh_Hant: 啟用飛書串流回覆模式 ja_JP: ストリーミング返信モードを有効化 description: - en_US: If enabled, the bot will use the stream of lark reply mode - zh_Hans: 如果启用,将使用飞书流式方式来回复内容 - zh_Hant: 如果啟用,將使用飛書串流方式來回覆內容 - ja_JP: 有効にすると、ボットはストリーミングモードでメッセージに返信します + en_US: Used for streaming replies from pipelines or plugins. If enabled, the bot will use the stream of lark reply mode + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,将使用飞书流式方式来回复内容 + zh_Hant: 用於流水線和插件主動發起的串流回覆。如果啟用,將使用飛書串流方式來回覆內容 + ja_JP: パイプラインやプラグインのストリーミング返信に適用されます。有効にすると、ボットはストリーミングモードでメッセージに返信します type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/sources/mattermost.yaml b/src/langbot/pkg/platform/sources/mattermost.yaml index 391f06902..f46b125bb 100644 --- a/src/langbot/pkg/platform/sources/mattermost.yaml +++ b/src/langbot/pkg/platform/sources/mattermost.yaml @@ -64,8 +64,8 @@ spec: vi_VN: Bật phản hồi luồng es_ES: Activar respuesta en streaming description: - en_US: Update a Mattermost post while LangBot generates a response - zh_Hans: 在 LangBot 生成回复时持续更新同一条 Mattermost 消息 + en_US: Used for streaming replies from pipelines or plugins. Update a Mattermost post while LangBot generates a response + zh_Hans: 用于流水线和插件主动发起的流式回复。在 LangBot 生成回复时持续更新同一条 Mattermost 消息 type: boolean required: false default: true diff --git a/src/langbot/pkg/platform/sources/qqofficial.yaml b/src/langbot/pkg/platform/sources/qqofficial.yaml index 46842b1f5..1d235d008 100644 --- a/src/langbot/pkg/platform/sources/qqofficial.yaml +++ b/src/langbot/pkg/platform/sources/qqofficial.yaml @@ -90,9 +90,9 @@ spec: zh_Hans: 启用流式回复模式 zh_Hant: 啟用串流回覆模式 description: - en_US: If enabled, the bot will use streaming mode to reply messages (C2C only) - zh_Hans: 如果启用,机器人将使用流式方式回复消息(仅私聊) - zh_Hant: 如果啟用,機器人將使用串流方式回覆訊息(僅私聊) + en_US: Used for streaming replies from pipelines or plugins. If enabled, the bot will use streaming mode to reply messages (C2C only) + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,机器人将使用流式方式回复消息(仅私聊) + zh_Hant: 用於流水線和插件主動發起的串流回覆。如果啟用,機器人將使用串流方式回覆訊息(僅私聊) type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/sources/telegram.yaml b/src/langbot/pkg/platform/sources/telegram.yaml index 810d94ac2..7f85c62ac 100644 --- a/src/langbot/pkg/platform/sources/telegram.yaml +++ b/src/langbot/pkg/platform/sources/telegram.yaml @@ -63,13 +63,13 @@ spec: vi_VN: Bật chế độ trả lời trực tuyến es_ES: Habilitar modo de respuesta en streaming description: - en_US: If enabled, the bot will use the stream of telegram reply mode - zh_Hans: 如果启用,将使用电报流式方式来回复内容 - zh_Hant: 如果啟用,將使用 Telegram 串流方式來回覆內容 - ja_JP: 有効にすると、ボットはストリーミングモードでメッセージに返信します - th_TH: หากเปิดใช้งาน บอทจะใช้โหมดสตรีมของ Telegram ในการตอบกลับ - vi_VN: Nếu bật, bot sẽ sử dụng chế độ trả lời trực tuyến của Telegram - es_ES: Si está habilitado, el bot usará el modo de respuesta en streaming de Telegram + en_US: Used for streaming replies from pipelines or plugins. If enabled, the bot will use the stream of telegram reply mode + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,将使用电报流式方式来回复内容 + zh_Hant: 用於流水線和插件主動發起的串流回覆。如果啟用,將使用 Telegram 串流方式來回覆內容 + ja_JP: パイプラインやプラグインのストリーミング返信に適用されます。有効にすると、ボットはストリーミングモードでメッセージに返信します + th_TH: ใช้กับการตอบกลับแบบสตรีมจากไปป์ไลน์หรือปลั๊กอิน หากเปิดใช้งาน บอทจะใช้โหมดสตรีมของ Telegram ในการตอบกลับ + vi_VN: Áp dụng cho phản hồi streaming từ pipeline hoặc plugin. Nếu bật, bot sẽ sử dụng chế độ trả lời trực tuyến của Telegram + es_ES: Se aplica a respuestas en streaming de pipelines o plugins. Si está habilitado, el bot usará el modo de respuesta en streaming de Telegram type: boolean required: true default: false diff --git a/src/langbot/pkg/platform/sources/wecombot.yaml b/src/langbot/pkg/platform/sources/wecombot.yaml index a12e6305d..51961d357 100644 --- a/src/langbot/pkg/platform/sources/wecombot.yaml +++ b/src/langbot/pkg/platform/sources/wecombot.yaml @@ -142,9 +142,9 @@ spec: zh_Hans: 启用流式回复 zh_Hant: 啟用串流回覆 description: - en_US: If enabled, the bot will use streaming mode to reply messages - zh_Hans: 如果启用,机器人将使用流式模式回复消息 - zh_Hant: 如果啟用,機器人將使用串流模式回覆訊息 + en_US: Used for streaming replies from pipelines or plugins. If enabled, the bot will use streaming mode to reply messages + zh_Hans: 用于流水线和插件主动发起的流式回复。如果启用,机器人将使用流式模式回复消息 + zh_Hant: 用於流水線和插件主動發起的串流回覆。如果啟用,機器人將使用串流模式回覆訊息 type: boolean required: false default: true diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index afdbc540f..1b9c64855 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -1511,6 +1511,56 @@ class RuntimeConnectionHandler(handler.Handler): }, ) + async def reply_stream(data: dict[str, Any]) -> handler.ActionResponse: + """Explicit reply delivery authorized by the frozen event_reply permission.""" + from ..agent.runner.reply_stream import ReplyStreamRequest + + action_context = self._require_runtime_action_context() + session, error = await _validate_run_authorization( + data.get('run_id'), + 'tool', + 'event_reply', + self.ap, + data.get('caller_plugin_identity'), + operation='call', + ) + if error: + return error + source_ref, error = _validate_frozen_tool_source_identity(session, 'event_reply', self.ap) + if error: + return error + if source_ref is None or source_ref['source'] != 'platform': + return handler.ActionResponse.error('event_reply must be a Host platform tool') + query = session.get('execution_query') + context = self._execution_context(action_context) + if query is None or any( + getattr(query, field, None) != getattr(context, field, None) + for field in ('instance_uuid', 'workspace_uuid', 'placement_generation') + ): + return handler.ActionResponse.error('Reply stream run belongs to another execution scope') + streams = session.get('reply_streams') + if streams is None: + return handler.ActionResponse.error('Streaming replies are unavailable for this run') + try: + if not streams.mock: + bot_id = session['authorization'].get('bot_id') + bot = await self.ap.platform_mgr.get_bot_by_uuid(context, bot_id) + if bot is None or bot.adapter is not streams.adapter: + return handler.ActionResponse.error('The reply adapter is no longer active') + if 'send_message' not in bot.adapter.get_supported_apis(): + return handler.ActionResponse.error('The reply adapter no longer supports sending messages') + request = ReplyStreamRequest.model_validate( + {key: data[key] for key in ('stream_id', 'operation', 'text') if key in data} + ) + result = await streams.apply(request) + return handler.ActionResponse.success(data={'result': result}) + except Exception as exc: + return handler.ActionResponse.error(f'Streaming reply failed: {exc}') + + reply_stream_action = getattr(PluginToRuntimeAction, 'REPLY_STREAM', None) + if reply_stream_action is not None: + self.action(reply_stream_action)(reply_stream) + @self.action(PluginToRuntimeAction.CALL_TOOL) async def call_tool(data: dict[str, Any]) -> handler.ActionResponse: """Call a tool diff --git a/tests/unit_tests/agent/test_reply_stream.py b/tests/unit_tests/agent/test_reply_stream.py new file mode 100644 index 000000000..5eeb1589a --- /dev/null +++ b/tests/unit_tests/agent/test_reply_stream.py @@ -0,0 +1,270 @@ +"""Explicit streaming delivery across SDK, Host lifecycle, and adapter boundaries.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +from langbot_plugin.api.entities.builtin.platform import events, entities, message +from langbot_plugin.api.entities.builtin.agent_runner.context_access import ContextAPICapabilities +from langbot_plugin.api.proxies.agent_run import AgentRunAPIProxy +from langbot_plugin.api.proxies.agent_run.common import PermissionDeniedError +from langbot_plugin.entities.io.actions.enums import PluginToRuntimeAction + +from langbot.pkg.agent.runner.reply_stream import ReplyStreamRequest, ReplyStreamSession + + +def make_session(*, native=True, source=True, mock=False): + event = SimpleNamespace( + delivery=SimpleNamespace( + surface='webui' if mock else 'platform', + platform_capabilities={'debug_mock': mock}, + reply_target={'target_type': 'person', 'target_id': 'user-1'}, + ) + ) + incoming = ( + events.MessageReceivedEvent( + message_id='source-1', + sender=entities.User(id='user-1'), + chat_id='user-1', + chat_type=entities.ChatType.PRIVATE, + message_chain=message.MessageChain([message.Plain(text='hello')]), + source_platform_object=object(), + ) + if source + else None + ) + adapter = SimpleNamespace( + is_stream_output_supported=AsyncMock(return_value=native), + create_message_card=AsyncMock(return_value=True), + reply_message_chunk=AsyncMock(), + send_message=AsyncMock(), + reply_message=AsyncMock(), + ) + return ReplyStreamSession(event, adapter, incoming), adapter, incoming + + +def request(key, operation='update', text='hello'): + return ReplyStreamRequest(stream_id=key, operation=operation, text=text) + + +def proxy_for(session, *, allowed=True, advertised=True): + if not hasattr(AgentRunAPIProxy, 'reply_stream'): + pytest.skip('SDK does not provide the optional streaming reply API') + context = SimpleNamespace( + run_id='run-1', + runtime=SimpleNamespace(deadline_at=None), + context=SimpleNamespace(available_apis=ContextAPICapabilities(reply_stream=advertised)), + resources=SimpleNamespace( + models=[], + knowledge_bases=[], + tools=[SimpleNamespace(tool_name='event_reply', operations=['call'])] if allowed else [], + ), + ) + + async def action(action, data, timeout): + assert action == PluginToRuntimeAction.REPLY_STREAM + assert data['run_id'] == 'run-1' + return { + 'result': await session.apply( + ReplyStreamRequest.model_validate({k: v for k, v in data.items() if k != 'run_id'}) + ) + } + + transport = SimpleNamespace(call_action=AsyncMock(side_effect=action)) + return AgentRunAPIProxy(context, transport), transport + + +@pytest.mark.parametrize( + 'native,source,mock', + [ + (True, True, False), + (False, True, False), + (True, False, False), + (True, True, True), + ], +) +async def test_sdk_stream_reuses_adapter_or_sends_one_final_message(native, source, mock): + session, adapter, incoming = make_session(native=native, source=source, mock=mock) + api, transport = proxy_for(session) + async with api.reply_stream() as stream: + await stream.update('hello') + await stream.update('hello world') + adapter.send_message.assert_not_awaited() + adapter.reply_message.assert_not_awaited() + assert stream.result['status'] == 'completed' + assert stream.result['text'] == 'hello world' + assert transport.call_action.await_count == 3 + if mock: + adapter.create_message_card.assert_not_awaited() + adapter.reply_message_chunk.assert_not_awaited() + adapter.send_message.assert_not_awaited() + assert stream.result['mock'] is True + elif native and source: + adapter.create_message_card.assert_awaited_once() + delivered_source = adapter.create_message_card.await_args.args[1] + assert delivered_source.source_platform_object is incoming.source_platform_object + chunks = adapter.reply_message_chunk.await_args_list + assert [c.kwargs['bot_message'].all_content for c in chunks] == ['hello', 'hello world', 'hello world'] + assert [c.kwargs['is_final'] for c in chunks] == [False, False, True] + assert chunks[-1].kwargs['bot_message'].tool_calls is None + else: + adapter.reply_message_chunk.assert_not_awaited() + if source: + adapter.reply_message.assert_awaited_once() + assert adapter.reply_message.await_args.kwargs['message'][0].text == 'hello world' + else: + adapter.send_message.assert_awaited_once() + assert adapter.send_message.await_args.args[2][0].text == 'hello world' + await session.close() + + +@pytest.mark.parametrize('native', [True, False]) +@pytest.mark.parametrize('error', [RuntimeError, asyncio.CancelledError]) +async def test_exception_finalizes_visible_card_without_sending_buffered_partial(native, error): + session, adapter, _ = make_session(native=native) + api, _ = proxy_for(session) + with pytest.raises(error): + async with api.reply_stream() as stream: + await stream.update('partial') + raise error() + if native: + assert adapter.reply_message_chunk.await_args.kwargs['is_final'] is True + adapter.reply_message.assert_not_awaited() + adapter.send_message.assert_not_awaited() + count = adapter.reply_message_chunk.await_count + await session.close() + assert adapter.reply_message_chunk.await_count == count + + +@pytest.mark.parametrize('allowed,advertised', [(False, True), (True, False)]) +async def test_missing_permission_or_old_host_fails_before_delivery(allowed, advertised): + session, _, _ = make_session() + api, transport = proxy_for(session, allowed=allowed, advertised=advertised) + with pytest.raises(PermissionDeniedError): + async with api.reply_stream(): + pytest.fail('Not authorized') + transport.call_action.assert_not_awaited() + + +async def test_empty_stream_and_duplicate_finish_do_not_send_twice(): + session, adapter, _ = make_session(native=False) + empty = uuid4() + await session.apply(request(empty, 'finish', '')) + adapter.reply_message.assert_not_awaited() + key = uuid4() + await session.apply(request(key)) + first = await session.apply(request(key, 'finish')) + assert await session.apply(request(key, 'finish')) == first + adapter.reply_message.assert_awaited_once() + with pytest.raises(ValueError, match='closed'): + await session.apply(request(key)) + await session.close() + with pytest.raises(ValueError, match='ended'): + await session.apply(request(uuid4())) + + +async def test_host_cleanup_closes_stream_when_plugin_disappears(): + session, adapter, _ = make_session() + await session.apply(request(uuid4())) + await session.close() + assert adapter.reply_message_chunk.await_args.kwargs['is_final'] is True + + +async def test_uncertain_final_send_is_not_retried_by_finish_or_cleanup(): + session, adapter, _ = make_session(native=False) + adapter.reply_message.side_effect = TimeoutError('Response lost') + key = uuid4() + with pytest.raises(TimeoutError): + await session.apply(request(key, 'finish')) + with pytest.raises(ValueError, match='closed'): + await session.apply(request(key, 'finish')) + await session.close() + adapter.reply_message.assert_awaited_once() + + +async def test_failed_update_closes_existing_card_during_run_cleanup(): + session, adapter, _ = make_session() + adapter.reply_message_chunk.side_effect = [RuntimeError('update failed'), None] + with pytest.raises(RuntimeError): + await session.apply(request(uuid4())) + await session.close() + assert adapter.reply_message_chunk.await_args.kwargs['is_final'] is True + + +async def test_streams_are_isolated_by_run_and_bounded(): + first, a, _ = make_session(native=False) + second, b, _ = make_session(native=False) + key = uuid4() + await first.apply(request(key, 'update', 'first')) + await second.apply(request(key, 'finish', 'second')) + assert b.reply_message.await_args.kwargs['message'][0].text == 'second' + a.reply_message.assert_not_awaited() + for _ in range(15): + await first.apply(request(uuid4(), 'finish', '')) + with pytest.raises(ValueError, match='at most'): + await first.apply(request(uuid4())) + await first.close() + a.reply_message.assert_not_awaited() + + +async def test_event_processor_uses_shared_sdk_api_and_emits_one_trace_for_the_stream(): + from unittest.mock import Mock + from langbot_plugin.api.definition.components.event_processor import EventProcessor + + session, adapter, incoming = make_session() + api, _ = proxy_for(session) + processor = EventProcessor() + processor.get_run_api = Mock(return_value=api) + + @processor.handler(events.MessageReceivedEvent) + async def handle(ctx): + async with ctx.reply_stream() as stream: + await stream.update('one') + await stream.update('one two') + + context = SimpleNamespace( + run_id='run-1', + config={}, + event=SimpleNamespace( + data=incoming.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'}) + ), + ) + results = [result async for result in processor.run(context)] + assert [r.type for r in results] == ['tool.call.started', 'tool.call.completed', 'run.completed'] + assert results[1].data['result']['text'] == 'one two' + assert adapter.reply_message_chunk.await_count == 3 + await session.close() + + +async def test_shared_adapter_uses_host_ids_to_isolate_identical_plugin_stream_ids(): + first, adapter, _ = make_session() + second, _, _ = make_session() + second.adapter = adapter + key = uuid4() + await first.apply(request(key, text='first')) + await second.apply(request(key, text='second')) + ids = [c.args[0] for c in adapter.create_message_card.await_args_list] + assert len(set(ids)) == 2 + assert str(key) not in ids + await first.close() + await second.close() + + +async def test_run_cleanup_cancels_inflight_update_and_finalizes_card(): + session, adapter, _ = make_session() + started = asyncio.Event() + + async def update(**kwargs): + if not kwargs['is_final']: + started.set() + await asyncio.Event().wait() + + adapter.reply_message_chunk.side_effect = update + task = asyncio.create_task(session.apply(request(uuid4()))) + await asyncio.wait_for(started.wait(), 1) + await asyncio.wait_for(session.close(), 1) + with pytest.raises(asyncio.CancelledError): + await task + assert adapter.reply_message_chunk.await_args.kwargs['is_final'] is True diff --git a/tests/unit_tests/api/service/test_agent_service.py b/tests/unit_tests/api/service/test_agent_service.py index 31eea9bee..66778bb9e 100644 --- a/tests/unit_tests/api/service/test_agent_service.py +++ b/tests/unit_tests/api/service/test_agent_service.py @@ -151,7 +151,18 @@ class TestAgentServiceMetadata: class TestAgentServiceDebug: @pytest.mark.parametrize('streaming', [False, True]) - async def test_debug_agent_runs_configured_runner_with_synthetic_event(self, streaming): + @pytest.mark.parametrize( + 'result_type', + [ + 'tool.call.started', + 'tool.call.completed', + 'message.delta', + 'message.completed', + 'processor.log', + 'run.completed', + ], + ) + async def test_debug_agent_runs_configured_runner_with_synthetic_event(self, streaming, result_type): app = _make_app() agent_config = _agent_row().config agent_config['allowed_platform_tools'] = ['platform_get_user_info'] @@ -162,7 +173,7 @@ class TestAgentServiceDebug: agent_config['allowed_tools'] = ['exec', 'weather'] visible_event = { - 'type': 'tool.call.started', + 'type': result_type, 'data': {'tool_name': 'exec', 'parameters': {'command': 'echo hi'}}, } observer = AsyncMock() if streaming else None @@ -171,6 +182,9 @@ class TestAgentServiceDebug: assert binding.delivery_policy.enable_streaming is streaming await adapter_context['_result_observer']({**visible_event, 'private_context': 'must not leak'}) await adapter_context['_result_observer']({'type': 'state.updated', 'data': {'private': True}}) + if streaming: + # Debug events reach the client before the runner returns its final output. + observer.assert_awaited_once_with(visible_event) yield SimpleNamespace( role='assistant', content='debug result', diff --git a/tests/unit_tests/api/test_bots_controller_errors.py b/tests/unit_tests/api/test_bots_controller_errors.py new file mode 100644 index 000000000..12920a83b --- /dev/null +++ b/tests/unit_tests/api/test_bots_controller_errors.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import sys +import types +from importlib import import_module +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import quart + + +core_app_module = types.ModuleType('langbot.pkg.core.app') +core_app_module.Application = object +sys.modules.setdefault('langbot.pkg.core.app', core_app_module) + + +pytestmark = pytest.mark.asyncio + + +async def _create_test_client(bot_service: SimpleNamespace): + app = quart.Quart(__name__) + account = SimpleNamespace( + uuid='account-test', + user='test@example.com', + ) + user_service = SimpleNamespace( + get_authenticated_account=AsyncMock(return_value=account), + ) + access = SimpleNamespace( + workspace=SimpleNamespace(uuid='workspace-test'), + membership=SimpleNamespace( + uuid='membership-test', + role='developer', + projection_revision=1, + ), + execution=SimpleNamespace( + instance_uuid='instance-test', + placement_generation=1, + ), + ) + ap = SimpleNamespace( + bot_service=bot_service, + user_service=user_service, + apikey_service=SimpleNamespace(authenticate_api_key=AsyncMock(return_value=None)), + workspace_collaboration_service=SimpleNamespace(resolve_account_workspace=AsyncMock(return_value=access)), + ) + BotsRouterGroup = import_module('langbot.pkg.api.http.controller.groups.platform.bots').BotsRouterGroup + group = BotsRouterGroup(ap, app) + await group.initialize() + return app.test_client() + + +@pytest.mark.parametrize('method,path', [('post', '/api/v1/platform/bots'), ('put', '/api/v1/platform/bots/bot-1')]) +async def test_bot_config_error_preserves_details(method, path): + error = ValueError('Lark missing required config: app_id, app_secret, bot_name') + service = SimpleNamespace(create_bot=AsyncMock(side_effect=error), update_bot=AsyncMock(side_effect=error)) + client = await _create_test_client(service) + response = await getattr(client, method)( + path, json={'adapter_config': {}}, headers={'Authorization': 'Bearer token'} + ) + assert response.status_code == 400 + body = await response.get_json() + assert body['code'] == 'invalid_bot_config' + assert body['msg'] == str(error) + + +async def test_bot_apply_failure_identifies_persisted_record(): + from langbot.pkg.api.http.service.bot_errors import BotApplyError + + service = SimpleNamespace(create_bot=AsyncMock(side_effect=BotApplyError('Missing app_id', 'saved-bot'))) + client = await _create_test_client(service) + response = await client.post('/api/v1/platform/bots', json={}, headers={'Authorization': 'Bearer token'}) + assert response.status_code == 400 + body = await response.get_json() + assert body.pop('request_id') + assert body == { + 'code': 'bot_apply_failed', + 'msg': 'Missing app_id', + 'data': {'uuid': 'saved-bot'}, + } + + +async def test_unexpected_failure_keeps_request_reference_without_exception_details(): + client = await _create_test_client( + SimpleNamespace(update_bot=AsyncMock(side_effect=RuntimeError('database password'))) + ) + response = await client.put('/api/v1/platform/bots/bot-1', json={}, headers={'Authorization': 'Bearer token'}) + body = await response.get_json() + assert response.status_code == 500 + assert body['request_id'] + assert 'database password' not in str(body) + + +async def test_invalid_request_body_is_actionable(): + service = SimpleNamespace(update_bot=AsyncMock()) + client = await _create_test_client(service) + response = await client.put('/api/v1/platform/bots/bot-1', json=[], headers={'Authorization': 'Bearer token'}) + assert response.status_code == 400 + service.update_bot.assert_not_awaited() + + +async def test_runtime_error_redacts_persisted_secrets_on_partial_update(): + from langbot.pkg.api.http.service.bot import BotService + from langbot.pkg.api.http.service.bot_errors import BotApplyError + + ap = SimpleNamespace( + persistence_mgr=SimpleNamespace(execute_async=AsyncMock(return_value=SimpleNamespace(rowcount=1))), + platform_mgr=SimpleNamespace( + remove_bot=AsyncMock(), + load_bot=AsyncMock(side_effect=ValueError('Invalid app_secret: persisted-secret-value')), + ), + ) + service = BotService(ap) + service.get_bot = AsyncMock( + return_value={'uuid': 'bot-1', 'adapter_config': {'app_secret': 'persisted-secret-value'}} + ) + with pytest.raises(BotApplyError) as captured: + await service.update_bot('workspace-test', 'bot-1', {'enable': True}) + assert str(captured.value) == 'Invalid app_secret: ***' + assert captured.value.bot_uuid == 'bot-1' + ap.persistence_mgr.execute_async.assert_awaited_once() + + +async def test_validation_error_does_not_include_input_values(): + import pydantic + from langbot.pkg.api.http.service.bot_errors import bot_error_message + + class Config(pydantic.BaseModel): + app_secret: int + + try: + Config(app_secret='private-value') + except pydantic.ValidationError as error: + result = bot_error_message(error, {'app_secret': 'private-value'}) + assert 'app_secret' in result + assert 'private-value' not in result + assert 'input_value' not in result diff --git a/tests/unit_tests/platform/test_routing_rules.py b/tests/unit_tests/platform/test_routing_rules.py index 65a2c6682..96da96e3d 100644 --- a/tests/unit_tests/platform/test_routing_rules.py +++ b/tests/unit_tests/platform/test_routing_rules.py @@ -447,7 +447,7 @@ class TestEBAEventBindings: 'event_get_actor', ] assert binding.delivery_policy.enable_streaming is False - assert binding.delivery_policy.enable_reply is True + assert binding.delivery_policy.enable_reply is False assert binding.delivery_policy.enable_interactions is True assert binding.state_policy.state_scopes == ['conversation', 'actor', 'subject', 'runner'] assert binding.agent_id == 'agent-1' @@ -651,7 +651,9 @@ class TestInteractionResumeRouting: assert binding.processor_type == 'agent' assert binding.processor_id == 'agent-1' assert adapter_context == {'_delivery_adapter': adapter} - adapter.send_message.assert_awaited_once() + assert envelope.delivery.supports_streaming is False + assert binding.delivery_policy.enable_reply is False + adapter.send_message.assert_not_awaited() def test_agent_product_to_binding_does_not_fallback_to_component_ref(self): """An empty config runner stays unconfigured even if component_ref is stale.""" @@ -762,3 +764,89 @@ async def test_bound_event_processor_receives_one_complete_typed_event(): assert envelope.data['type'] == 'group.member_joined' assert 'source_platform_object' not in envelope.data bot.ap.plugin_connector.emit_event.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('kind', ['agent', 'event_processor']) +@pytest.mark.parametrize('output_kind', ['message', 'chunks', 'tool_rounds']) +@pytest.mark.parametrize('explicit_reply', [False, True]) +@pytest.mark.parametrize('runner_fails', [False, True]) +async def test_processor_outputs_require_explicit_platform_actions(kind, output_kind, explicit_reply, runner_fails): + """Draining runner results must not send text, duplicate replies, or stream cards.""" + from langbot_plugin.api.entities.builtin.platform import entities, events, message + from langbot_plugin.api.entities.builtin.provider import message as provider_message + + from langbot.pkg.agent.runner.platform_tools import execute_platform_tool, freeze_platform_context + + bot = TestEventRouteTrace._make_bot( + [{'id': 'binding-1', 'event_pattern': 'message.received', 'target_type': kind, 'target_uuid': 'agent-1'}] + ) + adapter = SimpleNamespace( + get_supported_apis=lambda: ['send_message'], + is_stream_output_supported=AsyncMock(return_value=True), + send_message=AsyncMock(return_value='message-2'), + create_message_card=AsyncMock(), + reply_message_chunk=AsyncMock(), + ) + completed = [] + + async def run(envelope, binding, adapter_context): + assert envelope.delivery.supports_streaming is False + assert binding.delivery_policy.enable_streaming is False + assert binding.delivery_policy.enable_reply is False + assert adapter_context['_delivery_adapter'] is adapter + if explicit_reply: + session = {'authorization': {'bot_id': 'bot-1', 'platform_context': freeze_platform_context(envelope)}} + await execute_platform_tool(bot.ap, TEST_CONTEXT, session, 'event_reply', {'text': 'Working on it'}) + # Progress arrives while the runner is still working. + adapter.send_message.assert_awaited_once() + assert completed == [] + if output_kind == 'chunks': + yield provider_message.MessageChunk(role='assistant', content='Done', all_content='Done') + yield provider_message.MessageChunk(role='assistant', content='.', all_content='Done.', is_final=True) + elif output_kind == 'tool_rounds': + yield provider_message.Message(role='assistant', content='Checking the request') + yield provider_message.Message(role='assistant', content='Done.') + else: + yield provider_message.Message(role='assistant', content='Done.') + if runner_fails: + raise RuntimeError('Runner failed after producing text') + completed.append(True) + + bot.ap = SimpleNamespace( + workspace_service=active_workspace_service(), + platform_mgr=SimpleNamespace(get_bot_by_uuid=AsyncMock(return_value=SimpleNamespace(adapter=adapter))), + agent_service=SimpleNamespace( + get_agent=AsyncMock( + return_value={ + 'uuid': 'agent-1', + 'kind': kind, + 'enabled': True, + 'supported_event_patterns': ['message.received'], + 'config': {'runner': {'id': 'runner-1'}, 'runner_config': {'runner-1': {}}}, + } + ) + ), + agent_run_orchestrator=SimpleNamespace(run=run), + ) + event = events.MessageReceivedEvent( + message_id='message-1', + message_chain=message.MessageChain([message.Plain(text='hello')]), + sender=entities.User(id='user-1', nickname='QA'), + chat_type=entities.ChatType.PRIVATE, + chat_id='user-1', + ) + trace = await bot._dispatch_eba_event_to_processor(event, adapter) + + assert trace['status'] == ('failed' if runner_fails else 'delivered') + if runner_fails: + assert trace['failure_code'] == 'runner_failed' + assert completed == ([] if runner_fails else [True]) + assert adapter.send_message.await_count == int(explicit_reply) + if explicit_reply: + kwargs = adapter.send_message.await_args.kwargs + assert kwargs['target_type'] == 'person' + assert kwargs['target_id'] == 'user-1' + assert kwargs['message'][0].text == 'Working on it' + adapter.create_message_card.assert_not_awaited() + adapter.reply_message_chunk.assert_not_awaited() diff --git a/tests/unit_tests/plugin/test_handler_actions.py b/tests/unit_tests/plugin/test_handler_actions.py index be3784c4c..1e3da2d6c 100644 --- a/tests/unit_tests/plugin/test_handler_actions.py +++ b/tests/unit_tests/plugin/test_handler_actions.py @@ -1575,9 +1575,7 @@ class TestAgentRunProxyActions: query = build_execution_query(event, []) app.box_service = SimpleNamespace( available=True, - get_backend_status=AsyncMock( - return_value={'backend': {'available': True}} - ), + get_backend_status=AsyncMock(return_value={'backend': {'available': True}}), execute_tool=AsyncMock( return_value={ 'ok': True, @@ -1683,3 +1681,88 @@ class TestAgentRunProxyActions: provider.invoke_rerank.assert_awaited_once() kwargs = provider.invoke_rerank.await_args.kwargs assert kwargs['extra_args'] == {'top_n': 2, 'return_documents': False} + + +@pytest.mark.parametrize( + 'case', + [ + 'valid', + 'expired', + 'plugin', + 'workspace', + 'permission', + 'shadowed', + 'native', + 'bot_removed', + 'adapter_replaced', + 'api_revoked', + ], +) +@pytest.mark.skipif(not hasattr(PluginToRuntimeAction, 'REPLY_STREAM'), reason='SDK does not support streaming replies') +async def test_reply_stream_authorization_is_run_and_workspace_scoped(case): + from uuid import uuid4 + from langbot.pkg.agent.runner.session_registry import get_session_registry + + app = SimpleNamespace(logger=Mock(), _test_plugin_identity='test/runner') + runtime_handler = make_handler(app) + query = SimpleNamespace( + **{ + field: getattr(TEST_EXECUTION_CONTEXT, field) + for field in ('instance_uuid', 'workspace_uuid', 'placement_generation') + } + ) + if case == 'workspace': + query.workspace_uuid = 'another-workspace' + streams = SimpleNamespace(mock=True, apply=AsyncMock(return_value={'status': 'completed'})) + if case in {'native', 'bot_removed', 'adapter_replaced', 'api_revoked'}: + streams.mock = False + streams.adapter = SimpleNamespace(get_supported_apis=lambda: [] if case == 'api_revoked' else ['send_message']) + bot = ( + None + if case == 'bot_removed' + else SimpleNamespace(adapter=object() if case == 'adapter_replaced' else streams.adapter) + ) + app.platform_mgr = SimpleNamespace(get_bot_by_uuid=AsyncMock(return_value=bot)) + registry = get_session_registry() + run_id = str(uuid4()) + resources = make_agent_resources( + tools=[] + if case == 'permission' + else [ + { + 'tool_name': 'event_reply', + 'operations': ['call'], + 'source': 'mcp' if case == 'shadowed' else 'platform', + 'source_id': 'event_reply', + } + ] + ) + await registry.register( + run_id=run_id, + runner_id='plugin:test/runner/default', + query_id=None, + plugin_identity='other/plugin' if case == 'plugin' else 'test/runner', + resources=resources, + execution_query=query, + reply_streams=streams, + ) + try: + if case == 'expired': + await registry.unregister(run_id) + response = await runtime_handler.actions[PluginToRuntimeAction.REPLY_STREAM.value]( + { + 'run_id': run_id, + 'caller_plugin_identity': 'test/runner', + 'stream_id': str(uuid4()), + 'operation': 'finish', + 'text': 'hello', + } + ) + if case in {'valid', 'native'}: + assert response.code == 0 + streams.apply.assert_awaited_once() + else: + assert response.code != 0 + streams.apply.assert_not_awaited() + finally: + await registry.unregister(run_id) diff --git a/uv.lock b/uv.lock index 0a4bdf08b..2faa92602 100644 --- a/uv.lock +++ b/uv.lock @@ -2119,7 +2119,7 @@ requires-dist = [ { name = "ebooklib", specifier = ">=0.18" }, { name = "gewechat-client", specifier = ">=0.1.5" }, { name = "html2text", specifier = ">=2024.2.26" }, - { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk?rev=f82b3ce935f9a33afee389fb39fe8dc29a45b615" }, + { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk?rev=ca0671b81db5b2ed937d01158a3982de3fd8500f" }, { name = "langchain", specifier = ">=1.3.9" }, { name = "langchain-core", specifier = ">=1.3.3" }, { name = "langchain-text-splitters", specifier = ">=1.1.2" }, @@ -2186,7 +2186,7 @@ dev = [ [[package]] name = "langbot-plugin" version = "0.5.5" -source = { git = "https://github.com/langbot-app/langbot-plugin-sdk?rev=f82b3ce935f9a33afee389fb39fe8dc29a45b615#f82b3ce935f9a33afee389fb39fe8dc29a45b615" } +source = { git = "https://github.com/langbot-app/langbot-plugin-sdk?rev=ca0671b81db5b2ed937d01158a3982de3fd8500f#ca0671b81db5b2ed937d01158a3982de3fd8500f" } dependencies = [ { name = "aiofiles" }, { name = "aiohttp" }, diff --git a/web/src/app/home/bots/BotDetailContent.tsx b/web/src/app/home/bots/BotDetailContent.tsx index 47fdd9a14..a42a02b09 100644 --- a/web/src/app/home/bots/BotDetailContent.tsx +++ b/web/src/app/home/bots/BotDetailContent.tsx @@ -24,6 +24,7 @@ import { useTranslation } from 'react-i18next'; import { Settings, FileText, Users, RefreshCw, Trash2 } from 'lucide-react'; import { cn } from '@/lib/utils'; import { toast } from 'sonner'; +import { showBotError } from './bot-error'; import { useCurrentWorkspace } from '@/app/infra/http'; import { Bot } from '@/app/infra/entities/api'; import EntityBasicInfoDialog, { @@ -91,9 +92,9 @@ export default function BotDetailContent({ id }: { id: string }) { current ? { ...current, enable: checked } : current, ); refreshBots(); - } catch { + } catch (error) { setBotEnabled(prev); - toast.error(t('bots.setBotEnableError')); + showBotError(error, t('bots.setBotEnableError'), t); } }, [id, botEnabled, refreshBots, t], @@ -129,11 +130,7 @@ export default function BotDetailContent({ id }: { id: string }) { await refreshBots(); toast.success(t('bots.saveSuccess')); } catch (error) { - const message = - typeof error === 'object' && error && 'msg' in error - ? String((error as { msg?: string }).msg || '') - : ''; - toast.error(t('bots.saveError') + message); + showBotError(error, t('bots.saveError'), t); throw error; } } diff --git a/web/src/app/home/bots/bot-error.ts b/web/src/app/home/bots/bot-error.ts new file mode 100644 index 000000000..a0ca6d642 --- /dev/null +++ b/web/src/app/home/bots/bot-error.ts @@ -0,0 +1,31 @@ +import type { TFunction } from 'i18next'; +import { toast } from 'sonner'; + +export function showBotError(error: unknown, title: string, t: TFunction) { + const detail = + error && typeof error === 'object' + ? (error as { + code?: string; + msg?: string; + message?: string; + request_id?: string; + }) + : {}; + const message = detail.msg || detail.message || ''; + const description = + detail.code === 'internal_error' + ? [ + t('bots.internalErrorHint'), + detail.request_id && + t('bots.errorReference', { id: detail.request_id }), + ] + .filter(Boolean) + .join('\n') + : message; + toast.error( + detail.code === 'bot_apply_failed' + ? t('bots.applyFailed') + : title.replace(/[::]\s*$/, ''), + { description, duration: 10000 }, + ); +} diff --git a/web/src/app/home/bots/components/bot-form/BotForm.tsx b/web/src/app/home/bots/components/bot-form/BotForm.tsx index 9585d1582..dd110edf3 100644 --- a/web/src/app/home/bots/components/bot-form/BotForm.tsx +++ b/web/src/app/home/bots/components/bot-form/BotForm.tsx @@ -1,3 +1,4 @@ +import { showBotError } from '../../bot-error'; import React, { forwardRef, useEffect, @@ -413,7 +414,7 @@ const BotForm = forwardRef(function BotForm( toast.success(t('bots.saveSuccess')); }) .catch((err) => { - toast.error(t('bots.saveError') + err.msg); + showBotError(err, t('bots.saveError'), t); }) .finally(() => { setIsLoading(false); @@ -438,7 +439,10 @@ const BotForm = forwardRef(function BotForm( onNewBotCreated(res.uuid); }) .catch((err) => { - toast.error(t('bots.createError') + err.msg); + showBotError(err, t('bots.createError'), t); + if (err.code === 'bot_apply_failed' && err.data?.uuid) { + onNewBotCreated(err.data.uuid); + } }) .finally(() => { setIsLoading(false); diff --git a/web/src/app/infra/http/BaseHttpClient.ts b/web/src/app/infra/http/BaseHttpClient.ts index ec2988ea1..7d3747fcd 100644 --- a/web/src/app/infra/http/BaseHttpClient.ts +++ b/web/src/app/infra/http/BaseHttpClient.ts @@ -139,6 +139,9 @@ export abstract class BaseHttpClient { code: data?.code || status, msg: errMsg, data: data?.data || null, + request_id: + (data as { request_id?: string })?.request_id || + error.response.headers['x-request-id'], }); } diff --git a/web/src/i18n/locales/en-US.ts b/web/src/i18n/locales/en-US.ts index a1da79b87..47f869482 100644 --- a/web/src/i18n/locales/en-US.ts +++ b/web/src/i18n/locales/en-US.ts @@ -340,6 +340,11 @@ const enUS = { }, }, bots: { + applyFailed: 'Configuration saved, but could not be applied', + internalErrorHint: + 'An unexpected error occurred. Check the backend logs using the reference below.', + errorReference: 'Error reference: {{id}}', + title: 'Bots', description: 'Create and manage bots, which are the entry points for LangBot to connect with various platforms', diff --git a/web/src/i18n/locales/es-ES.ts b/web/src/i18n/locales/es-ES.ts index bd4ae544b..e28dabde3 100644 --- a/web/src/i18n/locales/es-ES.ts +++ b/web/src/i18n/locales/es-ES.ts @@ -348,6 +348,11 @@ const esES = { "Uses the Workspace owner's LangBot Account billing and credits.", }, bots: { + applyFailed: 'Configuración guardada, pero no se pudo aplicar', + internalErrorHint: + 'Se produjo un error interno. Consulta los registros del servidor con esta referencia.', + errorReference: 'Referencia del error: {{id}}', + adapterEventDebugAction: 'Probar escucha', title: 'Bots', description: diff --git a/web/src/i18n/locales/ja-JP.ts b/web/src/i18n/locales/ja-JP.ts index 7e5b0255b..2e89194c8 100644 --- a/web/src/i18n/locales/ja-JP.ts +++ b/web/src/i18n/locales/ja-JP.ts @@ -346,6 +346,11 @@ const jaJP = { }, }, bots: { + applyFailed: '設定を保存しましたが、適用に失敗しました', + internalErrorHint: + '内部エラーが発生しました。エラー番号でバックエンドのログを確認してください。', + errorReference: 'エラー番号: {{id}}', + title: 'ボット', description: 'ボットの作成と管理を行います。LangBotと各プラットフォームを接続するためのエントリーポイントです', diff --git a/web/src/i18n/locales/ru-RU.ts b/web/src/i18n/locales/ru-RU.ts index 0ed094338..c7ea219fc 100644 --- a/web/src/i18n/locales/ru-RU.ts +++ b/web/src/i18n/locales/ru-RU.ts @@ -346,6 +346,11 @@ const ruRU = { "Uses the Workspace owner's LangBot Account billing and credits.", }, bots: { + applyFailed: 'Настройки сохранены, но не применены', + internalErrorHint: + 'Внутренняя ошибка. Проверьте журналы сервера по указанному идентификатору.', + errorReference: 'Идентификатор ошибки: {{id}}', + adapterEventDebugAction: 'Тест прослушивания', title: 'Боты', description: diff --git a/web/src/i18n/locales/th-TH.ts b/web/src/i18n/locales/th-TH.ts index c2027cd9c..cf1dd7dfc 100644 --- a/web/src/i18n/locales/th-TH.ts +++ b/web/src/i18n/locales/th-TH.ts @@ -333,6 +333,11 @@ const thTH = { "Uses the Workspace owner's LangBot Account billing and credits.", }, bots: { + applyFailed: 'บันทึกการตั้งค่าแล้ว แต่ไม่สามารถนำไปใช้ได้', + internalErrorHint: + 'เกิดข้อผิดพลาดภายใน โปรดตรวจสอบบันทึกของเซิร์ฟเวอร์ด้วยหมายเลขอ้างอิง', + errorReference: 'หมายเลขข้อผิดพลาด: {{id}}', + adapterEventDebugAction: 'ทดสอบการรับเหตุการณ์', title: 'บอท', description: diff --git a/web/src/i18n/locales/vi-VN.ts b/web/src/i18n/locales/vi-VN.ts index ea13094fe..a0d3522f0 100644 --- a/web/src/i18n/locales/vi-VN.ts +++ b/web/src/i18n/locales/vi-VN.ts @@ -342,6 +342,11 @@ const viVN = { "Uses the Workspace owner's LangBot Account billing and credits.", }, bots: { + applyFailed: 'Đã lưu cấu hình nhưng không thể áp dụng', + internalErrorHint: + 'Đã xảy ra lỗi nội bộ. Hãy kiểm tra nhật ký máy chủ bằng mã lỗi.', + errorReference: 'Mã lỗi: {{id}}', + adapterEventDebugAction: 'Kiểm tra lắng nghe', title: 'Bot', description: diff --git a/web/src/i18n/locales/zh-Hans.ts b/web/src/i18n/locales/zh-Hans.ts index 6add74047..c6f7d480b 100644 --- a/web/src/i18n/locales/zh-Hans.ts +++ b/web/src/i18n/locales/zh-Hans.ts @@ -325,6 +325,10 @@ const zhHans = { }, }, bots: { + applyFailed: '配置已保存,但应用失败', + internalErrorHint: '发生内部错误,请通过错误编号查看后端日志。', + errorReference: '错误编号:{{id}}', + title: '机器人', description: '创建和管理机器人,这是 LangBot 与各个平台连接的入口', createBot: '创建机器人', diff --git a/web/src/i18n/locales/zh-Hant.ts b/web/src/i18n/locales/zh-Hant.ts index 75edf36ff..2c4c0924c 100644 --- a/web/src/i18n/locales/zh-Hant.ts +++ b/web/src/i18n/locales/zh-Hant.ts @@ -322,6 +322,10 @@ const zhHant = { "Uses the Workspace owner's LangBot Account billing and credits.", }, bots: { + applyFailed: '設定已儲存,但套用失敗', + internalErrorHint: '發生內部錯誤,請透過錯誤編號查看後端日誌。', + errorReference: '錯誤編號:{{id}}', + adapterEventDebugAction: '測試監聽', title: '機器人', description: '建立和管理機器人,這是 LangBot 與各個平台連接的入口', diff --git a/web/tests/e2e/bot-save-errors.spec.ts b/web/tests/e2e/bot-save-errors.spec.ts new file mode 100644 index 000000000..fd2da4ea9 --- /dev/null +++ b/web/tests/e2e/bot-save-errors.spec.ts @@ -0,0 +1,55 @@ +import { expect, test } from '@playwright/test'; +import { installLangBotApiMocks } from './fixtures/langbot-api'; + +for (const failure of [ + { + status: 400, + code: 'invalid_bot_config', + msg: 'Lark missing required config: app_id, app_secret, bot_name', + }, + { + status: 400, + code: 'bot_apply_failed', + msg: 'Lark missing required config: app_id, app_secret, bot_name', + }, + { + status: 500, + code: 'internal_error', + msg: 'Internal server error', + request_id: 'bot-save-test-reference', + }, +]) { + test(`bot save displays actionable ${failure.code}`, async ({ page }) => { + await installLangBotApiMocks(page, { authenticated: true }); + await page.goto('/home/bots?id=new'); + await page.getByRole('combobox').click(); + await page.getByRole('option', { name: 'Playwright Adapter' }).click(); + await page.locator('input[name="name"]').fill('Error Test Bot'); + await page.getByRole('button', { name: /^Submit$/ }).click(); + await expect(page).toHaveURL(/\/home\/bots\?id=bot-1$/); + await page.route('**/api/v1/platform/bots/bot-1', async (route) => { + if (route.request().method() !== 'PUT') return route.fallback(); + await route.fulfill({ + status: failure.status, + contentType: 'application/json', + body: JSON.stringify(failure), + }); + }); + await page.getByRole('button', { name: 'Edit basic information' }).click(); + const dialog = page.getByRole('dialog'); + await dialog.getByLabel('Name', { exact: true }).fill('Edited Bot'); + await dialog.getByRole('button', { name: /^Save$/ }).click(); + if (failure.code === 'internal_error') { + await expect( + page.getByText('Error reference: bot-save-test-reference'), + ).toBeVisible(); + } else { + await expect(page.getByText(failure.msg, { exact: true })).toBeVisible(); + } + if (failure.code === 'bot_apply_failed') { + await expect( + page.getByText('Configuration saved, but could not be applied'), + ).toBeVisible(); + } + }); +}