fix(agent): enable reliable cross-platform card interactions

This commit is contained in:
fdc310
2026-10-27 16:03:11 +08:00
parent 71b5b70633
commit de565d0bcc
11 changed files with 271 additions and 39 deletions
@@ -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
@@ -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(
@@ -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',
@@ -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
]
@@ -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()
@@ -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}
+13 -7
View File
@@ -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
@@ -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)
@@ -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."""
@@ -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(
@@ -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'))