feat(interactions): resume runner in place after card submission

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):

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,

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

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)

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

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(

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:

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)

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()

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,

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(

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:

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
             ]

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:

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

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]:

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,

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]:

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

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]:

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]:

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]:

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]:

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

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"""

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

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'

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(

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),
     )