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 9daf15900..e7fb25019 100644 --- a/src/langbot/libs/wecom_ai_bot_api/ws_client.py +++ b/src/langbot/libs/wecom_ai_bot_api/ws_client.py @@ -60,6 +60,7 @@ _UPLOAD_CHUNK_SIZE = 512 * 1024 _DEDUP_CACHE_MAX = 4096 _STREAM_CACHE_MAX = 1024 +_INTERACTION_STREAM_MAX_AGE_SECONDS = 570 _FEEDBACK_CACHE_MAX = 4096 _PENDING_FORM_MAX = 1024 _PENDING_FORM_TTL_SECONDS = 1800 @@ -140,6 +141,7 @@ class WecomBotWsClient: self._completed_stream_content: dict[str, str] = {} # Stream session info for feedback tracking self._stream_sessions: dict[str, dict] = {} # msg_id -> session info + self._interaction_stream_origins: dict[str, dict] = {} # msg_id -> original callback context # Feedback tracking: feedback_id -> session info self._feedback_sessions: dict[str, dict] = {} # feedback_id -> {msg_id, user_id, chat_id, stream_id, req_id} # msg_id -> feedback_id (for associating feedback with message) @@ -177,6 +179,72 @@ class WecomBotWsClient: if task_id: self._pending_forms_by_task.pop(task_id, None) + async def finish_stream_for_interaction(self, msg_id: str) -> bool: + """Release an unused stream or finish text already shown before a card.""" + key = self._stream_ids.get(msg_id) + if not key: + return True + session = self._stream_sessions.get(msg_id, {}) + if not self._stream_sessions.get(msg_id, {}).get('stream_started'): + self._remember_interaction_stream_origin(msg_id, session) + self._stream_ids.pop(msg_id, None) + self._stream_last_content.pop(msg_id, None) + self._stream_sessions.pop(msg_id, None) + return True + req_id, stream_id = key.split('|', 1) + content = self._stream_last_content.get(msg_id, '') + try: + ack = await self.reply_stream(req_id, stream_id, content, finish=True) + except Exception: + await self.logger.warning(f'Failed to finish WeCom loading stream: {traceback.format_exc()}') + return False + if not isinstance(ack, dict) or ack.get('errcode', 0) != 0: + detail = ack.get('errmsg', ack.get('errcode')) if isinstance(ack, dict) else 'no acknowledgement' + await self.logger.warning(f'Failed to finish WeCom loading stream: {detail}') + return False + self._remember_interaction_stream_origin(msg_id, session) + self._stream_ids.pop(msg_id, None) + self._stream_last_content.pop(msg_id, None) + self._stream_sessions.pop(msg_id, None) + return True + + def _remember_interaction_stream_origin(self, msg_id: str, session: dict) -> None: + if session.get('req_id'): + self._interaction_stream_origins[msg_id] = { + key: session.get(key) for key in ('req_id', 'user_id', 'chat_id', 'created_at') + } + self._cap_mapping(self._interaction_stream_origins, _STREAM_CACHE_MAX) + + def begin_interaction_stream(self, msg_id: str, target_type: str, target_id: str) -> bool: + """Use a fresh stream ID so the resumed answer appears after the card.""" + origin = self._interaction_stream_origins.get(msg_id) + if not origin or msg_id in self._stream_ids: + return False + age = time.monotonic() - float(origin.get('created_at') or 0) + if age >= _INTERACTION_STREAM_MAX_AGE_SECONDS: + self._interaction_stream_origins.pop(msg_id, None) + return False + expected = origin.get('chat_id') if target_type == 'group' else origin.get('user_id') + if target_type not in {'group', 'person', 'private'} or str(expected or '') != target_id: + return False + stream_id = _generate_req_id('stream') + self._stream_ids[msg_id] = f'{origin["req_id"]}|{stream_id}' + self._stream_sessions[msg_id] = {**origin, 'stream_id': stream_id, 'stream_started': False} + self._prune_stream_state() + return True + + async def push_interaction_stream_chunk(self, msg_id: str, content: str, is_final: bool) -> bool: + first_frame = not self._stream_sessions.get(msg_id, {}).get('stream_started') + if await self.push_stream_chunk(msg_id, content, is_final=is_final): + if first_frame and (is_final or self._stream_sessions.get(msg_id, {}).get('stream_started')): + await self.logger.info(f'WeCom interaction reply stream accepted for message {msg_id}') + return True + await self.logger.warning(f'WeCom interaction reply stream unavailable for message {msg_id}') + self._stream_ids.pop(msg_id, None) + self._stream_last_content.pop(msg_id, None) + self._stream_sessions.pop(msg_id, None) + return False + def _prune_pending_forms(self) -> None: cutoff = time.monotonic() - _PENDING_FORM_TTL_SECONDS for task_id, pending in tuple(self._pending_forms_by_task.items()): @@ -254,6 +322,7 @@ class WecomBotWsClient: self._stream_ids.clear() self._stream_last_content.clear() self._stream_sessions.clear() + self._interaction_stream_origins.clear() self._feedback_sessions.clear() self._msg_feedback_ids.clear() self._pending_forms_by_task.clear() @@ -727,8 +796,13 @@ class WecomBotWsClient: # WeCom replaces the displayed stream content on each refresh, so # every frame must contain the complete snapshot, not only a delta. - await self.reply_stream(req_id, stream_id, next_content, finish=is_final, feedback_id=feedback_id) + ack = await self.reply_stream(req_id, stream_id, next_content, finish=is_final, feedback_id=feedback_id) + if not isinstance(ack, dict) or ack.get('errcode', 0) != 0: + detail = ack.get('errmsg', ack.get('errcode')) if isinstance(ack, dict) else 'no acknowledgement' + raise RuntimeError(f'WeCom stream frame rejected: {detail}') self._stream_last_content[msg_id] = next_content + if msg_id in self._stream_sessions: + self._stream_sessions[msg_id]['stream_started'] = True 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) @@ -962,15 +1036,11 @@ class WecomBotWsClient: 'user_id': message_data.get('userid', ''), 'chat_id': message_data.get('chatid', ''), 'chat_type': message_data.get('type', 'single'), + 'stream_started': False, + 'created_at': time.monotonic(), } + self._remember_interaction_stream_origin(msg_id, self._stream_sessions[msg_id]) self._prune_stream_state() - # Send an initial empty stream frame so the WeCom client - # shows its built-in loading spinner while the pipeline - # processes the message (e.g. RAG retrieval). - try: - await self.reply_stream(req_id, stream_id, '', finish=False) - except Exception: - await self.logger.warning(f'Failed to send initial stream frame: {traceback.format_exc()}') message_data['stream_id'] = stream_id message_data['req_id'] = req_id diff --git a/src/langbot/pkg/agent/runner/interaction_manager.py b/src/langbot/pkg/agent/runner/interaction_manager.py index 7553c3450..083f8516b 100644 --- a/src/langbot/pkg/agent/runner/interaction_manager.py +++ b/src/langbot/pkg/agent/runner/interaction_manager.py @@ -402,7 +402,12 @@ class InteractionManager: except pydantic.ValidationError: return - async def acknowledge_submission(self, record: dict[str, typing.Any], adapter: typing.Any) -> None: + async def acknowledge_submission( + self, + record: dict[str, typing.Any], + adapter: typing.Any, + callback_context: dict[str, typing.Any] | None = None, + ) -> None: """Best-effort transition of submitted controls into a read-only state.""" delivery_result = record.get('delivery_result') if not isinstance(delivery_result, dict) or not self._supports_platform_api( @@ -416,6 +421,7 @@ class InteractionManager: 'request': record.get('request') or {}, 'submission': record.get('submission') or {}, 'update_target': delivery_result, + **({'callback_context': callback_context} if callback_context else {}), }, ) if isinstance(delivery_result, dict): diff --git a/src/langbot/pkg/agent/runner/interaction_store.py b/src/langbot/pkg/agent/runner/interaction_store.py index ccd8178b9..ae110644a 100644 --- a/src/langbot/pkg/agent/runner/interaction_store.py +++ b/src/langbot/pkg/agent/runner/interaction_store.py @@ -375,6 +375,103 @@ class InteractionStore: raise InteractionNotFoundError('Interaction disappeared after submission') return self._to_dict(refreshed) + async def consume_text_reply( + self, + *, + workspace_id: str, + bot_id: str, + conversation_id: str, + actor_id: str, + binding_id: str, + runner_id: str, + processor_id: str, + text: str, + message_id: str, + message_time: int | float | None = None, + ) -> dict[str, typing.Any] | None: + """Consume the latest scoped, plain-text form waiting for this user.""" + answer = text.strip() + if ( + not answer + or len(answer) > 20000 + or not all( + (workspace_id, bot_id, conversation_id, actor_id, binding_id, runner_id, processor_id, message_id) + ) + ): + return None + now = _utc_now() + source_marker = 'text-reply:' + _token_hash(message_id) + scope = ( + AgentInteraction.workspace_id == workspace_id, + AgentInteraction.bot_id == bot_id, + AgentInteraction.conversation_id == conversation_id, + AgentInteraction.actor_id == actor_id, + AgentInteraction.binding_id == binding_id, + AgentInteraction.runner_id == runner_id, + AgentInteraction.processor_type == 'agent', + AgentInteraction.processor_id == processor_id, + ) + async with self._session_factory() as session: + duplicate = await session.execute( + sqlalchemy.select(AgentInteraction.id) + .where( + *scope, + AgentInteraction.status == 'submitted', + AgentInteraction.status_reason == source_marker, + ) + .limit(1) + ) + if duplicate.scalar_one_or_none() is not None: + return {'duplicate_text_reply': True} + result = await session.execute( + sqlalchemy.select(AgentInteraction) + .where(*scope) + .order_by(AgentInteraction.created_at.desc(), AgentInteraction.id.desc()) + .limit(1) + ) + row = result.scalar_one_or_none() + if row is None or row.status != 'pending' or (row.expires_at and _as_utc(row.expires_at) <= now): + return None + if message_time is not None and float(message_time) < _as_utc(row.created_at).timestamp() - 1: + return None + delivery = _json_loads(row.delivery_result_json, None) + request = _json_loads(row.request_json, {}) + fields = request.get('fields') if isinstance(request, dict) else None + if not (isinstance(delivery, dict) and delivery.get('ok') is True and delivery.get('rich') is False): + return None + if not ( + isinstance(fields, list) + and len(fields) == 1 + and isinstance(fields[0], dict) + and fields[0].get('type') in {'text', 'textarea'} + and fields[0].get('id') + and not request.get('actions') + ): + return None + submission = { + 'interaction_id': row.interaction_id, + 'action_id': None, + 'values': {fields[0]['id']: answer}, + 'submitted_at': int(now.timestamp()), + } + updated = await session.execute( + sqlalchemy.update(AgentInteraction) + .where(AgentInteraction.id == row.id, AgentInteraction.status == 'pending') + .values( + status='submitted', + status_reason=source_marker, + submitted_at=_db_datetime(now), + submission_json=_json_dumps(submission), + updated_at=_db_datetime(now), + ) + ) + if updated.rowcount != 1: + await session.rollback() + return None + await session.commit() + refreshed = await session.get(AgentInteraction, row.id) + return self._to_dict(refreshed) + async def mark_delivery_failed(self, run_id: str, interaction_id: str, reason: str) -> bool: """Mark an undelivered pending interaction terminal.""" now = _utc_now() diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index cec54c631..4909f3dcc 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -539,6 +539,16 @@ class AgentRunOrchestrator: auto_reply_deferred = False explicit_reply_text = None + if auto_reply_enabled and result_type == 'run.completed' and auto_reply_stream_open: + await reply_streams.apply( + ReplyStreamRequest( + stream_id=auto_reply_stream_id, + operation='finish', + text=auto_reply_stream_text, + ) + ) + auto_reply_stream_open = False + if result is not None: yield result diff --git a/src/langbot/pkg/agent/runner/reply_stream.py b/src/langbot/pkg/agent/runner/reply_stream.py index b9beb8817..609c5f9c8 100644 --- a/src/langbot/pkg/agent/runner/reply_stream.py +++ b/src/langbot/pkg/agent/runner/reply_stream.py @@ -3,6 +3,8 @@ from __future__ import annotations import asyncio +import logging +import time from dataclasses import dataclass, field from typing import Literal from uuid import UUID, uuid4 @@ -12,6 +14,9 @@ from langbot_plugin.api.entities.builtin.platform import events, message from langbot_plugin.api.entities.builtin.provider import message as provider_message +logger = logging.getLogger(__name__) + + class ReplyStreamRequest(pydantic.BaseModel): model_config = pydantic.ConfigDict(extra='forbid') @@ -26,6 +31,8 @@ class _Reply: text: str = '' started: bool = False native: bool = False + interaction_origin: str = '' + last_sent_at: float = 0.0 final_attempted: bool = False sequence: int = 0 status: str = 'open' @@ -41,6 +48,11 @@ class ReplyStreamSession: if not isinstance(self.source, events.MessageEvent): self.source = None self.target = dict(event.delivery.reply_target or {}) + self.interaction_origin = ( + str(self.target.get('message_id') or '') + if getattr(event, 'event_type', None) == 'interaction.submitted' + else '' + ) self.mock = ( event.delivery.surface == 'webui' and (event.delivery.platform_capabilities or {}).get('debug_mock') is True ) @@ -91,12 +103,24 @@ class ReplyStreamSession: and callable(getattr(self.adapter, 'create_message_card_for_target', None)) and callable(getattr(self.adapter, 'reply_message_chunk_for_target', None)) ) + interaction_streaming = ( + self.source is None + and bool(self.interaction_origin) + and callable(getattr(self.adapter, 'begin_interaction_stream', None)) + and callable(getattr(self.adapter, 'reply_interaction_stream_chunk', None)) + ) reply.native = ( - self.source is not None or target_streaming + self.source is not None or target_streaming or interaction_streaming ) and await self.adapter.is_stream_output_supported() if reply.native: if self.source is not None: await self.adapter.create_message_card(reply.delivery_id, self.source) + elif interaction_streaming: + reply.native = await self.adapter.begin_interaction_stream( + self.interaction_origin, self.target + ) + if reply.native: + reply.interaction_origin = self.interaction_origin else: target_type, target_id = self.target.get('target_type'), self.target.get('target_id') if not target_type or not target_id: @@ -105,8 +129,30 @@ class ReplyStreamSession: reply.delivery_id, str(target_type), str(target_id) ) 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: + should_refresh = ( + request.operation != 'update' + or not reply.interaction_origin + or reply.sequence == 0 + or time.monotonic() - reply.last_sent_at >= 0.5 + ) + if should_refresh: + try: + await self._update_native(key, reply, final=request.operation != 'update') + reply.last_sent_at = time.monotonic() + except Exception: + if not reply.interaction_origin: + raise + logger.warning( + 'WeCom interaction stream failed; sending the final reply proactively', exc_info=True + ) + reply.native = False + if ( + reply.started + and not reply.native + and 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 @@ -154,6 +200,12 @@ class ReplyStreamSession: quote_origin=False, is_final=final, ) + elif reply.interaction_origin: + await self.adapter.reply_interaction_stream_chunk( + source_message_id=reply.interaction_origin, + content=reply.text, + is_final=final, + ) else: await self.adapter.reply_message_chunk_for_target( message_id=reply.delivery_id, content=reply.text, is_final=final diff --git a/src/langbot/pkg/platform/adapters/wecombot/adapter.py b/src/langbot/pkg/platform/adapters/wecombot/adapter.py index 90ee727e7..579715ff9 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/adapter.py +++ b/src/langbot/pkg/platform/adapters/wecombot/adapter.py @@ -18,6 +18,7 @@ from langbot.pkg.platform.adapters.wecombot.event_converter import WecomBotEvent from langbot.pkg.platform.adapters.wecombot.message_converter import WecomBotMessageConverter from langbot.pkg.platform.adapters.wecombot.platform_api import PLATFORM_API_MAP from langbot.pkg.platform.adapters.wecombot.interaction import ( + acknowledge_interaction, interaction_delivery_capabilities, interaction_event_from_native, send_interaction, @@ -123,7 +124,7 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo 'call_platform_api', ] if not self.config.get('enable-webhook', False) and hasattr(self.bot, 'send_template_card'): - apis.append('interaction.request') + apis.extend(['interaction.request', 'interaction.acknowledge']) return apis def get_interaction_capabilities(self) -> dict[str, typing.Any]: @@ -174,6 +175,9 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo if list(self._iter_media_components(items)): raise NotSupportedError('send_message:media requires an originating message callback') raw = await self.bot.send_message(str(target_id), content) + if not isinstance(raw, dict) or raw.get('errcode', 0) != 0: + detail = raw.get('errmsg', raw.get('errcode')) if isinstance(raw, dict) else 'no acknowledgement' + raise RuntimeError(f'WeCom message delivery failed: {detail}') return platform_events.MessageResult(raw={'result': raw}) async def reply_to_event( @@ -222,8 +226,8 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo if not self.config.get('enable-webhook', False) and event.get('req_id'): if 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) + if not finished and content.strip(' \t\r\n\u200b\u200c\u200d\ufeff'): + raw = await self.bot.send_message(str(event.get('chatid') or event.get('userid') or ''), content) for item in self._iter_media_components(items): await self._send_media(self.bot, event.get('req_id'), item) else: @@ -248,8 +252,14 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo 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) + if ( + not success + and is_final + and content.strip(' \t\r\n\u200b\u200c\u200d\ufeff') + and not self.config.get('enable-webhook', False) + and event.get('req_id') + ): + await self.bot.send_message(str(event.get('chatid') or event.get('userid') or ''), content) if is_final: if not self.config.get('enable-webhook', False) and event.get('req_id'): for item in self._iter_media_components(items): @@ -261,9 +271,29 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo async def is_stream_output_supported(self) -> bool: return self.config.get('enable-stream-reply', True) + async def begin_interaction_stream(self, source_message_id: str, target: dict) -> bool: + if self.config.get('enable-webhook', False): + return False + started = self.bot.begin_interaction_stream( + source_message_id, + str(target.get('target_type') or ''), + str(target.get('target_id') or ''), + ) + if not started: + await self.logger.warning('WeCom interaction stream context unavailable; buffering final reply') + return started + + async def reply_interaction_stream_chunk(self, *, source_message_id: str, content: str, is_final: bool) -> None: + from .stream_text import format_stream_text + + if not await self.bot.push_interaction_stream_chunk(source_message_id, format_stream_text(content), is_final): + raise RuntimeError('WeCom rejected the new interaction reply stream') + async def call_platform_api(self, action: str, params: dict = {}) -> dict: if action == 'interaction.request' and 'interaction.request' in self.get_supported_apis(): return await send_interaction(self, params) + if action == 'interaction.acknowledge' and 'interaction.acknowledge' in self.get_supported_apis(): + return await acknowledge_interaction(self, params) handler = PLATFORM_API_MAP.get(action) if handler is None: raise NotSupportedError(f'call_platform_api:{action}') diff --git a/src/langbot/pkg/platform/adapters/wecombot/interaction.py b/src/langbot/pkg/platform/adapters/wecombot/interaction.py index 4e5cf7663..2efe2194b 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/interaction.py +++ b/src/langbot/pkg/platform/adapters/wecombot/interaction.py @@ -7,7 +7,7 @@ import time import typing import uuid -from langbot.libs.wecom_ai_bot_api.api import extract_template_card_event_payload +from langbot.libs.wecom_ai_bot_api.api import extract_template_card_action, extract_template_card_event_payload from langbot.libs.wecom_ai_bot_api.wecombotevent import WecomBotEvent from langbot_plugin.api.entities.builtin.platform import events as platform_events @@ -126,7 +126,59 @@ async def send_interaction(adapter: typing.Any, params: dict[str, typing.Any]) - if not isinstance(raw, dict) or raw.get('errcode', 0) != 0: detail = raw.get('errmsg', raw.get('errcode')) if isinstance(raw, dict) else 'no acknowledgement' raise RuntimeError(f'WeCom template card delivery failed: {detail}') - return {'ok': True, 'message_id': None, 'rich': True, 'raw': raw} + source_message_id = str(reply_target.get('message_id') or '') + if source_message_id and hasattr(adapter.bot, 'finish_stream_for_interaction'): + await adapter.bot.finish_stream_for_interaction(source_message_id) + return { + 'ok': True, + 'message_id': None, + 'rich': True, + 'task_id': card['template_card']['task_id'], + 'raw': raw, + } + + +async def acknowledge_interaction(adapter: typing.Any, params: dict[str, typing.Any]) -> dict[str, typing.Any]: + request = params.get('request') + submission = params.get('submission') + target = params.get('update_target') + callback = params.get('callback_context') + if not all(isinstance(value, dict) for value in (request, submission, target, callback)): + raise ValueError('interaction.acknowledge requires request, submission, update_target, and callback_context') + task_id = str(callback.get('task_id') or '') + req_id = str(callback.get('req_id') or '') + if not req_id or not task_id or task_id != target.get('task_id') or target.get('rich') is not True: + raise ValueError('WeCom interaction callback does not match the delivered card') + + card = build_interaction_card(request, 'submitted') + if card is None: + raise ValueError('WeCom interaction has no card to update') + updated = card['template_card'] + updated['task_id'] = task_id + action_id = submission.get('action_id') + if updated['card_type'] == 'button_interaction': + for index, action in enumerate((request.get('actions') or [])[:6]): + updated['button_list'][index]['key'] = 'submitted' + if isinstance(action, dict) and action.get('id') == action_id: + button = updated['button_list'][index] + button['text'] = f'✓ {button["text"]}' + button['replace_text'] = button['text'] + break + else: + field = request['fields'][0] + selected = (submission.get('values') or {}).get(field['id']) + for index, option in enumerate(field.get('options') or []): + if option.get('value') == selected: + updated['select_list'][0]['selected_id'] = f'opt_{index + 1}' + break + updated['select_list'][0]['disable'] = True + updated['submit_button']['text'] = '✓ Submitted' + updated['submit_button']['key'] = 'submitted' + raw = await adapter.bot.update_template_card(req_id, updated) + if not isinstance(raw, dict) or raw.get('errcode', 0) != 0: + detail = raw.get('errmsg', raw.get('errcode')) if isinstance(raw, dict) else 'no acknowledgement' + raise RuntimeError(f'WeCom template card update failed: {detail}') + return {**target, 'updated': True, 'raw': raw} def _template_card_event(event: WecomBotEvent) -> dict[str, typing.Any]: @@ -227,6 +279,7 @@ def interaction_event_from_native( if not callback_key.startswith('lbi:'): return None parsed = parse_callback_key(callback_key) + task_id, _, _ = extract_template_card_action(_template_card_event(event)) if parsed.pop('select_submit', False): parsed['field_ref'] = 0 parsed['option_ref'] = _selected_option_ref(_template_card_event(event)) @@ -242,6 +295,8 @@ def interaction_event_from_native( 'target_type': target_type, 'target_id': target_id, 'display_text': 'submitted', + 'callback_req_id': str(event.get('req_id') or ''), + 'callback_task_id': task_id, }, timestamp=time.time(), source_platform_object=event, diff --git a/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml b/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml index d5b7c1eeb..ae21b4d7d 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/wecombot/manifest.yaml @@ -191,6 +191,7 @@ spec: - get_group_member_list - call_platform_api - interaction.request + - interaction.acknowledge platform_specific_apis: - action: is_websocket_mode diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index e50fc7ed4..eb280527d 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -1304,6 +1304,58 @@ class RuntimeBot: try: with trace_mod.scope(route_ref=self._route_ref(event_binding, target_type, target_uuid)): + if ( + target_type == 'agent' + and isinstance(event, platform_events.MessageReceivedEvent) + and event.adapter_name == 'wecombot-omni' + and event.message_chain + and all( + isinstance( + component, + ( + platform_message.Source, + platform_message.Plain, + platform_message.At, + platform_message.AtAll, + ), + ) + for component in event.message_chain + ) + ): + answer = ''.join( + component.text + for component in event.message_chain + if isinstance(component, platform_message.Plain) + ) + record = await self.ap.agent_run_orchestrator.interaction_manager.store.consume_text_reply( + workspace_id=self.execution_context.workspace_uuid, + bot_id=self.bot_entity.uuid, + conversation_id=envelope.conversation_id or '', + actor_id=envelope.actor.actor_id if envelope.actor else '', + binding_id=binding.binding_id, + runner_id=binding.runner_id, + processor_id=target_uuid, + text=answer, + message_id=str(event.message_id or self._extract_message_id(event.message_chain)), + message_time=event.timestamp, + ) + if record is not None: + if not record.get('duplicate_text_reply'): + await self._resume_agent_interaction( + record, + event, + adapter, + envelope.actor.actor_id if envelope.actor else None, + execution_id, + ) + return await self._record_event_route_trace( + event_type=event_type, + status='delivered', + binding=event_binding, + target_type=target_type, + target_uuid=target_uuid, + text=f'Event {event_type} handled text interaction reply', + ) from .processing_indicator import processing_indicator reply_type, reply_id = (None, None) @@ -1535,7 +1587,15 @@ class RuntimeBot: # Only a validated callback resolving an original processor is eligible. # Admit before acknowledgement and target lookup to preserve failures. trace_mod.admit() - await self.ap.agent_run_orchestrator.interaction_manager.acknowledge_submission(record, adapter) + callback_context = None + if getattr(event, 'adapter_name', None) == 'wecombot-omni': + callback_context = { + 'req_id': data.get('callback_req_id'), + 'task_id': data.get('callback_task_id'), + } + await self.ap.agent_run_orchestrator.interaction_manager.acknowledge_submission( + record, adapter, callback_context + ) if record['processor_type'] == 'agent': await self._resume_agent_interaction(record, event, adapter, actor_id, execution_id) @@ -1615,7 +1675,7 @@ class RuntimeBot: async def _resume_agent_interaction( self, record: dict[str, typing.Any], - event: platform_events.PlatformSpecificEvent, + event: platform_events.EBAEvent, adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter, actor_id: str | None, execution_id: str | None = None, @@ -1639,9 +1699,13 @@ class RuntimeBot: binding.binding_id = record['binding_id'] submission = record['submission'] + event_data = getattr(event, 'data', None) + values = (submission or {}).get('values') or {} + text_answer = next(iter(values.values()), None) if len(values) == 1 else None display_text = str( (submission or {}).get('action_id') - or (event.data if isinstance(event.data, dict) else {}).get('display_text') + or (event_data if isinstance(event_data, dict) else {}).get('display_text') + or text_answer or 'submitted' ) input_data: dict[str, typing.Any] = { diff --git a/tests/unit_tests/agent/test_interaction_manager.py b/tests/unit_tests/agent/test_interaction_manager.py index 13ca440f8..f17e16678 100644 --- a/tests/unit_tests/agent/test_interaction_manager.py +++ b/tests/unit_tests/agent/test_interaction_manager.py @@ -190,6 +190,63 @@ async def test_structured_delivery_uses_frozen_target_and_persists_request(store assert record['delivery_result']['message_id'] == 'message-1' +@pytest.mark.asyncio +async def test_text_reply_consumes_latest_scoped_plain_form_once(store): + async def create(interaction_id, *, field_type='textarea', rich=False): + await store.create_request( + interaction_id=interaction_id, + run_id=f'run-{interaction_id}', + binding_id='binding-1', + runner_id='runner-1', + processor_type='agent', + processor_id='agent-1', + request={ + 'fields': [{'id': 'question', 'type': field_type}], + 'actions': [], + }, + delivery_target={'target_type': 'person', 'target_id': 'user-1'}, + workspace_id='workspace-1', + bot_id='bot-1', + conversation_id='person_user-1', + actor_id='user-1', + expires_at=time.time() + 600, + ) + await store.record_delivery_success(f'run-{interaction_id}', interaction_id, {'ok': True, 'rich': rich}) + + async def reply(**overrides): + scope = dict( + workspace_id='workspace-1', + bot_id='bot-1', + conversation_id='person_user-1', + actor_id='user-1', + binding_id='binding-1', + runner_id='runner-1', + processor_id='agent-1', + text=' answer ', + message_id='message-1', + ) + scope.update(overrides) + return await store.consume_text_reply(**scope) + + await create('old') + await create('new') + assert await reply(actor_id='user-2') is None + assert await reply(conversation_id='person_user-2') is None + assert await reply(workspace_id='workspace-2') is None + assert await reply(text=' ') is None + submitted = await reply() + assert submitted['interaction_id'] == 'new' + assert submitted['submission']['values'] == {'question': 'answer'} + assert (await reply())['duplicate_text_reply'] is True + assert await reply(message_id='message-2') is None + assert (await store.get_request('run-old', 'old'))['status'] == 'pending' + + await create('rich', rich=True) + assert await reply(message_id='message-3') is None + await create('select', field_type='select') + assert await reply(message_id='message-4') is None + + @pytest.mark.asyncio async def test_agent_tool_card_uses_host_token_and_persists_resume_target(store, event): adapter = FakeAdapter(supports_interactions=True) diff --git a/tests/unit_tests/agent/test_orchestrator_integration.py b/tests/unit_tests/agent/test_orchestrator_integration.py index ecef29cc0..a49dc7702 100644 --- a/tests/unit_tests/agent/test_orchestrator_integration.py +++ b/tests/unit_tests/agent/test_orchestrator_integration.py @@ -1271,7 +1271,10 @@ class TestQueryEntrySessionQueryId: @pytest.mark.asyncio @pytest.mark.parametrize('event_type', ['message.received', 'interaction.submitted']) -async def test_direct_agent_streams_output_to_the_originating_platform(clean_agent_state, event_type): +@pytest.mark.parametrize('completed_message', [False, True]) +async def test_direct_agent_streams_output_to_the_originating_platform( + clean_agent_state, event_type, completed_message +): from langbot.pkg.agent.runner.host_models import ( AgentBinding, AgentEventEnvelope, @@ -1283,25 +1286,27 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age from langbot_plugin.api.entities.builtin.runner.delivery import DeliveryContext from langbot_plugin.api.entities.builtin.runner.input import AgentInput - connector = FakePluginConnector( - results=[ - { - 'type': 'message.delta', - 'data': { - 'chunk': { - 'role': 'assistant', - 'content': 'hello', - 'all_content': 'hello', - } - }, + runner_results = [ + { + 'type': 'message.delta', + 'data': { + 'chunk': { + 'role': 'assistant', + 'content': 'hello', + 'all_content': 'hello', + } }, + }, + ] + if completed_message: + runner_results.append( { 'type': 'message.completed', 'data': {'message': {'role': 'assistant', 'content': 'hello world'}}, - }, - {'type': 'run.completed', 'data': {'finish_reason': 'stop'}}, - ] - ) + } + ) + runner_results.append({'type': 'run.completed', 'data': {'finish_reason': 'stop'}}) + connector = FakePluginConnector(results=runner_results) ap = FakeApplication(connector, clean_agent_state) orchestrator = AgentRunOrchestrator(ap, FakeRegistry(make_descriptor())) adapter = types.SimpleNamespace( @@ -1358,11 +1363,13 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age ) ] - assert [item.content for item in results] == ['hello', 'hello world'] + assert [item.content for item in results] == (['hello', 'hello world'] if completed_message else ['hello']) if event_type == 'message.received': adapter.create_message_card.assert_awaited_once() chunks = adapter.reply_message_chunk.await_args_list - assert [call.kwargs['bot_message'].all_content for call in chunks] == ['hello', 'hello world'] + assert [call.kwargs['bot_message'].all_content for call in chunks] == ( + ['hello', 'hello world'] if completed_message else ['hello', 'hello'] + ) assert [call.kwargs['is_final'] for call in chunks] == [False, True] adapter.reply_message.assert_not_awaited() adapter.send_message.assert_not_awaited() @@ -1371,7 +1378,7 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age adapter.reply_message_chunk.assert_not_awaited() adapter.reply_message.assert_not_awaited() adapter.send_message.assert_awaited_once() - assert adapter.send_message.await_args.args[2][0].text == 'hello world' + assert adapter.send_message.await_args.args[2][0].text == ('hello world' if completed_message else 'hello') @pytest.mark.asyncio diff --git a/tests/unit_tests/agent/test_reply_stream.py b/tests/unit_tests/agent/test_reply_stream.py index 5ffba91fc..2c4761a81 100644 --- a/tests/unit_tests/agent/test_reply_stream.py +++ b/tests/unit_tests/agent/test_reply_stream.py @@ -142,6 +142,97 @@ async def test_target_streaming_adapter_updates_card_without_source_message(): await session.close() +async def test_interaction_callback_buffers_one_final_message_after_card_update(): + event = SimpleNamespace( + event_type='interaction.submitted', + delivery=SimpleNamespace( + surface='platform', + platform_capabilities={}, + reply_target={'target_type': 'person', 'target_id': 'user-1'}, + ), + ) + source = events.PlatformSpecificEvent( + type='platform.specific', + adapter_name='wecombot-omni', + action='interaction.submitted', + data={'callback_req_id': 'callback-1'}, + ) + adapter = SimpleNamespace( + is_stream_output_supported=AsyncMock(return_value=True), + send_message=AsyncMock(), + ) + session = ReplyStreamSession(event, adapter, source) + key = uuid4() + await session.apply(request(key, text='partial')) + result = await session.apply(request(key, 'finish', 'complete')) + + assert result['delivery'] == 'buffered' + adapter.send_message.assert_awaited_once() + assert adapter.send_message.await_args.args[2][0].text == 'complete' + await session.close() + + +async def test_interaction_reply_is_sent_after_card_as_new_message(): + event = SimpleNamespace( + event_type='interaction.submitted', + delivery=SimpleNamespace( + surface='platform', + platform_capabilities={}, + reply_target={'target_type': 'person', 'target_id': 'user-1', 'message_id': 'original-1'}, + ), + ) + adapter = SimpleNamespace( + is_stream_output_supported=AsyncMock(return_value=True), + send_message=AsyncMock(), + ) + session = ReplyStreamSession(event, adapter, source=None) + key = uuid4() + await session.apply(request(key, text='partial')) + result = await session.apply(request(key, 'finish', 'complete')) + + assert result['delivery'] == 'buffered' + adapter.send_message.assert_awaited_once() + assert adapter.send_message.await_args.args[2][0].text == 'complete' + await session.close() + + +@pytest.mark.parametrize('stream_fails', [False, True]) +async def test_interaction_uses_new_stream_with_buffered_fallback(stream_fails): + event = SimpleNamespace( + event_type='interaction.submitted', + delivery=SimpleNamespace( + surface='platform', + platform_capabilities={}, + reply_target={'target_type': 'person', 'target_id': 'user-1', 'message_id': 'original-1'}, + ), + ) + adapter = SimpleNamespace( + is_stream_output_supported=AsyncMock(return_value=True), + begin_interaction_stream=AsyncMock(return_value=True), + reply_interaction_stream_chunk=AsyncMock(side_effect=RuntimeError('invalid req_id') if stream_fails else None), + send_message=AsyncMock(), + ) + session = ReplyStreamSession(event, adapter) + key = uuid4() + await session.apply(request(key, text='part')) + result = await session.apply(request(key, 'finish', 'part done')) + + adapter.begin_interaction_stream.assert_awaited_once_with('original-1', event.delivery.reply_target) + if stream_fails: + assert result['delivery'] == 'buffered' + adapter.send_message.assert_awaited_once() + assert adapter.send_message.await_args.args[2][0].text == 'part done' + else: + assert result['delivery'] == 'streaming' + assert [call.kwargs['content'] for call in adapter.reply_interaction_stream_chunk.await_args_list] == [ + 'part', + 'part done', + ] + assert adapter.reply_interaction_stream_chunk.await_args.kwargs['is_final'] is True + adapter.send_message.assert_not_awaited() + 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): diff --git a/tests/unit_tests/platform/test_routing_rules.py b/tests/unit_tests/platform/test_routing_rules.py index 139c3afb9..147e41bd1 100644 --- a/tests/unit_tests/platform/test_routing_rules.py +++ b/tests/unit_tests/platform/test_routing_rules.py @@ -589,6 +589,21 @@ class TestInteractionResumeRouting: actor_id='user-1', ) + @pytest.mark.asyncio + async def test_wecom_callback_passes_native_card_update_context(self): + record = self._record('pipeline') + bot = self._make_bot(record) + adapter = SimpleNamespace() + event = self._event() + event.adapter_name = 'wecombot-omni' + event.data.update({'callback_req_id': 'callback-req', 'callback_task_id': 'task-1'}) + + await bot._handle_interaction_submission(event, adapter) + + bot.ap.agent_run_orchestrator.interaction_manager.acknowledge_submission.assert_awaited_once_with( + record, adapter, {'req_id': 'callback-req', 'task_id': 'task-1'} + ) + @pytest.mark.asyncio async def test_callback_forwards_compact_platform_references(self): bot = self._make_bot(self._record('pipeline')) @@ -676,6 +691,27 @@ class TestInteractionResumeRouting: assert binding.delivery_policy.enable_reply is True adapter.send_message.assert_not_awaited() + from langbot_plugin.api.entities.builtin.platform import entities, events, message + + text_record = self._record('agent') + text_record['submission'] = { + 'interaction_id': 'form-1', + 'action_id': None, + 'values': {'name': 'My question'}, + 'submitted_at': 1235, + } + text_event = events.MessageReceivedEvent( + timestamp=1235, + message_id='message-2', + message_chain=message.MessageChain([message.Plain(text='My question')]), + sender=entities.User(id='user-1'), + chat_type=entities.ChatType.GROUP, + chat_id='chat-1', + ) + await bot._resume_agent_interaction(text_record, text_event, adapter, 'user-1') + assert captured[1][0].input.text == 'My question' + assert captured[1][0].input.interaction.values == {'name': 'My question'} + 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.""" from langbot.pkg.platform.botmgr import RuntimeBot @@ -782,6 +818,53 @@ async def test_bound_event_processor_receives_one_complete_typed_event(): bot.ap.plugin_connector.emit_event.assert_not_called() +@pytest.mark.asyncio +async def test_wecom_plain_reply_resumes_pending_agent_interaction(): + from datetime import datetime + + from langbot_plugin.api.entities.builtin.platform import entities, events, message + + bot = TestEventRouteTrace._make_bot( + [{'id': 'binding-1', 'event_pattern': 'message.received', 'target_type': 'agent', 'target_uuid': 'agent-1'}] + ) + run = Mock() + consume = AsyncMock(return_value={'interaction_id': 'form-1'}) + bot.ap = SimpleNamespace( + agent_service=SimpleNamespace( + get_agent=AsyncMock( + return_value={ + 'uuid': 'agent-1', + 'kind': 'agent', + 'supported_event_patterns': ['message.received'], + 'config': {'runner': {'id': 'runner-1'}, 'runner_config': {'runner-1': {}}}, + } + ) + ), + agent_run_orchestrator=SimpleNamespace( + run=run, interaction_manager=SimpleNamespace(store=SimpleNamespace(consume_text_reply=consume)) + ), + ) + bot._resume_agent_interaction = AsyncMock() + event = events.MessageReceivedEvent( + adapter_name='wecombot-omni', + message_id='message-1', + message_chain=message.MessageChain( + [message.Source(id='message-1', time=datetime.now()), message.Plain(text='answer')] + ), + sender=entities.User(id='user-1'), + chat_type=entities.ChatType.PRIVATE, + chat_id='user-1', + ) + + result = await bot._dispatch_eba_event_to_processor(event, SimpleNamespace(get_supported_apis=lambda: [])) + + assert result['status'] == 'delivered' + assert consume.await_args.kwargs['text'] == 'answer' + assert consume.await_args.kwargs['conversation_id'] == 'person_user-1' + bot._resume_agent_interaction.assert_awaited_once() + run.assert_not_called() + + @pytest.mark.asyncio @pytest.mark.parametrize('kind', ['agent', 'event_processor']) @pytest.mark.parametrize('output_kind', ['message', 'chunks', 'tool_rounds']) diff --git a/tests/unit_tests/platform/test_wecombot_eba_adapter.py b/tests/unit_tests/platform/test_wecombot_eba_adapter.py index 2cfec1729..a0c8b2de5 100644 --- a/tests/unit_tests/platform/test_wecombot_eba_adapter.py +++ b/tests/unit_tests/platform/test_wecombot_eba_adapter.py @@ -1,7 +1,7 @@ from __future__ import annotations import pathlib -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, Mock, patch import pytest import yaml @@ -48,8 +48,12 @@ class DummyWecomBotWsClient: self.send_message = AsyncMock(return_value={'ok': True}) self.reply_text = AsyncMock(return_value={'reply': True}) self.push_stream_chunk = AsyncMock(return_value=True) + self.finish_stream_for_interaction = AsyncMock(return_value=True) + self.begin_interaction_stream = Mock(return_value=True) + self.push_interaction_stream_chunk = AsyncMock(return_value=True) self.set_message = AsyncMock(return_value={'set': True}) self.send_template_card = AsyncMock(return_value={'sent': True}) + self.update_template_card = AsyncMock(return_value={'errcode': 0}) def on_message(self, msg_type: str): def decorator(func): @@ -109,6 +113,30 @@ def make_adapter(enable_webhook: bool = False) -> WecomBotAdapter: return WecomBotAdapter(config, DummyLogger()) +@pytest.mark.asyncio +async def test_interaction_card_finishes_original_loading_stream(): + adapter = make_adapter() + await adapter.call_platform_api( + 'interaction.request', + { + 'request': {'title': 'Continue', 'actions': [{'id': 'ok', 'label': 'OK'}]}, + 'reply_target': {'target_type': 'person', 'target_id': 'user-1', 'message_id': 'original-1'}, + 'callback_token': 'token-1', + }, + ) + adapter.bot.finish_stream_for_interaction.assert_awaited_once_with('original-1') + + +@pytest.mark.asyncio +async def test_interaction_reply_uses_new_wecom_stream(): + adapter = make_adapter() + target = {'target_type': 'person', 'target_id': 'user-1', 'message_id': 'original-1'} + assert await adapter.begin_interaction_stream('original-1', target) + adapter.bot.begin_interaction_stream.assert_called_once_with('original-1', 'person', 'user-1') + await adapter.reply_interaction_stream_chunk(source_message_id='original-1', content='part', is_final=False) + adapter.bot.push_interaction_stream_chunk.assert_awaited_once_with('original-1', 'part', False) + + def wecombot_event(**overrides) -> WecomBotEvent: event_type = overrides.get('type', 'single') payload = { @@ -216,6 +244,7 @@ async def test_wecombot_select_card_submits_native_dropdown_value(): 'eventtype': 'template_card_event', 'template_card_event': { 'card_type': 'multiple_interaction', + 'task_id': result['task_id'], 'event_key': card['submit_button']['key'], 'selected_items': { 'selected_item': [ @@ -226,6 +255,8 @@ async def test_wecombot_select_card_submits_native_dropdown_value(): } event = interaction_event_from_native(native) assert event.data['callback_token'] == 'callback-token' + assert event.data['callback_req_id'] == 'req-1' + assert event.data['callback_task_id'] == result['task_id'] resolved = InteractionManager._resolve_submission_refs( {'interaction_id': request['interaction_id'], 'request': request}, event.data ) @@ -238,6 +269,70 @@ async def test_wecombot_select_card_submits_native_dropdown_value(): assert listener.await_args.args[0].data['option_ref'] == 1 +@pytest.mark.asyncio +@pytest.mark.parametrize('kind', ['button', 'select']) +async def test_wecombot_acknowledges_submitted_card(kind): + adapter = make_adapter() + request = {'title': 'Choose', 'actions': [{'id': 'yes', 'label': 'Yes'}, {'id': 'no', 'label': 'No'}]} + submission = {'action_id': 'no', 'values': {}} + if kind == 'select': + request = { + 'title': 'Choose', + 'actions': [], + 'fields': [ + { + 'id': 'choice', + 'type': 'select', + 'label': 'Choice', + 'options': [{'value': 'a', 'label': 'A'}, {'value': 'b', 'label': 'B'}], + } + ], + } + submission = {'action_id': None, 'values': {'choice': 'b'}} + delivered = await adapter.call_platform_api( + 'interaction.request', + { + 'callback_token': 'token-1', + 'reply_target': {'target_type': 'person', 'target_id': 'user-1'}, + 'request': request, + }, + ) + + result = await adapter.call_platform_api( + 'interaction.acknowledge', + { + 'request': request, + 'submission': submission, + 'update_target': delivered, + 'callback_context': {'req_id': 'callback-req', 'task_id': delivered['task_id']}, + }, + ) + + assert result['updated'] is True + req_id, card = adapter.bot.update_template_card.await_args.args + assert req_id == 'callback-req' + assert card['task_id'] == delivered['task_id'] + if kind == 'select': + assert card['select_list'][0]['selected_id'] == 'opt_2' + assert card['select_list'][0]['disable'] is True + assert card['submit_button']['key'] == 'submitted' + else: + assert card['button_list'][1]['replace_text'].endswith('No') + assert all(button['key'] == 'submitted' for button in card['button_list']) + + with pytest.raises(ValueError, match='does not match'): + await adapter.call_platform_api( + 'interaction.acknowledge', + { + 'request': request, + 'submission': submission, + 'update_target': delivered, + 'callback_context': {'req_id': 'callback-req', 'task_id': 'another-card'}, + }, + ) + assert adapter.bot.update_template_card.await_count == 1 + + def test_wecombot_select_rejects_more_than_ten_options(): with pytest.raises(ValueError, match='at most 10 options'): build_interaction_card( @@ -475,6 +570,29 @@ async def test_wecombot_send_reply_feedback_and_platform_api_use_underlying_clie assert feedback_calls[0].inaccurate_reasons == ['1', '2'] +@pytest.mark.asyncio +async def test_wecombot_expired_stream_falls_back_to_proactive_reply(): + adapter = make_adapter() + adapter.bot.push_stream_chunk.return_value = False + source_event = await WecomBotEventConverter().target2yiri(wecombot_event()) + message = platform_message.MessageChain([platform_message.Plain(text='late answer')]) + + await adapter.reply_message_chunk(source_event, None, message, is_final=True) + + adapter.bot.send_message.assert_awaited_once_with('user-1', 'late answer') + adapter.bot.reply_text.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_wecombot_send_message_rejects_failed_ack(): + adapter = make_adapter() + adapter.bot.send_message.return_value = {'errcode': 400, 'errmsg': 'invalid message'} + with pytest.raises(RuntimeError, match='invalid message'): + await adapter.send_message( + 'person', 'user-1', platform_message.MessageChain([platform_message.Plain(text='hello')]) + ) + + @pytest.mark.asyncio async def test_wecombot_webhook_mode_rejects_proactive_send(): adapter = make_adapter(enable_webhook=True) diff --git a/tests/unit_tests/platform/test_wecombot_template_card.py b/tests/unit_tests/platform/test_wecombot_template_card.py index d27a6650c..d83d0f22b 100644 --- a/tests/unit_tests/platform/test_wecombot_template_card.py +++ b/tests/unit_tests/platform/test_wecombot_template_card.py @@ -1,14 +1,10 @@ -import sys +import time import types -from unittest.mock import Mock +from unittest.mock import AsyncMock, Mock import pytest -logger_module = types.ModuleType('langbot.pkg.platform.logger') -logger_module.EventLogger = object -sys.modules.setdefault('langbot.pkg.platform.logger', logger_module) - from langbot.libs.wecom_ai_bot_api.api import ( # noqa: E402 WecomBotClient, build_button_interaction_payload, @@ -45,7 +41,7 @@ def test_webhook_dispatch_tasks_are_bounded(): @pytest.mark.asyncio -async def test_ws_initial_stream_frame_precedes_pipeline_dispatch(monkeypatch): +async def test_ws_message_waits_for_real_content_before_creating_stream(monkeypatch): from langbot.libs.wecom_ai_bot_api import ws_client as ws_client_module order = [] @@ -61,9 +57,7 @@ async def test_ws_initial_stream_frame_precedes_pipeline_dispatch(monkeypatch): return {'msgid': 'msg-1', 'type': 'single', 'userid': 'user-1'} async def reply_stream(*args, **kwargs): - del args, kwargs - order.append('initial-frame') - return {} + raise AssertionError('An empty stream must not be created') async def dispatch_event(event): del event @@ -76,11 +70,126 @@ async def test_ws_initial_stream_frame_precedes_pipeline_dispatch(monkeypatch): await client._handle_message_callback({'headers': {'req_id': 'req-1'}, 'body': {}}) - assert order == ['initial-frame', 'pipeline-dispatch'] + assert order == ['pipeline-dispatch'] + assert client._stream_sessions['msg-1']['stream_started'] is False @pytest.mark.asyncio -async def test_ws_initial_stream_failure_still_dispatches_message(monkeypatch): +async def test_ws_interaction_releases_unused_stream_without_visible_message(): + client = WecomBotWsClient('bot-id', 'secret', object()) + client._stream_ids['msg-1'] = 'req-1|stream-1' + client._stream_sessions['msg-1'] = {'user_id': 'user-1'} + sent = [] + + async def reply_stream(req_id, stream_id, content, finish=False): + sent.append((req_id, stream_id, content, finish)) + return {'errcode': 0} + + client.reply_stream = reply_stream + assert await client.finish_stream_for_interaction('msg-1') + assert sent == [] + assert 'msg-1' not in client._stream_ids + assert 'msg-1' not in client._stream_sessions + + +@pytest.mark.asyncio +async def test_ws_interaction_starts_a_new_stream_after_card(): + client = WecomBotWsClient('bot-id', 'secret', types.SimpleNamespace(info=AsyncMock())) + client._stream_ids['msg-1'] = 'req-1|old-stream' + client._stream_sessions['msg-1'] = { + 'req_id': 'req-1', + 'user_id': 'user-1', + 'chat_id': '', + 'created_at': time.monotonic(), + 'stream_started': False, + } + sent = [] + + async def reply_stream(req_id, stream_id, content, finish=False, feedback_id=''): + sent.append((req_id, stream_id, content, finish)) + return {'errcode': 0} + + client.reply_stream = reply_stream + assert await client.finish_stream_for_interaction('msg-1') + assert sent == [] + assert client.begin_interaction_stream('msg-1', 'person', 'user-1') + assert not client.begin_interaction_stream('msg-1', 'person', 'user-1') + assert await client.push_interaction_stream_chunk('msg-1', 'part', False) + assert await client.push_interaction_stream_chunk('msg-1', 'part done', True) + assert [item[2:] for item in sent] == [('part', False), ('part done', True)] + assert sent[0][0] == sent[1][0] == 'req-1' + assert sent[0][1] == sent[1][1] + assert sent[0][1] != 'old-stream' + + +def test_ws_interaction_stream_checks_target_and_expiry(): + client = WecomBotWsClient('bot-id', 'secret', object()) + client._interaction_stream_origins['msg-1'] = { + 'req_id': 'req-1', + 'user_id': 'user-1', + 'chat_id': 'group-1', + 'created_at': time.monotonic(), + } + assert not client.begin_interaction_stream('msg-1', 'person', 'other-user') + assert not client.begin_interaction_stream('msg-1', 'group', 'other-group') + client._interaction_stream_origins['msg-1']['created_at'] -= 571 + assert not client.begin_interaction_stream('msg-1', 'person', 'user-1') + assert 'msg-1' not in client._interaction_stream_origins + + +@pytest.mark.asyncio +async def test_ws_interaction_preserves_existing_stream_text_when_finishing(): + client = WecomBotWsClient('bot-id', 'secret', object()) + client._stream_ids['msg-1'] = 'req-1|stream-1' + client._stream_sessions['msg-1'] = {'stream_started': True} + client._stream_last_content['msg-1'] = 'Earlier answer' + sent = [] + + async def reply_stream(req_id, stream_id, content, finish=False): + sent.append((content, finish)) + return {'errcode': 0} + + client.reply_stream = reply_stream + assert await client.finish_stream_for_interaction('msg-1') + assert sent == [('Earlier answer', True)] + + +@pytest.mark.asyncio +async def test_ws_interaction_finish_failure_keeps_stream_state(): + logger = types.SimpleNamespace(warning=AsyncMock()) + client = WecomBotWsClient('bot-id', 'secret', logger) + client._stream_ids['msg-1'] = 'req-1|stream-1' + client._stream_sessions['msg-1'] = {'stream_started': True} + + async def reject(*args, **kwargs): + raise ConnectionError('disconnected') + + client.reply_stream = reject + assert not await client.finish_stream_for_interaction('msg-1') + assert client._stream_ids['msg-1'] == 'req-1|stream-1' + + +@pytest.mark.asyncio +async def test_ws_rejected_stream_refresh_does_not_complete_session(): + logger = types.SimpleNamespace(error=Mock()) + client = WecomBotWsClient('bot-id', 'secret', logger) + client._stream_ids['msg-1'] = 'req-1|stream-1' + client._stream_sessions['msg-1'] = {'user_id': 'user-1'} + + async def reject(*args, **kwargs): + return {'errcode': 400, 'errmsg': 'invalid req_id'} + + async def record_error(message): + assert 'invalid req_id' in message + + client.reply_stream = reject + logger.error = record_error + assert not await client.push_stream_chunk('msg-1', 'answer', is_final=True) + assert 'msg-1' in client._stream_ids + + +@pytest.mark.asyncio +async def test_ws_stream_starts_with_first_real_content(monkeypatch): from langbot.libs.wecom_ai_bot_api import ws_client as ws_client_module dispatched = [] @@ -105,9 +214,11 @@ async def test_ws_initial_stream_failure_still_dispatches_message(monkeypatch): del args, kwargs return {'msgid': 'msg-1', 'type': 'single', 'userid': 'user-1'} - async def reply_stream(*args, **kwargs): - del args, kwargs - raise ConnectionError('simulated reply failure') + sent = [] + + async def reply_stream(req_id, stream_id, content, finish=False, feedback_id=''): + sent.append((req_id, content, finish)) + return {'errcode': 0} async def dispatch_event(event): dispatched.append(event) @@ -118,10 +229,17 @@ async def test_ws_initial_stream_failure_still_dispatches_message(monkeypatch): client._dispatch_event = dispatch_event await client._handle_message_callback({'headers': {'req_id': 'req-1'}, 'body': {}}) + assert sent == [] + assert await client.push_stream_chunk('msg-1', 'hello') assert len(dispatched) == 1 - assert len(logger.warnings) == 1 - assert 'simulated reply failure' in logger.warnings[0] + assert sent == [('req-1', 'hello', False)] + assert client._stream_sessions['msg-1']['stream_started'] is True + assert logger.warnings == [] + + assert await client.push_stream_chunk('msg-1', 'hello', is_final=True) + assert 'msg-1' not in client._stream_ids + assert client.begin_interaction_stream('msg-1', 'person', 'user-1') def test_extract_template_card_action_supports_nested_button_key():