From 5eb2b90ae6eb0e48364b03d4071b29ad7a12edb9 Mon Sep 17 00:00:00 2001 From: fdc310 <2213070223@qq.com> Date: Sat, 11 Jul 2026 20:46:42 +0800 Subject: [PATCH] feat: Refactor pending form handling to isolate by bot and pipeline --- src/langbot/pkg/provider/runners/difysvapi.py | 83 +++++++++------- .../provider/runners/test_difysvapi_runner.py | 98 ++++++++++++++----- 2 files changed, 121 insertions(+), 60 deletions(-) diff --git a/src/langbot/pkg/provider/runners/difysvapi.py b/src/langbot/pkg/provider/runners/difysvapi.py index b1845f0aa..7226bf9fe 100644 --- a/src/langbot/pkg/provider/runners/difysvapi.py +++ b/src/langbot/pkg/provider/runners/difysvapi.py @@ -22,11 +22,11 @@ from langbot.libs.dify_service_api.v1 import client, errors import httpx -# Module-level store for paused-workflow form state, keyed by session key -# (launcher_type_value + "_" + launcher_id). Each session holds an -# insertion-ordered dict of form_token -> form_data, allowing multiple -# Dify workflows to be paused simultaneously for the same session. -_PENDING_FORMS: dict[str, 'OrderedDict[str, dict[str, typing.Any]]'] = {} +# Module-level store for paused-workflow form state. The key isolates the bot, +# pipeline, adapter, and launcher; each value holds an insertion-ordered map of +# form_token -> form_data so one conversation can pause multiple workflows. +PendingFormKey = tuple[str, str, str, str, str] +_PENDING_FORMS: dict[PendingFormKey, 'OrderedDict[str, dict[str, typing.Any]]'] = {} _PENDING_FORM_DEFAULT_TTL = 30 * 60 # 30 minutes safety cap _STREAM_FORM_PLACEHOLDER = '\u200b' @@ -43,10 +43,23 @@ def _merge_stream_text(accumulated: str, incoming: typing.Any) -> str: return accumulated + incoming_text -def _session_key_from_query(query: pipeline_query.Query) -> str: +def _dify_user_from_query(query: pipeline_query.Query) -> str: return f'{query.session.launcher_type.value}_{query.session.launcher_id}' +def _session_key_from_query(query: pipeline_query.Query) -> PendingFormKey: + """Build a process-local pending-form key isolated by bot and pipeline.""" + adapter = getattr(query, 'adapter', None) + adapter_type = f'{type(adapter).__module__}.{type(adapter).__qualname__}' + return ( + str(getattr(query, 'bot_uuid', '') or ''), + str(getattr(query, 'pipeline_uuid', '') or ''), + adapter_type, + str(query.session.launcher_type.value), + str(query.session.launcher_id), + ) + + def _prune_pending_forms(now: float | None = None) -> None: if now is None: now = time.time() @@ -59,7 +72,7 @@ def _prune_pending_forms(now: float | None = None) -> None: _PENDING_FORMS.pop(session_key, None) -def _set_pending_form(session_key: str, form_data: dict[str, typing.Any]) -> None: +def _set_pending_form(session_key: PendingFormKey, form_data: dict[str, typing.Any]) -> None: _prune_pending_forms() stored = dict(form_data) expiration_time = stored.get('expiration_time') @@ -75,7 +88,7 @@ def _set_pending_form(session_key: str, form_data: dict[str, typing.Any]) -> Non forms[form_token] = stored -def _get_pending_form_by_token(session_key: str, form_token: str) -> dict[str, typing.Any] | None: +def _get_pending_form_by_token(session_key: PendingFormKey, form_token: str) -> dict[str, typing.Any] | None: _prune_pending_forms() forms = _PENDING_FORMS.get(session_key) if not forms or not form_token: @@ -83,7 +96,7 @@ def _get_pending_form_by_token(session_key: str, form_token: str) -> dict[str, t return forms.get(form_token) -def _get_pending_form_by_w_suffix(session_key: str, w_suffix: str) -> dict[str, typing.Any] | None: +def _get_pending_form_by_w_suffix(session_key: PendingFormKey, w_suffix: str) -> dict[str, typing.Any] | None: """Look up a pending form whose workflow_run_id ends with the given suffix. Used by adapters (e.g. Telegram) whose callback payload is too small to @@ -100,7 +113,7 @@ def _get_pending_form_by_w_suffix(session_key: str, w_suffix: str) -> dict[str, return None -def _get_latest_pending_form(session_key: str) -> dict[str, typing.Any] | None: +def _get_latest_pending_form(session_key: PendingFormKey) -> dict[str, typing.Any] | None: _prune_pending_forms() forms = _PENDING_FORMS.get(session_key) if not forms: @@ -108,7 +121,7 @@ def _get_latest_pending_form(session_key: str) -> dict[str, typing.Any] | None: return forms[next(reversed(forms))] -def _iter_pending_forms(session_key: str) -> typing.Iterator[dict[str, typing.Any]]: +def _iter_pending_forms(session_key: PendingFormKey) -> typing.Iterator[dict[str, typing.Any]]: """Iterate pending forms for a session, newest-first.""" _prune_pending_forms() forms = _PENDING_FORMS.get(session_key) @@ -118,7 +131,7 @@ def _iter_pending_forms(session_key: str) -> typing.Iterator[dict[str, typing.An yield forms[token] -def _clear_pending_form(session_key: str, form_token: str | None = None) -> None: +def _clear_pending_form(session_key: PendingFormKey, form_token: str | None = None) -> None: """Clear one specific pending form (by token) or all forms for the session.""" forms = _PENDING_FORMS.get(session_key) if not forms: @@ -683,7 +696,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): """ plain_text = '' upload_files: list[dict] = [] - user_tag = f'{query.session.launcher_type.value}_{query.session.launcher_id}' + user_tag = _dify_user_from_query(query) async def upload_file_bytes(file_name: str, file_bytes: bytes, content_type: str) -> str: file_name = file_name or 'file' @@ -823,7 +836,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): values = dict(pending_form.get('inputs') or {}) keyed_values = _extract_key_value_inputs(user_text) - user = pending_form.get('user') or _session_key_from_query(query) + user = pending_form.get('user') or _dify_user_from_query(query) current_field_name = str(pending_form.get('current_input_field') or '').strip() file_fields = [field for field in input_defs if _field_type(field) in {'file', 'file-list'}] @@ -933,7 +946,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): form_action, ) return - async for msg in self._submit_workflow_form_blocking(form_action): + async for msg in self._submit_workflow_form_blocking(form_action, session_key): yield msg _clear_pending_form(session_key, form_action.get('form_token') or None) return @@ -965,7 +978,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): async for chunk in self.dify_client.chat_messages( inputs=inputs, query=plain_text, - user=f'{query.session.launcher_type.value}_{query.session.launcher_id}', + user=_dify_user_from_query(query), conversation_id=cov_id, files=files, timeout=120, @@ -982,7 +995,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): for reason in reasons: if reason.get('TYPE') != 'human_input_required': continue - user = f'{query.session.launcher_type.value}_{query.session.launcher_id}' + user = _dify_user_from_query(query) form_snapshot, raw_form_content, input_defs, _ = _extract_form_snapshot( workflow_run_id, reason, @@ -1063,7 +1076,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): async for chunk in self.dify_client.chat_messages( inputs=inputs, query=plain_text, - user=f'{query.session.launcher_type.value}_{query.session.launcher_id}', + user=_dify_user_from_query(query), response_mode='streaming', conversation_id=cov_id, files=files, @@ -1131,7 +1144,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): query.session.using_conversation.uuid = chunk['conversation_id'] async def _submit_workflow_form_blocking( - self, form_action: dict + self, form_action: dict, session_key: PendingFormKey ) -> typing.AsyncGenerator[provider_message.Message, None]: """Submit human input to resume a paused Dify workflow (non-streaming).""" @@ -1206,7 +1219,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): actions = form_snapshot.get('actions', []) paused_node_title = form_snapshot.get('node_title', '') - _set_pending_form(user, form_snapshot) + _set_pending_form(session_key, form_snapshot) display_text = _format_human_input_text( paused_node_title, @@ -1226,7 +1239,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): 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: + def _resolve_pending_form(self, session_key: PendingFormKey, form_action: dict) -> dict | None: """Locate the pending form this action targets. Tries identifiers in order of specificity: form_token, full @@ -1253,7 +1266,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): return _get_latest_pending_form(session_key) - def _merge_pending_form_action(self, session_key: str, form_action: dict | None) -> dict | None: + def _merge_pending_form_action(self, session_key: PendingFormKey, form_action: dict | None) -> dict | None: """Backfill resume fields from the matching pending form.""" if not form_action: return None @@ -1310,7 +1323,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): async def _match_pending_form_action( self, query: pipeline_query.Query, - session_key: str, + session_key: PendingFormKey, user_text: str, ) -> dict | None: """Match plain text replies against pending Dify form actions. @@ -1428,7 +1441,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): form_action, ) return - async for msg in self._submit_workflow_form_blocking(form_action): + async for msg in self._submit_workflow_form_blocking(form_action, session_key): yield msg _clear_pending_form(session_key, form_action.get('form_token') or None) return @@ -1463,7 +1476,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): async for chunk in self.dify_client.workflow_run( inputs=inputs, - user=f'{query.session.launcher_type.value}_{query.session.launcher_id}', + user=_dify_user_from_query(query), files=files, timeout=120, ): @@ -1476,7 +1489,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): workflow_run_id = chunk['data'].get('workflow_run_id', '') for reason in reasons: if reason.get('TYPE') == 'human_input_required': - user = f'{query.session.launcher_type.value}_{query.session.launcher_id}' + user = _dify_user_from_query(query) form_snapshot, raw_form_content, input_defs, _ = _extract_form_snapshot( workflow_run_id, reason, @@ -1561,7 +1574,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): form_action, ) return - async for msg in self._submit_workflow_form(form_action): + async for msg in self._submit_workflow_form(form_action, session_key): yield msg _clear_pending_form(session_key, form_action.get('form_token') or None) return @@ -1607,7 +1620,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): async for chunk in self.dify_client.chat_messages( inputs=inputs, query=plain_text, - user=f'{query.session.launcher_type.value}_{query.session.launcher_id}', + user=_dify_user_from_query(query), conversation_id=cov_id, files=files, timeout=120, @@ -1639,7 +1652,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): for reason in reasons: if reason.get('TYPE') != 'human_input_required': continue - user = f'{query.session.launcher_type.value}_{query.session.launcher_id}' + user = _dify_user_from_query(query) form_snapshot, raw_form_content, input_defs, display_form_content = _extract_form_snapshot( workflow_run_id, reason, @@ -1759,7 +1772,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): async for chunk in self.dify_client.chat_messages( inputs=inputs, query=plain_text, - user=f'{query.session.launcher_type.value}_{query.session.launcher_id}', + user=_dify_user_from_query(query), response_mode='streaming', conversation_id=cov_id, files=files, @@ -1846,7 +1859,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): query.session.using_conversation.uuid = chunk['conversation_id'] async def _submit_workflow_form( - self, form_action: dict + self, form_action: dict, session_key: PendingFormKey ) -> typing.AsyncGenerator[provider_message.MessageChunk, None]: """Submit human input to resume a paused Dify workflow.""" @@ -1901,7 +1914,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): # card still shows which step the user acted on. paused_node_title = form_snapshot.get('node_title', '') - _set_pending_form(user, form_snapshot) + _set_pending_form(session_key, form_snapshot) repause_form_data = _initial_interactive_form_data(form_snapshot) or { 'form_content': display_form_content, @@ -2014,7 +2027,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): ) return # Resume paused workflow via submit endpoint - async for msg in self._submit_workflow_form(form_action): + async for msg in self._submit_workflow_form(form_action, session_key): yield msg _clear_pending_form(session_key, form_action.get('form_token') or None) return @@ -2060,7 +2073,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): remove_think = self.pipeline_config['output'].get('misc', {}).get('remove-think') async for chunk in self.dify_client.workflow_run( inputs=inputs, - user=f'{query.session.launcher_type.value}_{query.session.launcher_id}', + user=_dify_user_from_query(query), files=files, timeout=120, ): @@ -2075,7 +2088,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): workflow_run_id = chunk['data'].get('workflow_run_id', workflow_run_id) for reason in reasons: if reason.get('TYPE') == 'human_input_required': - user = f'{query.session.launcher_type.value}_{query.session.launcher_id}' + user = _dify_user_from_query(query) form_snapshot, raw_form_content, input_defs, display_form_content = _extract_form_snapshot( workflow_run_id, reason, diff --git a/tests/unit_tests/provider/runners/test_difysvapi_runner.py b/tests/unit_tests/provider/runners/test_difysvapi_runner.py index 911e61fe9..2efff4cd3 100644 --- a/tests/unit_tests/provider/runners/test_difysvapi_runner.py +++ b/tests/unit_tests/provider/runners/test_difysvapi_runner.py @@ -196,6 +196,49 @@ class TestDifyHumanInputForms: runner.dify_client.upload_file = AsyncMock(return_value={'id': 'upload-1'}) return runner + def test_pending_forms_are_isolated_by_bot_and_pipeline(self): + from langbot.pkg.provider.runners import difysvapi + + query_a = MagicMock() + query_a.bot_uuid = 'bot-a' + query_a.pipeline_uuid = 'pipeline-a' + query_a.session.launcher_type.value = 'person' + query_a.session.launcher_id = 'shared-user' + + query_b = MagicMock() + query_b.bot_uuid = 'bot-b' + query_b.pipeline_uuid = 'pipeline-a' + query_b.session.launcher_type.value = 'person' + query_b.session.launcher_id = 'shared-user' + + query_c = MagicMock() + query_c.bot_uuid = 'bot-a' + query_c.pipeline_uuid = 'pipeline-b' + query_c.session.launcher_type.value = 'person' + query_c.session.launcher_id = 'shared-user' + + key_a = difysvapi._session_key_from_query(query_a) + key_b = difysvapi._session_key_from_query(query_b) + key_c = difysvapi._session_key_from_query(query_c) + difysvapi._PENDING_FORMS.clear() + difysvapi._set_pending_form(key_a, {'form_token': 'token-a', 'workflow_run_id': 'run-a'}) + difysvapi._set_pending_form(key_b, {'form_token': 'token-b', 'workflow_run_id': 'run-b'}) + difysvapi._set_pending_form(key_c, {'form_token': 'token-c', 'workflow_run_id': 'run-c'}) + + assert key_a != key_b + assert key_a != key_c + assert difysvapi._get_pending_form_by_token(key_a, 'token-a') is not None + assert difysvapi._get_pending_form_by_token(key_a, 'token-b') is None + assert difysvapi._get_pending_form_by_token(key_a, 'token-c') is None + assert difysvapi._get_pending_form_by_token(key_b, 'token-b') is not None + assert difysvapi._get_pending_form_by_token(key_c, 'token-c') is not None + 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._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_format_human_input_text_includes_field_help(self): from langbot.pkg.provider.runners.difysvapi import _format_human_input_text @@ -737,7 +780,8 @@ class TestDifyHumanInputForms: 'user': 'person_user-1', 'action_id': 'yes', 'inputs': {}, - } + }, + ('bot-1', 'pipeline-1', 'adapter.Type', 'person', 'user-1'), ) ] @@ -748,18 +792,7 @@ class TestDifyHumanInputForms: 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 @@ -772,12 +805,25 @@ class TestDifyHumanInputForms: query = MagicMock() query.session.launcher_type.value = 'person' query.session.launcher_id = 'user-1' + query.bot_uuid = 'bot-1' + query.pipeline_uuid = 'pipeline-1' + session_key = difysvapi._session_key_from_query(query) + difysvapi._set_pending_form( + session_key, + { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'actions': [{'id': 'yes', 'title': 'Yes'}], + 'inputs': {}, + 'user': 'person_user-1', + }, + ) query.variables = { '_dify_form_action': { 'form_token': 'token-1', 'workflow_run_id': 'run-1', 'action_id': 'yes', - 'user': session_key, + 'user': 'person_user-1', 'inputs': {}, } } @@ -793,18 +839,7 @@ class TestDifyHumanInputForms: 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 @@ -815,12 +850,25 @@ class TestDifyHumanInputForms: query = MagicMock() query.session.launcher_type.value = 'person' query.session.launcher_id = 'user-1' + query.bot_uuid = 'bot-1' + query.pipeline_uuid = 'pipeline-1' + session_key = difysvapi._session_key_from_query(query) + difysvapi._set_pending_form( + session_key, + { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'actions': [{'id': 'yes', 'title': 'Yes'}], + 'inputs': {}, + 'user': 'person_user-1', + }, + ) query.variables = { '_dify_form_action': { 'form_token': 'token-1', 'workflow_run_id': 'run-1', 'action_id': 'yes', - 'user': session_key, + 'user': 'person_user-1', 'inputs': {}, } }