Fix DingTalk A2UI form submissions and agent replies

This commit is contained in:
fdc310
2026-10-28 02:18:17 +08:00
parent d68b371734
commit 21307f9b5d
8 changed files with 382 additions and 54 deletions
+78 -22
View File
@@ -257,11 +257,17 @@ class AgentRunOrchestrator:
auto_reply_enabled = (
binding.processor_type == 'agent'
and binding.delivery_policy.enable_reply
and event.event_type == 'message.received'
and event.event_type in {'message.received', 'interaction.submitted'}
and reply_streams.adapter is not None
)
auto_reply_stream_id = uuid4()
explicit_delivery_completed = False
auto_reply_stream_text = ''
auto_reply_stream_open = False
auto_reply_cumulative_text = ''
auto_reply_prefix = ''
auto_reply_deferred = False
explicit_reply_text: str | None = None
pending_explicit_replies: dict[str, str] = {}
# A run reached without a platform ingress (WebUI debug, service API)
# owns its own chain; a run inside an ingress reuses that chain.
trace_binding: trace_mod.TraceBinding | None = None
@@ -370,14 +376,38 @@ class AgentRunOrchestrator:
):
continue
if result_type == 'tool.call.completed':
if result_type == 'tool.call.started':
result_data = result_dict.get('data') or {}
tool_definition = PLATFORM_TOOLS_BY_NAME.get(result_data.get('tool_name'))
if tool_definition is not None and tool_definition.api in {
'send_message',
'interaction.request',
}:
explicit_delivery_completed = True
parameters = result_data.get('parameters') or {}
if tool_definition is not None and tool_definition.api == 'send_message':
text = parameters.get('text') if isinstance(parameters, dict) else None
if isinstance(text, str):
pending_explicit_replies[str(result_data.get('tool_call_id') or '')] = text
if result_type == 'tool.call.completed' and auto_reply_enabled:
if auto_reply_stream_open:
await reply_streams.apply(
ReplyStreamRequest(
stream_id=auto_reply_stream_id,
operation='finish',
text=auto_reply_stream_text,
)
)
auto_reply_prefix = auto_reply_cumulative_text
auto_reply_stream_id = uuid4()
auto_reply_stream_text = ''
auto_reply_stream_open = False
result_data = result_dict.get('data') or {}
tool_definition = PLATFORM_TOOLS_BY_NAME.get(result_data.get('tool_name'))
auto_reply_deferred = (
tool_definition is not None
and tool_definition.api == 'send_message'
and not result_data.get('error')
)
explicit_reply_text = pending_explicit_replies.pop(
str(result_data.get('tool_call_id') or ''), None
)
journal_started = time.monotonic()
await self.journal.append_run_result(
@@ -465,23 +495,49 @@ class AgentRunOrchestrator:
async with binding_for(execution_query).lock:
result.attachments = exported_message(execution_query, file_ids, consume=True)
if auto_reply_enabled and not explicit_delivery_completed and result is not None:
reply_text = None
operation = None
if auto_reply_enabled and result is not None:
if result_type == 'message.delta' and isinstance(result, provider_message.MessageChunk):
reply_text = result.all_content or result.content
operation = 'update'
cumulative_text = result.all_content or result.content
if isinstance(cumulative_text, str):
if auto_reply_prefix and cumulative_text.startswith(auto_reply_prefix):
reply_text = cumulative_text[len(auto_reply_prefix) :]
else:
auto_reply_prefix = ''
reply_text = cumulative_text
auto_reply_cumulative_text = cumulative_text
auto_reply_stream_text = reply_text
if reply_text.strip() and not auto_reply_deferred:
await reply_streams.apply(
ReplyStreamRequest(
stream_id=auto_reply_stream_id,
operation='update',
text=reply_text,
)
)
auto_reply_stream_open = True
elif has_completed_message:
reply_text = result.content
operation = 'finish'
if isinstance(reply_text, str) and reply_text.strip() and operation is not None:
await reply_streams.apply(
ReplyStreamRequest(
stream_id=auto_reply_stream_id,
operation=operation,
text=reply_text,
)
)
if isinstance(reply_text, str):
if auto_reply_prefix and reply_text.startswith(auto_reply_prefix):
reply_text = reply_text[len(auto_reply_prefix) :]
if reply_text.strip() and (
not auto_reply_deferred
or explicit_reply_text is None
or reply_text.strip() != explicit_reply_text.strip()
):
await reply_streams.apply(
ReplyStreamRequest(
stream_id=auto_reply_stream_id,
operation='finish',
text=reply_text,
)
)
auto_reply_stream_id = uuid4()
auto_reply_stream_text = ''
auto_reply_stream_open = False
auto_reply_prefix = auto_reply_cumulative_text
auto_reply_deferred = False
explicit_reply_text = None
if result is not None:
yield result
@@ -126,9 +126,9 @@ PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = (
{'zh_Hans': '回复当前会话', 'en_US': 'Reply to current conversation'},
{
'zh_Hans': '向触发当前事件的会话发送回复或任务进度。目标由 LangBot 固定,Agent 无法改写。'
'需要发送消息时请调用此工具,Agent 的输出文本不会自动发送。',
'需要显式发送消息时请调用此工具;独立 Agent 的文本输出也可能自动回复。',
'en_US': 'Send a reply or task progress to the conversation that triggered this run. LangBot fixes the target. '
'Call this tool to send a message; Agent output text is not sent automatically.',
'Call this tool to send an explicit message. Direct Agent text output may also be delivered automatically.',
},
_object_schema({'text': {**_TEXT, 'description': 'Reply text'}}, ['text']),
('message.*', 'friend.*', 'group.*', 'feedback.*', 'interaction.submitted'),
@@ -282,9 +282,9 @@ PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = (
'write',
{'zh_Hans': '发送消息', 'en_US': 'Send message'},
{
'zh_Hans': '使用当前机器人向指定用户或群组发送回复或任务进度。Agent 的输出文本不会自动发送。',
'zh_Hans': '使用当前机器人向指定用户或群组发送回复或任务进度。独立 Agent 的文本输出也可能自动回复。',
'en_US': 'Send a reply or task progress to a specified person or group using the current bot. '
'Agent output text is not sent automatically.',
'Direct Agent text output may also be delivered automatically.',
},
_object_schema(
{'target_type': _TARGET_TYPE, 'target_id': _ID, 'text': _TEXT}, ['target_type', 'target_id', 'text']
@@ -1,5 +1,6 @@
from __future__ import annotations
import json
import time
import traceback
import typing
@@ -34,27 +35,59 @@ class DingTalkCardCallbackHandler(dingtalk_stream.CallbackHandler):
self.adapter = adapter
async def process(self, message: dingtalk_stream.CallbackMessage):
callback = dingtalk_stream.CardCallbackMessage.from_dict(message.data or {})
content = dict(callback.content) if isinstance(callback.content, dict) else {}
if isinstance(message.data, dict):
content['_raw_callback'] = message.data
raw = message.data if isinstance(message.data, dict) else {}
content = raw.get('content')
if isinstance(content, str):
try:
content = json.loads(content)
except json.JSONDecodeError:
content = {}
content = dict(content) if isinstance(content, dict) else {}
content['_raw_callback'] = raw
extension = raw.get('extension')
if isinstance(extension, str):
try:
extension = json.loads(extension)
except json.JSONDecodeError:
extension = {}
track_id = str(raw.get('outTrackId') or '')
matched = track_id in self.adapter.interaction_callback_contexts
self.logger.info(
'DingTalk card callback: topic=%s track=%s keys=%s content_keys=%s context_matched=%s',
getattr(message.headers, 'topic', None),
track_id,
sorted(raw),
sorted(content),
matched,
)
event = DingTalkEvent.from_payload(
{
'conversation_type': 'CardCallback',
'Type': 'card_callback',
'timestamp': float(message.headers.time or 0) / 1000,
'CardCallback': {
'extension': callback.extension,
'corp_id': callback.corp_id,
'user_id': callback.user_id,
'extension': extension if isinstance(extension, dict) else {},
'corp_id': raw.get('corpId'),
'user_id': raw.get('userId'),
'content': content,
'space_id': callback.space_id,
'card_instance_id': callback.card_instance_id,
'space_id': raw.get('openSpaceId') or raw.get('spaceId'),
'card_instance_id': track_id,
},
}
)
if event is not None:
await self.adapter._handle_native_event(event)
interaction = interaction_event_from_native(event, self.adapter.interaction_callback_contexts)
if interaction is not None:
try:
await self.adapter._dispatch_eba_event(interaction)
except Exception:
self.logger.exception('DingTalk card callback dispatch failed: track=%s', track_id)
return 500, 'Card callback dispatch failed'
elif matched:
self.logger.warning('DingTalk card callback could not be parsed: track=%s', track_id)
return 500, 'Card callback could not be parsed'
else:
self.logger.info('DingTalk card callback has no active interaction: track=%s', track_id)
return dingtalk_stream.AckMessage.STATUS_OK, 'OK'
@@ -216,6 +216,8 @@ async def send_interaction(adapter: typing.Any, params: dict[str, typing.Any]) -
adapter.interaction_callback_contexts[out_track_id] = {
'callback_token': callback_token,
'field_types': field_types,
'target_type': target_type,
'target_id': target_id,
'expires_at': time.monotonic() + 30 * 60,
}
return {
@@ -340,6 +342,12 @@ def interaction_event_from_native(
if callback_data:
parsed = parse_callback_data(callback_data)
raw_values = _mapping(_find_named_value(callback, {'values'}))
if 'field_ref' in parsed and 'option_ref' not in parsed:
# The A2UI form submit button uses f:0, but its values already
# contain the selected option; option refs belong to native pickers.
if not raw_values:
return None
parsed.pop('field_ref')
context: dict[str, typing.Any] = {}
if callback_contexts is not None:
track_id = _callback_track_id(callback)
@@ -348,7 +356,6 @@ def interaction_event_from_native(
return None
if parsed.get('callback_token') != context.get('callback_token'):
return None
callback_contexts.pop(track_id, None)
if raw_values:
field_types = context.get('field_types') if isinstance(context.get('field_types'), dict) else {}
values = {}
@@ -370,10 +377,12 @@ def interaction_event_from_native(
'callback_token': str(context.get('callback_token') or ''),
'values': {str(context.get('field_id') or ''): value},
}
contexts.pop(track_id, None)
space_id = str(callback.get('space_id') or '')
normalized_space_id = space_id.lower()
if 'im_group.' in normalized_space_id:
if context.get('target_type') in {'group', 'person'} and context.get('target_id'):
target_type = str(context['target_type'])
target_id = str(context['target_id'])
elif 'im_group.' in normalized_space_id:
target_type = 'group'
target_id = space_id[normalized_space_id.index('im_group.') + len('im_group.') :]
elif 'im_robot.' in normalized_space_id:
+3 -2
View File
@@ -899,7 +899,8 @@ class RuntimeBot:
state_policy=StatePolicy(state_scopes=['conversation', 'actor', 'subject', 'runner']),
delivery_policy=DeliveryPolicy(
enable_streaming=agent.get('kind') == 'agent' and event_type == 'message.received',
enable_reply=agent.get('kind') == 'agent' and event_type == 'message.received',
enable_reply=agent.get('kind') == 'agent'
and event_type in {'message.received', 'interaction.submitted'},
enable_interactions=agent.get('kind') != 'event_processor',
),
agent_id=agent.get('uuid'),
@@ -1700,7 +1701,7 @@ class RuntimeBot:
'_execution_context': self.execution_context,
},
):
# Resuming an interaction follows the same explicit-action delivery policy.
# The orchestrator delivers Direct Agent output unless a platform action already did.
pass
async def _dispatch_eba_message_to_pipeline(
@@ -1270,7 +1270,8 @@ class TestQueryEntrySessionQueryId:
@pytest.mark.asyncio
async def test_direct_agent_streams_output_to_the_originating_platform(clean_agent_state):
@pytest.mark.parametrize('event_type', ['message.received', 'interaction.submitted'])
async def test_direct_agent_streams_output_to_the_originating_platform(clean_agent_state, event_type):
from langbot.pkg.agent.runner.host_models import (
AgentBinding,
AgentEventEnvelope,
@@ -1318,6 +1319,115 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age
message_chain=platform_message.MessageChain([platform_message.Plain(text='hi')]),
source_platform_object=object(),
)
event = AgentEventEnvelope(
event_id='event-1',
event_type=event_type,
source='platform',
bot_id='bot_001',
workspace_id=TEST_CONTEXT.workspace_uuid,
conversation_id='person_user-1',
input=AgentInput(text='hi'),
delivery=DeliveryContext(
surface='platform',
reply_target={'target_type': 'person', 'target_id': 'user-1', 'message_id': 'message-1'},
supports_streaming=event_type == 'message.received',
),
)
binding = AgentBinding(
binding_id='binding-1',
scope=BindingScope(scope_type='bot', scope_id='bot_001'),
event_types=[event_type],
runner_id=RUNNER_ID,
resource_policy=ResourcePolicy(),
state_policy=StatePolicy(),
delivery_policy=DeliveryPolicy(enable_streaming=event_type == 'message.received', enable_reply=True),
processor_type='agent',
processor_id='agent-1',
)
results = [
item
async for item in orchestrator.run(
event,
binding,
adapter_context={
'_delivery_adapter': adapter,
'_platform_event': incoming if event_type == 'message.received' else None,
'_execution_context': TEST_CONTEXT,
},
)
]
assert [item.content for item in results] == ['hello', 'hello world']
if event_type == 'message.received':
adapter.create_message_card.assert_awaited_once()
chunks = adapter.reply_message_chunk.await_args_list
assert [call.kwargs['bot_message'].all_content for call in chunks] == ['hello', 'hello world']
assert [call.kwargs['is_final'] for call in chunks] == [False, True]
adapter.reply_message.assert_not_awaited()
adapter.send_message.assert_not_awaited()
else:
adapter.create_message_card.assert_not_awaited()
adapter.reply_message_chunk.assert_not_awaited()
adapter.reply_message.assert_not_awaited()
adapter.send_message.assert_awaited_once()
assert adapter.send_message.await_args.args[2][0].text == 'hello world'
@pytest.mark.asyncio
@pytest.mark.parametrize('tool_name', ['present_interactive_card', 'event_reply'])
async def test_direct_agent_delivers_text_after_card_without_repeating_explicit_reply(clean_agent_state, tool_name):
from langbot.pkg.agent.runner.host_models import (
AgentBinding,
AgentEventEnvelope,
BindingScope,
DeliveryPolicy,
ResourcePolicy,
StatePolicy,
)
from langbot_plugin.api.entities.builtin.runner.delivery import DeliveryContext
from langbot_plugin.api.entities.builtin.runner.input import AgentInput
results = []
if tool_name == 'present_interactive_card':
results.append({'type': 'message.delta', 'data': {'chunk': {'role': 'assistant', 'content': 'Before'}}})
results.extend(
[
{
'type': 'tool.call.started',
'data': {'tool_call_id': 'call-1', 'tool_name': tool_name, 'parameters': {'text': 'After'}},
},
{
'type': 'tool.call.completed',
'data': {'tool_call_id': 'call-1', 'tool_name': tool_name, 'result': {'ok': True}, 'error': None},
},
]
)
if tool_name == 'present_interactive_card':
results.append({'type': 'message.delta', 'data': {'chunk': {'role': 'assistant', 'content': 'BeforeAfter'}}})
results.extend(
[
{'type': 'message.completed', 'data': {'message': {'role': 'assistant', 'content': 'After'}}},
{'type': 'run.completed', 'data': {'finish_reason': 'stop'}},
]
)
connector = FakePluginConnector(results=results)
orchestrator = AgentRunOrchestrator(FakeApplication(connector, clean_agent_state), FakeRegistry(make_descriptor()))
adapter = types.SimpleNamespace(
is_stream_output_supported=AsyncMock(return_value=True),
create_message_card=AsyncMock(return_value=True),
reply_message_chunk=AsyncMock(),
reply_message=AsyncMock(),
send_message=AsyncMock(),
)
incoming = platform_events.MessageReceivedEvent(
message_id='message-1',
sender=platform_entities.User(id='user-1'),
chat_type=platform_entities.ChatType.PRIVATE,
chat_id='user-1',
message_chain=platform_message.MessageChain([platform_message.Plain(text='hi')]),
source_platform_object=object(),
)
event = AgentEventEnvelope(
event_id='event-1',
event_type='message.received',
@@ -1344,7 +1454,7 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age
processor_id='agent-1',
)
results = [
_ = [
item
async for item in orchestrator.run(
event,
@@ -1357,13 +1467,20 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age
)
]
assert [item.content for item in results] == ['hello', 'hello world']
adapter.create_message_card.assert_awaited_once()
chunks = adapter.reply_message_chunk.await_args_list
assert [call.kwargs['bot_message'].all_content for call in chunks] == ['hello', 'hello world']
assert [call.kwargs['is_final'] for call in chunks] == [False, True]
adapter.reply_message.assert_not_awaited()
adapter.send_message.assert_not_awaited()
if tool_name == 'present_interactive_card':
assert adapter.create_message_card.await_count == 2
chunks = adapter.reply_message_chunk.await_args_list
assert [call.kwargs['bot_message'].all_content for call in chunks] == [
'Before',
'Before',
'After',
'After',
]
assert [call.kwargs['is_final'] for call in chunks] == [False, True, False, True]
assert chunks[0].kwargs['bot_message'].resp_message_id != chunks[2].kwargs['bot_message'].resp_message_id
else:
adapter.create_message_card.assert_not_awaited()
adapter.reply_message_chunk.assert_not_awaited()
class TestQueryEntryAdapterParams:
@@ -1,5 +1,6 @@
from __future__ import annotations
import json
import pathlib
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock, patch
@@ -9,7 +10,8 @@ import yaml
from langbot.libs.dingtalk_api.api import DingTalkClient
from langbot.libs.dingtalk_api.dingtalkevent import DingTalkEvent
from langbot.pkg.platform.adapters.dingtalk.adapter import DingTalkAdapter
from langbot.pkg.agent.runner.interaction_manager import InteractionManager
from langbot.pkg.platform.adapters.dingtalk.adapter import DingTalkAdapter, DingTalkCardCallbackHandler
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
@@ -411,7 +413,7 @@ async def test_dingtalk_native_input_uses_template_variables_and_submits_value()
assert event.data['callback_token'] == 'callback-token'
assert event.data['values'] == {'comment': 'Looks good'}
assert result['message_id'] not in adapter.interaction_callback_contexts
assert result['message_id'] in adapter.interaction_callback_contexts
@pytest.mark.asyncio
@@ -467,6 +469,116 @@ async def test_dingtalk_native_select_uses_select_block_and_normalizes_callback(
assert event.data['target_type'] == 'person'
assert event.data['target_id'] == 'user-1'
assert event.data['values'] == {'priority': 'urgent'}
assert 'field_ref' not in event.data
resolved = InteractionManager._resolve_submission_refs(
{
'interaction_id': 'form-1',
'request': {
'fields': [
{'id': 'priority', 'type': 'select', 'options': [{'value': 'urgent'}]},
]
},
},
{'values': event.data['values']},
)
assert resolved['values'] == {'priority': 'urgent'}
@pytest.mark.asyncio
async def test_dingtalk_stream_callback_submits_dropdown_without_space_id():
adapter = make_adapter()
received = []
async def listener(event, _adapter):
received.append(event)
adapter.register_listener(platform_events.PlatformSpecificEvent, listener)
result = await adapter.call_platform_api(
'interaction.request',
{
'callback_token': 'callback-token',
'reply_target': {'target_type': 'person', 'target_id': 'user-1'},
'request': {
'title': 'Priority',
'fields': [
{
'id': 'priority',
'type': 'select',
'options': [{'value': 'urgent', 'label': 'Urgent'}],
}
],
},
},
)
raw_callback = {
'outTrackId': result['message_id'],
'userId': 'user-1',
'corpId': 'corp-1',
'extension': '{}',
'content': json.dumps(
{
'actionData': {
'a2uiAction': {
'context': {
'actionId': 'lbi:callback-token:f:0',
'values': {'priority': ['urgent']},
}
}
}
}
),
}
await DingTalkCardCallbackHandler(adapter).process(
SimpleNamespace(data=raw_callback, headers=SimpleNamespace(time=1_714_000_000_000))
)
assert len(received) == 1
assert received[0].action == 'interaction.submitted'
assert received[0].data['target_type'] == 'person'
assert received[0].data['target_id'] == 'user-1'
assert received[0].data['values'] == {'priority': 'urgent'}
assert result['message_id'] in adapter.interaction_callback_contexts
@pytest.mark.asyncio
async def test_dingtalk_stream_callback_accepts_object_content_and_retries_unparsed_action():
adapter = make_adapter()
received = []
async def listener(event, _adapter):
received.append(event)
adapter.register_listener(platform_events.PlatformSpecificEvent, listener)
result = await adapter.call_platform_api(
'interaction.request',
{
'callback_token': 'callback-token',
'reply_target': {'target_type': 'person', 'target_id': 'user-1'},
'request': {'fields': [{'id': 'priority', 'type': 'select', 'options': [{'value': 'urgent'}]}]},
},
)
raw = {
'outTrackId': result['message_id'],
'userId': 'user-1',
'content': {
'actionData': {
'a2uiAction': {
'context': {
'actionId': 'lbi:callback-token:f:0',
'values': {'priority': ['urgent']},
}
}
}
},
}
handler = DingTalkCardCallbackHandler(adapter)
message = SimpleNamespace(data=raw, headers=SimpleNamespace(time=1_714_000_000_000))
assert (await handler.process(message))[0] == 200
assert received[0].data['values'] == {'priority': 'urgent'}
raw['content'] = {'actionData': {'unexpected': True}}
assert (await handler.process(message))[0] == 500
@pytest.mark.asyncio
@@ -571,7 +683,7 @@ async def test_dingtalk_adapter_dispatches_native_input_submission():
assert calls[0].action == 'interaction.submitted'
assert calls[0].data['callback_token'] == 'callback-token'
assert calls[0].data['values'] == {'score': 42.5}
assert result['message_id'] not in adapter.interaction_callback_contexts
assert result['message_id'] in adapter.interaction_callback_contexts
def test_dingtalk_subscription_member_and_group_mapping():
@@ -673,7 +673,7 @@ class TestInteractionResumeRouting:
'_execution_context': TEST_CONTEXT,
}
assert envelope.delivery.supports_streaming is False
assert binding.delivery_policy.enable_reply is False
assert binding.delivery_policy.enable_reply is True
adapter.send_message.assert_not_awaited()
def test_agent_product_to_binding_does_not_fallback_to_component_ref(self):