diff --git a/scripts/build_dingtalk_card_template.py b/scripts/build_dingtalk_card_template.py index c07c73d33..8508ac0c1 100644 --- a/scripts/build_dingtalk_card_template.py +++ b/scripts/build_dingtalk_card_template.py @@ -68,6 +68,210 @@ def markdown_block(node_id, variable='content'): } +def _dynamic_string_var(variable): + return {'type': 'dynamicString', 'content': '', 'i18n': False, 'variable': variable, 'variableType': 'global'} + + +def _dynamic_visible_var(variable): + return { + 'type': 'dynamicVisible', + 'value': True, + 'valueType': 'variable', + 'variable': variable, + 'variableType': 'global', + 'condition': {'op': 'and', 'conditions': []}, + } + + +SELECT_OPTION_LOCALES = ( + 'zh_CN', + 'zh_TW', + 'en_US', + 'ja_JP', + 'vi_VN', + 'th_TH', + 'id_ID', + 'ne_NP', + 'ms_MY', + 'ko_KR', + 'ru_RU', + 'es_EA', + 'tr_TR', + 'fr_FR', + 'pt_BR', +) + + +def _empty_select_option(): + return {'value': '', 'text': {locale: '' for locale in SELECT_OPTION_LOCALES}} + + +def _select_options_variable(): + return { + 'name': 'index_o', + 'private': False, + 'type': 'selectOptions', + 'id': 'index_o', + 'description': 'Select options', + 'editorVarType': 'variables', + 'disabled': False, + 'schema': [ + { + 'id': 'index_o.value', + 'type': 'string', + 'name': 'value', + 'private': False, + 'editorVarType': 'variables', + 'disabled': True, + 'description': '', + }, + { + 'id': 'index_o.text', + 'type': 'object', + 'name': 'text', + 'private': False, + 'editorVarType': 'variables', + 'disabled': True, + 'description': '', + 'schema': [ + { + 'id': f'index_o.{locale}', + 'type': 'string', + 'name': locale, + 'private': False, + 'editorVarType': 'variables', + 'disabled': True, + 'description': '', + } + for locale in SELECT_OPTION_LOCALES + ], + }, + ], + } + + +def input_block(node_id): + return { + 'componentName': 'Input', + 'id': node_id, + 'props': { + 'placeholder': _dynamic_string_var('input_placeholder'), + 'currentValue': _dynamic_string_var('input_value'), + 'message': _dynamic_string_var('input_placeholder'), + 'title': _dynamic_string_var('input_title'), + 'id': {'type': 'dynamicString', 'content': 'input', 'i18n': False}, + 'params': [ + { + 'type': 'builtIn', + 'variable': '', + 'value': '', + 'name': 'input', + 'variableType': 'global', + 'id': '__built_in_inputResult__', + } + ], + 'visible': _dynamic_visible_var('input_visible'), + 'status': { + 'type': 'dynamicSelect', + 'valueType': 'fixed', + 'value': 'normal', + 'variable': '', + 'variableType': 'global', + }, + 'actionType': 'request', + 'localVarAction': {'type': 'variableValue', 'variableType': 'global', 'variable': ''}, + 'keyOfDynamicObject': {'type': 'dynamicString', 'content': '', 'i18n': False}, + 'inlineMode': False, + 'textArea': True, + 'minRows': { + 'type': 'dynamicNumber', + 'valueType': 'fixed', + 'value': 2, + 'variable': '', + 'variableType': 'global', + }, + 'maxRows': { + 'type': 'dynamicNumber', + 'valueType': 'fixed', + 'value': 6, + 'variable': '', + 'variableType': 'global', + }, + 'marginLeft': 12, + 'marginRight': 12, + 'marginTop': 6, + 'marginBottom': 6, + 'margin': 12, + 'innerOffset': 0, + }, + 'title': 'Text input', + 'hidden': False, + 'isLocked': False, + 'condition': True, + 'conditionGroup': '', + } + + +def select_block(node_id): + return { + 'componentName': 'SelectBlock', + 'id': node_id, + 'props': { + 'id': {'type': 'dynamicString', 'content': 'select', 'i18n': False}, + 'placeholder': _dynamic_string_var('select_placeholder'), + 'currentIndex': { + 'type': 'dynamicNumber', + 'valueType': 'variable', + 'value': -1, + 'variable': 'select_index', + 'variableType': 'global', + }, + 'options': { + 'type': 'dynamicSelectOptions', + 'valueType': 'variable', + 'value': [], + 'variable': 'index_o', + 'variableType': 'global', + }, + 'optionLabelMaxLines': 3, + 'params': [ + { + 'type': 'builtIn', + 'variable': '', + 'value': '{"index": ${index}, "value": "${value}"}', + 'name': 'select', + 'variableType': 'global', + 'id': '__built_in_selectResult__', + } + ], + 'actionType': 'request', + 'localVarAction': {'type': 'variableValue', 'variableType': 'global', 'variable': ''}, + 'keyOfDynamicObject': {'type': 'dynamicString', 'content': '', 'i18n': False}, + 'status': { + 'type': 'dynamicSelect', + 'valueType': 'fixed', + 'value': 'normal', + 'variable': '', + 'variableType': 'global', + }, + 'visible': _dynamic_visible_var('select_visible'), + 'marginLeft': 12, + 'marginRight': 12, + 'marginTop': 6, + 'marginBottom': 6, + 'pullOptionsWhileOpen': False, + 'pullOptionsRequestParams': [], + 'margin': 12, + 'innerOffset': 0, + }, + 'title': 'Select', + 'hidden': False, + 'isLocked': False, + 'condition': True, + 'conditionGroup': '', + } + + def text_block( node_id, text, @@ -280,6 +484,8 @@ def build_editor_data(): 'ButtonGroup', 'MarkdownBlock', 'Avatar', + 'Input', + 'SelectBlock', ] components_map = [ { @@ -396,6 +602,8 @@ def build_editor_data(): 'conditionGroup': '', 'children': [ avatar('node_avatar', name='LangBot'), + input_block('node_input'), + select_block('node_select'), markdown_block('node_text_content', variable='content'), button_group('node_btn_group'), ], @@ -716,6 +924,15 @@ def build_editor_data(): 'cardData': { 'flowStatus': 3, 'content': '请审核以下报销申请:\n\n- 申请人:张三\n- 金额:¥1,200\n- 类别:差旅', + 'input_visible': '', + 'input_title': '', + 'input_placeholder': '', + 'input_value': '', + 'select_visible': '', + 'select_placeholder': '', + 'index_o': [_empty_select_option()], + 'select_options': [], + 'select_index': -1, 'btns': [ { 'text': '通过', @@ -783,6 +1000,79 @@ def build_editor_data(): 'editorVarType': 'variables', 'disabled': False, }, + { + 'id': 'input_visible', + 'type': 'string', + 'name': 'input_visible', + 'description': 'Whether to show the text input component', + 'private': False, + 'editorVarType': 'variables', + 'disabled': False, + }, + { + 'id': 'input_title', + 'type': 'string', + 'name': 'input_title', + 'description': 'Text input title', + 'private': False, + 'editorVarType': 'variables', + 'disabled': False, + }, + { + 'id': 'input_placeholder', + 'type': 'string', + 'name': 'input_placeholder', + 'description': 'Text input placeholder', + 'private': False, + 'editorVarType': 'variables', + 'disabled': False, + }, + { + 'id': 'input_value', + 'type': 'string', + 'name': 'input_value', + 'description': 'Text input current value', + 'private': False, + 'editorVarType': 'variables', + 'disabled': False, + }, + { + 'id': 'select_visible', + 'type': 'string', + 'name': 'select_visible', + 'description': 'Whether to show the select component', + 'private': False, + 'editorVarType': 'variables', + 'disabled': False, + }, + { + 'id': 'select_placeholder', + 'type': 'string', + 'name': 'select_placeholder', + 'description': 'Select placeholder', + 'private': False, + 'editorVarType': 'variables', + 'disabled': False, + }, + _select_options_variable(), + { + 'id': 'select_options', + 'type': 'array', + 'name': 'select_options', + 'description': 'Legacy select options', + 'private': False, + 'editorVarType': 'variables', + 'disabled': False, + }, + { + 'id': 'select_index', + 'type': 'number', + 'name': 'select_index', + 'description': 'Current select index', + 'private': False, + 'editorVarType': 'variables', + 'disabled': False, + }, btns_var, ], 'formList': [], diff --git a/src/langbot/libs/qq_official_api/api.py b/src/langbot/libs/qq_official_api/api.py index 7e2b8bcd0..69ab414a8 100644 --- a/src/langbot/libs/qq_official_api/api.py +++ b/src/langbot/libs/qq_official_api/api.py @@ -12,6 +12,78 @@ import traceback from cryptography.hazmat.primitives.asymmetric import ed25519 +QQ_SELECT_ACTION_PREFIX = '__langbot_select__:' + + +def get_select_field_options(form_data: dict) -> tuple[str, list[str]]: + """Return the active select field name and its display/submission values.""" + field_name = str(form_data.get('_current_input_field') or '').strip() + if not field_name: + return '', [] + + field = next( + ( + item + for item in form_data.get('input_defs') or [] + if str(item.get('output_variable_name') or '').strip() == field_name + ), + None, + ) + if not field or str(field.get('type') or '').strip().lower() != 'select': + return '', [] + + source = field.get('option_source') or {} + source_value = source.get('value') if isinstance(source, dict) else None + if isinstance(source_value, list): + return field_name, [str(item) for item in source_value] + if isinstance(source_value, str): + return field_name, [part.strip() for part in source_value.splitlines() if part.strip()] + + options = field.get('options') + if not isinstance(options, list): + return field_name, [] + values = [] + for item in options: + if isinstance(item, dict): + values.append(str(item.get('label') or item.get('value') or '')) + else: + values.append(str(item)) + return field_name, [value for value in values if value] + + +def build_keyboard_from_select_field(form_data: dict, *, buttons_per_row: int | None = None) -> dict: + """Build callback buttons for the currently active Dify select field.""" + _, options = get_select_field_options(form_data) + visible_options = options[:25] + if buttons_per_row is None: + # Keep small choices readable while fitting up to QQ's 5x5 limit. + buttons_per_row = min(5, max(2, (len(visible_options) + 4) // 5)) + selection_actions = [ + { + 'id': f'{QQ_SELECT_ACTION_PREFIX}{idx}', + 'title': option, + 'button_style': 'secondary', + } + for idx, option in enumerate(visible_options) + ] + return build_keyboard_from_form({'actions': selection_actions}, buttons_per_row=buttons_per_row) + + +def resolve_select_button_action(form_data: dict, action_id: str) -> tuple[str, str] | None: + """Resolve a select-button callback to ``(field_name, option_value)``.""" + if not action_id.startswith(QQ_SELECT_ACTION_PREFIX): + return None + try: + option_index = int(action_id[len(QQ_SELECT_ACTION_PREFIX) :]) + except ValueError: + return None + + field_name, options = get_select_field_options(form_data) + if not field_name or option_index < 0 or option_index >= len(options) or option_index >= 25: + return None + return field_name, options[option_index] + + def build_keyboard_from_form(form_data: dict, *, buttons_per_row: int = 2) -> dict: """Build a QQ keyboard JSON payload from a Dify human-input form_data. diff --git a/src/langbot/pkg/platform/sources/dingtalk.py b/src/langbot/pkg/platform/sources/dingtalk.py index 09e52c6a5..943cc46a7 100644 --- a/src/langbot/pkg/platform/sources/dingtalk.py +++ b/src/langbot/pkg/platform/sources/dingtalk.py @@ -1203,6 +1203,11 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): if not raw_action_id: await self.logger.warning(f'DingTalk: card action with no action_id, payload={payload}') return + if raw_action_id not in known_action_ids: + await self.logger.warning( + f'DingTalk: card action_id={raw_action_id!r} is not present on out_track_id={out_track_id}' + ) + return action_id = raw_action_id action_title = raw_action_id @@ -1217,7 +1222,17 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): else provider_session.LauncherTypes.PERSON ) launcher_id = state.get('launcher_id', '') - sender_user_id = state.get('sender_user_id') or payload.get('user_id') or launcher_id + initiator_user_id = str(state.get('sender_user_id') or '') + actor_user_id = str(payload.get('user_id') or initiator_user_id or launcher_id) + if ( + launcher_type == provider_session.LauncherTypes.PERSON + and initiator_user_id + and actor_user_id != initiator_user_id + ): + await self.logger.warning( + f'DingTalk: user {actor_user_id} cannot act on private form created for {initiator_user_id}' + ) + return form_action_data = { 'form_token': state.get('form_token', ''), @@ -1234,7 +1249,7 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): if launcher_type == provider_session.LauncherTypes.GROUP: synthetic_event = platform_events.GroupMessage( sender=platform_entities.GroupMember( - id=sender_user_id, + id=actor_user_id, member_name='', permission=platform_entities.Permission.Member, group=platform_entities.Group( @@ -1251,7 +1266,7 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): else: synthetic_event = platform_events.FriendMessage( sender=platform_entities.Friend( - id=sender_user_id, + id=actor_user_id, nickname='', remark='', ), @@ -1273,13 +1288,14 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): self.ap.logger.info( f'DingTalk _on_card_action enqueuing form action: action_id={action_id!r} ' f'action_title={action_title!r} launcher_type={launcher_type.value} ' - f'launcher_id={launcher_id} bot_uuid={bot_uuid} pipeline_uuid={pipeline_uuid}' + f'launcher_id={launcher_id} actor_user_id={actor_user_id} ' + f'bot_uuid={bot_uuid} pipeline_uuid={pipeline_uuid}' ) await self.ap.query_pool.add_query( bot_uuid=bot_uuid, launcher_type=launcher_type, launcher_id=launcher_id, - sender_id=sender_user_id, + sender_id=actor_user_id, message_event=synthetic_event, message_chain=message_chain, adapter=self, @@ -1336,7 +1352,17 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): else provider_session.LauncherTypes.PERSON ) launcher_id = state.get('launcher_id', '') - sender_user_id = state.get('sender_user_id') or payload.get('user_id') or launcher_id + initiator_user_id = str(state.get('sender_user_id') or '') + actor_user_id = str(payload.get('user_id') or initiator_user_id or launcher_id) + if ( + launcher_type == provider_session.LauncherTypes.PERSON + and initiator_user_id + and actor_user_id != initiator_user_id + ): + await self.logger.warning( + f'DingTalk: user {actor_user_id} cannot update private form created for {initiator_user_id}' + ) + return form_action_data = { 'form_token': state.get('form_token', ''), 'workflow_run_id': state.get('workflow_run_id', ''), @@ -1353,7 +1379,7 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): if launcher_type == provider_session.LauncherTypes.GROUP: synthetic_event = platform_events.GroupMessage( sender=platform_entities.GroupMember( - id=sender_user_id, + id=actor_user_id, member_name='', permission=platform_entities.Permission.Member, group=platform_entities.Group( @@ -1370,7 +1396,7 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): else: synthetic_event = platform_events.FriendMessage( sender=platform_entities.Friend( - id=sender_user_id, + id=actor_user_id, nickname='', remark='', ), @@ -1394,7 +1420,7 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): bot_uuid=bot_uuid, launcher_type=launcher_type, launcher_id=launcher_id, - sender_id=sender_user_id, + sender_id=actor_user_id, message_event=synthetic_event, message_chain=message_chain, adapter=self, diff --git a/src/langbot/pkg/platform/sources/discord.py b/src/langbot/pkg/platform/sources/discord.py index 0b22067c5..9613e9bcc 100644 --- a/src/langbot/pkg/platform/sources/discord.py +++ b/src/langbot/pkg/platform/sources/discord.py @@ -1453,7 +1453,7 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): # Already responded somehow — proceed regardless. pass - pending = self._pending_forms.pop(session_key, None) + pending = self._pending_forms.get(session_key) if not pending: if self.ap is not None: self.ap.logger.warning( @@ -1462,24 +1462,33 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): await self._lock_view_message(interaction, view, action_title, stale=True) return - # Lock the buttons in place: disable everything, mark chosen one. - await self._lock_view_message(interaction, view, action_title) - form_data: dict = pending.get('form_data') or {} guild_id = pending.get('guild_id', '') channel_id = pending.get('channel_id', '') - sender_id = pending.get('sender_id', '') + initiator_id = str(pending.get('sender_id', '') or '') + actor_id = str(interaction.user.id) if interaction.user is not None else initiator_id + if not guild_id and initiator_id and actor_id != initiator_id: + if self.ap is not None: + self.ap.logger.warning( + f'Discord: user {actor_id} cannot act on private form created for {initiator_id}' + ) + await self._lock_view_message(interaction, view, action_title, stale=True) + return - # In group context the launcher is the CHANNEL (not the user who - # clicked) — matches how the original message was routed through - # the pipeline. Using the clicker's user id would mismatch the - # Dify session and produce "Workflow run not found". + self._pending_forms.pop(session_key, None) + + # Lock the buttons in place: disable everything, mark chosen one. + await self._lock_view_message(interaction, view, action_title) + + # In group context the launcher remains the channel so Dify resumes + # the original group session. The synthetic sender is still the real + # clicker, preserving actor identity for auditing and routing rules. if guild_id: launcher_type = provider_session.LauncherTypes.GROUP launcher_id = channel_id else: launcher_type = provider_session.LauncherTypes.PERSON - launcher_id = sender_id or str(interaction.user.id) + launcher_id = initiator_id or actor_id form_action_data = { 'form_token': form_data.get('form_token', ''), @@ -1500,7 +1509,7 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): if launcher_type == provider_session.LauncherTypes.GROUP: synthetic_event: platform_events.MessageEvent = platform_events.GroupMessage( sender=platform_entities.GroupMember( - id=sender_id, + id=actor_id, member_name=interaction.user.display_name if interaction.user else '', permission='MEMBER', group=platform_entities.Group( @@ -1517,7 +1526,7 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): else: synthetic_event = platform_events.FriendMessage( sender=platform_entities.Friend( - id=sender_id, + id=actor_id, nickname=interaction.user.display_name if interaction.user else '', remark='', ), @@ -1553,7 +1562,7 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): bot_uuid=bot_uuid, launcher_type=launcher_type, launcher_id=launcher_id, - sender_id=sender_id, + sender_id=actor_id, message_event=synthetic_event, message_chain=message_chain, adapter=self, @@ -1565,7 +1574,8 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): ) if self.ap is not None: self.ap.logger.info( - f'Discord: button-click query enqueued action_id={action_id!r} session={session_key}' + f'Discord: button-click query enqueued action_id={action_id!r} ' + f'session={session_key} actor_id={actor_id}' ) except Exception: if self.ap is not None: diff --git a/src/langbot/pkg/platform/sources/qqofficial.py b/src/langbot/pkg/platform/sources/qqofficial.py index e9f07af5b..db823eef7 100644 --- a/src/langbot/pkg/platform/sources/qqofficial.py +++ b/src/langbot/pkg/platform/sources/qqofficial.py @@ -11,7 +11,13 @@ import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platf import langbot_plugin.api.entities.builtin.platform.message as platform_message import langbot_plugin.api.entities.builtin.platform.events as platform_events import langbot_plugin.api.entities.builtin.platform.entities as platform_entities -from langbot.libs.qq_official_api.api import QQOfficialClient, build_keyboard_from_form +from langbot.libs.qq_official_api.api import ( + QQ_SELECT_ACTION_PREFIX, + QQOfficialClient, + build_keyboard_from_form, + build_keyboard_from_select_field, + resolve_select_button_action, +) from langbot.libs.qq_official_api.qqofficialevent import QQOfficialEvent from ...utils import image from ..logger import EventLogger @@ -840,7 +846,8 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter parts.append('请点击下方按钮选择:') markdown_content = '\n\n'.join(parts) - if is_field_step: + keyboard = build_keyboard_from_select_field(form_data) if is_field_step else None + if is_field_step and not keyboard.get('content', {}).get('rows'): field_parts = parts[:-1] if len(parts) > 1 else parts text_msg = platform_message.MessageChain([platform_message.Plain(text='\n\n'.join(field_parts))]) try: @@ -874,7 +881,8 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter ) return - keyboard = build_keyboard_from_form(form_data, buttons_per_row=2) + if keyboard is None: + keyboard = build_keyboard_from_form(form_data, buttons_per_row=2) if not keyboard.get('content', {}).get('rows'): # No actions to render — fall back to plain text. text_msg = platform_message.MessageChain([platform_message.Plain(text=markdown_content)]) @@ -1018,7 +1026,7 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter session_key = f'{target_type}_{target_id}' self._prune_pending_forms() - pending = self._pending_forms.pop(session_key, None) + pending = self._pending_forms.get(session_key) if not pending: await self.logger.warning( f'QQ Official: no pending form for session {session_key}; click ignored (action_id={action_id!r})' @@ -1046,13 +1054,27 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter form_data: dict = pending.get('form_data') or {} actions = form_data.get('actions') or [] - matched = next( - (a for a in actions if str(a.get('id', '')) == action_id), - None, - ) - action_title = (matched or {}).get('title') or action_id + select_choice = resolve_select_button_action(form_data, action_id) + if action_id.startswith(QQ_SELECT_ACTION_PREFIX) and select_choice is None: + await self.logger.warning(f'QQ Official: invalid select action_id={action_id!r} for {session_key}') + return - sender_id = pending.get('sender_id') or event_data.get('user_openid') or event_data.get('member_openid') or '' + matched = None + if select_choice is None: + matched = next( + (a for a in actions if str(a.get('id', '')) == action_id), + None, + ) + if matched is None: + await self.logger.warning( + f'QQ Official: action_id={action_id!r} is not present on pending form for {session_key}' + ) + return + self._pending_forms.pop(session_key, None) + action_title = select_choice[1] if select_choice else matched.get('title') or action_id + + initiator_id = str(pending.get('sender_id') or '') + actor_id = str(event_data.get('member_openid') or event_data.get('user_openid') or initiator_id) # Build resume payload matching the shape every other adapter uses # (DingTalk / Lark / Telegram / WeCom). The runner's @@ -1062,24 +1084,28 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter launcher_id = target_id else: launcher_type = provider_session.LauncherTypes.PERSON - launcher_id = sender_id or target_id + launcher_id = target_id form_action_data = { 'form_token': form_data.get('form_token', ''), 'workflow_run_id': form_data.get('workflow_run_id', ''), - 'action_id': action_id, + 'action_id': '' if select_choice else action_id, 'action_title': action_title, 'node_title': form_data.get('node_title', ''), 'user': f'{launcher_type.value}_{launcher_id}', - 'inputs': {}, + 'inputs': {'select': select_choice[1]} if select_choice else {}, } + if select_choice: + form_action_data['_current_input_field'] = select_choice[0] + form_action_data['_input_progress'] = True - message_chain = platform_message.MessageChain([platform_message.Plain(text=f'[Form Action: {action_title}]')]) + event_label = 'Form Select' if select_choice else 'Form Action' + message_chain = platform_message.MessageChain([platform_message.Plain(text=f'[{event_label}: {action_title}]')]) if launcher_type == provider_session.LauncherTypes.GROUP: synthetic_event: platform_events.MessageEvent = platform_events.GroupMessage( sender=platform_entities.GroupMember( - id=sender_id or launcher_id, + id=actor_id or launcher_id, member_name='', permission='MEMBER', group=platform_entities.Group( @@ -1096,7 +1122,7 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter else: synthetic_event = platform_events.FriendMessage( sender=platform_entities.Friend( - id=sender_id or launcher_id, + id=actor_id or launcher_id, nickname='', remark='', ), @@ -1122,7 +1148,7 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter bot_uuid=bot_uuid, launcher_type=launcher_type, launcher_id=launcher_id, - sender_id=sender_id or launcher_id, + sender_id=actor_id or launcher_id, message_event=synthetic_event, message_chain=message_chain, adapter=self, @@ -1133,7 +1159,8 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter }, ) await self.logger.info( - f'QQ Official: button-click query enqueued action_id={action_id!r} session={session_key}' + f'QQ Official: button-click query enqueued action_id={action_id!r} ' + f'session={session_key} actor_id={actor_id}' ) except Exception: await self.logger.error(f'QQ Official: enqueue button-click query failed: {traceback.format_exc()}') diff --git a/src/langbot/pkg/platform/sources/telegram.py b/src/langbot/pkg/platform/sources/telegram.py index 135f0127c..afb583674 100644 --- a/src/langbot/pkg/platform/sources/telegram.py +++ b/src/langbot/pkg/platform/sources/telegram.py @@ -3,7 +3,7 @@ from __future__ import annotations import telegram import telegram.ext -from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup +from telegram import ForceReply, InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, CallbackQueryHandler, filters import telegramify_markdown import typing @@ -20,6 +20,61 @@ import langbot_plugin.api.entities.builtin.platform.entities as platform_entitie import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger +def _telegram_select_field_options(form_data: dict) -> tuple[str, list[str]]: + """Return the active select field and its option values.""" + field_name = str(form_data.get('_current_input_field') or '').strip() + if not field_name: + return '', [] + field = next( + ( + item + for item in form_data.get('input_defs') or [] + if str(item.get('output_variable_name') or '').strip() == field_name + ), + None, + ) + if not field or str(field.get('type') or '').strip().lower() != 'select': + return '', [] + + source = field.get('option_source') or {} + source_value = source.get('value') if isinstance(source, dict) else None + if isinstance(source_value, list): + return field_name, [str(item) for item in source_value] + if isinstance(source_value, str): + return field_name, [part.strip() for part in source_value.splitlines() if part.strip()] + + options = field.get('options') + if not isinstance(options, list): + return field_name, [] + values = [] + for item in options: + if isinstance(item, dict): + values.append(str(item.get('label') or item.get('value') or '')) + else: + values.append(str(item)) + return field_name, [value for value in values if value] + + +def _telegram_form_action_from_callback(data: dict) -> dict | None: + """Translate compact Telegram callback data into a runner form action.""" + if 'x' not in data: + return { + 'action_id': str(data.get('action_id') or data.get('a') or ''), + 'inputs': {}, + } + try: + option_index = int(data['x']) + except (TypeError, ValueError): + return None + if option_index < 0: + return None + return { + 'action_id': '', + 'inputs': {'select': str(option_index)}, + '_input_progress': True, + } + + class TelegramMessageConverter(abstract_platform_adapter.AbstractMessageConverter): @staticmethod async def yiri2target(message_chain: platform_message.MessageChain, bot: telegram.Bot) -> list[dict]: @@ -205,7 +260,7 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None], ] = {} - _form_action_titles: typing.Dict[str, str] = {} # action_id -> action_title mapping + _form_action_titles: typing.Dict[str, str] = {} # callback_data -> display title def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger): async def telegram_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): @@ -240,11 +295,13 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): # for Telegram's 64-byte limit). Only w_suffix is sent; # the runner resolves the full run id from _PENDING_FORMS. w_suffix = data.get('w', '') - action_id = data.get('action_id') or data.get('a', '') session_key = data.get('session_key') or data.get('s', '') - + callback_action = _telegram_form_action_from_callback(data) + if callback_action is None or query.data not in self._form_action_titles: + await self.logger.warning(f'Invalid or stale Telegram form callback: {query.data!r}') + return # Show selected action feedback by editing the original message - action_title = self._form_action_titles.get(action_id, action_id) + action_title = self._form_action_titles[query.data] try: original_text = query.message.text or '' selected_text = f'{original_text}\n\n✅ Selected: {action_title}' @@ -254,7 +311,7 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): pass finally: # Clean up the stored title - self._form_action_titles.pop(action_id, None) + self._form_action_titles.pop(query.data, None) if session_key.startswith('group_') or session_key.startswith('g:'): launcher_type = provider_session.LauncherTypes.GROUP @@ -286,13 +343,13 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): # workflow_run_id is intentionally omitted; the runner # resolves it from w_suffix via _PENDING_FORMS. 'w_suffix': w_suffix, - 'action_id': action_id, 'user': f'{launcher_type.value}_{launcher_id}', - 'inputs': {}, + **callback_action, } + event_label = 'Form Select' if callback_action.get('_input_progress') else 'Form Action' message_chain = platform_message.MessageChain( - [platform_message.Plain(text=f'[Form Action: {action_id}]')] + [platform_message.Plain(text=f'[{event_label}: {action_title}]')] ) if launcher_type == provider_session.LauncherTypes.GROUP: @@ -578,7 +635,7 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): form_data: dict, edit_message_id: int | None = None, ): - """Send inline keyboard buttons for Dify human_input_required form actions.""" + """Send inline keyboard buttons for Dify form fields or actions.""" actions = form_data.get('actions', []) node_title = form_data.get('node_title', '') form_content = form_data.get('form_content', '') @@ -593,48 +650,86 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): else: session_key = f'p:{message_source.sender.id}' + current_field = str(form_data.get('_current_input_field') or '').strip() + is_field_step = bool(current_field) and not form_data.get('_action_select_only') + select_field, select_options = _telegram_select_field_options(form_data) + is_select_field = bool(select_field and select_options) + if is_select_field: + choices = [(option, {'x': idx}) for idx, option in enumerate(select_options)] + elif is_field_step: + choices = [] + else: + choices = [(action.get('title', action.get('id', '')), {'a': action.get('id', '')}) for action in actions] + keyboard = [] + pending_title_mappings: dict[str, str] = {} oversized = False - for action in actions: - action_id = action.get('id', '') - action_title = action.get('title', action_id) - # Store action_id -> title mapping for displaying selection feedback - self._form_action_titles[action_id] = action_title - callback_payload = {'f': 1, 'a': action_id, 's': session_key} + buttons_per_row = 2 if is_select_field else 1 + current_row = [] + for title, choice_data in choices: + callback_payload = {'f': 1, **choice_data, 's': session_key} if w_suffix: callback_payload['w'] = w_suffix callback_data = json.dumps(callback_payload, separators=(',', ':')) if len(callback_data.encode('utf-8')) > 64: oversized = True break - keyboard.append([InlineKeyboardButton(action_title, callback_data=callback_data)]) + pending_title_mappings[callback_data] = str(title) + current_row.append(InlineKeyboardButton(str(title), callback_data=callback_data)) + if len(current_row) == buttons_per_row: + keyboard.append(current_row) + current_row = [] + if current_row and not oversized: + keyboard.append(current_row) update = message_source.source_platform_object chat_id = update.effective_chat.id effective_message = update.effective_message message_thread_id = getattr(effective_message, 'message_thread_id', None) if effective_message else None - text_lines = [f'[{node_title}] Please select an action:'] + if is_select_field: + prompt = f'Please select {select_field}:' + elif is_field_step: + prompt = f'Please reply with a value for {current_field}.' + else: + prompt = 'Please select an action:' + text_lines = [f'[{node_title}] {prompt}'] if form_content: text_lines.insert(0, form_content) if oversized: # callback_data exceeds Telegram's 64-byte limit — fall back to # a plain-text numbered list so the user can reply by number. - for idx, action in enumerate(actions, start=1): - title = action.get('title') or action.get('id') or '' + for idx, (title, _) in enumerate(choices, start=1): text_lines.append(f' {idx}. {title}') args = { 'chat_id': chat_id, 'text': '\n\n'.join(text_lines), } - else: + elif keyboard: + self._form_action_titles.update(pending_title_mappings) reply_markup = InlineKeyboardMarkup(keyboard) args = { 'chat_id': chat_id, 'text': '\n\n'.join(text_lines), 'reply_markup': reply_markup, } + elif is_field_step: + args = { + 'chat_id': chat_id, + 'text': '\n\n'.join(text_lines), + # Telegram privacy-mode bots receive replies to ForceReply + # prompts even when they cannot read ordinary group messages. + 'reply_markup': ForceReply( + selective=False, + input_field_placeholder=f'Enter {current_field}', + ), + } + else: + args = { + 'chat_id': chat_id, + 'text': '\n\n'.join(text_lines), + } if message_thread_id: args['message_thread_id'] = message_thread_id @@ -645,8 +740,7 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): 'message_id': edit_message_id, 'text': args['text'], } - if 'reply_markup' in args: - edit_args['reply_markup'] = args['reply_markup'] + edit_args['reply_markup'] = args.get('reply_markup') try: await self.bot.edit_message_text(**edit_args) return diff --git a/src/langbot/pkg/provider/runners/difysvapi.py b/src/langbot/pkg/provider/runners/difysvapi.py index 226bbfaae..d7b365a55 100644 --- a/src/langbot/pkg/provider/runners/difysvapi.py +++ b/src/langbot/pkg/provider/runners/difysvapi.py @@ -31,6 +31,18 @@ _PENDING_FORM_DEFAULT_TTL = 30 * 60 # 30 minutes safety cap _STREAM_FORM_PLACEHOLDER = '\u200b' +def _merge_stream_text(accumulated: str, incoming: typing.Any) -> str: + """Merge either a delta chunk or a cumulative stream snapshot.""" + incoming_text = '' if incoming is None else str(incoming) + if not incoming_text: + return accumulated + if not accumulated: + return incoming_text + if len(incoming_text) > len(accumulated) and incoming_text.startswith(accumulated): + return incoming_text + return accumulated + incoming_text + + def _session_key_from_query(query: pipeline_query.Query) -> str: return f'{query.session.launcher_type.value}_{query.session.launcher_id}' @@ -892,9 +904,9 @@ class DifyServiceAPIRunner(runner.RequestRunner): form_action, ) return - _clear_pending_form(session_key, form_action.get('form_token') or None) async for msg in self._submit_workflow_form_blocking(form_action): yield msg + _clear_pending_form(session_key, form_action.get('form_token') or None) return cov_id = query.session.using_conversation.uuid or None @@ -977,7 +989,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): ) elif mode == 'basic': if chunk['event'] == 'message': - basic_mode_pending_chunk += chunk['answer'] + basic_mode_pending_chunk = _merge_stream_text(basic_mode_pending_chunk, chunk['answer']) elif chunk['event'] == 'message_end': content, _ = self._process_thinking_content(basic_mode_pending_chunk) yield provider_message.Message( @@ -1034,7 +1046,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): continue if chunk['event'] == 'agent_message' or chunk['event'] == 'message': - pending_agent_message += chunk['answer'] + pending_agent_message = _merge_stream_text(pending_agent_message, chunk['answer']) else: if pending_agent_message.strip() != '': pending_agent_message = pending_agent_message.replace('Action:', '') @@ -1099,6 +1111,9 @@ class DifyServiceAPIRunner(runner.RequestRunner): user = form_action['user'] action_id = form_action.get('action_id', '') inputs = form_action.get('inputs', {}) + pending_content = '' + saw_event = False + answer_node_seen = False async for chunk in self.dify_client.workflow_submit( form_token=form_token, @@ -1108,21 +1123,49 @@ class DifyServiceAPIRunner(runner.RequestRunner): action=action_id, timeout=120, ): + saw_event = True self.ap.logger.debug('dify-workflow-submit-chunk: ' + str(chunk)) + event = chunk.get('event') - if chunk['event'] == 'workflow_finished': - if chunk['data'].get('error'): - raise errors.DifyAPIError(chunk['data']['error']) - content, _ = self._process_thinking_content(chunk['data']['outputs']['summary']) + if event == 'error': + raise errors.DifyAPIError(chunk.get('message') or 'Dify workflow resume failed') + + if event in ('message', 'agent_message') and not answer_node_seen: + pending_content = _merge_stream_text( + pending_content, + self._extract_dify_text_output(chunk.get('answer')), + ) + + if event == 'text_chunk': + pending_content = _merge_stream_text( + pending_content, + self._extract_dify_text_output(chunk.get('data', {}).get('text')), + ) + + if event == 'node_finished' and chunk.get('data', {}).get('node_type') == 'answer': + answer = self._extract_dify_text_output(chunk.get('data', {}).get('outputs', {}).get('answer')) + if answer: + # Answer-node output is the complete answer and may duplicate + # preceding message events, so prefer it over the accumulator. + pending_content = answer + answer_node_seen = True + + if event == 'workflow_finished': + data = chunk.get('data', {}) + if data.get('error'): + raise errors.DifyAPIError(data['error']) + if not pending_content: + pending_content = self._extract_dify_text_output(data.get('outputs', {}).get('summary')) + content, _ = self._process_thinking_content(pending_content) yield provider_message.Message( role='assistant', content=content, ) return - if chunk['event'] == 'workflow_paused': - reasons = chunk['data'].get('reasons', []) - new_run_id = chunk['data'].get('workflow_run_id', workflow_run_id) + if event == 'workflow_paused': + reasons = chunk.get('data', {}).get('reasons', []) + new_run_id = chunk.get('data', {}).get('workflow_run_id', workflow_run_id) for reason in reasons: if reason.get('TYPE') != 'human_input_required': continue @@ -1148,6 +1191,12 @@ class DifyServiceAPIRunner(runner.RequestRunner): ) return + if not saw_event: + raise errors.DifyAPIError('Dify API did not return any workflow resume events') + if pending_content: + content, _ = self._process_thinking_content(pending_content) + yield provider_message.Message(role='assistant', content=content) + def _resolve_pending_form(self, session_key: str, form_action: dict) -> dict | None: """Locate the pending form this action targets. @@ -1350,9 +1399,9 @@ class DifyServiceAPIRunner(runner.RequestRunner): form_action, ) return - _clear_pending_form(session_key, form_action.get('form_token') or None) async for msg in self._submit_workflow_form_blocking(form_action): yield msg + _clear_pending_form(session_key, form_action.get('form_token') or None) return if not query.session.using_conversation.uuid: @@ -1483,9 +1532,9 @@ class DifyServiceAPIRunner(runner.RequestRunner): form_action, ) return - _clear_pending_form(session_key, form_action.get('form_token') or None) async for msg in self._submit_workflow_form(form_action): yield msg + _clear_pending_form(session_key, form_action.get('form_token') or None) return cov_id = query.session.using_conversation.uuid or None @@ -1513,14 +1562,19 @@ class DifyServiceAPIRunner(runner.RequestRunner): chunk = None # 初始化chunk变量,防止在没有响应时引用错误 is_final = False - think_start = False - think_end = False yielded_final = False human_input_yielded = False pending_form_data = None remove_think = self.pipeline_config['output'].get('misc', {}).get('remove-think') + def visible_content(content: str) -> str: + if not remove_think: + return content + if '' in content and '' not in content: + return content.split('', 1)[0].rstrip() + return self._process_thinking_content(content)[0] + async for chunk in self.dify_client.chat_messages( inputs=inputs, query=plain_text, @@ -1539,23 +1593,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): if chunk['event'] == 'message': message_idx += 1 - if remove_think: - if '' in chunk['answer'] and not think_start: - think_start = True - continue - if '' in chunk['answer'] and not think_end: - import re - - content = re.sub(r'^\n', '', chunk['answer']) - basic_mode_pending_chunk += content - think_end = True - elif think_end: - basic_mode_pending_chunk += chunk['answer'] - if think_start: - continue - - else: - basic_mode_pending_chunk += chunk['answer'] + basic_mode_pending_chunk = _merge_stream_text(basic_mode_pending_chunk, chunk['answer']) if chunk['event'] == 'message_end': is_final = True @@ -1620,7 +1658,11 @@ class DifyServiceAPIRunner(runner.RequestRunner): and (is_final or message_idx % 8 == 0) and (basic_mode_pending_chunk != '' or is_final) ): - final_content = basic_mode_pending_chunk if basic_mode_pending_chunk.strip() else '' + final_content = visible_content(basic_mode_pending_chunk) + if not final_content.strip() and is_final and pending_form_data: + final_content = _STREAM_FORM_PLACEHOLDER + if not final_content and not is_final: + continue msg = provider_message.MessageChunk( role='assistant', content=final_content, @@ -1637,9 +1679,10 @@ class DifyServiceAPIRunner(runner.RequestRunner): # workflow_finished event, yield a final chunk so the adapter # can update the card and add buttons. if human_input_yielded and not yielded_final: + final_content = visible_content(basic_mode_pending_chunk) msg = provider_message.MessageChunk( role='assistant', - content=basic_mode_pending_chunk or '', + content=final_content or _STREAM_FORM_PLACEHOLDER, is_final=True, ) msg._form_data = pending_form_data @@ -1708,15 +1751,15 @@ class DifyServiceAPIRunner(runner.RequestRunner): import re content = re.sub(r'^\n', '', chunk['answer']) - pending_agent_message += content + pending_agent_message = _merge_stream_text(pending_agent_message, content) think_end = True elif think_end or not think_start: - pending_agent_message += chunk['answer'] + pending_agent_message = _merge_stream_text(pending_agent_message, chunk['answer']) if think_start and not think_end: continue else: - pending_agent_message += chunk['answer'] + pending_agent_message = _merge_stream_text(pending_agent_message, chunk['answer']) elif chunk['event'] == 'message_end': is_final = True else: @@ -1865,11 +1908,11 @@ class DifyServiceAPIRunner(runner.RequestRunner): workflow_contents += content think_end = True elif think_end: - workflow_contents += chunk['data']['text'] + workflow_contents = _merge_stream_text(workflow_contents, chunk['data']['text']) if think_start: continue else: - workflow_contents += chunk['data']['text'] + workflow_contents = _merge_stream_text(workflow_contents, chunk['data']['text']) if messsage_idx % 8 == 0: yield_this_iteration = True @@ -1890,11 +1933,11 @@ class DifyServiceAPIRunner(runner.RequestRunner): workflow_contents += content think_end = True elif think_end: - workflow_contents += answer + workflow_contents = _merge_stream_text(workflow_contents, answer) if think_start: continue else: - workflow_contents += answer + workflow_contents = _merge_stream_text(workflow_contents, answer) if messsage_idx % 8 == 0: yield_this_iteration = True @@ -1941,10 +1984,10 @@ class DifyServiceAPIRunner(runner.RequestRunner): form_action, ) return - _clear_pending_form(session_key, form_action.get('form_token') or None) # Resume paused workflow via submit endpoint async for msg in self._submit_workflow_form(form_action): yield msg + _clear_pending_form(session_key, form_action.get('form_token') or None) return if not query.session.using_conversation.uuid: @@ -2057,12 +2100,12 @@ class DifyServiceAPIRunner(runner.RequestRunner): workflow_contents += content think_end = True elif think_end: - workflow_contents += chunk['data']['text'] + workflow_contents = _merge_stream_text(workflow_contents, chunk['data']['text']) if think_start: continue else: - workflow_contents += chunk['data']['text'] + workflow_contents = _merge_stream_text(workflow_contents, chunk['data']['text']) if chunk['event'] == 'node_started': if chunk['data']['node_type'] == 'start' or chunk['data']['node_type'] == 'end': diff --git a/tests/unit_tests/platform/test_dingtalk_adapter.py b/tests/unit_tests/platform/test_dingtalk_adapter.py index 3eede810b..8551e613a 100644 --- a/tests/unit_tests/platform/test_dingtalk_adapter.py +++ b/tests/unit_tests/platform/test_dingtalk_adapter.py @@ -1,6 +1,12 @@ """Tests for DingTalk adapter helper behavior.""" +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + from langbot.pkg.platform.sources.dingtalk import ( + DingTalkAdapter, _dingtalk_clean_form_content, _dingtalk_completed_input_lines, _dingtalk_extract_component_inputs, @@ -102,3 +108,69 @@ def test_dingtalk_clean_form_content_uses_all_input_defs(): ) assert content == 'Hello' + + +def _build_card_action_adapter() -> DingTalkAdapter: + adapter = DingTalkAdapter.model_construct( + card_state={ + 'card-1': { + 'session_key': 'group_group-1', + 'launcher_type': 'group', + 'launcher_id': 'group-1', + 'sender_user_id': 'initiator-1', + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'actions': [{'id': 'approve', 'title': 'Approve'}], + 'node_title': 'Review', + 'form_content': 'Please review', + 'input_defs': [], + 'inputs': {}, + } + }, + active_turn_card={}, + active_turn_text={}, + ) + adapter.logger = AsyncMock() + adapter.ap = MagicMock() + adapter.ap.platform_mgr.bots = [] + adapter.ap.query_pool.add_query = AsyncMock() + return adapter + + +@pytest.mark.asyncio +async def test_dingtalk_group_card_action_uses_clicker_as_sender(): + adapter = _build_card_action_adapter() + + with patch.object(DingTalkAdapter, '_mark_card_resolved', new=AsyncMock()) as mark_resolved: + await adapter._on_card_action( + { + 'out_track_id': 'card-1', + 'user_id': 'reviewer-2', + 'action_id': 'approve', + 'params': {}, + } + ) + await asyncio.sleep(0) + + call = adapter.ap.query_pool.add_query.await_args + assert call.kwargs['launcher_id'] == 'group-1' + assert call.kwargs['sender_id'] == 'reviewer-2' + assert call.kwargs['message_event'].sender.id == 'reviewer-2' + mark_resolved.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_dingtalk_unknown_card_action_is_rejected(): + adapter = _build_card_action_adapter() + + await adapter._on_card_action( + { + 'out_track_id': 'card-1', + 'user_id': 'reviewer-2', + 'action_id': 'not-on-card', + 'params': {}, + } + ) + + adapter.ap.query_pool.add_query.assert_not_awaited() + adapter.logger.warning.assert_awaited_once() diff --git a/tests/unit_tests/platform/test_qqofficial_api.py b/tests/unit_tests/platform/test_qqofficial_api.py new file mode 100644 index 000000000..df63f583b --- /dev/null +++ b/tests/unit_tests/platform/test_qqofficial_api.py @@ -0,0 +1,117 @@ +"""Tests for QQ Official keyboard payload helpers.""" + +import asyncio +import time +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from langbot.libs.qq_official_api.api import ( + QQ_SELECT_ACTION_PREFIX, + build_keyboard_from_select_field, + get_select_field_options, + resolve_select_button_action, +) + + +def _select_form_data() -> dict: + return { + '_current_input_field': 'choice', + 'input_defs': [ + { + 'output_variable_name': 'choice', + 'type': 'select', + 'option_source': {'type': 'constant', 'value': ['A', 'B', 'C']}, + } + ], + } + + +def test_qq_select_field_builds_callback_buttons(): + keyboard = build_keyboard_from_select_field(_select_form_data(), buttons_per_row=2) + + rows = keyboard['content']['rows'] + assert [[button['render_data']['label'] for button in row['buttons']] for row in rows] == [ + ['A', 'B'], + ['C'], + ] + assert rows[0]['buttons'][0]['action']['data'] == f'{QQ_SELECT_ACTION_PREFIX}0' + assert rows[0]['buttons'][1]['action']['data'] == f'{QQ_SELECT_ACTION_PREFIX}1' + + +def test_qq_select_button_resolves_field_and_value(): + form_data = _select_form_data() + + assert get_select_field_options(form_data) == ('choice', ['A', 'B', 'C']) + assert resolve_select_button_action(form_data, f'{QQ_SELECT_ACTION_PREFIX}1') == ('choice', 'B') + assert resolve_select_button_action(form_data, f'{QQ_SELECT_ACTION_PREFIX}99') is None + + +def test_qq_select_keyboard_fits_twenty_five_options(): + form_data = _select_form_data() + form_data['input_defs'][0]['option_source']['value'] = [f'Option {idx}' for idx in range(25)] + + rows = build_keyboard_from_select_field(form_data)['content']['rows'] + + assert len(rows) == 5 + assert all(len(row['buttons']) == 5 for row in rows) + + +def test_qq_non_select_field_does_not_build_keyboard(): + form_data = { + '_current_input_field': 'comment', + 'input_defs': [{'output_variable_name': 'comment', 'type': 'paragraph'}], + } + + assert build_keyboard_from_select_field(form_data)['content']['rows'] == [] + + +@pytest.mark.asyncio +async def test_qq_select_click_enqueues_input_progress_query(): + import langbot.pkg.core.app # noqa: F401 + from langbot.pkg.platform.sources.qqofficial import QQOfficialAdapter + + adapter = QQOfficialAdapter.model_construct() + adapter.logger = AsyncMock() + adapter.bot = MagicMock() + adapter.bot.ack_interaction = AsyncMock() + adapter.ap = MagicMock() + adapter.ap.platform_mgr.bots = [] + adapter.ap.query_pool.add_query = AsyncMock() + adapter._pending_forms = { + 'group_group-1': { + 'form_data': { + **_select_form_data(), + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'node_title': 'Review', + 'actions': [{'id': 'approve', 'title': 'Approve'}], + }, + 'sender_id': 'initiator-1', + 'posted_at': time.time(), + } + } + adapter._session_event_ids = {} + adapter._anchor_msg_seq = {} + + await adapter._handle_interaction_create( + { + 'id': 'interaction-1', + 'chat_type': 1, + 'group_openid': 'group-1', + 'member_openid': 'reviewer-2', + 'data': {'resolved': {'button_data': f'{QQ_SELECT_ACTION_PREFIX}1'}}, + }, + ws_event_id='event-1', + ) + await asyncio.sleep(0) + + call = adapter.ap.query_pool.add_query.await_args + form_action = call.kwargs['variables']['_dify_form_action'] + assert call.kwargs['launcher_id'] == 'group-1' + assert call.kwargs['sender_id'] == 'reviewer-2' + assert form_action['action_id'] == '' + assert form_action['inputs'] == {'select': 'B'} + assert form_action['_current_input_field'] == 'choice' + assert form_action['_input_progress'] is True + adapter.bot.ack_interaction.assert_awaited_once_with('interaction-1', code=0) diff --git a/tests/unit_tests/platform/test_telegram_adapter.py b/tests/unit_tests/platform/test_telegram_adapter.py new file mode 100644 index 000000000..247d51b16 --- /dev/null +++ b/tests/unit_tests/platform/test_telegram_adapter.py @@ -0,0 +1,120 @@ +"""Tests for Telegram Dify form callback helpers.""" + +import json +from unittest.mock import AsyncMock, MagicMock + +import pytest +from telegram import ForceReply + +import langbot_plugin.api.entities.builtin.platform.entities as platform_entities +import langbot_plugin.api.entities.builtin.platform.events as platform_events +import langbot_plugin.api.entities.builtin.platform.message as platform_message +from langbot.pkg.platform.sources.telegram import ( + TelegramAdapter, + _telegram_form_action_from_callback, + _telegram_select_field_options, +) + + +def _select_form_data() -> dict: + return { + '_current_input_field': 'choice', + 'input_defs': [ + { + 'output_variable_name': 'choice', + 'type': 'select', + 'option_source': {'type': 'constant', 'value': ['A', 'B', 'C']}, + } + ], + } + + +def test_telegram_select_field_options_are_extracted(): + assert _telegram_select_field_options(_select_form_data()) == ('choice', ['A', 'B', 'C']) + + +def test_telegram_select_callback_becomes_input_progress(): + assert _telegram_form_action_from_callback({'f': 1, 'x': 1}) == { + 'action_id': '', + 'inputs': {'select': '1'}, + '_input_progress': True, + } + + +def test_telegram_action_callback_remains_final_action(): + assert _telegram_form_action_from_callback({'f': 1, 'a': 'approve'}) == { + 'action_id': 'approve', + 'inputs': {}, + } + + +def test_telegram_invalid_select_callback_is_rejected(): + assert _telegram_form_action_from_callback({'f': 1, 'x': -1}) is None + assert _telegram_form_action_from_callback({'f': 1, 'x': 'invalid'}) is None + + +@pytest.mark.asyncio +async def test_telegram_select_field_sends_two_column_inline_keyboard(): + bot = MagicMock() + bot.send_message = AsyncMock() + adapter = TelegramAdapter.model_construct(bot=bot, config={}, msg_stream_id={}, seq=1, listeners={}) + adapter._form_action_titles = {} + + update = MagicMock() + update.effective_chat.id = 123 + update.effective_message.message_thread_id = None + event = platform_events.FriendMessage( + sender=platform_entities.Friend(id='user-1', nickname='', remark=''), + message_chain=platform_message.MessageChain([]), + source_platform_object=update, + ) + form_data = { + **_select_form_data(), + 'node_title': 'Review', + 'form_content': 'Choose one', + 'workflow_run_id': 'workflow-run-12345678', + 'actions': [{'id': 'approve', 'title': 'Approve'}], + } + + await adapter._send_form_action_buttons(event, form_data) + + args = bot.send_message.await_args.kwargs + rows = args['reply_markup'].inline_keyboard + assert [[button.text for button in row] for row in rows] == [['A', 'B'], ['C']] + callback_data = rows[0][1].callback_data + assert len(callback_data.encode('utf-8')) <= 64 + assert json.loads(callback_data)['x'] == 1 + assert callback_data in adapter._form_action_titles + + +@pytest.mark.asyncio +async def test_telegram_text_field_does_not_show_action_buttons(): + bot = MagicMock() + bot.send_message = AsyncMock() + adapter = TelegramAdapter.model_construct(bot=bot, config={}, msg_stream_id={}, seq=1, listeners={}) + adapter._form_action_titles = {} + + update = MagicMock() + update.effective_chat.id = 123 + update.effective_message.message_thread_id = None + event = platform_events.FriendMessage( + sender=platform_entities.Friend(id='user-1', nickname='', remark=''), + message_chain=platform_message.MessageChain([]), + source_platform_object=update, + ) + form_data = { + '_current_input_field': 'us_input', + 'input_defs': [{'output_variable_name': 'us_input', 'type': 'paragraph'}], + 'node_title': '人工介入', + 'form_content': 'us_input (paragraph): reply "us_input: "', + 'workflow_run_id': 'workflow-run-12345678', + 'actions': [{'id': 'yes', 'title': 'yes'}, {'id': 'no', 'title': 'no'}], + } + + await adapter._send_form_action_buttons(event, form_data) + + args = bot.send_message.await_args.kwargs + assert isinstance(args['reply_markup'], ForceReply) + assert args['reply_markup'].selective is False + assert 'Please reply with a value for us_input.' in args['text'] + assert adapter._form_action_titles == {} diff --git a/tests/unit_tests/provider/runners/test_difysvapi_runner.py b/tests/unit_tests/provider/runners/test_difysvapi_runner.py index 8db124386..80e77b563 100644 --- a/tests/unit_tests/provider/runners/test_difysvapi_runner.py +++ b/tests/unit_tests/provider/runners/test_difysvapi_runner.py @@ -128,7 +128,7 @@ class TestDifyRunnerConfigValidation: mock_app = MagicMock() - for app_type in ['chat', 'agent', 'workflow']: + for app_type in ['chat', 'agent', 'workflow', 'chatflow']: pipeline_config = { 'ai': { 'dify-service-api': { @@ -669,3 +669,205 @@ class TestDifyHumanInputForms: assert second['_partial'] is True assert second['inputs'] == {'comment': 'looks good', 'choice': 'B'} assert second['_form_data']['_action_select_only'] is True + + @pytest.mark.asyncio + async def test_blocking_resume_uses_chatflow_answer_node_output(self): + runner = self._create_runner() + + async def workflow_submit(**kwargs): + del kwargs + yield {'event': 'message', 'answer': 'partial'} + yield { + 'event': 'node_finished', + 'data': { + 'node_type': 'answer', + 'outputs': {'answer': {'content': 'Final chatflow answer'}}, + }, + } + yield {'event': 'workflow_finished', 'data': {'error': None, 'outputs': {}}} + + runner.dify_client.workflow_submit = workflow_submit + + messages = [ + message + async for message in runner._submit_workflow_form_blocking( + { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'user': 'person_user-1', + 'action_id': 'yes', + 'inputs': {}, + } + ) + ] + + assert [message.content for message in messages] == ['Final chatflow answer'] + + @pytest.mark.asyncio + async def test_successful_blocking_resume_clears_pending_form(self): + from langbot.pkg.provider.runners import difysvapi + + runner = self._create_runner() + session_key = 'person_user-1' + difysvapi._PENDING_FORMS.clear() + difysvapi._set_pending_form( + session_key, + { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'actions': [{'id': 'yes', 'title': 'Yes'}], + 'inputs': {}, + 'user': session_key, + }, + ) + + async def workflow_submit(**kwargs): + del kwargs + yield { + 'event': 'workflow_finished', + 'data': {'error': None, 'outputs': {'summary': 'Completed'}}, + } + + runner.dify_client.workflow_submit = workflow_submit + query = MagicMock() + query.session.launcher_type.value = 'person' + query.session.launcher_id = 'user-1' + query.variables = { + '_dify_form_action': { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'action_id': 'yes', + 'user': session_key, + 'inputs': {}, + } + } + + messages = [message async for message in runner._workflow_messages(query)] + + assert [message.content for message in messages] == ['Completed'] + assert difysvapi._get_pending_form_by_token(session_key, 'token-1') is None + + @pytest.mark.asyncio + async def test_failed_blocking_resume_keeps_pending_form_for_retry(self): + from langbot.libs.dify_service_api.v1.errors import DifyAPIError + from langbot.pkg.provider.runners import difysvapi + + runner = self._create_runner() + session_key = 'person_user-1' + difysvapi._PENDING_FORMS.clear() + difysvapi._set_pending_form( + session_key, + { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'actions': [{'id': 'yes', 'title': 'Yes'}], + 'inputs': {}, + 'user': session_key, + }, + ) + + async def workflow_submit(**kwargs): + del kwargs + raise DifyAPIError('temporary failure') + yield + + runner.dify_client.workflow_submit = workflow_submit + query = MagicMock() + query.session.launcher_type.value = 'person' + query.session.launcher_id = 'user-1' + query.variables = { + '_dify_form_action': { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'action_id': 'yes', + 'user': session_key, + 'inputs': {}, + } + } + + with pytest.raises(DifyAPIError, match='temporary failure'): + _ = [message async for message in runner._workflow_messages(query)] + + assert difysvapi._get_pending_form_by_token(session_key, 'token-1') is not None + difysvapi._PENDING_FORMS.clear() + + +class TestDifyCumulativeStreaming: + def _create_runner(self, *, remove_think: bool = False): + from langbot.pkg.provider.runners.difysvapi import DifyServiceAPIRunner + + mock_app = MagicMock() + mock_app.logger = MagicMock() + runner = DifyServiceAPIRunner( + mock_app, + { + 'ai': { + 'dify-service-api': { + 'app-type': 'chat', + 'api-key': 'test-key', + 'base-url': 'https://api.dify.ai', + 'base-prompt': '', + } + }, + 'output': {'misc': {'remove-think': remove_think}}, + }, + ) + runner.dify_client = MagicMock() + return runner + + @staticmethod + def _query(): + query = MagicMock() + query.session.launcher_type.value = 'person' + query.session.launcher_id = 'user-1' + query.session.using_conversation.uuid = 'conversation-1' + query.variables = {} + query.user_message.content = 'hello' + return query + + def test_merge_stream_text_accepts_deltas_and_snapshots(self): + from langbot.pkg.provider.runners.difysvapi import _merge_stream_text + + assert _merge_stream_text('Hello', ' world') == 'Hello world' + assert _merge_stream_text('one', 'one two') == 'one two' + assert _merge_stream_text('ha', 'ha') == 'haha' + + @pytest.mark.asyncio + async def test_chat_stream_deduplicates_cumulative_snapshots(self): + runner = self._create_runner() + snapshots = [f'Reasoning{"." * idx}' for idx in range(1, 10)] + snapshots.append(f'{snapshots[-1]}\nHello!') + + async def chat_messages(**kwargs): + del kwargs + for snapshot in snapshots: + yield {'event': 'message', 'answer': snapshot, 'conversation_id': 'conversation-2'} + yield {'event': 'message_end', 'conversation_id': 'conversation-2'} + + runner.dify_client.chat_messages = chat_messages + + chunks = [chunk async for chunk in runner._chat_messages_chunk(self._query())] + + assert chunks[-1].content == snapshots[-1] + assert chunks[-1].content.count('') == 1 + + @pytest.mark.asyncio + async def test_chat_stream_removes_think_from_cumulative_snapshots(self): + runner = self._create_runner(remove_think=True) + + async def chat_messages(**kwargs): + del kwargs + yield {'event': 'message', 'answer': 'Reasoning', 'conversation_id': 'conversation-2'} + yield { + 'event': 'message', + 'answer': 'Reasoning complete\nHello!', + 'conversation_id': 'conversation-2', + } + yield {'event': 'message_end', 'conversation_id': 'conversation-2'} + + runner.dify_client.chat_messages = chat_messages + + chunks = [chunk async for chunk in runner._chat_messages_chunk(self._query())] + + assert chunks[-1].content == 'Hello!' + assert all('' not in chunk.content for chunk in chunks if isinstance(chunk.content, str))