From de565d0bccbbf3d148b8f0835552037a061426ea Mon Sep 17 00:00:00 2001 From: fdc310 <2213070223@qq.com> Date: Tue, 27 Oct 2026 16:03:11 +0800 Subject: [PATCH] fix(agent): enable reliable cross-platform card interactions --- .../pkg/agent/runner/interaction_manager.py | 43 +++++++++++-- .../pkg/agent/runner/platform_tools.py | 2 +- .../pkg/agent/runner/query_entry_adapter.py | 18 ++++-- .../pkg/agent/runner/resource_builder.py | 2 +- .../platform/adapters/qqofficial/adapter.py | 16 +++++ .../adapters/qqofficial/interaction.py | 14 +++- src/langbot/pkg/platform/botmgr.py | 20 ++++-- .../agent/test_interaction_manager.py | 55 ++++++++++++++++ .../unit_tests/agent/test_resource_builder.py | 64 ++++++++++++++++++- .../platform/test_qqofficial_eba_adapter.py | 63 +++++++++++++----- .../unit_tests/platform/test_routing_rules.py | 13 ++++ 11 files changed, 271 insertions(+), 39 deletions(-) diff --git a/src/langbot/pkg/agent/runner/interaction_manager.py b/src/langbot/pkg/agent/runner/interaction_manager.py index 54b024283..7553c3450 100644 --- a/src/langbot/pkg/agent/runner/interaction_manager.py +++ b/src/langbot/pkg/agent/runner/interaction_manager.py @@ -218,10 +218,11 @@ class InteractionManager: authorization = session.get('authorization') if isinstance(session, typing.Mapping) else None if not isinstance(authorization, dict): raise ValueError('Interactive card run authorization is unavailable') - if authorization.get('processor_type') != 'agent': - raise ValueError('Interactive card tools are currently available only to independent Agents') + processor_type = authorization.get('processor_type') + if processor_type not in {'agent', 'pipeline'}: + raise ValueError('Interactive card tools are unavailable for this processor type') if authorization.get('interactions_enabled') is not True: - raise ValueError('Interactive cards are disabled for this Agent binding') + raise ValueError('Interactive cards are disabled for this processor binding') run_id = str(session.get('run_id') or '').strip() runner_id = str(session.get('runner_id') or '').strip() @@ -256,13 +257,18 @@ class InteractionManager: raise ValueError('Interactive card has no authorized reply target') callback_conversation_id = f'{target_type}_{target_id}' actor = platform_context.get('actor') if isinstance(platform_context.get('actor'), dict) else {} + pipeline_admission = self._pipeline_tool_admission( + session=session, + authorization=authorization, + runner_id=runner_id, + ) _, callback_token = await self.store.create_request( interaction_id=request.interaction_id, run_id=run_id, binding_id=binding_id, runner_id=runner_id, - processor_type='agent', + processor_type=processor_type, processor_id=processor_id, request=request_data, delivery_target=reply_target, @@ -272,6 +278,7 @@ class InteractionManager: thread_id=authorization.get('thread_id'), actor_id=actor.get('actor_id'), expires_at=expires_at, + **pipeline_admission, ) try: if not self._supports_interaction_delivery(adapter): @@ -348,6 +355,34 @@ class InteractionManager: return {'expected_config': expected, 'authority_check': authority} + @staticmethod + def _pipeline_tool_admission(session, authorization, runner_id): + if authorization.get('processor_type') != 'pipeline': + return {} + + query = session.get('execution_query') + expected = copy.deepcopy(getattr(query, 'pipeline_config', None)) + query_session = getattr(query, 'session', None) + conversation = getattr(query_session, 'using_conversation', None) + processor_id = str(authorization.get('processor_id') or '').strip() + + def authority(): + from .config_resolver import RunnerConfigResolver + + current_config = getattr(query, 'pipeline_config', None) + return ( + query is not None + and query_session is not None + and conversation is not None + and isinstance(expected, dict) + and query_session.using_conversation is conversation + and getattr(query, 'pipeline_uuid', None) == processor_id + and current_config == expected + and RunnerConfigResolver.resolve_runner_id(expected) == runner_id + ) + + return {'expected_config': expected, 'authority_check': authority} + async def restore_legacy_identity(self, event: AgentEventEnvelope, binding: AgentBinding) -> None: """Resume only from the scoped, consumed Host record, never callback fields.""" event.legacy_identity = None diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index 6f7a98527..b75b4009b 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -100,7 +100,7 @@ PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = ( '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 Agent.' + 'The user submission will start a new interaction.submitted turn for this processor.' ), }, _object_schema( diff --git a/src/langbot/pkg/agent/runner/query_entry_adapter.py b/src/langbot/pkg/agent/runner/query_entry_adapter.py index a3ecbfe16..cdc4051ab 100644 --- a/src/langbot/pkg/agent/runner/query_entry_adapter.py +++ b/src/langbot/pkg/agent/runner/query_entry_adapter.py @@ -31,6 +31,7 @@ from .host_models import ( ) from .config_resolver import RunnerConfigResolver from .resource_policy import ResourcePolicyProjector +from .platform_tools import PRESENT_INTERACTIVE_CARD_TOOL, resolve_agent_platform_tool_names from . import events as runner_events from ...pipeline.pool import get_query_execution_context from ...provider.tools.toolmgr import TOOL_SOURCE_REFS_QUERY_KEY @@ -131,6 +132,15 @@ class QueryEntryAdapter: pipeline_config = query.pipeline_config or {} runner_config = RunnerConfigResolver.resolve_runner_config(pipeline_config, runner_id) agent_id = getattr(query, 'pipeline_uuid', None) + variables = getattr(query, 'variables', None) or {} + event_type = ( + runner_events.INTERACTION_SUBMITTED + if isinstance(variables.get('_interaction_submission'), dict) + else runner_events.MESSAGE_RECEIVED + ) + platform_tool_names = [ + name for name in resolve_agent_platform_tool_names({}, event_type) if name == PRESENT_INTERACTIVE_CARD_TOOL + ] resource_policy = ResourcePolicyProjector.from_runner_config( runner_config, @@ -139,6 +149,7 @@ class QueryEntryAdapter: resolved_tool_sources=cls._extract_allowed_tool_sources(query), resolved_kb_uuids=cls._extract_allowed_kbs(query), resolved_skill_names=cls._extract_allowed_skills(query), + allowed_platform_tool_names=platform_tool_names, ) # Build state policy @@ -153,13 +164,6 @@ class QueryEntryAdapter: enable_reply=True, enable_interactions=True, ) - variables = getattr(query, 'variables', None) or {} - event_type = ( - runner_events.INTERACTION_SUBMITTED - if isinstance(variables.get('_interaction_submission'), dict) - else runner_events.MESSAGE_RECEIVED - ) - return AgentConfig( agent_id=agent_id, processor_type='pipeline', diff --git a/src/langbot/pkg/agent/runner/resource_builder.py b/src/langbot/pkg/agent/runner/resource_builder.py index 8a4dc457f..7cd8de0ed 100644 --- a/src/langbot/pkg/agent/runner/resource_builder.py +++ b/src/langbot/pkg/agent/runner/resource_builder.py @@ -91,7 +91,7 @@ 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 != 'agent' or not binding.delivery_policy.enable_interactions: + if binding.processor_type not in {'agent', 'pipeline'} or not binding.delivery_policy.enable_interactions: selected_platform_tools = [ name for name in selected_platform_tools if name != PRESENT_INTERACTIVE_CARD_TOOL ] diff --git a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py index 7703a5935..ce25f1aee 100644 --- a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py +++ b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py @@ -56,9 +56,11 @@ class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPl _stream_ctx_ts: dict[str, float] = {} _fallback_text: dict[str, str] = {} _fallback_text_ts: dict[str, float] = {} + _anchor_msg_seq: dict[str, int] = {} _ws_task: asyncio.Task | None = None _STREAM_CTX_TTL = 300 + _MAX_REPLIES_PER_ANCHOR = 5 class Config: arbitrary_types_allowed = True @@ -93,6 +95,7 @@ class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPl _stream_ctx_ts={}, _fallback_text={}, _fallback_text_ts={}, + _anchor_msg_seq={}, _ws_task=None, ) self._register_native_handlers() @@ -128,6 +131,18 @@ class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPl def get_interaction_capabilities(self) -> dict[str, typing.Any]: return interaction_delivery_capabilities() + def next_interaction_msg_seq(self, anchor: str) -> int | None: + """Allocate QQ's per-anchor passive reply sequence for interaction cards.""" + if not anchor: + return 1 + used = self._anchor_msg_seq.get(anchor, 0) + if used >= self._MAX_REPLIES_PER_ANCHOR: + return None + self._anchor_msg_seq[anchor] = used + 1 + while len(self._anchor_msg_seq) > 4096: + self._anchor_msg_seq.pop(next(iter(self._anchor_msg_seq)), None) + return used + 1 + @staticmethod def _plain_message(text: str) -> platform_message.MessageChain: return platform_message.MessageChain([platform_message.Plain(text=text)]) @@ -242,6 +257,7 @@ class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPl pass self._ws_task = None await self.bot.close() + self._anchor_msg_seq.clear() self._stream_ctx.clear() self._stream_ctx_ts.clear() self._fallback_text.clear() diff --git a/src/langbot/pkg/platform/adapters/qqofficial/interaction.py b/src/langbot/pkg/platform/adapters/qqofficial/interaction.py index be28492fa..2223cfe64 100644 --- a/src/langbot/pkg/platform/adapters/qqofficial/interaction.py +++ b/src/langbot/pkg/platform/adapters/qqofficial/interaction.py @@ -105,12 +105,24 @@ async def send_interaction(adapter: typing.Any, params: dict[str, typing.Any]) - adapter._plain_message(_text(request, False)), ) return {'ok': True, 'message_id': result.message_id, 'rich': False} + msg_id = reply_target.get('message_id') + if msg_id in (None, '', -1, '-1'): + msg_id = None + event_id = reply_target.get('event_id') + if event_id in (None, '', -1, '-1'): + event_id = None + anchor = str(msg_id or event_id or '') + msg_seq = adapter.next_interaction_msg_seq(anchor) + if msg_seq is None: + raise ValueError('QQ interaction reply limit reached for this message') raw = await adapter.bot.send_markdown_keyboard( target_type='c2c' if target_type == 'person' else 'group', target_id=target_id, markdown_content=_text(request, True), keyboard=keyboard, - msg_id=reply_target.get('message_id'), + msg_id=msg_id, + event_id=event_id, + msg_seq=msg_seq, ) return {'ok': True, 'message_id': raw.get('id') if isinstance(raw, dict) else None, 'rich': True} diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index cd3e150fe..f7ae30188 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -35,6 +35,7 @@ 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 ..entity.persistence import workspace as persistence_workspace from ..api.http.context import ExecutionContext, PrincipalContext, PrincipalType, RequestContext @@ -1512,13 +1513,18 @@ class RuntimeBot: if data.get(ref_name) is not None: submission[ref_name] = data[ref_name] - record = await self.ap.agent_run_orchestrator.interaction_manager.consume_callback( - callback_token=callback_token, - submission=submission, - bot_id=self.bot_entity.uuid, - conversation_id=conversation_id, - actor_id=actor_id, - ) + try: + record = await self.ap.agent_run_orchestrator.interaction_manager.consume_callback( + callback_token=callback_token, + submission=submission, + bot_id=self.bot_entity.uuid, + conversation_id=conversation_id, + actor_id=actor_id, + ) + except InteractionAlreadySubmittedError: + # Platform retries and repeated clicks are successful no-ops. The + # first callback remains the only turn allowed to resume execution. + return if record['processor_type'] not in {'agent', 'pipeline'}: raise ValueError(f'Unsupported interaction processor type: {record["processor_type"]}') from ..telemetry import trace as trace_mod diff --git a/tests/unit_tests/agent/test_interaction_manager.py b/tests/unit_tests/agent/test_interaction_manager.py index 06eb8e516..13ca440f8 100644 --- a/tests/unit_tests/agent/test_interaction_manager.py +++ b/tests/unit_tests/agent/test_interaction_manager.py @@ -250,6 +250,61 @@ async def test_agent_tool_card_uses_host_token_and_persists_resume_target(store, assert record['actor_id'] == 'user-1' +@pytest.mark.asyncio +async def test_pipeline_tool_card_preserves_pipeline_authority(store, event): + adapter = FakeAdapter(supports_interactions=True) + manager = InteractionManager(SimpleNamespace(), store=store) + event.delivery.interactions = InteractionDeliveryCapabilities( + field_types=['select'], + action_styles=['default', 'primary', 'danger'], + supports_updates=True, + max_fields=1, + ) + conversation = SimpleNamespace(pipeline_uuid='pipeline-1') + execution_query = SimpleNamespace( + pipeline_uuid='pipeline-1', + pipeline_config=CONFIG, + session=SimpleNamespace(using_conversation=conversation), + ) + session = { + 'run_id': 'pipeline-tool-run-1', + 'runner_id': 'plugin:test/ApprovalRunner/default', + 'execution_query': execution_query, + 'authorization': { + 'binding_id': 'pipeline-binding-1', + 'processor_type': 'pipeline', + 'processor_id': 'pipeline-1', + 'interactions_enabled': True, + 'bot_id': 'bot-1', + 'workspace_id': 'workspace-1', + 'thread_id': None, + 'platform_context': { + 'actor': {'actor_type': 'user', 'actor_id': 'user-1'}, + 'delivery': event.delivery.model_dump(mode='json'), + }, + }, + } + request = { + 'interaction_id': 'pipeline-card-host-generated', + 'kind': 'choice', + 'title': 'Continue?', + 'actions': [ + {'id': 'confirm', 'label': 'Confirm', 'style': 'primary'}, + {'id': 'cancel', 'label': 'Cancel', 'style': 'default'}, + ], + 'fallback_text': 'Reply confirm or cancel.', + } + + result = await manager.present_tool_request(request_payload=request, session=session, adapter=adapter) + + assert result['status'] == 'pending' + record = await store.get_request('pipeline-tool-run-1', 'pipeline-card-host-generated') + assert record is not None + assert record['processor_type'] == 'pipeline' + assert record['processor_id'] == 'pipeline-1' + assert record['binding_id'] == 'pipeline-binding-1' + + @pytest.mark.asyncio async def test_continuous_interaction_reuses_submitted_platform_presentation(store, event, binding): adapter = FakeAdapter(supports_interactions=True) diff --git a/tests/unit_tests/agent/test_resource_builder.py b/tests/unit_tests/agent/test_resource_builder.py index 3273a6293..13cfdcda3 100644 --- a/tests/unit_tests/agent/test_resource_builder.py +++ b/tests/unit_tests/agent/test_resource_builder.py @@ -12,7 +12,13 @@ from langbot.pkg.agent.runner.descriptor import RunnerDescriptor from langbot.pkg.agent.runner.binding_resolver import AgentBindingResolver from langbot.pkg.agent.runner.query_entry_adapter import QueryEntryAdapter from langbot.pkg.agent.runner.resource_builder import AgentResourceBuilder -from langbot.pkg.agent.runner.host_models import AgentBinding, AgentEventEnvelope, BindingScope, ResourcePolicy +from langbot.pkg.agent.runner.host_models import ( + AgentBinding, + AgentEventEnvelope, + BindingScope, + DeliveryPolicy, + ResourcePolicy, +) from langbot.pkg.api.http.context import ExecutionContext from langbot.pkg.provider.tools.toolmgr import ToolManager @@ -240,6 +246,62 @@ async def test_platform_tools_are_not_claimed_when_runner_disables_tool_calling( ] +@pytest.mark.asyncio +async def test_pipeline_receives_interactive_card_tool_for_supported_adapter(app): + query = make_query({}) + query.adapter = SimpleNamespace( + get_supported_apis=lambda: ['interaction.request'], + get_interaction_capabilities=lambda: { + 'field_types': ['select'], + 'action_styles': ['default', 'primary', 'danger'], + 'supports_updates': True, + 'max_fields': 10, + }, + ) + + resources = await build_resources( + app, + query, + make_descriptor(capabilities={'tool_calling': True}), + ) + + assert [tool['tool_name'] for tool in resources['tools']] == ['present_interactive_card'] + assert resources['platform_capabilities']['authorized_tools'] == ['present_interactive_card'] + + +@pytest.mark.asyncio +async def test_event_processor_cannot_receive_interactive_card_tool(app): + event = AgentEventEnvelope( + event_id='event-processor-card', + event_type='message.received', + source='platform', + bot_id='bot-1', + input=AgentInput(text='hello'), + delivery=DeliveryContext( + surface='platform', + reply_target={'target_type': 'person', 'target_id': 'user-1'}, + platform_capabilities={'supported_apis': ['interaction.request']}, + ), + ) + binding = AgentBinding( + binding_id='event-processor-binding', + scope=BindingScope(scope_type='global'), + runner_id=RUNNER_ID, + processor_type='event_processor', + resource_policy=ResourcePolicy(allowed_platform_tool_names=['present_interactive_card']), + delivery_policy=DeliveryPolicy(enable_interactions=True), + ) + + resources = await AgentResourceBuilder(app).build_resources_from_binding( + execution_context=TEST_CONTEXT, + event=event, + binding=binding, + descriptor=make_descriptor(capabilities={'tool_calling': True}), + ) + + assert resources['tools'] == [] + + @pytest.mark.asyncio async def test_build_models_authorizes_rerank_and_llm_refs_from_config(app): """Config-selected model references are projected regardless of method granularity.""" diff --git a/tests/unit_tests/platform/test_qqofficial_eba_adapter.py b/tests/unit_tests/platform/test_qqofficial_eba_adapter.py index a5aee2934..cd29fe3b6 100644 --- a/tests/unit_tests/platform/test_qqofficial_eba_adapter.py +++ b/tests/unit_tests/platform/test_qqofficial_eba_adapter.py @@ -191,26 +191,27 @@ def test_qqofficial_interaction_keyboard_contains_only_host_token_and_index(): @pytest.mark.asyncio async def test_qqofficial_interaction_delivery_and_callback_event(): adapter = make_adapter() - result = await adapter.call_platform_api( - 'interaction.request', - { - 'callback_token': 'callback-token', - 'reply_target': { - 'target_type': 'group', - 'target_id': 'group-openid', - 'message_id': 'msg-1', - }, - 'request': { - 'interaction_id': 'form-1', - 'title': 'Approve?', - 'actions': [{'id': 'approve', 'label': 'Approve'}], - 'fallback_text': 'Reply approve.', - }, + params = { + 'callback_token': 'callback-token', + 'reply_target': { + 'target_type': 'group', + 'target_id': 'group-openid', + 'message_id': 'msg-1', }, - ) + 'request': { + 'interaction_id': 'form-1', + 'title': 'Approve?', + 'actions': [{'id': 'approve', 'label': 'Approve'}], + 'fallback_text': 'Reply approve.', + }, + } + result = await adapter.call_platform_api('interaction.request', params) + await adapter.call_platform_api('interaction.request', params) assert result == {'ok': True, 'message_id': 'keyboard-message', 'rich': True} - assert adapter.bot.send_markdown_keyboard.await_args.kwargs['msg_id'] == 'msg-1' + calls = adapter.bot.send_markdown_keyboard.await_args_list + assert [call.kwargs['msg_id'] for call in calls] == ['msg-1', 'msg-1'] + assert [call.kwargs['msg_seq'] for call in calls] == [1, 2] event = interaction_event_from_payload( { @@ -226,6 +227,34 @@ async def test_qqofficial_interaction_delivery_and_callback_event(): assert event.data['actor_id'] == 'user-openid' +@pytest.mark.asyncio +async def test_qqofficial_interaction_omits_synthetic_negative_message_id(): + adapter = make_adapter() + + await adapter.call_platform_api( + 'interaction.request', + { + 'callback_token': 'callback-token', + 'reply_target': { + 'target_type': 'person', + 'target_id': 'user-openid', + 'message_id': -1, + }, + 'request': { + 'interaction_id': 'form-1', + 'title': 'Approve?', + 'actions': [{'id': 'approve', 'label': 'Approve'}], + 'fallback_text': 'Reply approve.', + }, + }, + ) + + call = adapter.bot.send_markdown_keyboard.await_args + assert call.kwargs['msg_id'] is None + assert call.kwargs['event_id'] is None + assert call.kwargs['msg_seq'] == 1 + + @pytest.mark.asyncio async def test_qqofficial_message_converter_maps_common_components_to_send_payloads(): payload = await QQOfficialMessageConverter.yiri2target( diff --git a/tests/unit_tests/platform/test_routing_rules.py b/tests/unit_tests/platform/test_routing_rules.py index 30455643d..48951c08b 100644 --- a/tests/unit_tests/platform/test_routing_rules.py +++ b/tests/unit_tests/platform/test_routing_rules.py @@ -605,6 +605,19 @@ class TestInteractionResumeRouting: assert submission['action_ref'] == 1 assert submission['interaction_id'] is None + @pytest.mark.asyncio + async def test_replayed_callback_is_an_idempotent_noop(self): + from langbot.pkg.agent.runner.interaction_store import InteractionAlreadySubmittedError + + bot = self._make_bot(self._record('pipeline')) + manager = bot.ap.agent_run_orchestrator.interaction_manager + manager.consume_callback.side_effect = InteractionAlreadySubmittedError('Interaction was already submitted') + + await bot._handle_interaction_submission(self._event(), SimpleNamespace()) + + manager.acknowledge_submission.assert_not_awaited() + bot.ap.msg_aggregator.add_message.assert_not_awaited() + @pytest.mark.asyncio async def test_pipeline_callback_rejects_changed_runner(self): bot = self._make_bot(self._record('pipeline'))