feat: Implement workflow form handling for paused workflows

- Added module-level storage for pending forms to manage state across sessions.
- Introduced functions to set, get, and clear pending forms with expiration handling.
- Enhanced DifyServiceAPIRunner to support resuming paused workflows via form actions.
- Implemented logic to yield human input requests and display appropriate messages.
- Updated workflow submission methods to handle paused states and resume actions.
- Ensured proper merging of pending form actions with user inputs for seamless interaction.

src/langbot/pkg/provider/runners/difysvapi.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import typing
 import json
+import time
 import uuid
 import base64
 import mimetypes
@@ -16,6 +17,46 @@ 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). Keeps state out of the SDK
+# Conversation type, which may not accept arbitrary attribute assignment.
+_PENDING_FORMS: dict[str, dict[str, typing.Any]] = {}
+_PENDING_FORM_DEFAULT_TTL = 30 * 60  # 30 minutes safety cap
+
+
+def _session_key_from_query(query: pipeline_query.Query) -> str:
+    return f'{query.session.launcher_type.value}_{query.session.launcher_id}'
+
+
+def _prune_pending_forms(now: float | None = None) -> None:
+    if now is None:
+        now = time.time()
+    expired = [key for key, data in _PENDING_FORMS.items() if data.get('_expires_at', 0) <= now]
+    for key in expired:
+        _PENDING_FORMS.pop(key, None)
+
+
+def _set_pending_form(session_key: str, form_data: dict[str, typing.Any]) -> None:
+    _prune_pending_forms()
+    stored = dict(form_data)
+    expiration_time = stored.get('expiration_time')
+    try:
+        expiration_ts = float(expiration_time) if expiration_time is not None else 0.0
+    except (TypeError, ValueError):
+        expiration_ts = 0.0
+    stored['_expires_at'] = expiration_ts or (time.time() + _PENDING_FORM_DEFAULT_TTL)
+    _PENDING_FORMS[session_key] = stored
+
+
+def _get_pending_form(session_key: str) -> dict[str, typing.Any] | None:
+    _prune_pending_forms()
+    return _PENDING_FORMS.get(session_key)
+
+
+def _clear_pending_form(session_key: str) -> None:
+    _PENDING_FORMS.pop(session_key, None)
+
+
 @runner.runner_class('dify-service-api')
 class DifyServiceAPIRunner(runner.RequestRunner):
     """Dify Service API 对话请求器"""
@@ -335,11 +376,107 @@ class DifyServiceAPIRunner(runner.RequestRunner):
 
         query.session.using_conversation.uuid = chunk['conversation_id']
 
+    async def _submit_workflow_form_blocking(
+        self, form_action: dict
+    ) -> typing.AsyncGenerator[provider_message.Message, None]:
+        """Submit human input to resume a paused Dify workflow (non-streaming)."""
+
+        form_token = form_action['form_token']
+        workflow_run_id = form_action['workflow_run_id']
+        user = form_action['user']
+        action_id = form_action.get('action_id', '')
+        inputs = form_action.get('inputs', {})
+
+        async for chunk in self.dify_client.workflow_submit(
+            form_token=form_token,
+            workflow_run_id=workflow_run_id,
+            inputs=inputs,
+            user=user,
+            action=action_id,
+            timeout=120,
+        ):
+            self.ap.logger.debug('dify-workflow-submit-chunk: ' + str(chunk))
+
+            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'])
+                yield provider_message.Message(
+                    role='assistant',
+                    content=content,
+                )
+
+    def _merge_pending_form_action(self, form_action: dict | None, pending_form: dict | None) -> dict | None:
+        """Backfill resume fields from the pending form stored on the conversation."""
+        if not form_action:
+            return None
+
+        merged_action = dict(form_action)
+        if pending_form:
+            merged_action.setdefault('form_token', pending_form.get('form_token', ''))
+            merged_action.setdefault('workflow_run_id', pending_form.get('workflow_run_id', ''))
+            merged_action.setdefault('inputs', pending_form.get('inputs', {}))
+            merged_action.setdefault('user', pending_form.get('user', ''))
+
+            # Resolve clicked action's display title from the stored actions list
+            if 'action_title' not in merged_action:
+                clicked_id = merged_action.get('action_id', '')
+                for action in pending_form.get('actions', []):
+                    if str(action.get('id', '')) == str(clicked_id):
+                        merged_action['action_title'] = action.get('title', clicked_id)
+                        break
+
+        return merged_action
+
+    def _match_pending_form_action(self, user_text: str, pending_form: dict | None) -> dict | None:
+        """Match plain text replies against pending Dify form actions."""
+        if not pending_form:
+            return None
+
+        normalized_text = user_text.strip().lower()
+        if not normalized_text:
+            return None
+
+        for action in pending_form.get('actions', []):
+            titles = {
+                str(action.get('title', '')).strip().lower(),
+                str(action.get('id', '')).strip().lower(),
+            }
+            if normalized_text in titles:
+                return {
+                    'form_token': pending_form.get('form_token', ''),
+                    'workflow_run_id': pending_form.get('workflow_run_id', ''),
+                    'action_id': action.get('id', ''),
+                    'action_title': action.get('title', action.get('id', '')),
+                    'inputs': pending_form.get('inputs', {}),
+                    'user': pending_form.get('user', ''),
+                }
+
+        return None
+
     async def _workflow_messages(
         self, query: pipeline_query.Query
     ) -> typing.AsyncGenerator[provider_message.Message, None]:
         """调用工作流"""
 
