From 1bdf371b5da0a3d797faaa13d34f9d88b67832a0 Mon Sep 17 00:00:00 2001 From: fdc310 <2213070223@qq.com> Date: Sun, 12 Jul 2026 19:40:59 +0800 Subject: [PATCH] feat: Add pipeline_uuid handling across multiple adapters and update related tests --- src/langbot/pkg/platform/sources/dingtalk.py | 10 +++-- src/langbot/pkg/platform/sources/discord.py | 4 +- src/langbot/pkg/platform/sources/lark.py | 10 +++-- .../pkg/platform/sources/qqofficial.py | 4 +- .../pkg/platform/sources/qqofficial.yaml | 6 +-- src/langbot/pkg/platform/sources/telegram.py | 40 ++++++++++++------- src/langbot/pkg/platform/sources/wecombot.py | 5 ++- src/langbot/pkg/provider/runners/difysvapi.py | 13 ++++++ .../platform/test_telegram_adapter.py | 15 +++++++ .../provider/runners/test_difysvapi_runner.py | 38 ++++++++++++++++++ 10 files changed, 114 insertions(+), 31 deletions(-) diff --git a/src/langbot/pkg/platform/sources/dingtalk.py b/src/langbot/pkg/platform/sources/dingtalk.py index e9e1f71e3..996187f08 100644 --- a/src/langbot/pkg/platform/sources/dingtalk.py +++ b/src/langbot/pkg/platform/sources/dingtalk.py @@ -939,6 +939,7 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): 'sender_user_id': sender_user_id, 'form_token': form_data.get('form_token', ''), 'workflow_run_id': form_data.get('workflow_run_id', ''), + 'pipeline_uuid': form_data.get('pipeline_uuid', ''), 'actions': actions if should_show_actions else [], 'all_actions': actions, 'node_title': node_title, @@ -1046,6 +1047,7 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): 'sender_user_id': sender_user_id, 'form_token': form_data.get('form_token', ''), 'workflow_run_id': form_data.get('workflow_run_id', ''), + 'pipeline_uuid': form_data.get('pipeline_uuid', ''), 'actions': actions if should_show_actions else [], 'all_actions': actions, 'node_title': node_title, @@ -1318,12 +1320,12 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): ) bot_uuid = '' - pipeline_uuid = None + pipeline_uuid = state.get('pipeline_uuid') or None if self.ap is not None: for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid - pipeline_uuid = bot.bot_entity.use_pipeline_uuid + pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break try: @@ -1451,11 +1453,11 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): return bot_uuid = '' - pipeline_uuid = None + pipeline_uuid = state.get('pipeline_uuid') or None for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid - pipeline_uuid = bot.bot_entity.use_pipeline_uuid + pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break try: await self.ap.query_pool.add_query( diff --git a/src/langbot/pkg/platform/sources/discord.py b/src/langbot/pkg/platform/sources/discord.py index 3cb13a251..fae65f6a0 100644 --- a/src/langbot/pkg/platform/sources/discord.py +++ b/src/langbot/pkg/platform/sources/discord.py @@ -1540,11 +1540,11 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): return bot_uuid = '' - pipeline_uuid = None + pipeline_uuid = form_data.get('pipeline_uuid') or None for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid - pipeline_uuid = bot.bot_entity.use_pipeline_uuid + pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break # Remember the channel so _reply_synthetic and _handle_form_chunk diff --git a/src/langbot/pkg/platform/sources/lark.py b/src/langbot/pkg/platform/sources/lark.py index bf22dffa5..fdc5cb506 100644 --- a/src/langbot/pkg/platform/sources/lark.py +++ b/src/langbot/pkg/platform/sources/lark.py @@ -1074,11 +1074,11 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): # Find the bot entity to get bot_uuid and pipeline_uuid bot_uuid = '' - pipeline_uuid = None + pipeline_uuid = action_value_obj.get('pipeline_uuid') or None for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid - pipeline_uuid = bot.bot_entity.use_pipeline_uuid + pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break form_action_data = { @@ -2396,6 +2396,7 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): actions = form_data.get('actions', []) form_token = form_data.get('form_token', '') workflow_run_id = form_data.get('workflow_run_id', '') + pipeline_uuid = form_data.get('pipeline_uuid', '') node_title = form_data.get('node_title', '') or 'Human Input Required' form_content = form_data.get('form_content', '') input_defs = _lark_form_input_defs(form_data) @@ -2459,6 +2460,7 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): 'form_action': True, 'form_token': form_token, 'workflow_run_id': workflow_run_id, + 'pipeline_uuid': pipeline_uuid, 'action_id': action_id, 'session_key': session_key, 'card_id': card_id, @@ -2940,11 +2942,11 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): ) bot_uuid = '' - pipeline_uuid = None + pipeline_uuid = action_value_obj.get('pipeline_uuid') or None for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid - pipeline_uuid = bot.bot_entity.use_pipeline_uuid + pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break await self.ap.query_pool.add_query( diff --git a/src/langbot/pkg/platform/sources/qqofficial.py b/src/langbot/pkg/platform/sources/qqofficial.py index 3369391e8..35fd81f76 100644 --- a/src/langbot/pkg/platform/sources/qqofficial.py +++ b/src/langbot/pkg/platform/sources/qqofficial.py @@ -1108,11 +1108,11 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter return bot_uuid = '' - pipeline_uuid = None + pipeline_uuid = form_data.get('pipeline_uuid') or None for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid - pipeline_uuid = bot.bot_entity.use_pipeline_uuid + pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break try: diff --git a/src/langbot/pkg/platform/sources/qqofficial.yaml b/src/langbot/pkg/platform/sources/qqofficial.yaml index c9a1bfe79..c3f193be9 100644 --- a/src/langbot/pkg/platform/sources/qqofficial.yaml +++ b/src/langbot/pkg/platform/sources/qqofficial.yaml @@ -37,9 +37,9 @@ spec: zh_Hans: 一键扫码绑定 zh_Hant: 一鍵掃碼綁定 description: - en_US: Scan QR code with mobile QQ to auto-fill AppID and Secret (Token still needs to be filled manually) - zh_Hans: 使用手机 QQ 扫码绑定，自动填写 AppID 和密钥（Token 仍需手动填写） - zh_Hant: 使用手機 QQ 掃碼綁定，自動填寫 AppID 和密鑰（Token 仍需手動填寫） + en_US: Scan QR code with mobile QQ to auto-fill AppID and Secret (Token is not used and can be left blank) + zh_Hans: 使用手机 QQ 扫码绑定，自动填写 AppID 和密钥（当前未使用 Token，可留空） + zh_Hant: 使用手機 QQ 掃碼綁定，自動填寫 AppID 和密鑰（目前未使用 Token，可留空） type: qr-code-login login_platform: qqofficial required: false diff --git a/src/langbot/pkg/platform/sources/telegram.py b/src/langbot/pkg/platform/sources/telegram.py index 9d58c9287..e7ad626bb 100644 --- a/src/langbot/pkg/platform/sources/telegram.py +++ b/src/langbot/pkg/platform/sources/telegram.py @@ -263,37 +263,46 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): ] = {} _FORM_ACTION_CACHE_TTL = 30 * 60 - # callback_data -> (display title, expiration monotonic time, form group id) - _form_action_titles: typing.Dict[str, tuple[str, float, str]] = {} + # callback_data -> (display title, pipeline UUID, expiration time, form group id) + _form_action_titles: typing.Dict[str, tuple[str, str, float, str]] = {} def _prune_form_action_titles(self, now: float | None = None) -> None: now = time.monotonic() if now is None else now - expired = [key for key, (_, expires_at, _) in self._form_action_titles.items() if expires_at <= now] + expired = [key for key, (_, _, expires_at, _) in self._form_action_titles.items() if expires_at <= now] for key in expired: self._form_action_titles.pop(key, None) - def _cache_form_action_titles(self, mappings: dict[str, str], now: float | None = None) -> None: + def _cache_form_action_titles( + self, + mappings: dict[str, str], + pipeline_uuid: str = '', + now: float | None = None, + ) -> None: now = time.monotonic() if now is None else now self._prune_form_action_titles(now) group_id = uuid.uuid4().hex expires_at = now + self._FORM_ACTION_CACHE_TTL self._form_action_titles.update( - {callback_data: (title, expires_at, group_id) for callback_data, title in mappings.items()} + {callback_data: (title, pipeline_uuid, expires_at, group_id) for callback_data, title in mappings.items()} ) - def _take_form_action_title(self, callback_data: str, now: float | None = None) -> str | None: + def _take_form_action_context(self, callback_data: str, now: float | None = None) -> tuple[str, str] | None: """Consume a callback and invalidate every button from the same form.""" self._prune_form_action_titles(now) entry = self._form_action_titles.get(callback_data) if entry is None: return None - title, _, group_id = entry + title, pipeline_uuid, _, group_id = entry group_keys = [ - key for key, (_, _, cached_group_id) in self._form_action_titles.items() if cached_group_id == group_id + key for key, (_, _, _, cached_group_id) in self._form_action_titles.items() if cached_group_id == group_id ] for key in group_keys: self._form_action_titles.pop(key, None) - return title + return title, pipeline_uuid + + def _take_form_action_title(self, callback_data: str, now: float | None = None) -> str | None: + context = self._take_form_action_context(callback_data, now) + return context[0] if context else None def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger): async def telegram_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): @@ -330,10 +339,11 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): w_suffix = data.get('w', '') session_key = data.get('session_key') or data.get('s', '') callback_action = _telegram_form_action_from_callback(data) - action_title = self._take_form_action_title(query.data) if callback_action is not None else None - if callback_action is None or action_title is None: + action_context = self._take_form_action_context(query.data) if callback_action is not None else None + if callback_action is None or action_context is None: await self.logger.warning(f'Invalid or stale Telegram form callback: {query.data!r}') return + action_title, pipeline_uuid = action_context # Show selected action feedback by editing the original message try: original_text = query.message.text or '' @@ -362,11 +372,10 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): # Find bot_uuid and pipeline_uuid bot_uuid = '' - pipeline_uuid = None for b in self.ap.platform_mgr.bots: if b.adapter is self: bot_uuid = b.bot_entity.uuid - pipeline_uuid = b.bot_entity.use_pipeline_uuid + pipeline_uuid = pipeline_uuid or b.bot_entity.use_pipeline_uuid break form_action_data = { @@ -732,7 +741,10 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): 'text': '\n\n'.join(text_lines), } elif keyboard: - self._cache_form_action_titles(pending_title_mappings) + self._cache_form_action_titles( + pending_title_mappings, + str(form_data.get('pipeline_uuid') or ''), + ) reply_markup = InlineKeyboardMarkup(keyboard) args = { 'chat_id': chat_id, diff --git a/src/langbot/pkg/platform/sources/wecombot.py b/src/langbot/pkg/platform/sources/wecombot.py index 4175817e2..d0febad9e 100644 --- a/src/langbot/pkg/platform/sources/wecombot.py +++ b/src/langbot/pkg/platform/sources/wecombot.py @@ -838,6 +838,7 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): 'node_title': form.get('node_title', ''), 'workflow_run_id': form.get('workflow_run_id', ''), 'form_token': form.get('form_token', ''), + 'pipeline_uuid': form.get('pipeline_uuid', ''), '_action_select_only': True, } target_chat_id = session.chat_id or session.user_id or '' @@ -924,11 +925,11 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): return bot_uuid = '' - pipeline_uuid = None + pipeline_uuid = form.get('pipeline_uuid') or None for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid - pipeline_uuid = bot.bot_entity.use_pipeline_uuid + pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break try: diff --git a/src/langbot/pkg/provider/runners/difysvapi.py b/src/langbot/pkg/provider/runners/difysvapi.py index afee2f68f..7566d20aa 100644 --- a/src/langbot/pkg/provider/runners/difysvapi.py +++ b/src/langbot/pkg/provider/runners/difysvapi.py @@ -74,6 +74,8 @@ def _prune_pending_forms(now: float | None = None) -> None: def _set_pending_form(session_key: PendingFormKey, form_data: dict[str, typing.Any]) -> None: _prune_pending_forms() + if isinstance(session_key, tuple) and len(session_key) > 1: + form_data['pipeline_uuid'] = session_key[1] stored = dict(form_data) expiration_time = stored.get('expiration_time') try: @@ -514,6 +516,7 @@ def _field_input_form_data(pending_form: dict[str, typing.Any], field: dict[str, 'node_title': pending_form.get('node_title', ''), 'workflow_run_id': pending_form.get('workflow_run_id', ''), 'form_token': pending_form.get('form_token', ''), + 'pipeline_uuid': pending_form.get('pipeline_uuid', ''), '_current_input_field': _field_name(field), } @@ -533,6 +536,7 @@ def _action_select_form_data(pending_form: dict[str, typing.Any]) -> dict[str, t 'node_title': pending_form.get('node_title', ''), 'workflow_run_id': pending_form.get('workflow_run_id', ''), 'form_token': pending_form.get('form_token', ''), + 'pipeline_uuid': pending_form.get('pipeline_uuid', ''), '_action_select_only': True, } @@ -1341,6 +1345,8 @@ class DifyServiceAPIRunner(runner.RequestRunner): if form: return form + if form_token or workflow_run_id or w_suffix: + return None return _get_latest_pending_form(session_key) def _merge_pending_form_action(self, session_key: PendingFormKey, form_action: dict | None) -> dict | None: @@ -1351,6 +1357,10 @@ class DifyServiceAPIRunner(runner.RequestRunner): merged_action = dict(form_action) merged_action.pop('w_suffix', None) pending_form = self._resolve_pending_form(session_key, form_action) + if pending_form is None and any( + form_action.get(identifier) for identifier in ('form_token', 'workflow_run_id', 'w_suffix') + ): + return None if pending_form: merged_action['form_token'] = merged_action.get('form_token') or pending_form.get('form_token', '') merged_action['workflow_run_id'] = merged_action.get('workflow_run_id') or pending_form.get( @@ -1771,6 +1781,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): 'node_title': node_title, 'workflow_run_id': workflow_run_id, 'form_token': reason.get('form_token', ''), + 'pipeline_uuid': form_snapshot.get('pipeline_uuid', ''), } human_input_yielded = True @@ -2010,6 +2021,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): 'node_title': paused_node_title, 'workflow_run_id': new_run_id, 'form_token': reason.get('form_token', ''), + 'pipeline_uuid': form_snapshot.get('pipeline_uuid', ''), } # Ensure the final chunk has non-empty content so # ResponseWrapper (which skips empty-content chunks) lets it @@ -2208,6 +2220,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): 'node_title': node_title, 'workflow_run_id': workflow_run_id, 'form_token': reason.get('form_token', ''), + 'pipeline_uuid': form_snapshot.get('pipeline_uuid', ''), } human_input_yielded = True diff --git a/tests/unit_tests/platform/test_telegram_adapter.py b/tests/unit_tests/platform/test_telegram_adapter.py index c97349f51..6e2262283 100644 --- a/tests/unit_tests/platform/test_telegram_adapter.py +++ b/tests/unit_tests/platform/test_telegram_adapter.py @@ -73,6 +73,21 @@ def test_telegram_form_callback_cache_prunes_expired_entries(): assert adapter._form_action_titles == {} +def test_telegram_form_callback_cache_preserves_pipeline_uuid(): + adapter = TelegramAdapter.model_construct() + adapter._form_action_titles = {} + adapter._cache_form_action_titles( + {'callback-a': 'Approve'}, + pipeline_uuid='pipeline-routed', + now=100.0, + ) + + assert adapter._take_form_action_context('callback-a', now=101.0) == ( + 'Approve', + 'pipeline-routed', + ) + + @pytest.mark.asyncio async def test_telegram_select_field_sends_two_column_inline_keyboard(): bot = MagicMock() diff --git a/tests/unit_tests/provider/runners/test_difysvapi_runner.py b/tests/unit_tests/provider/runners/test_difysvapi_runner.py index 2263267c1..f75938209 100644 --- a/tests/unit_tests/provider/runners/test_difysvapi_runner.py +++ b/tests/unit_tests/provider/runners/test_difysvapi_runner.py @@ -291,10 +291,48 @@ class TestDifyHumanInputForms: assert difysvapi._get_latest_pending_form(key_a)['workflow_run_id'] == 'run-a' assert difysvapi._get_latest_pending_form(key_b)['workflow_run_id'] == 'run-b' assert difysvapi._get_latest_pending_form(key_c)['workflow_run_id'] == 'run-c' + assert difysvapi._get_latest_pending_form(key_a)['pipeline_uuid'] == 'pipeline-a' + assert difysvapi._get_latest_pending_form(key_c)['pipeline_uuid'] == 'pipeline-b' assert difysvapi._dify_user_from_query(query_a) == difysvapi._dify_user_from_query(query_b) assert difysvapi._dify_user_from_query(query_a) == difysvapi._dify_user_from_query(query_c) difysvapi._PENDING_FORMS.clear() + def test_interactive_form_data_preserves_pipeline_uuid(self): + from langbot.pkg.provider.runners import difysvapi + + pending_form = { + 'pipeline_uuid': 'pipeline-routed', + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'input_defs': [{'output_variable_name': 'comment', 'type': 'paragraph'}], + 'actions': [{'id': 'approve', 'title': 'Approve'}], + } + + field_form = difysvapi._field_input_form_data(pending_form, pending_form['input_defs'][0]) + action_form = difysvapi._action_select_form_data(pending_form) + + assert field_form['pipeline_uuid'] == 'pipeline-routed' + assert action_form['pipeline_uuid'] == 'pipeline-routed' + + def test_explicit_stale_form_identifiers_do_not_fall_back_to_latest(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': 'current-token', 'workflow_run_id': 'current-run'}, + ) + + assert runner._resolve_pending_form(session_key, {'form_token': 'stale-token'}) is None + assert runner._resolve_pending_form(session_key, {'workflow_run_id': 'stale-run'}) is None + assert runner._resolve_pending_form(session_key, {'w_suffix': 'stale-suffix'}) is None + assert runner._merge_pending_form_action(session_key, {'form_token': 'stale-token'}) is None + assert runner._resolve_pending_form(session_key, {})['form_token'] == 'current-token' + + difysvapi._PENDING_FORMS.clear() + def test_format_human_input_text_includes_field_help(self): from langbot.pkg.provider.runners.difysvapi import _format_human_input_text