diff --git a/plugins/Runner/claude-code-agent/pkg/native_cli.py b/plugins/Runner/claude-code-agent/pkg/native_cli.py index 112454a1d..36ab73de1 100644 --- a/plugins/Runner/claude-code-agent/pkg/native_cli.py +++ b/plugins/Runner/claude-code-agent/pkg/native_cli.py @@ -238,7 +238,9 @@ _LANGBOT_MCP_SYSTEM_PROMPT = ( "any mcp__langbot_agent__* tool, then call that tool directly. In coordinator mode, " "the main session may expose only delegation tools; delegate the MCP request to one " "worker and require that worker to perform the same ToolSearch-then-call flow. Never " - "claim a LangBot tool result unless the direct or delegated tool call completed." + "claim a LangBot tool result unless the direct or delegated tool call completed. " + "For user questions, call mcp__langbot_agent__ask_user_question so the answer returns " + "inside the current CLI process; do not call the built-in AskUserQuestion tool." ) @@ -450,7 +452,25 @@ class ClaudeCodeExternalTools(AgentRunExternalTools): args_model=AskUserQuestionArgs, ) async def ask_user_question(self, args: AskUserQuestionArgs) -> dict[str, typing.Any]: - return {"status": "paused", "question_count": len(args.questions)} + request, continuation = _interaction_from_claude_tool( + uuid.uuid4().hex, args.model_dump(mode='json', by_alias=True) + ) + answers: list[dict[str, typing.Any]] = [] + for field in request.fields: + step = request.model_copy(update={ + 'interaction_id': f'{request.interaction_id}-{field.id}'[:255], + 'fields': [field], + 'actions': [], + }) + result = await self.ctx.request_interaction(step, timeout=3600, on_unsupported='degrade') + if result.status != 'submitted' or result.submission is None: + raise NativeCliError(f'User interaction {result.status}', code='claude_code.interaction_expired') + answers.append({ + 'id': field.id, + 'question': next((item['question'] for item in continuation['questions'] if item['id'] == field.id), field.label), + 'answer': result.submission.values.get(field.id), + }) + return {'status': 'submitted', 'answers': answers} class NativeClaudeCodeRunner(Runner): diff --git a/plugins/Runner/codex-agent/pkg/native_cli.py b/plugins/Runner/codex-agent/pkg/native_cli.py index e8111b1b2..571ca09c2 100644 --- a/plugins/Runner/codex-agent/pkg/native_cli.py +++ b/plugins/Runner/codex-agent/pkg/native_cli.py @@ -1112,6 +1112,7 @@ async def _run_cli_process( _run_cli_process_events( command, args, + ctx=ctx, cwd=cwd, env=env, timeout=timeout, @@ -1165,6 +1166,8 @@ class _CodexAppServerClient: streaming: bool, approval_grant: dict[str, str] | None, approval_policy: str | None, + ctx: RunnerContext | None = None, + operation_timeout: float = 120, ) -> None: self.process = process self.streaming = streaming @@ -1183,8 +1186,39 @@ class _CodexAppServerClient: self.interaction_continuation: dict[str, typing.Any] | None = None self.approval_grant = dict(approval_grant) if approval_grant else None self.approval_policy = approval_policy + self.ctx = ctx + self.operation_timeout = operation_timeout + self.timeout_scope: asyncio.Timeout | None = None self.items: dict[str, dict[str, typing.Any]] = {} + async def _wait_for_user(self, request: InteractionRequest) -> InteractionSubmission: + if self.ctx is None: + raise NativeCliError('No active run for interaction', code='codex.interaction_invalid') + loop = asyncio.get_running_loop() + if self.timeout_scope is not None: + self.timeout_scope.reschedule(loop.time() + 3600 + self.operation_timeout) + try: + if request.actions and not request.fields: + result = await self.ctx.request_interaction(request, timeout=3600, on_unsupported='degrade') + if result.status != 'submitted' or result.submission is None: + raise NativeCliError(f'User interaction {result.status}', code='codex.interaction_expired') + return result.submission + values: dict[str, typing.Any] = {} + for field in request.fields: + step = request.model_copy(update={ + 'interaction_id': f'{request.interaction_id}-{field.id}'[:255], + 'fields': [field], + 'actions': [], + }) + result = await self.ctx.request_interaction(step, timeout=3600, on_unsupported='degrade') + if result.status != 'submitted' or result.submission is None: + raise NativeCliError(f'User interaction {result.status}', code='codex.interaction_expired') + values[field.id] = result.submission.values.get(field.id) + return InteractionSubmission(interaction_id=request.interaction_id, values=values) + finally: + if self.timeout_scope is not None: + self.timeout_scope.reschedule(loop.time() + self.operation_timeout) + async def _write_json(self, payload: dict[str, typing.Any]) -> None: assert self.process.stdin is not None data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + b"\n" @@ -1364,6 +1398,12 @@ class _CodexAppServerClient: await self.respond(request_id, {"decision": "accept"}) return request, continuation = _interaction_from_codex_approval(method, params, item) + if self.ctx is not None: + submission = await self._wait_for_user(request) + await self.respond(request_id, { + 'decision': 'accept' if submission.action_id == APPROVE_ONCE_ACTION else 'cancel' + }) + return self.interaction_request = request self.interaction_continuation = continuation await self.respond(request_id, {"decision": "cancel"}) @@ -1381,6 +1421,21 @@ class _CodexAppServerClient: except NativeCliError as exc: await self.respond_error(request_id, exc.message) return + if self.ctx is not None: + submission = await self._wait_for_user(request) + if method == 'item/tool/call': + await self.respond(request_id, { + 'success': True, + 'contentItems': [{'type': 'inputText', 'text': json.dumps(submission.values, ensure_ascii=False)}], + }) + else: + await self.respond(request_id, { + 'answers': { + field_id: {'answers': value if isinstance(value, list) else [str(value)]} + for field_id, value in submission.values.items() + } + }) + return self.interaction_request = request self.interaction_continuation = continuation if method == "item/tool/call": @@ -1560,6 +1615,7 @@ async def _run_cli_process_events( command: str, args: list[str], *, + ctx: RunnerContext | None = None, cwd: str | None, env: dict[str, str], timeout: float, @@ -1598,6 +1654,8 @@ async def _run_cli_process_events( streaming=streaming, approval_grant=approval_grant, approval_policy=approval_policy, + ctx=ctx, + operation_timeout=timeout, ) reader_task = asyncio.create_task(client.read_stdout()) stderr_task = asyncio.create_task(process.stderr.read()) @@ -1606,7 +1664,8 @@ async def _run_cli_process_events( if initial_stdin: process.stdin.write(initial_stdin) await process.stdin.drain() - async with asyncio.timeout(timeout): + async with asyncio.timeout(timeout) as timeout_scope: + client.timeout_scope = timeout_scope await client.initialize() thread_id = await client.start_or_resume_thread( resume_session_id, diff --git a/plugins/Runner/dify-agent/README.md b/plugins/Runner/dify-agent/README.md index 486d906f2..6988a7d48 100644 --- a/plugins/Runner/dify-agent/README.md +++ b/plugins/Runner/dify-agent/README.md @@ -56,8 +56,8 @@ events, which are used only when no explicit answer was produced. When Dify emits `workflow_paused` with a `human_input_required` reason, or a direct `human_input_required` event, the runner asks LangBot to deliver a -structured interaction. A validated `interaction.submitted` event resumes Dify -through `/form/human_input/{form_token}`. The submission response may itself +structured interaction and waits for the validated submission in the same run. +It then resumes Dify through `/form/human_input/{form_token}`. The response may itself stream events; otherwise the runner reads `/workflow/{workflow_run_id}/events`. Workflows that pause again produce a new interaction and can be resumed again. Forms are emitted as atomic steps: the runner collects one field at a time and @@ -65,10 +65,9 @@ then emits the action buttons. This matches LangBot delivery adapters that support a single select or an action-only confirmation. Provider continuation data is not included in the interaction payload. The -runner stores `form_token`, `workflow_run_id`, the Dify user tag, and field/action -mappings in authorized plugin storage under a random LangBot `interaction_id`. -The continuation is deleted after a successful terminal event or after it is -replaced by a later pause. +runner keeps `form_token`, `workflow_run_id`, the Dify user tag, and field/action +mappings in the active run. Previously issued interactions can still resume +through their stored continuation records during migration. The selected Runner binding must allow interactions, and the delivery adapter must support the requested field/action shape. Adapters with a smaller diff --git a/plugins/Runner/dify-agent/components/runner/default.py b/plugins/Runner/dify-agent/components/runner/default.py index 1fd8276e9..e0a581a1c 100644 --- a/plugins/Runner/dify-agent/components/runner/default.py +++ b/plugins/Runner/dify-agent/components/runner/default.py @@ -907,6 +907,45 @@ class DefaultRunner(Runner): await self._store_interaction_continuation(ctx, continuation) return RunnerResult.interaction_requested(ctx.run_id, request) + async def _wait_and_resume_workflow( + self, ctx: RunnerContext, client: AsyncDifyClient, + event: dict[str, typing.Any], user: str, remove_think: bool, + ) -> typing.AsyncGenerator[RunnerResult, None]: + reason = _human_input_reason(event) + if reason is None: + raise DifyAPIError('Dify paused without a supported human-input reason', code='dify.response_invalid') + _, continuation = self._build_interaction_request(reason, _workflow_run_id(event), user) + while continuation['phase'] == 'field': + request = self._request_from_continuation(continuation) + result = await ctx.request_interaction(request, timeout=3600, on_unsupported='degrade') + if result.status != 'submitted' or result.submission is None: + raise DifyAPIError(f'Dify human input {result.status}', code='dify.interaction_expired') + field_map = continuation['field_map'] + inputs = continuation['default_inputs'] + for field_id, value in result.submission.values.items(): + if field_id in field_map: + inputs[field_map[field_id]] = value + current_field = InteractionField.model_validate(continuation['current_fields'][0]) + provider_name = field_map.get(current_field.id) + if current_field.required and provider_name and inputs.get(provider_name) in (None, '', []): + raise DifyAPIError(f'Dify human-input field is required: {current_field.label}', code='dify.interaction_invalid') + remaining = continuation['remaining_fields'] + continuation['interaction_id'] = f'dify-{uuid.uuid4().hex}' + if remaining: + continuation['current_fields'] = [remaining.pop(0)] + else: + continuation['phase'] = 'action' + continuation['current_fields'] = [] + request = self._request_from_continuation(continuation) + result = await ctx.request_interaction(request, timeout=3600, on_unsupported='degrade') + if result.status != 'submitted' or result.submission is None: + raise DifyAPIError(f'Dify human input {result.status}', code='dify.interaction_expired') + async for item in self._resume_workflow_once( + ctx, client, result.submission, remove_think, + continuation=continuation, blocking=True, + ): + yield item + async def run(self, ctx: RunnerContext) -> typing.AsyncGenerator[RunnerResult, None]: """Run the Dify agent. @@ -1079,13 +1118,8 @@ class DefaultRunner(Runner): final_conversation_id, scope="conversation", ) - interaction_result = await self._interaction_result_for_pause(ctx, event, user) - if interaction_result is None: - raise DifyAPIError( - "Dify paused the workflow without a supported reason", - code="dify.response_invalid", - ) - yield interaction_result + async for result in self._wait_and_resume_workflow(ctx, client, event, user, remove_think): + yield result return elif event_type == "message" or event_type == "agent_message": @@ -1342,13 +1376,8 @@ class DefaultRunner(Runner): external_conv_id, scope="conversation", ) - interaction_result = await self._interaction_result_for_pause(ctx, event, user) - if interaction_result is None: - raise DifyAPIError( - "Dify paused the workflow without a supported reason", - code="dify.response_invalid", - ) - yield interaction_result + async for result in self._wait_and_resume_workflow(ctx, client, event, user, remove_think): + yield result return elif event_type == "workflow_finished": @@ -1426,9 +1455,11 @@ class DefaultRunner(Runner): client: AsyncDifyClient, submission: typing.Any, remove_think: bool, + continuation: dict[str, typing.Any] | None = None, + blocking: bool = False, ) -> typing.AsyncGenerator[RunnerResult, None]: interaction_id = str(submission.interaction_id) - continuation = await self._load_interaction_continuation(ctx, interaction_id) + continuation = continuation if continuation is not None else await self._load_interaction_continuation(ctx, interaction_id) if continuation.get("phase") == "field": yield await self._advance_field_interaction(ctx, continuation, submission) return @@ -1449,7 +1480,8 @@ class DefaultRunner(Runner): # Persist consumption BEFORE the external effect. An ambiguous failure # remains consumed, never silently retries an approval at the vendor. continuation["consumed"] = True - await self._store_interaction_continuation(ctx, continuation) + if not blocking: + await self._store_interaction_continuation(ctx, continuation) pending_content = "" streamed_content = "" @@ -1526,18 +1558,22 @@ class DefaultRunner(Runner): **data, "workflow_run_id": str(continuation["workflow_run_id"]), } - interaction_result = await self._interaction_result_for_pause( - ctx, - event, - str(continuation["user"]), - ) - if interaction_result is None: - raise DifyAPIError( - "Dify paused the workflow without a supported reason", - code="dify.response_invalid", + if blocking: + async for result in self._wait_and_resume_workflow( + ctx, client, event, str(continuation['user']), remove_think, + ): + yield result + else: + interaction_result = await self._interaction_result_for_pause( + ctx, event, str(continuation['user']), ) - await self._delete_interaction_continuation(ctx, interaction_id) - yield interaction_result + if interaction_result is None: + raise DifyAPIError( + 'Dify paused the workflow without a supported reason', + code='dify.response_invalid', + ) + await self._delete_interaction_continuation(ctx, interaction_id) + yield interaction_result return elif event_type == "workflow_finished": data = event.get("data", {}) @@ -1574,5 +1610,6 @@ class DefaultRunner(Runner): "Dify workflow returned no response after human input", code="dify.api_error", ) - await self._delete_interaction_continuation(ctx, interaction_id) + if not blocking: + await self._delete_interaction_continuation(ctx, interaction_id) yield RunnerResult.run_completed(ctx.run_id, usage=usage) diff --git a/plugins/Runner/dify-agent/readme/README_en_US.md b/plugins/Runner/dify-agent/readme/README_en_US.md index 4b62497fc..e471df244 100644 --- a/plugins/Runner/dify-agent/readme/README_en_US.md +++ b/plugins/Runner/dify-agent/readme/README_en_US.md @@ -38,8 +38,8 @@ Configuration is **static** and should not contain runtime state. Only the follo ## Workflow Human Input When Dify emits `workflow_paused` with a `human_input_required` reason, the -runner asks LangBot to deliver a structured interaction. A validated -`interaction.submitted` event resumes Dify through +runner asks LangBot to deliver a structured interaction and waits for the +validated submission in the same run. It resumes Dify through `/form/human_input/{form_token}` and `/workflow/{workflow_run_id}/events`. Workflows that pause again produce a new interaction and can be resumed again. Forms are emitted as atomic steps: the runner collects one field at a time and @@ -47,10 +47,9 @@ then emits the action buttons. This matches LangBot delivery adapters that support a single select or an action-only confirmation. Provider continuation data is not included in the interaction payload. The -runner stores `form_token`, `workflow_run_id`, the Dify user tag, and field/action -mappings in authorized plugin storage under a random LangBot `interaction_id`. -The continuation is deleted after a successful terminal event or after it is -replaced by a later pause. +runner keeps `form_token`, `workflow_run_id`, the Dify user tag, and field/action +mappings in the active run. Previously issued interactions can still resume +through their stored continuation records during migration. The selected Runner binding must allow interactions, and the delivery adapter must support the requested field/action shape. Adapters with a smaller diff --git a/plugins/Runner/tests/test_plugin_layout.py b/plugins/Runner/tests/test_plugin_layout.py index ea2bc0d95..2ce7aa751 100644 --- a/plugins/Runner/tests/test_plugin_layout.py +++ b/plugins/Runner/tests/test_plugin_layout.py @@ -18,6 +18,7 @@ from langbot_plugin.api.entities.builtin.runner import ( AgentTrigger, DeliveryContext, InteractionSubmission, + InteractionResult, RunnerContext, ) @@ -852,12 +853,20 @@ def _write_fake_codex_app_server(path: Path) -> None: " method = request.get('method')", " request_id = request.get('id')", " params = request.get('params') or {}", + " if request_id == 700 and method is None:", + " answer = (request.get('result') or {}).get('contentItems') or []", + " send({'jsonrpc': '2.0', 'method': 'item/completed', 'params': {'threadId': thread_id, 'item': {'id': 'answer-result', 'type': 'agentMessage', 'text': 'USER_ANSWER:' + str(answer), 'phase': 'final_answer'}}})", + " send({'jsonrpc': '2.0', 'method': 'turn/completed', 'params': {'threadId': thread_id, 'turn': {'id': 'turn-1', 'status': 'completed'}}})", + " continue", " if request_id == 701 and method is None and pending_approval_prompt:", " decision = (request.get('result') or {}).get('decision')", " if decision == 'accept':", " send({'jsonrpc': '2.0', 'method': 'item/completed', 'params': {'threadId': thread_id, 'item': {'id': 'command-1', 'type': 'commandExecution', 'command': 'Write approval-probe.txt', 'cwd': '.', 'commandActions': [], 'status': 'completed', 'aggregatedOutput': 'created'}}})", " send({'jsonrpc': '2.0', 'method': 'item/completed', 'params': {'threadId': thread_id, 'item': {'id': 'approval-result', 'type': 'agentMessage', 'text': 'APPROVAL_EXECUTED', 'phase': 'final_answer'}}})", " send({'jsonrpc': '2.0', 'method': 'turn/completed', 'params': {'threadId': thread_id, 'turn': {'id': 'turn-approval', 'status': 'completed'}}})", + " else:", + " send({'jsonrpc': '2.0', 'method': 'item/completed', 'params': {'threadId': thread_id, 'item': {'id': 'rejection-result', 'type': 'agentMessage', 'text': 'APPROVAL_REJECTED', 'phase': 'final_answer'}}})", + " send({'jsonrpc': '2.0', 'method': 'turn/completed', 'params': {'threadId': thread_id, 'turn': {'id': 'turn-approval', 'status': 'completed'}}})", " pending_approval_prompt = ''", " continue", " if request_id is not None and method == 'initialize':", @@ -1238,11 +1247,27 @@ def test_codex_runner_defaults_to_non_interactive_full_access(tmp_path: Path) -> assert config["sandbox_mode"] == "danger-full-access" -def test_codex_runner_pauses_for_dynamic_question_and_resumes_thread(tmp_path: Path) -> None: +def _interaction_context(config, text, *, action_id=None, values=None): + ctx = _native_ctx(config, text=text) + + async def request_interaction(request, **kwargs): + return InteractionResult( + status='submitted', + submission=InteractionSubmission( + interaction_id=request.interaction_id, + action_id=action_id, + values=values or {}, + ), + ) + + ctx._api = types.SimpleNamespace(request_interaction=request_interaction) + return ctx + + +def test_codex_runner_answers_dynamic_question_in_live_turn(tmp_path: Path) -> None: fake_cli = tmp_path / "fake_codex_app_server.py" _write_fake_codex_app_server(fake_cli) module = _load_runner_module("codex-agent") - native = sys.modules["pkg.native_cli"] runner = object.__new__(module.DefaultRunner) runner.get_plugin_config = lambda: {} runner.get_run_api = lambda ctx: None @@ -1255,37 +1280,12 @@ def test_codex_runner_pauses_for_dynamic_question_and_resumes_thread(tmp_path: P "sandbox-mode": "read-only", } - paused = asyncio.run(_collect_async(runner.run(_native_ctx(config, text="ASK_INTERACTION")))) + resumed = asyncio.run(_collect_async(runner.run(_interaction_context( + config, 'ASK_INTERACTION', values={'environment': 'production'}, + )))) - assert [item.type for item in paused] == ["state.updated", "state.updated", "action.requested"] - pending = paused[1].data["value"] - request = paused[2].data["payload"] - assert paused[2].data["action"] == "interaction.requested" - assert request["fields"][0]["type"] == "select" - assert pending["provider_call_id"] == "call-question-1" - - resumed = asyncio.run( - _collect_async( - runner.run( - _native_ctx( - config, - conversation_state={ - native.SESSION_STATE_KEY: "thread-123", - native.PENDING_INTERACTION_STATE_KEY: pending, - }, - interaction=InteractionSubmission( - interaction_id=request["interaction_id"], - action_id="submit", - values={"environment": "production"}, - ), - ) - ) - ) - ) - - assert resumed[0].data["key"] == native.PENDING_INTERACTION_STATE_KEY - assert resumed[0].data["value"] is None assert resumed[-1].type == "run.completed" + assert not any(item.type == 'action.requested' for item in resumed) assert any( "production" in str(item.data.get("chunk", {}).get("content") or "") for item in resumed @@ -1293,11 +1293,10 @@ def test_codex_runner_pauses_for_dynamic_question_and_resumes_thread(tmp_path: P ) -def test_codex_runner_pauses_for_native_approval_and_approves_exact_retry(tmp_path: Path) -> None: +def test_codex_runner_approves_native_request_in_live_turn(tmp_path: Path) -> None: fake_cli = tmp_path / "fake_codex_app_server.py" _write_fake_codex_app_server(fake_cli) module = _load_runner_module("codex-agent") - native = sys.modules["pkg.native_cli"] runner = object.__new__(module.DefaultRunner) runner.get_plugin_config = lambda: {} runner.get_run_api = lambda ctx: None @@ -1310,39 +1309,12 @@ def test_codex_runner_pauses_for_native_approval_and_approves_exact_retry(tmp_pa "sandbox-mode": "read-only", } - paused = asyncio.run(_collect_async(runner.run(_native_ctx(config, text="TRIGGER_COMMAND_APPROVAL")))) + resumed = asyncio.run(_collect_async(runner.run(_interaction_context( + config, 'TRIGGER_COMMAND_APPROVAL', action_id='approve_once', + )))) - assert [item.type for item in paused] == ["state.updated", "state.updated", "action.requested"] - pending = paused[1].data["value"] - request = paused[2].data["payload"] - assert pending["kind"] == "approval" - assert pending["approval_category"] == "command" - assert request["kind"] == "confirmation" - assert request["fields"] == [] - assert [action["id"] for action in request["actions"]] == ["approve_once", "reject"] - assert "Write approval-probe.txt" in request["description"] - - resumed = asyncio.run( - _collect_async( - runner.run( - _native_ctx( - config, - conversation_state={ - native.SESSION_STATE_KEY: "thread-123", - native.PENDING_INTERACTION_STATE_KEY: pending, - }, - interaction=InteractionSubmission( - interaction_id=request["interaction_id"], - action_id="approve_once", - ), - ) - ) - ) - ) - - assert resumed[0].data["key"] == native.PENDING_INTERACTION_STATE_KEY - assert resumed[0].data["value"] is None assert resumed[-1].type == "run.completed" + assert not any(item.type == 'action.requested' for item in resumed) assert any( "APPROVAL_EXECUTED" in str(item.data.get("chunk", {}).get("content") or "") for item in resumed @@ -1350,11 +1322,10 @@ def test_codex_runner_pauses_for_native_approval_and_approves_exact_retry(tmp_pa ) -def test_codex_runner_resumes_after_native_approval_rejection(tmp_path: Path) -> None: +def test_codex_runner_rejects_native_request_in_live_turn(tmp_path: Path) -> None: fake_cli = tmp_path / "fake_codex_app_server.py" _write_fake_codex_app_server(fake_cli) module = _load_runner_module("codex-agent") - native = sys.modules["pkg.native_cli"] runner = object.__new__(module.DefaultRunner) runner.get_plugin_config = lambda: {} runner.get_run_api = lambda ctx: None @@ -1366,27 +1337,9 @@ def test_codex_runner_resumes_after_native_approval_rejection(tmp_path: Path) -> "approval-policy": "on-request", "sandbox-mode": "read-only", } - paused = asyncio.run(_collect_async(runner.run(_native_ctx(config, text="TRIGGER_COMMAND_APPROVAL")))) - pending = paused[1].data["value"] - request = paused[2].data["payload"] - - resumed = asyncio.run( - _collect_async( - runner.run( - _native_ctx( - config, - conversation_state={ - native.SESSION_STATE_KEY: "thread-123", - native.PENDING_INTERACTION_STATE_KEY: pending, - }, - interaction=InteractionSubmission( - interaction_id=request["interaction_id"], - action_id="reject", - ), - ) - ) - ) - ) + resumed = asyncio.run(_collect_async(runner.run(_interaction_context( + config, 'TRIGGER_COMMAND_APPROVAL', action_id='reject', + )))) assert resumed[-1].type == "run.completed" assert any( diff --git a/plugins/Runner/tests/test_traditional_runner_protocol_fixes.py b/plugins/Runner/tests/test_traditional_runner_protocol_fixes.py index 9fb3efbae..0e4dc471d 100644 --- a/plugins/Runner/tests/test_traditional_runner_protocol_fixes.py +++ b/plugins/Runner/tests/test_traditional_runner_protocol_fixes.py @@ -22,6 +22,7 @@ from langbot_plugin.api.entities.builtin.runner import ( ConversationContext, DeliveryContext, InteractionSubmission, + InteractionResult, RunnerContext, ) @@ -92,6 +93,25 @@ def _ctx( ) +def _dify_interaction_ctx(): + ctx = _ctx() + requests = [] + + async def request_interaction(request, **kwargs): + requests.append(request) + return InteractionResult( + status='submitted', + submission=InteractionSubmission( + interaction_id=request.interaction_id, + action_id=request.actions[0].id if request.actions else None, + values={request.fields[0].id: 'high'} if request.fields else {}, + ), + ) + + ctx._api = types.SimpleNamespace(request_interaction=request_interaction) + return ctx, requests + + def _text_and_image_ctx(config: dict[str, Any]) -> RunnerContext: return _ctx( config=config, @@ -231,25 +251,25 @@ def test_dify_workflow_pause_requests_host_interaction_and_hides_provider_tokens }, } - results = asyncio.run(_collect_async(runner._run_workflow(_ctx(), FakeClient(), {}, "hello", "user_1", [], False))) + async def workflow_submit(self, **kwargs): + assert kwargs['form_token'] == 'form-private' + assert kwargs['workflow_run_id'] == 'workflow-private' + assert kwargs['inputs']['priority'] == 'high' + assert kwargs['action'] == 'approve-private' + yield {'event': 'node_finished', 'data': {'node_type': 'answer', 'outputs': {'answer': 'Approved'}}} + yield {'event': 'workflow_finished', 'data': {'outputs': {}}} + + ctx, requests = _dify_interaction_ctx() + results = asyncio.run(_collect_async(runner._run_workflow(ctx, FakeClient(), {}, "hello", "user_1", [], False))) - interaction_result = results[-1] assert _type(results[0]) == "tool.call.started" assert results[0].data["tool_name"] == "Manual review" - assert all(_type(result) != "message.delta" for result in results) - assert _type(interaction_result) == "action.requested" - assert interaction_result.data["action"] == "interaction.requested" - request = interaction_result.data["payload"] - assert request["fields"][0]["options"][0]["value"] == "high" - assert request["actions"] == [] - assert "form-private" not in str(request) - assert "workflow-private" not in str(request) - assert len(storage.values) == 1 - continuation = json.loads(next(iter(storage.values.values()))) - assert continuation["form_token"] == "form-private" - assert continuation["workflow_run_id"] == "workflow-private" - assert continuation["interaction_actions"][0]["label"] == "Approve" - assert all(_type(result) != "run.completed" for result in results) + assert requests[0].fields[0].options[0].value == 'high' + assert requests[1].actions[0].label == 'Approve' + assert 'form-private' not in str(requests) + assert 'workflow-private' not in str(requests) + assert storage.values == {} + assert _type(results[-1]) == 'run.completed' def test_dify_current_human_input_event_requests_host_interaction() -> None: @@ -278,18 +298,20 @@ def test_dify_current_human_input_event_requests_host_interaction() -> None: }, } - results = asyncio.run(_collect_async(runner._run_workflow(_ctx(), FakeClient(), {}, "hello", "user_1", [], False))) + async def workflow_submit(self, **kwargs): + assert kwargs['inputs']['priority'] == 'high' + yield {'event': 'node_finished', 'data': {'node_type': 'answer', 'outputs': {'answer': 'Approved'}}} + yield {'event': 'workflow_finished', 'data': {'outputs': {}}} - assert _type(results[-1]) == "action.requested" - assert all(_type(result) != "run.completed" for result in results) - request = results[-1].data["payload"] - assert request["title"] == "Manual review" - assert request["fields"][0]["options"][0]["value"] == "high" - assert "form-private" not in str(request) - assert "workflow-private" not in str(request) - continuation = json.loads(next(iter(storage.values.values()))) - assert continuation["form_token"] == "form-private" - assert continuation["workflow_run_id"] == "workflow-private" + ctx, requests = _dify_interaction_ctx() + results = asyncio.run(_collect_async(runner._run_workflow(ctx, FakeClient(), {}, "hello", "user_1", [], False))) + + assert _type(results[-1]) == 'run.completed' + assert requests[0].title == 'Manual review' + assert requests[0].fields[0].options[0].value == 'high' + assert 'form-private' not in str(requests) + assert 'workflow-private' not in str(requests) + assert storage.values == {} def test_dify_pause_accepts_single_reason_and_camel_case_fields() -> None: diff --git a/src/langbot/pkg/agent/runner/interaction_manager.py b/src/langbot/pkg/agent/runner/interaction_manager.py index 083f8516b..78d16855a 100644 --- a/src/langbot/pkg/agent/runner/interaction_manager.py +++ b/src/langbot/pkg/agent/runner/interaction_manager.py @@ -3,6 +3,7 @@ from __future__ import annotations import copy +import asyncio import json import time import typing @@ -10,6 +11,12 @@ import typing import pydantic from langbot_plugin.api.entities.builtin.platform import message as platform_message +from langbot_plugin.api.entities.builtin.runner.interaction import ( + InteractionDeliveryCapabilities, + InteractionHandle, + InteractionRequest, + InteractionSubmission, +) from .descriptor import RunnerDescriptor from .errors import RunnerProtocolError @@ -69,7 +76,7 @@ class HostInteractionRequest(pydantic.BaseModel): fields: list[HostInteractionField] = pydantic.Field(default_factory=list, max_length=50) actions: list[HostInteractionAction] = pydantic.Field(default_factory=list, max_length=20) expires_at: int | None = None - fallback_text: str = pydantic.Field(min_length=1, max_length=20000) + fallback_text: str | None = pydantic.Field(default=None, min_length=1, max_length=20000) model_config = pydantic.ConfigDict(extra='forbid') @@ -81,6 +88,7 @@ class HostInteractionSubmission(pydantic.BaseModel): action_id: str | None = pydantic.Field(default=None, max_length=128) values: dict[str, typing.Any] = pydantic.Field(default_factory=dict) submitted_at: int | None = None + label: str | None = None model_config = pydantic.ConfigDict(extra='forbid') @@ -91,6 +99,342 @@ class InteractionManager: def __init__(self, ap: typing.Any, store: InteractionStore | None = None): self.ap = ap self.store = store or InteractionStore(ap.persistence_mgr.get_db_engine()) + self._waiters: dict[tuple[str, str], asyncio.Future[dict[str, typing.Any]]] = {} + self._waiter_scopes: dict[tuple[str, str], tuple[str, str]] = {} + self._responders: dict[tuple[str, str], str | list[str]] = {} + self._callback_contexts: dict[tuple[str, str], dict[str, typing.Any]] = {} + self._text_waiters: dict[tuple[str, str], tuple[str, str, str, str, InteractionRequest]] = {} + + async def request_interaction( + self, + *, + session: typing.Mapping[str, typing.Any], + adapter: typing.Any, + request_payload: dict[str, typing.Any], + timeout: float, + on_unsupported: str, + responders: str | list[str], + target_type: str | None = None, + target_id: str | None = None, + ) -> dict[str, typing.Any]: + """Deliver an interaction and wait for its validated callback in this run.""" + authorization = session['authorization'] + if not authorization.get('interactions_enabled') or not authorization.get('interactions_authorized'): + raise ValueError('Runner did not declare interaction capability and permission') + if not 0 < timeout <= MAX_INTERACTION_TTL_SECONDS: + raise ValueError('Interaction timeout must be within 24 hours') + if on_unsupported not in {'error', 'degrade'}: + raise ValueError('Invalid interaction fallback policy') + valid_responders = ( + responders in {'actor', 'any'} + if isinstance(responders, str) + else ( + isinstance(responders, list) + and bool(responders) + and all(isinstance(item, str) and item for item in responders) + ) + ) + if not valid_responders: + raise ValueError('Invalid interaction responders') + + request = InteractionRequest.model_validate(request_payload) + if request.fields and request.actions: + raise ValueError('Interaction cannot combine fields and actions') + self._validate_request_identity( + HostInteractionRequest.model_validate(request.model_dump(mode='json')), session['runner_id'] + ) + run_id = session['run_id'] + key = (run_id, request.interaction_id) + bot_id = authorization.get('bot_id') + workspace_id = authorization.get('workspace_id') + workspace_waiters = sum(scope[0] == workspace_id for scope in self._waiter_scopes.values()) + bot_waiters = sum(scope == (workspace_id, bot_id) for scope in self._waiter_scopes.values()) + if workspace_waiters >= 256 or bot_waiters >= 32: + raise ValueError('Too many pending interactions') + if key in self._waiters: + raise ValueError('Interaction is already waiting') + platform_context = authorization.get('platform_context') or {} + delivery = platform_context.get('delivery') or {} + reply_target = delivery.get('reply_target') or {} + selected_type = target_type or reply_target.get('target_type') + selected_id = target_id or reply_target.get('target_id') + if not selected_type or not selected_id: + raise ValueError('Interaction has no delivery target') + if target_type is not None and (target_type, target_id) != ( + reply_target.get('target_type'), + reply_target.get('target_id'), + ): + from .platform_tools import resolve_platform_api_call + + resolve_platform_api_call( + session, + bot_id, + 'send_message', + { + 'target_type': target_type, + 'target_id': target_id, + 'message': platform_message.MessageChain( + [platform_message.Plain(text='Interaction request')] + ).model_dump(mode='json'), + }, + ) + actor = platform_context.get('actor') or {} + actor_id = actor.get('actor_id') + if responders == 'actor' and not actor_id: + raise ValueError('Interaction has no actor') + capabilities = adapter.get_interaction_capabilities() + if not isinstance(capabilities, InteractionDeliveryCapabilities): + capabilities = InteractionDeliveryCapabilities.model_validate(capabilities) + supported_apis = adapter.get_supported_apis() if callable(getattr(adapter, 'get_supported_apis', None)) else [] + if supported_apis and 'interaction.acknowledge' not in supported_apis: + capabilities.supports_updates = False + normalized, degraded, use_text, issues = self._normalize_request(request, capabilities, on_unsupported) + if issues: + return {'unsupported': issues} + request_data = normalized.model_dump(mode='json') + request_data['expires_at'] = int(time.time() + timeout + 1) + if len(json.dumps(request_data, ensure_ascii=False).encode('utf-8')) > INTERACTION_PAYLOAD_MAX_BYTES: + raise ValueError('Interaction request exceeds 256 KiB') + conversation_id = f'{selected_type}_{selected_id}' + _, callback_token = await self.store.create_request( + interaction_id=request.interaction_id, + run_id=run_id, + binding_id=authorization['binding_id'], + runner_id=session['runner_id'], + processor_type=authorization['processor_type'], + processor_id=authorization['processor_id'], + request=request_data, + delivery_target={'target_type': selected_type, 'target_id': selected_id}, + bot_id=bot_id, + workspace_id=workspace_id, + conversation_id=conversation_id, + thread_id=authorization.get('thread_id'), + actor_id=actor_id if responders == 'actor' else None, + expires_at=request_data['expires_at'], + blocking=True, + **self._pipeline_tool_admission(session, authorization, session['runner_id']), + ) + future = asyncio.get_running_loop().create_future() + self._waiters[key] = future + self._responders[key] = responders + self._waiter_scopes[key] = (workspace_id, bot_id) + if use_text: + self._text_waiters[key] = (workspace_id, bot_id, conversation_id, callback_token, normalized) + runtime = session.get('runtime_context') + original_deadline = runtime.get('deadline_at') if isinstance(runtime, dict) else None + started_at = time.time() + if original_deadline is not None: + runtime['deadline_at'] = float(original_deadline) + timeout + 30 + handle = None + terminal = None + try: + if use_text: + await adapter.send_message( + str(selected_type), + str(selected_id), + platform_message.MessageChain( + [platform_message.Plain(text=normalized.fallback_text or normalized.title)] + ), + ) + else: + handle = await adapter.send_interaction( + str(selected_type), + str(selected_id), + normalized, + callback_token, + reply_to=reply_target.get('message_id'), + ) + if isinstance(handle, dict): + handle = InteractionHandle.model_validate(handle) + await self.store.record_delivery_success(run_id, request.interaction_id, handle.model_dump(mode='json')) + try: + submission = await asyncio.wait_for(future, timeout=timeout) + terminal = 'submitted' + except asyncio.TimeoutError: + terminal = 'expired' + await self.store.mark_terminal(run_id, request.interaction_id, terminal) + submission = None + if handle is not None and capabilities.supports_updates: + handle.context['callback_context'] = self._callback_contexts.get(key, {}) + try: + await adapter.update_interaction( + handle, + terminal, + InteractionSubmission.model_validate(submission) if submission else None, + ) + except Exception as exc: + self._warning(f'Failed to update interaction card: {exc}') + if original_deadline is not None: + runtime['deadline_at'] = float(original_deadline) + time.time() - started_at + return { + 'status': 'timeout' if terminal == 'expired' else 'submitted', + 'submission': submission, + 'degraded': degraded, + 'deadline_at': runtime.get('deadline_at') if isinstance(runtime, dict) else None, + } + except asyncio.CancelledError: + terminal = 'cancelled' + await self.store.mark_terminal(run_id, request.interaction_id, terminal) + if handle is not None and capabilities.supports_updates: + try: + await adapter.update_interaction(handle, terminal) + except Exception as exc: + self._warning(f'Failed to cancel interaction card: {exc}') + raise + except Exception as exc: + if terminal is None: + await self.store.mark_delivery_failed(run_id, request.interaction_id, str(exc)) + raise + finally: + if original_deadline is not None: + runtime['deadline_at'] = float(original_deadline) + time.time() - started_at + self._waiters.pop(key, None) + self._responders.pop(key, None) + self._callback_contexts.pop(key, None) + self._text_waiters.pop(key, None) + self._waiter_scopes.pop(key, None) + + @staticmethod + def _normalize_request(request, capabilities, policy): + normalized = request.model_copy(deep=True) + issues = [] + degraded = [] + native = bool(capabilities.field_types or capabilities.max_actions or capabilities.action_styles) + if not native: + issues.append({'path': 'platform', 'reason': 'native_interactions_unavailable'}) + for index, field in enumerate(request.fields): + if field.type not in capabilities.field_types: + issues.append({'path': f'fields[{index}].type', 'reason': 'unsupported_field_type'}) + if capabilities.max_options is not None and len(field.options) > capabilities.max_options: + issues.append({'path': f'fields[{index}].options', 'reason': 'too_many_options'}) + if field.placeholder: + degraded.append( + {'path': f'fields[{index}].placeholder', 'reason': 'unsupported_placeholder', 'applied': 'omit'} + ) + normalized.fields[index].placeholder = None + if field.default is not None: + degraded.append( + {'path': f'fields[{index}].default', 'reason': 'unsupported_default', 'applied': 'omit'} + ) + normalized.fields[index].default = None + if capabilities.max_fields is not None and len(request.fields) > capabilities.max_fields: + issues.append({'path': 'fields', 'reason': 'too_many_fields'}) + if capabilities.max_actions is not None and len(request.actions) > capabilities.max_actions: + issues.append({'path': 'actions', 'reason': 'too_many_actions'}) + for index, action in enumerate(request.actions): + if action.style not in capabilities.action_styles: + degraded.append( + {'path': f'actions[{index}].style', 'reason': 'unsupported_style', 'applied': 'default'} + ) + normalized.actions[index].style = 'default' + if request.description and not capabilities.supports_markdown: + degraded.append({'path': 'description', 'reason': 'unsupported_markdown', 'applied': 'literal_text'}) + if issues and policy == 'error': + return normalized, degraded, False, issues + if issues and ( + not capabilities.text_fallback + or len(request.fields) > 1 + or any(field.type in {'file', 'multiselect'} for field in request.fields) + ): + return normalized, degraded, False, issues + use_text = bool(issues) + fallback_text = request.fallback_text or request.title + if use_text and request.actions: + choices = ' / '.join(f'{index + 1} {action.label}' for index, action in enumerate(request.actions)) + fallback_text = f'{fallback_text}\n{choices}' + elif use_text and request.fields: + field = request.fields[0] + fallback_text = f'{fallback_text}\n{field.label}' + if field.type == 'select': + fallback_text += '\n' + ' / '.join( + f'{index + 1} {option.label}' for index, option in enumerate(field.options) + ) + normalized.fallback_text = fallback_text + return ( + normalized, + degraded + + ( + [{'path': 'platform', 'reason': 'unsupported_structure', 'applied': 'text_fallback'}] + if use_text + else [] + ), + use_text, + [], + ) + + def wake_waiter( + self, + record: dict[str, typing.Any], + actor_id: str | None, + callback_context: dict[str, typing.Any] | None = None, + ) -> bool: + key = (record['run_id'], record['interaction_id']) + future = self._waiters.get(key) + if future is None or future.done(): + return False + responders = self._responders[key] + if isinstance(responders, list) and actor_id not in responders: + raise ValueError('Interaction actor is not an allowed responder') + self._callback_contexts[key] = callback_context or {} + future.set_result(record['submission']) + return True + + async def consume_text_message( + self, + *, + workspace_id: str, + bot_id: str, + conversation_id: str, + actor_id: str, + text: str, + ) -> bool: + answer = text.strip() + if not answer or len(answer) > 20000: + return False + for key, (workspace, bot, conversation, token, request) in reversed(list(self._text_waiters.items())): + if (workspace, bot, conversation) != (workspace_id, bot_id, conversation_id): + continue + responders = self._responders.get(key) + if responders == 'actor': + pending = await self.store.get_by_callback_token(token) + if pending is None or pending.get('actor_id') != actor_id: + continue + elif isinstance(responders, list) and actor_id not in responders: + continue + submission: dict[str, typing.Any] = {'interaction_id': request.interaction_id, 'values': {}} + if request.actions: + if not answer.isdigit() or not 1 <= int(answer) <= len(request.actions): + return False + submission['action_id'] = request.actions[int(answer) - 1].id + submission['label'] = request.actions[int(answer) - 1].label + elif request.fields: + field = request.fields[0] + value: typing.Any = answer + if field.type == 'number': + try: + value = float(answer) + except ValueError: + return False + elif field.type == 'boolean': + if answer.lower() not in {'yes', 'no', 'true', 'false', '1', '0'}: + return False + value = answer.lower() in {'yes', 'true', '1'} + elif field.type == 'select': + options = field.options + if answer.isdigit() and 1 <= int(answer) <= len(options): + value = options[int(answer) - 1].value + elif answer not in {option.value for option in options}: + return False + submission['values'] = {field.id: value} + record = await self.consume_callback( + callback_token=token, + submission=submission, + bot_id=bot_id, + conversation_id=conversation_id, + actor_id=actor_id, + ) + return self.wake_waiter(record, actor_id) + return False async def handle_result( self, @@ -496,6 +840,9 @@ class InteractionManager: from .interaction_store import InteractionNotFoundError raise InteractionNotFoundError('Unknown interaction callback token') + responders = self._responders.get((pending['run_id'], pending['interaction_id'])) + if isinstance(responders, list) and actor_id not in responders: + raise ValueError('Interaction actor is not an allowed responder') normalized_submission = self._resolve_submission_refs(pending, submission) try: normalized = HostInteractionSubmission.model_validate(normalized_submission) diff --git a/src/langbot/pkg/agent/runner/interaction_store.py b/src/langbot/pkg/agent/runner/interaction_store.py index 806ec258f..4faa42c23 100644 --- a/src/langbot/pkg/agent/runner/interaction_store.py +++ b/src/langbot/pkg/agent/runner/interaction_store.py @@ -126,6 +126,7 @@ class InteractionStore: expires_at: int | float | None = None, expected_config: dict | None = None, authority_check: typing.Callable[[], bool] | None = None, + blocking: bool = False, ) -> tuple[dict[str, typing.Any], str]: """Persist a request and return its record plus one-time callback token.""" if not interaction_id or not run_id or not binding_id or not runner_id or not processor_id: @@ -154,6 +155,7 @@ class InteractionStore: thread_id=thread_id, actor_id=actor_id, status='pending', + status_reason='blocking' if blocking else None, request_json=request_json, delivery_target_json=delivery_target_json, replaces_interaction_id=replaces_interaction_id, @@ -491,6 +493,36 @@ class InteractionStore: await session.commit() return result.rowcount == 1 + async def mark_terminal(self, run_id: str, interaction_id: str, status: str) -> bool: + if status not in {'expired', 'cancelled'}: + raise ValueError('Invalid terminal interaction status') + now = _utc_now() + async with self._session_factory() as session: + result = await session.execute( + sqlalchemy.update(AgentInteraction) + .where( + AgentInteraction.run_id == run_id, + AgentInteraction.interaction_id == interaction_id, + AgentInteraction.status == 'pending', + ) + .values(status=status, status_reason=f'interaction {status}', updated_at=_db_datetime(now)) + ) + await session.commit() + return result.rowcount == 1 + + async def cancel_orphaned_pending(self, active_run_ids: set[str]) -> int: + """Invalidate pending callbacks after Host restart or run cleanup.""" + now = _utc_now() + condition = ~AgentInteraction.run_id.in_(active_run_ids) if active_run_ids else sqlalchemy.true() + async with self._session_factory() as session: + result = await session.execute( + sqlalchemy.update(AgentInteraction) + .where(AgentInteraction.status == 'pending', AgentInteraction.status_reason == 'blocking', condition) + .values(status='cancelled', status_reason='run is no longer active', updated_at=_db_datetime(now)) + ) + await session.commit() + return int(result.rowcount or 0) + async def expire_pending(self, *, now: datetime.datetime | None = None) -> int: """Expire pending interactions whose deadline has elapsed.""" cutoff = _as_utc(now) or _utc_now() diff --git a/src/langbot/pkg/agent/runner/invoker.py b/src/langbot/pkg/agent/runner/invoker.py index c4e54307b..a32a9d8b3 100644 --- a/src/langbot/pkg/agent/runner/invoker.py +++ b/src/langbot/pkg/agent/runner/invoker.py @@ -94,17 +94,31 @@ class RunnerInvoker: await self._close_generator(gen, descriptor) raise asyncio.TimeoutError + next_result = asyncio.create_task(anext(gen)) try: - if remaining is None: - return await anext(gen) - return await asyncio.wait_for(anext(gen), timeout=remaining) + while True: + if remaining is None: + return await next_result + try: + return await asyncio.wait_for(asyncio.shield(next_result), timeout=remaining) + except asyncio.TimeoutError: + # The Host can extend the deadline while a blocking interaction + # is active. Keep the same pending read when that happens. + renewed = self._remaining_deadline_seconds(context) + if renewed is None or renewed <= 0: + raise + remaining = renewed except StopAsyncIteration: if self._is_deadline_exhausted(context): raise asyncio.TimeoutError raise except asyncio.TimeoutError: + next_result.cancel() await self._close_generator(gen, descriptor) raise + finally: + if not next_result.done(): + next_result.cancel() def _remaining_deadline_seconds( self, diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index 4909f3dcc..028640d36 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -313,6 +313,11 @@ class AgentRunOrchestrator: processor_type=binding.processor_type, processor_id=binding.processor_id or binding.agent_id, interactions_enabled=binding.delivery_policy.enable_interactions, + interactions_authorized=( + bool(getattr(descriptor.capabilities, 'interactions', False)) + and 'request' in (getattr(descriptor.permissions, 'interactions', []) or []) + ), + runtime_context=context.get('runtime'), ) event_log_id = await self.journal.write_event_log( diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index ff63b47c1..a5582f858 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -46,7 +46,7 @@ _TARGET_TYPE = {'type': 'string', 'enum': ['person', 'group']} _CHAT_TYPE = {'type': 'string', 'enum': ['person', 'private', 'group']} _APPROVE = {'type': 'boolean', 'description': 'true to accept; false to reject'} -PRESENT_INTERACTIVE_CARD_TOOL = 'present_interactive_card' +PRESENT_INTERACTIVE_CARD_TOOL = 'request_interaction' _INTERACTION_OPTION_SCHEMA = { 'type': 'object', @@ -94,13 +94,13 @@ PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = ( 'event', 'interaction', 'write', - {'en_US': 'Present interactive card'}, + {'en_US': 'Request user interaction'}, { 'en_US': ( 'Present a native interactive card in the current conversation. Define semantic buttons, selects, ' 'and input fields; LangBot owns the target, callback token, validation, and platform rendering. ' 'Each card is one step: provide either actions or fields, never both. ' - 'The user submission will start a new interaction.submitted turn for this processor.' + 'The tool waits for the user and returns the structured submission in this run.' ), }, _object_schema( @@ -111,6 +111,7 @@ PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = ( 'fields': {'type': 'array', 'items': _INTERACTION_FIELD_SCHEMA, 'maxItems': 50}, 'actions': {'type': 'array', 'items': _INTERACTION_ACTION_SCHEMA, 'maxItems': 20}, 'expires_in_seconds': {'type': 'integer', 'minimum': 60, 'maximum': 86400}, + 'responders': {'type': 'string', 'enum': ['actor', 'any']}, }, ['title'], ), @@ -536,10 +537,14 @@ def build_platform_tool_resources( reason = None if definition is None: reason = 'unknown_tool' - elif definition.api not in supported_apis and not ( - definition.name == 'event_reply' - and {'reply_message', 'get_message'} <= supported_apis - and (event.delivery.reply_target or {}).get('message_id') + elif ( + definition.api not in supported_apis + and not (definition.name == PRESENT_INTERACTIVE_CARD_TOOL and 'send_message' in supported_apis) + and not ( + definition.name == 'event_reply' + and {'reply_message', 'get_message'} <= supported_apis + and (event.delivery.reply_target or {}).get('message_id') + ) ): reason = 'adapter_api_unsupported' elif definition.scope == 'event' and not _event_matches(event.event_type, definition.event_patterns): @@ -873,10 +878,13 @@ async def execute_platform_tool( manager = getattr(getattr(ap, 'agent_run_orchestrator', None), 'interaction_manager', None) if manager is None: raise ValueError('Interactive card delivery is unavailable') - return await manager.present_tool_request( + return await manager.request_interaction( request_payload=_interactive_card_request(normalized), session=session, adapter=bot.adapter, + timeout=normalized.get('expires_in_seconds', 600), + on_unsupported='degrade', + responders=normalized.get('responders', 'actor'), ) api_func = getattr(bot.adapter, definition.api, None) if reply_via_source: diff --git a/src/langbot/pkg/agent/runner/resource_builder.py b/src/langbot/pkg/agent/runner/resource_builder.py index 7cd8de0ed..ec3a2b83c 100644 --- a/src/langbot/pkg/agent/runner/resource_builder.py +++ b/src/langbot/pkg/agent/runner/resource_builder.py @@ -91,7 +91,12 @@ class AgentResourceBuilder: ) runner_uses_host_tools = config_schema.uses_host_tools(descriptor) selected_platform_tools = list(resource_policy.allowed_platform_tool_names or []) - if binding.processor_type not in {'agent', 'pipeline'} or not binding.delivery_policy.enable_interactions: + if ( + binding.processor_type not in {'agent', 'pipeline'} + or not binding.delivery_policy.enable_interactions + or not getattr(descriptor.capabilities, 'interactions', False) + or 'request' not in (getattr(manifest_perms, 'interactions', []) or []) + ): selected_platform_tools = [ name for name in selected_platform_tools if name != PRESENT_INTERACTIVE_CARD_TOOL ] diff --git a/src/langbot/pkg/agent/runner/session_registry.py b/src/langbot/pkg/agent/runner/session_registry.py index 1af30d04a..40c8edb96 100644 --- a/src/langbot/pkg/agent/runner/session_registry.py +++ b/src/langbot/pkg/agent/runner/session_registry.py @@ -54,6 +54,7 @@ class RunAuthorizationSnapshot(typing.TypedDict): processor_type: str | None processor_id: str | None interactions_enabled: bool + interactions_authorized: bool SteeringQueueItem = dict[str, typing.Any] @@ -83,6 +84,7 @@ class AgentRunSession(typing.TypedDict): status: AgentRunSessionStatus reply_streams: typing.Any steering_queue: list[SteeringQueueItem] + runtime_context: dict[str, typing.Any] | None class AgentRunSessionRegistry: @@ -125,6 +127,8 @@ class AgentRunSessionRegistry: processor_type: str | None = None, processor_id: str | None = None, interactions_enabled: bool = False, + interactions_authorized: bool = False, + runtime_context: dict[str, typing.Any] | None = None, ) -> None: """Register a new agent run session. @@ -174,6 +178,7 @@ class AgentRunSessionRegistry: 'processor_type': processor_type, 'processor_id': processor_id, 'interactions_enabled': interactions_enabled, + 'interactions_authorized': interactions_authorized, } session: AgentRunSession = { @@ -189,6 +194,7 @@ class AgentRunSessionRegistry: 'last_activity_at': now, }, 'steering_queue': [], + 'runtime_context': runtime_context, } async with self._lock: diff --git a/src/langbot/pkg/core/stages/build_app.py b/src/langbot/pkg/core/stages/build_app.py index c42dd769e..42100f918 100644 --- a/src/langbot/pkg/core/stages/build_app.py +++ b/src/langbot/pkg/core/stages/build_app.py @@ -347,6 +347,7 @@ class BuildAppStage(stage.BootingStage): agent_run_orchestrator_inst = AgentRunOrchestrator(ap, runner_registry_inst) ap.agent_run_orchestrator = agent_run_orchestrator_inst + await agent_run_orchestrator_inst.interaction_manager.store.cancel_orphaned_pending(set()) ctrl = controller.Controller(ap) ap.ctrl = ctrl diff --git a/src/langbot/pkg/platform/adapters/dingtalk/adapter.py b/src/langbot/pkg/platform/adapters/dingtalk/adapter.py index 972d8eca1..2a39fbee4 100644 --- a/src/langbot/pkg/platform/adapters/dingtalk/adapter.py +++ b/src/langbot/pkg/platform/adapters/dingtalk/adapter.py @@ -15,6 +15,7 @@ from langbot.libs.dingtalk_api.dingtalkevent import DingTalkEvent import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger from langbot.pkg.platform.adapters.dingtalk.api_impl import DingTalkAPIMixin +from langbot.pkg.platform.adapters.interaction_api import InteractionAdapterMixin from langbot.pkg.platform.adapters.dingtalk.event_converter import DingTalkEventConverter from langbot.pkg.platform.adapters.dingtalk.message_converter import DingTalkMessageConverter from langbot.pkg.platform.adapters.dingtalk.platform_api import PLATFORM_API_MAP @@ -92,7 +93,7 @@ class DingTalkCardCallbackHandler(dingtalk_stream.CallbackHandler): return dingtalk_stream.AckMessage.STATUS_OK, 'OK' -class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter): +class DingTalkAdapter(DingTalkAPIMixin, InteractionAdapterMixin, abstract_platform_adapter.AbstractPlatformAdapter): bot: DingTalkClient = pydantic.Field(exclude=True) message_converter: DingTalkMessageConverter = DingTalkMessageConverter() @@ -165,9 +166,9 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo 'call_platform_api', ] if hasattr(self.bot, 'create_and_deliver_ai_card'): - apis.append('interaction.request') + apis.extend(['interaction.request', 'send_interaction', 'get_interaction_capabilities']) if hasattr(self.bot, 'update_ai_card'): - apis.append('interaction.acknowledge') + apis.extend(['interaction.acknowledge', 'update_interaction']) return apis def get_interaction_capabilities(self) -> dict[str, typing.Any]: diff --git a/src/langbot/pkg/platform/adapters/dingtalk/interaction.py b/src/langbot/pkg/platform/adapters/dingtalk/interaction.py index f87c01674..a46835ec8 100644 --- a/src/langbot/pkg/platform/adapters/dingtalk/interaction.py +++ b/src/langbot/pkg/platform/adapters/dingtalk/interaction.py @@ -303,7 +303,7 @@ async def send_interaction(adapter: typing.Any, params: dict[str, typing.Any]) - 'field_types': field_types, 'target_type': target_type, 'target_id': target_id, - 'expires_at': time.monotonic() + 30 * 60, + 'expires_at': time.monotonic() + max(1, int(request.get('expires_at') or time.time() + 30 * 60) - time.time()), } result = { 'ok': True, diff --git a/src/langbot/pkg/platform/adapters/discord/adapter.py b/src/langbot/pkg/platform/adapters/discord/adapter.py index 374c008cf..e5858225f 100644 --- a/src/langbot/pkg/platform/adapters/discord/adapter.py +++ b/src/langbot/pkg/platform/adapters/discord/adapter.py @@ -11,6 +11,7 @@ import pydantic import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger from langbot.pkg.platform.adapters.discord.api_impl import DiscordAPIMixin +from langbot.pkg.platform.adapters.interaction_api import InteractionAdapterMixin from langbot.pkg.platform.adapters.discord.event_converter import DiscordEventConverter from langbot.pkg.platform.adapters.discord.message_converter import DiscordMessageConverter from langbot.pkg.platform.adapters.discord.platform_api import PLATFORM_API_MAP @@ -24,7 +25,7 @@ from langbot_plugin.api.entities.builtin.platform import events as platform_even from langbot_plugin.api.entities.builtin.platform import message as platform_message -class DiscordAdapter(DiscordAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter): +class DiscordAdapter(DiscordAPIMixin, InteractionAdapterMixin, abstract_platform_adapter.AbstractPlatformAdapter): bot: discord.Client = pydantic.Field(exclude=True) message_converter: DiscordMessageConverter = DiscordMessageConverter() @@ -205,6 +206,8 @@ class DiscordAdapter(DiscordAPIMixin, abstract_platform_adapter.AbstractPlatform 'leave_group', 'call_platform_api', 'interaction.request', + 'send_interaction', + 'get_interaction_capabilities', ] def get_interaction_capabilities(self) -> dict[str, typing.Any]: diff --git a/src/langbot/pkg/platform/adapters/interaction_api.py b/src/langbot/pkg/platform/adapters/interaction_api.py new file mode 100644 index 000000000..735f04200 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/interaction_api.py @@ -0,0 +1,55 @@ +"""Standard interaction adapter methods backed by existing platform renderers.""" + +from __future__ import annotations + +from langbot_plugin.api.entities.builtin.runner.interaction import ( + InteractionHandle, + InteractionRequest, + InteractionSubmission, +) + + +class InteractionAdapterMixin: + async def send_interaction( + self, + target_type: str, + target_id: str, + request: InteractionRequest, + callback_token: str, + reply_to: str | None = None, + ) -> InteractionHandle: + result = await self.call_platform_api( + 'interaction.request', + { + 'request': request.model_dump(mode='json'), + 'callback_token': callback_token, + 'reply_target': { + 'target_type': target_type, + 'target_id': target_id, + 'message_id': reply_to, + }, + }, + ) + handle = InteractionHandle.model_validate(result) + handle.context['request'] = request.model_dump(mode='json') + return handle + + async def update_interaction( + self, + handle: InteractionHandle, + state: str, + submission: InteractionSubmission | None = None, + ) -> InteractionHandle: + if 'interaction.acknowledge' not in self.get_supported_apis(): + return handle + result = await self.call_platform_api( + 'interaction.acknowledge', + { + 'update_target': handle.model_dump(mode='json'), + 'request': handle.context.get('request') or {}, + 'callback_context': handle.context.get('callback_context') or {}, + 'submission': submission.model_dump(mode='json') if submission else {}, + 'state': state, + }, + ) + return InteractionHandle.model_validate(result) if isinstance(result, dict) else handle diff --git a/src/langbot/pkg/platform/adapters/lark/adapter.py b/src/langbot/pkg/platform/adapters/lark/adapter.py index 84d0e313a..f5e0e4c3f 100644 --- a/src/langbot/pkg/platform/adapters/lark/adapter.py +++ b/src/langbot/pkg/platform/adapters/lark/adapter.py @@ -1,6 +1,7 @@ from __future__ import annotations from langbot.pkg.i18n.adapters import adapter_t, localized_adapter +from langbot.pkg.platform.adapters.interaction_api import InteractionAdapterMixin from langbot.pkg.platform.sources.lark import ( LarkAdapter as LegacyLarkAdapter, @@ -113,7 +114,7 @@ async def _cancel_ws_cache_task(client: typing.Any) -> None: await asyncio.gather(cache_task, return_exceptions=True) -class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter): +class LarkAdapter(LarkAPIMixin, InteractionAdapterMixin, abstract_platform_adapter.AbstractPlatformAdapter): bot: lark_oapi.ws.Client = pydantic.Field(exclude=True) api_client: lark_oapi.Client = pydantic.Field(exclude=True) quart_app: quart.Quart = pydantic.Field(exclude=True) @@ -245,6 +246,9 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte 'call_platform_api', 'interaction.request', 'interaction.acknowledge', + 'send_interaction', + 'update_interaction', + 'get_interaction_capabilities', ] def get_interaction_capabilities(self) -> dict[str, typing.Any]: diff --git a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py index 80a414940..44d142393 100644 --- a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py +++ b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py @@ -10,6 +10,7 @@ import pydantic from langbot.libs.qq_official_api.api import QQOfficialClient from langbot.libs.qq_official_api.qqofficialevent import QQOfficialEvent from langbot.pkg.platform.adapters.qqofficial.api_impl import QQOfficialAPIMixin +from langbot.pkg.platform.adapters.interaction_api import InteractionAdapterMixin from langbot.pkg.platform.adapters.qqofficial.errors import NotSupportedError from langbot.pkg.platform.adapters.qqofficial.event_converter import ( BOT_INVITED_EVENT_TYPES, @@ -35,7 +36,7 @@ from langbot_plugin.api.entities.builtin.platform import events as platform_even from langbot_plugin.api.entities.builtin.platform import message as platform_message -class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter): +class QQOfficialAdapter(QQOfficialAPIMixin, InteractionAdapterMixin, abstract_platform_adapter.AbstractPlatformAdapter): bot: typing.Any = pydantic.Field(exclude=True) message_converter: QQOfficialMessageConverter = QQOfficialMessageConverter() @@ -126,6 +127,8 @@ class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPl 'get_group_member_info', 'call_platform_api', 'interaction.request', + 'send_interaction', + 'get_interaction_capabilities', ] def get_interaction_capabilities(self) -> dict[str, typing.Any]: diff --git a/src/langbot/pkg/platform/adapters/telegram/adapter.py b/src/langbot/pkg/platform/adapters/telegram/adapter.py index 5d6519267..497f691e9 100644 --- a/src/langbot/pkg/platform/adapters/telegram/adapter.py +++ b/src/langbot/pkg/platform/adapters/telegram/adapter.py @@ -30,6 +30,7 @@ import langbot_plugin.api.entities.builtin.platform.events as platform_events import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger from langbot.pkg.platform.adapters.telegram.message_converter import TelegramMessageConverter +from langbot.pkg.platform.adapters.interaction_api import InteractionAdapterMixin from langbot.pkg.platform.adapters.telegram.event_converter import TelegramEventConverter, LegacyEventConverter from langbot.pkg.platform.adapters.telegram.api_impl import TelegramAPIMixin from langbot.pkg.platform.adapters.telegram.platform_api import PLATFORM_API_MAP @@ -41,7 +42,7 @@ from langbot.pkg.platform.adapters.telegram.interaction import ( ) -class TelegramAdapter(TelegramAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter): +class TelegramAdapter(TelegramAPIMixin, InteractionAdapterMixin, abstract_platform_adapter.AbstractPlatformAdapter): """Telegram adapter (EBA version).""" bot: telegram.Bot = pydantic.Field(exclude=True) @@ -199,6 +200,8 @@ class TelegramAdapter(TelegramAPIMixin, abstract_platform_adapter.AbstractPlatfo 'leave_group', 'call_platform_api', 'interaction.request', + 'send_interaction', + 'get_interaction_capabilities', ] def get_interaction_capabilities(self) -> dict[str, typing.Any]: diff --git a/src/langbot/pkg/platform/adapters/wecombot/adapter.py b/src/langbot/pkg/platform/adapters/wecombot/adapter.py index 579715ff9..a73a1e3a4 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/adapter.py +++ b/src/langbot/pkg/platform/adapters/wecombot/adapter.py @@ -14,6 +14,7 @@ from langbot.libs.wecom_ai_bot_api.api import WecomBotClient from langbot.libs.wecom_ai_bot_api.wecombotevent import WecomBotEvent from langbot.libs.wecom_ai_bot_api.ws_client import WecomBotWsClient from langbot.pkg.platform.adapters.wecombot.api_impl import WecomBotAPIMixin +from langbot.pkg.platform.adapters.interaction_api import InteractionAdapterMixin from langbot.pkg.platform.adapters.wecombot.event_converter import WecomBotEventConverter from langbot.pkg.platform.adapters.wecombot.message_converter import WecomBotMessageConverter from langbot.pkg.platform.adapters.wecombot.platform_api import PLATFORM_API_MAP @@ -31,7 +32,7 @@ from langbot_plugin.api.entities.builtin.platform import message as platform_mes from langbot_plugin.api.entities.builtin.platform.errors import NotSupportedError -class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter): +class WecomBotAdapter(WecomBotAPIMixin, InteractionAdapterMixin, abstract_platform_adapter.AbstractPlatformAdapter): bot: typing.Any = pydantic.Field(exclude=True) message_converter: WecomBotMessageConverter = WecomBotMessageConverter() @@ -124,7 +125,15 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo 'call_platform_api', ] if not self.config.get('enable-webhook', False) and hasattr(self.bot, 'send_template_card'): - apis.extend(['interaction.request', 'interaction.acknowledge']) + apis.extend( + [ + 'interaction.request', + 'interaction.acknowledge', + 'send_interaction', + 'update_interaction', + 'get_interaction_capabilities', + ] + ) return apis def get_interaction_capabilities(self) -> dict[str, typing.Any]: diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index eb280527d..133ac148d 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -35,7 +35,8 @@ from ..agent.runner.host_models import ( from ..agent.runner.resource_policy import ResourcePolicyProjector from ..agent.runner.platform_tools import resolve_agent_platform_tool_names from ..agent.runner.query_entry_adapter import QueryEntryAdapter -from ..agent.runner.interaction_store import InteractionAlreadySubmittedError +from ..agent.runner.interaction_store import InteractionAlreadySubmittedError, InteractionExpiredError +from langbot_plugin.api.entities.builtin.runner.interaction import InteractionHandle from ..entity.persistence import workspace as persistence_workspace from ..api.http.context import ExecutionContext, PrincipalContext, PrincipalType, RequestContext @@ -1031,12 +1032,29 @@ class RuntimeBot: ) -> None: results: list[typing.Any] = [] try: + if isinstance(event, platform_events.PlatformSpecificEvent) and event.action == 'interaction.submitted': + callback_data = event.data if isinstance(event.data, dict) else {} + event = platform_events.InteractionCallbackEvent( + callback_token=str(callback_data.get('callback_token') or ''), + values=callback_data.get('values') if isinstance(callback_data.get('values'), dict) else {}, + action_id=callback_data.get('action_id'), + actor_id=str(callback_data.get('actor_id') or callback_data.get('user_id') or '') or None, + target_type=callback_data.get('target_type'), + target_id=callback_data.get('target_id') or callback_data.get('chat_id'), + raw_submission=callback_data, + timestamp=event.timestamp, + bot_uuid=event.bot_uuid, + adapter_name=event.adapter_name, + source_platform_object=event.source_platform_object, + ) await self._record_adapter_event(event, adapter) await self._persist_monitoring_ingress(event, adapter, execution_id) primary_handler = ( self._handle_interaction_submission - if isinstance(event, platform_events.PlatformSpecificEvent) and event.action == 'interaction.submitted' + if isinstance(event, platform_events.InteractionCallbackEvent) + or isinstance(event, platform_events.PlatformSpecificEvent) + and event.action == 'interaction.submitted' else self._dispatch_eba_event_to_processor ) subscriptions = self._get_event_bindings_from_value(getattr(self.bot_entity, 'plugin_processors', [])) @@ -1141,6 +1159,28 @@ class RuntimeBot: ) -> dict[str, typing.Any]: event_type = getattr(event, 'type', None) or event.__class__.__name__ + if isinstance(event, platform_events.MessageReceivedEvent): + interaction_manager = getattr(getattr(self.ap, 'agent_run_orchestrator', None), 'interaction_manager', None) + else: + interaction_manager = None + if callable(getattr(interaction_manager, 'consume_text_message', None)): + target_type_for_text, target_id_for_text, _ = self._infer_reply_target(event) + sender_for_text = getattr(event, 'sender', None) or getattr(event, 'user', None) + actor_id_for_text = self._get_entity_id(sender_for_text) + text_for_interaction = ''.join( + part.text for part in event.message_chain if isinstance(part, platform_message.Plain) + ) + if target_type_for_text and target_id_for_text and actor_id_for_text and text_for_interaction: + consumed = await interaction_manager.consume_text_message( + workspace_id=self.execution_context.workspace_uuid, + bot_id=self.bot_entity.uuid, + conversation_id=f'{target_type_for_text}_{target_id_for_text}', + actor_id=str(actor_id_for_text), + text=text_for_interaction, + ) + if consumed: + return {'status': 'delivered', 'interaction': 'text_reply'} + if event_binding is None: event_binding = self._resolve_eba_event_binding(event, event_type) if event_binding is None: @@ -1547,12 +1587,24 @@ class RuntimeBot: async def _handle_interaction_submission( self, - event: platform_events.PlatformSpecificEvent, + event: platform_events.PlatformSpecificEvent | platform_events.InteractionCallbackEvent, adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter, execution_id: str | None = None, ) -> None: """Consume a platform interaction callback and resume its original processor.""" - data = event.data if isinstance(event.data, dict) else {} + data = ( + event.data + if isinstance(event, platform_events.PlatformSpecificEvent) and isinstance(event.data, dict) + else { + **event.raw_submission, + 'callback_token': event.callback_token, + 'action_id': event.action_id, + 'values': event.values, + 'actor_id': event.actor_id, + 'target_type': event.target_type, + 'target_id': event.target_id, + } + ) callback_token = str(data.get('callback_token') or '') actor_id = str(data.get('actor_id') or data.get('user_id') or '') or None target_type = str(data.get('target_type') or '') or None @@ -1580,6 +1632,30 @@ class RuntimeBot: # Platform retries and repeated clicks are successful no-ops. The # first callback remains the only turn allowed to resume execution. return + except InteractionExpiredError: + stale = await self.ap.agent_run_orchestrator.interaction_manager.store.get_by_callback_token(callback_token) + delivery_result = stale.get('delivery_result') if stale else None + if isinstance(delivery_result, dict) and callable(getattr(adapter, 'update_interaction', None)): + try: + handle = InteractionHandle.model_validate(delivery_result) + handle.context['callback_context'] = { + 'req_id': data.get('callback_req_id'), + 'task_id': data.get('callback_task_id'), + } + await adapter.update_interaction(handle, 'cancelled') + except Exception as exc: + await self.logger.warning(f'Failed to invalidate stale interaction card: {exc}') + return + wake_waiter = getattr(self.ap.agent_run_orchestrator.interaction_manager, 'wake_waiter', None) + if callable(wake_waiter) and wake_waiter( + record, + actor_id, + { + 'req_id': data.get('callback_req_id'), + 'task_id': data.get('callback_task_id'), + }, + ): + return if record['processor_type'] not in {'agent', 'pipeline'}: raise ValueError(f'Unsupported interaction processor type: {record["processor_type"]}') from ..telemetry import trace as trace_mod diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index 05a17cbab..6206d2ee1 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -1364,6 +1364,41 @@ class RuntimeConnectionHandler(handler.Handler): }, ) + @self.action(PluginToRuntimeAction.REQUEST_INTERACTION) + async def request_interaction(data: dict[str, Any]) -> handler.ActionResponse: + action_context, _ = await self._require_plugin_action_context() + session, error = await _validate_agent_run_session( + data.get('run_id'), data.get('caller_plugin_identity'), self.ap, 'request_interaction' + ) + if error: + return error + authorization = session['authorization'] + if authorization.get('workspace_id') != action_context.workspace_uuid: + return handler.ActionResponse.error('Run belongs to another Workspace') + if not authorization.get('interactions_enabled') or not authorization.get('interactions_authorized'): + return handler.ActionResponse.error('Runner did not declare interaction capability and permission') + bot_id = authorization.get('bot_id') + if not bot_id: + return handler.ActionResponse.error('Interaction requires a platform bot') + bot = await self.ap.platform_mgr.get_bot_by_uuid(self._execution_context(action_context), bot_id) + if bot is None: + return handler.ActionResponse.error('Interaction bot is not running') + manager = self.ap.agent_run_orchestrator.interaction_manager + try: + result = await manager.request_interaction( + session=session, + adapter=bot.adapter, + request_payload=data['request'], + timeout=float(data['timeout']), + on_unsupported=data.get('on_unsupported', 'error'), + responders=data.get('responders', 'actor'), + target_type=data.get('target_type'), + target_id=data.get('target_id'), + ) + return handler.ActionResponse.success(data=result) + except (ValueError, KeyError, TypeError, pydantic.ValidationError) as exc: + return handler.ActionResponse.error(str(exc)) + @self.action(PluginToRuntimeAction.GET_LLM_MODELS) async def get_llm_models(data: dict[str, Any]) -> handler.ActionResponse: """Get llm models, returns list of UUID strings""" diff --git a/tests/unit_tests/agent/test_interaction_manager.py b/tests/unit_tests/agent/test_interaction_manager.py index f17e16678..37aa1345e 100644 --- a/tests/unit_tests/agent/test_interaction_manager.py +++ b/tests/unit_tests/agent/test_interaction_manager.py @@ -3,6 +3,7 @@ from __future__ import annotations from types import SimpleNamespace +import asyncio import time import pytest @@ -14,6 +15,7 @@ from langbot_plugin.api.entities.builtin.runner.event import ActorContext from langbot_plugin.api.entities.builtin.runner.input import AgentInput from langbot_plugin.api.entities.builtin.runner.interaction import ( InteractionDeliveryCapabilities, + InteractionHandle, InteractionSubmission, ) @@ -765,3 +767,295 @@ async def test_callback_rejects_option_not_present_in_request(store, event, bind conversation_id='group_chat-1', actor_id='user-1', ) + + +class BlockingAdapter: + def __init__(self): + self.sent = asyncio.Event() + self.token = None + self.updates = [] + + def get_interaction_capabilities(self): + return InteractionDeliveryCapabilities( + action_styles=['default', 'primary', 'danger'], + max_actions=6, + supports_updates=True, + ) + + async def send_interaction(self, target_type, target_id, request, callback_token, reply_to=None): + self.token = callback_token + self.sent.set() + return InteractionHandle(message_id='card-1') + + async def update_interaction(self, handle, state, submission=None): + self.updates.append(state) + return handle + + +class TextAdapter: + def __init__(self): + self.messages = [] + self.sent = asyncio.Event() + + def get_interaction_capabilities(self): + return InteractionDeliveryCapabilities() + + async def send_message(self, target_type, target_id, message): + self.messages.append(str(message)) + self.sent.set() + + +class BlockingStore: + def __init__(self): + self.record = None + self.token = 'test-token' + + async def create_request(self, **data): + self.record = { + **data, + 'status': 'pending', + 'submission': None, + } + return self.record, self.token + + async def record_delivery_success(self, run_id, interaction_id, delivery_result): + self.record['delivery_result'] = delivery_result + + async def get_by_callback_token(self, token): + return self.record if token == self.token else None + + async def consume_submission(self, *, callback_token, submission, bot_id, conversation_id, actor_id, **kwargs): + if self.record['status'] != 'pending': + raise ValueError('Interaction was already submitted') + if self.record['actor_id'] and self.record['actor_id'] != actor_id: + raise ValueError('Actor mismatch') + self.record['status'] = 'submitted' + self.record['submission'] = submission + return self.record + + async def mark_terminal(self, run_id, interaction_id, status): + self.record['status'] = status + return True + + async def mark_delivery_failed(self, run_id, interaction_id, reason): + self.record['status'] = 'delivery_failed' + return True + + +def _blocking_session(): + return { + 'run_id': 'blocking-run', + 'runner_id': 'plugin:test/ApprovalRunner/default', + 'runtime_context': {'deadline_at': time.time() + 300}, + 'authorization': { + 'binding_id': 'binding-1', + 'processor_type': 'agent', + 'processor_id': 'agent-1', + 'bot_id': 'bot-1', + 'workspace_id': 'workspace-1', + 'thread_id': None, + 'interactions_enabled': True, + 'interactions_authorized': True, + 'platform_context': { + 'actor': {'actor_id': 'user-1'}, + 'delivery': {'reply_target': {'target_type': 'group', 'target_id': 'chat-1'}}, + }, + }, + } + + +@pytest.mark.asyncio +async def test_blocking_request_returns_submission_in_same_run(): + adapter = BlockingAdapter() + store = BlockingStore() + manager = InteractionManager(SimpleNamespace(), store=store) + session = _blocking_session() + original_deadline = session['runtime_context']['deadline_at'] + task = asyncio.create_task( + manager.request_interaction( + session=session, + adapter=adapter, + request_payload={ + 'interaction_id': 'approval', + 'title': 'Approve?', + 'actions': [ + {'id': 'yes', 'label': 'Yes'}, + {'id': 'no', 'label': 'No'}, + ], + }, + timeout=1, + on_unsupported='error', + responders='actor', + ) + ) + await asyncio.wait_for(adapter.sent.wait(), 1) + record = await manager.consume_callback( + callback_token=adapter.token, + submission={'action_ref': 0}, + bot_id='bot-1', + conversation_id='group_chat-1', + actor_id='user-1', + ) + assert manager.wake_waiter(record, 'user-1') + result = await task + assert result['status'] == 'submitted' + assert result['submission']['action_id'] == 'yes' + assert adapter.updates == ['submitted'] + assert original_deadline <= result['deadline_at'] < original_deadline + 1 + with pytest.raises(Exception): + await manager.consume_callback( + callback_token=adapter.token, + submission={'action_ref': 1}, + bot_id='bot-1', + conversation_id='group_chat-1', + actor_id='user-1', + ) + + +@pytest.mark.asyncio +async def test_blocking_request_times_out_without_polling(): + adapter = BlockingAdapter() + store = BlockingStore() + manager = InteractionManager(SimpleNamespace(), store=store) + result = await manager.request_interaction( + session=_blocking_session(), + adapter=adapter, + request_payload={ + 'interaction_id': 'timeout', + 'title': 'Approve?', + 'actions': [ + {'id': 'yes', 'label': 'Yes'}, + ], + }, + timeout=0.02, + on_unsupported='error', + responders='actor', + ) + assert result['status'] == 'timeout' + assert adapter.updates == ['expired'] + record = await store.get_by_callback_token(adapter.token) + assert record['status'] == 'expired' + + +@pytest.mark.asyncio +async def test_blocking_request_cancel_invalidates_card(): + adapter = BlockingAdapter() + store = BlockingStore() + manager = InteractionManager(SimpleNamespace(), store=store) + task = asyncio.create_task( + manager.request_interaction( + session=_blocking_session(), + adapter=adapter, + request_payload={ + 'interaction_id': 'cancel', + 'title': 'Approve?', + 'actions': [ + {'id': 'yes', 'label': 'Yes'}, + ], + }, + timeout=1, + on_unsupported='error', + responders='actor', + ) + ) + await asyncio.wait_for(adapter.sent.wait(), 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert store.record['status'] == 'cancelled' + assert adapter.updates == ['cancelled'] + assert not manager._waiters + + +@pytest.mark.asyncio +async def test_blocking_request_rejects_unlisted_responder_before_consuming(): + adapter = BlockingAdapter() + store = BlockingStore() + manager = InteractionManager(SimpleNamespace(), store=store) + task = asyncio.create_task( + manager.request_interaction( + session=_blocking_session(), + adapter=adapter, + request_payload={ + 'interaction_id': 'restricted', + 'title': 'Approve?', + 'actions': [ + {'id': 'yes', 'label': 'Yes'}, + ], + }, + timeout=1, + on_unsupported='error', + responders=['user-1'], + ) + ) + await asyncio.wait_for(adapter.sent.wait(), 1) + with pytest.raises(ValueError, match='allowed responder'): + await manager.consume_callback( + callback_token=adapter.token, + submission={'action_ref': 0}, + bot_id='bot-1', + conversation_id='group_chat-1', + actor_id='user-2', + ) + assert store.record['status'] == 'pending' + record = await manager.consume_callback( + callback_token=adapter.token, + submission={'action_ref': 0}, + bot_id='bot-1', + conversation_id='group_chat-1', + actor_id='user-1', + ) + assert manager.wake_waiter(record, 'user-1') + assert (await task)['status'] == 'submitted' + + +@pytest.mark.asyncio +async def test_text_fallback_consumes_numbered_action(): + adapter = TextAdapter() + store = BlockingStore() + manager = InteractionManager(SimpleNamespace(), store=store) + task = asyncio.create_task( + manager.request_interaction( + session=_blocking_session(), + adapter=adapter, + request_payload={ + 'interaction_id': 'text-choice', + 'title': 'Approve?', + 'actions': [ + {'id': 'yes', 'label': 'Yes'}, + {'id': 'no', 'label': 'No'}, + ], + }, + timeout=1, + on_unsupported='degrade', + responders='actor', + ) + ) + await asyncio.wait_for(adapter.sent.wait(), 1) + assert '1 Yes / 2 No' in adapter.messages[0] + assert await manager.consume_text_message( + workspace_id='workspace-1', + bot_id='bot-1', + conversation_id='group_chat-1', + actor_id='user-1', + text='2', + ) + result = await task + assert result['submission']['action_id'] == 'no' + assert result['degraded'][-1]['applied'] == 'text_fallback' + + +@pytest.mark.asyncio +async def test_unsupported_error_does_not_persist_request(): + store = BlockingStore() + manager = InteractionManager(SimpleNamespace(), store=store) + result = await manager.request_interaction( + session=_blocking_session(), + adapter=TextAdapter(), + request_payload={'interaction_id': 'error', 'title': 'Approve?', 'actions': [{'id': 'yes', 'label': 'Yes'}]}, + timeout=1, + on_unsupported='error', + responders='actor', + ) + assert result['unsupported'][0]['path'] == 'platform' + assert store.record is None diff --git a/tests/unit_tests/agent/test_invoker_interaction_deadline.py b/tests/unit_tests/agent/test_invoker_interaction_deadline.py new file mode 100644 index 000000000..ec79b3919 --- /dev/null +++ b/tests/unit_tests/agent/test_invoker_interaction_deadline.py @@ -0,0 +1,34 @@ +"""Deadline behavior while a Runner blocks on a Host interaction.""" + +import asyncio +import time +from types import SimpleNamespace + +import pytest + +from langbot.pkg.agent.runner.invoker import RunnerInvoker + + +@pytest.mark.asyncio +async def test_next_result_survives_interaction_deadline_extension(): + gate = asyncio.Event() + context = {'runtime': {'deadline_at': time.time() + 0.02}} + + async def results(): + await gate.wait() + yield {'type': 'run.completed'} + + reader = RunnerInvoker(SimpleNamespace()) + task = asyncio.create_task( + reader._next_with_deadline( + results(), + SimpleNamespace(id='test'), + context, + ) + ) + await asyncio.sleep(0.01) + context['runtime']['deadline_at'] = time.time() + 0.1 + await asyncio.sleep(0.03) + gate.set() + + assert (await task)['type'] == 'run.completed' diff --git a/tests/unit_tests/agent/test_platform_tools.py b/tests/unit_tests/agent/test_platform_tools.py index d2dad499a..3a9fbfa47 100644 --- a/tests/unit_tests/agent/test_platform_tools.py +++ b/tests/unit_tests/agent/test_platform_tools.py @@ -395,11 +395,11 @@ async def test_event_reply_uses_run_reply_stream_when_available() -> None: @pytest.mark.asyncio -async def test_present_interactive_card_uses_host_interaction_manager() -> None: +async def test_request_interaction_uses_host_interaction_manager() -> None: adapter = SimpleNamespace(get_supported_apis=lambda: ['interaction.request']) manager = SimpleNamespace( - present_tool_request=AsyncMock( - return_value={'ok': True, 'status': 'pending', 'interaction_id': 'card-host-generated'} + request_interaction=AsyncMock( + return_value={'status': 'submitted', 'submission': {'interaction_id': 'card-host-generated'}} ) ) ap = SimpleNamespace( @@ -439,13 +439,14 @@ async def test_present_interactive_card_uses_host_interaction_manager() -> None: }, ) - assert result == {'ok': True, 'status': 'pending', 'interaction_id': 'card-host-generated'} - payload = manager.present_tool_request.await_args.kwargs['request_payload'] + assert result == {'status': 'submitted', 'submission': {'interaction_id': 'card-host-generated'}} + payload = manager.request_interaction.await_args.kwargs['request_payload'] assert payload['interaction_id'].startswith('card-') assert payload['fallback_text'].startswith('Choose deployment environment') assert 'callback_token' not in payload - assert manager.present_tool_request.await_args.kwargs['session'] is session - assert manager.present_tool_request.await_args.kwargs['adapter'] is adapter + assert manager.request_interaction.await_args.kwargs['session'] is session + assert manager.request_interaction.await_args.kwargs['adapter'] is adapter + assert manager.request_interaction.await_args.kwargs['on_unsupported'] == 'degrade' with pytest.raises(ValueError, match='cannot combine fields and actions'): await execute_platform_tool( diff --git a/tests/unit_tests/agent/test_resource_builder.py b/tests/unit_tests/agent/test_resource_builder.py index 13cfdcda3..ae8c56971 100644 --- a/tests/unit_tests/agent/test_resource_builder.py +++ b/tests/unit_tests/agent/test_resource_builder.py @@ -262,11 +262,14 @@ async def test_pipeline_receives_interactive_card_tool_for_supported_adapter(app resources = await build_resources( app, query, - make_descriptor(capabilities={'tool_calling': True}), + make_descriptor( + capabilities={'tool_calling': True, 'interactions': True}, + permissions={**FULL_PERMISSIONS, 'interactions': ['request']}, + ), ) - assert [tool['tool_name'] for tool in resources['tools']] == ['present_interactive_card'] - assert resources['platform_capabilities']['authorized_tools'] == ['present_interactive_card'] + assert [tool['tool_name'] for tool in resources['tools']] == ['request_interaction'] + assert resources['platform_capabilities']['authorized_tools'] == ['request_interaction'] @pytest.mark.asyncio @@ -288,7 +291,7 @@ async def test_event_processor_cannot_receive_interactive_card_tool(app): scope=BindingScope(scope_type='global'), runner_id=RUNNER_ID, processor_type='event_processor', - resource_policy=ResourcePolicy(allowed_platform_tool_names=['present_interactive_card']), + resource_policy=ResourcePolicy(allowed_platform_tool_names=['request_interaction']), delivery_policy=DeliveryPolicy(enable_interactions=True), )