+        # Check if this is a form action resume (button click or text match)
+        form_action = query.variables.get('_dify_form_action')
+        session_key = _session_key_from_query(query)
+        pending_form = _get_pending_form(session_key)
+
+        if form_action:
+            form_action = self._merge_pending_form_action(form_action, pending_form)
+            _clear_pending_form(session_key)
+        elif pending_form:
+            form_action = self._match_pending_form_action(str(query.message_chain), pending_form)
+            if form_action:
+                _clear_pending_form(session_key)
+
+        if form_action:
+            async for msg in self._submit_workflow_form_blocking(form_action):
+                yield msg
+            return
+
         if not query.session.using_conversation.uuid:
             query.session.using_conversation.uuid = str(uuid.uuid4())
 
@@ -366,6 +503,7 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         }
 
         inputs.update(query.variables)
+        human_input_yielded = False
 
         async for chunk in self.dify_client.workflow_run(
             inputs=inputs,
@@ -377,6 +515,46 @@ class DifyServiceAPIRunner(runner.RequestRunner):
             if chunk['event'] in ignored_events:
                 continue
 
+            if chunk['event'] == 'workflow_paused':
+                reasons = chunk['data'].get('reasons', [])
+                workflow_run_id = chunk['data'].get('workflow_run_id', '')
+                for reason in reasons:
+                    if reason.get('TYPE') == 'human_input_required':
+                        form_content = reason.get('form_content', '')
+                        actions = reason.get('actions', [])
+                        node_title = reason.get('node_title', '')
+
+                        _set_pending_form(
+                            _session_key_from_query(query),
+                            {
+                                'workflow_run_id': workflow_run_id,
+                                'form_id': reason.get('form_id'),
+                                'form_token': reason.get('form_token'),
+                                'node_id': reason.get('node_id'),
+                                'node_title': node_title,
+                                'form_content': form_content,
+                                'inputs': reason.get('inputs', {}),
+                                'actions': actions,
+                                'expiration_time': reason.get('expiration_time'),
+                                'user': f'{query.session.launcher_type.value}_{query.session.launcher_id}',
+                            },
+                        )
+
+                        query.variables['_dify_form_render'] = {
+                            'form_content': form_content,
+                            'actions': actions,
+                            'node_title': node_title,
+                        }
+
+                        action_lines = '\n'.join(f'- [{a.get("title", a.get("id", ""))}]' for a in actions)
+                        display_text = f'[Human Input Required] {node_title}\n{form_content}\n{action_lines}'
+
+                        human_input_yielded = True
+                        yield provider_message.Message(
+                            role='assistant',
+                            content=display_text,
+                        )
+
             if chunk['event'] == 'node_started':
                 if chunk['data']['node_type'] == 'start' or chunk['data']['node_type'] == 'end':
                     continue
@@ -399,6 +577,8 @@ class DifyServiceAPIRunner(runner.RequestRunner):
                 yield msg
 
             elif chunk['event'] == 'workflow_finished':
+                if human_input_yielded:
+                    break
                 if chunk['data']['error']:
                     raise errors.DifyAPIError(chunk['data']['error'])
                 content, _ = self._process_thinking_content(chunk['data']['outputs']['summary'])
@@ -636,11 +816,143 @@ class DifyServiceAPIRunner(runner.RequestRunner):
 
         query.session.using_conversation.uuid = chunk['conversation_id']
 
+    async def _submit_workflow_form(
+        self, form_action: dict
+    ) -> typing.AsyncGenerator[provider_message.MessageChunk, None]:
+        """Submit human input to resume a paused Dify workflow."""
+
+        form_token = form_action['form_token']
+        workflow_run_id = form_action['workflow_run_id']
+        user = form_action['user']
+        action_id = form_action.get('action_id', '')
+        action_title = form_action.get('action_title', '') or action_id
+        inputs = form_action.get('inputs', {})
+
+        messsage_idx = 0
+        is_final = False
+        think_start = False
+        think_end = False
+        workflow_contents = ''
+        repause_form_data: dict | None = None
+
+        remove_think = self.pipeline_config['output'].get('misc', {}).get('remove-think')
+        async for chunk in self.dify_client.workflow_submit(
+            form_token=form_token,
+            workflow_run_id=workflow_run_id,
+            inputs=inputs,
+            user=user,
+            action=action_id,
+            timeout=120,
+        ):
+            self.ap.logger.debug('dify-workflow-submit-chunk: ' + str(chunk))
+
+            yield_this_iteration = False
+
+            if chunk['event'] == 'workflow_finished':
+                is_final = True
+                yield_this_iteration = True
+                if chunk['data'].get('error'):
+                    raise errors.DifyAPIError(chunk['data']['error'])
+
+            if chunk['event'] == 'workflow_paused':
+                reasons = chunk['data'].get('reasons', [])
+                new_run_id = chunk['data'].get('workflow_run_id', workflow_run_id)
+                for reason in reasons:
+                    if reason.get('TYPE') != 'human_input_required':
+                        continue
+                    form_content = reason.get('form_content', '')
+                    actions = reason.get('actions', [])
+                    node_title = reason.get('node_title', '')
+                    raw_inputs = reason.get('inputs', {})
+
+                    _set_pending_form(
+                        user,
+                        {
+                            'workflow_run_id': new_run_id,
+                            'form_id': reason.get('form_id'),
+                            'form_token': reason.get('form_token'),
+                            'node_id': reason.get('node_id'),
+                            'node_title': node_title,
+                            'form_content': form_content,
+                            'inputs': raw_inputs if isinstance(raw_inputs, dict) else {},
+                            'actions': actions,
+                            'expiration_time': reason.get('expiration_time'),
+                            'user': user,
+                        },
+                    )
+
+                    repause_form_data = {
+                        'form_content': form_content,
+                        'actions': actions,
+                        'node_title': node_title,
+                        'workflow_run_id': new_run_id,
+                        'form_token': reason.get('form_token', ''),
+                    }
+                    is_final = True
+                    yield_this_iteration = True
+                    break
+
+            if chunk['event'] == 'text_chunk':
+                messsage_idx += 1
+                if remove_think:
+                    if '<think>' in chunk['data']['text'] and not think_start:
+                        think_start = True
+                        continue
+                    if '</think>' in chunk['data']['text'] and not think_end:
+                        import re
+
+                        content = re.sub(r'^\n</think>', '', chunk['data']['text'])
+                        workflow_contents += content
+                        think_end = True
+                    elif think_end:
+                        workflow_contents += chunk['data']['text']
+                    if think_start:
+                        continue
+                else:
+                    workflow_contents += chunk['data']['text']
+                if messsage_idx % 8 == 0:
+                    yield_this_iteration = True
+
+            if yield_this_iteration:
+                msg = provider_message.MessageChunk(
+                    role='assistant',
+                    content=workflow_contents,
+                    is_final=is_final,
+                )
+                msg._resume_from_form = True
+                if action_title:
+                    msg._resume_action_title = action_title
+                if is_final and repause_form_data:
+                    msg._form_data = repause_form_data
+                    msg._open_new_card = True
+                yield msg
+                if is_final:
+                    return
+
     async def _workflow_messages_chunk(
         self, query: pipeline_query.Query
     ) -> typing.AsyncGenerator[provider_message.MessageChunk, None]:
         """调用工作流"""
 
+        # Check if this is a form action resume (button click or text match)
+        form_action = query.variables.get('_dify_form_action')
+        session_key = _session_key_from_query(query)
+        pending_form = _get_pending_form(session_key)
+
+        if form_action:
+            form_action = self._merge_pending_form_action(form_action, pending_form)
+            _clear_pending_form(session_key)
+        elif pending_form:
+            form_action = self._match_pending_form_action(str(query.message_chain), pending_form)
+            if form_action:
+                _clear_pending_form(session_key)
+
+        if form_action:
+            # Resume paused workflow via submit endpoint
+            async for msg in self._submit_workflow_form(form_action):
+                yield msg
+            return
+
         if not query.session.using_conversation.uuid:
             query.session.using_conversation.uuid = str(uuid.uuid4())
 
@@ -672,6 +984,13 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         think_start = False
         think_end = False
         workflow_contents = ''
+        workflow_run_id = ''
+        human_input_yielded = False
+
+        # Saved form data to attach to the final MessageChunk so the adapter
+        # can detect it when is_final=True and render buttons.
+        pending_form_data = None
+        display_text = ''
 
         remove_think = self.pipeline_config['output'].get('misc', '').get('remove-think')
         async for chunk in self.dify_client.workflow_run(
@@ -682,7 +1001,62 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         ):
             self.ap.logger.debug('dify-workflow-chunk: ' + str(chunk))
             if chunk['event'] in ignored_events:
+                if chunk['event'] == 'workflow_started':
+                    workflow_run_id = chunk['data'].get('workflow_run_id', '')
                 continue
+
+            if chunk['event'] == 'workflow_paused':
+                reasons = chunk['data'].get('reasons', [])
+                workflow_run_id = chunk['data'].get('workflow_run_id', workflow_run_id)
+                for reason in reasons:
+                    if reason.get('TYPE') == 'human_input_required':
+                        form_content = reason.get('form_content', '')
+                        actions = reason.get('actions', [])
+                        node_title = reason.get('node_title', '')
+
+                        # Persist form state in module-level store keyed by session
+                        raw_inputs = reason.get('inputs', {})
+                        _set_pending_form(
+                            _session_key_from_query(query),
+                            {
+                                'workflow_run_id': workflow_run_id,
+                                'form_id': reason.get('form_id'),
+                                'form_token': reason.get('form_token'),
+                                'node_id': reason.get('node_id'),
+                                'node_title': node_title,
+                                'form_content': form_content,
+                                'inputs': raw_inputs if isinstance(raw_inputs, dict) else {},
+                                'actions': actions,
+                                'expiration_time': reason.get('expiration_time'),
+                                'user': f'{query.session.launcher_type.value}_{query.session.launcher_id}',
+                            },
+                        )
+
+                        # Pass form render metadata to downstream stages
+                        query.variables['_dify_form_render'] = {
+                            'form_content': form_content,
+                            'actions': actions,
+                            'node_title': node_title,
+                        }
+
+                        action_lines = '\n'.join(f'- [{a.get("title", a.get("id", ""))}]' for a in actions)
+                        display_text = f'[Human Input Required] {node_title}\n{form_content}\n{action_lines}'
+                        workflow_contents += display_text + '\n'
+
+                        # Save form data to attach to the final chunk later.
+                        # We do NOT yield here — the form content will be sent
+                        # as the final MessageChunk (with is_final=True and
+                        # _form_data) so the adapter can update the card and
+                        # add buttons in one pass.
+                        pending_form_data = {
+                            'form_content': form_content,
+                            'actions': actions,
+                            'node_title': node_title,
+                            'workflow_run_id': workflow_run_id,
+                            'form_token': reason.get('form_token', ''),
+                        }
+                        human_input_yielded = True
+
             if chunk['event'] == 'workflow_finished':
                 is_final = True
                 if chunk['data']['error']:
@@ -730,11 +1104,29 @@ class DifyServiceAPIRunner(runner.RequestRunner):
                 yield msg
 
             if messsage_idx % 8 == 0 or is_final:
-                yield provider_message.MessageChunk(
+                final_content = workflow_contents if workflow_contents.strip() else ''
+                msg = provider_message.MessageChunk(
                     role='assistant',
-                    content=workflow_contents,
+                    content=final_content,
                     is_final=is_final,
                 )
+                # Attach form data to the final chunk for the adapter
+                if is_final and pending_form_data:
+                    msg._form_data = pending_form_data
+                    pending_form_data = None
+                yield msg
+
+        # If the stream ended after workflow_paused without a
+        # workflow_finished event, yield a final chunk so the adapter
+        # can update the card and add buttons.
+        if human_input_yielded and not is_final:
+            msg = provider_message.MessageChunk(
+                role='assistant',
+                content=workflow_contents or display_text,
+                is_final=True,
+            )
+            msg._form_data = pending_form_data
+            yield msg
 
     async def run(self, query: pipeline_query.Query) -> typing.AsyncGenerator[provider_message.Message, None]:
         """运行请求"""