feat(dingtalk): deliver A2UI cards and restore agent replies

src/langbot/libs/dingtalk_api/a2ui.py

@@ -0,0 +1,127 @@
+"""DingTalk A2UI message builders shared by both DingTalk adapters."""
+
+from __future__ import annotations
+
+import json
+import typing
+
+
+A2UI_CATALOG_ID = 'https://dingtalk.com/card/a2ui/catalogs/public/catalog.json'
+A2UI_MESSAGE_VERSION = 'v1.0'
+A2UI_PROTOCOL_VERSION = '1.0'
+
+
+def binding(path: str) -> dict[str, str]:
+    return {'path': path}
+
+
+def serialize_messages(messages: typing.Iterable[dict | str]) -> list[str]:
+    serialized = [
+        message if isinstance(message, str) else json.dumps(message, ensure_ascii=False, separators=(',', ':'))
+        for message in messages
+    ]
+    if not serialized:
+        raise ValueError('DingTalk A2UI messages must not be empty')
+    return serialized
+
+
+def create_surface_messages(
+    surface_id: str,
+    *,
+    data: dict[str, typing.Any],
+    components: list[dict[str, typing.Any]],
+) -> list[dict[str, typing.Any]]:
+    return [
+        {
+            'version': A2UI_MESSAGE_VERSION,
+            'createSurface': {
+                'surfaceId': surface_id,
+                'catalogId': A2UI_CATALOG_ID,
+            },
+        },
+        {
+            'version': A2UI_MESSAGE_VERSION,
+            'updateDataModel': {
+                'surfaceId': surface_id,
+                'path': '/',
+                'value': data,
+            },
+        },
+        {
+            'version': A2UI_MESSAGE_VERSION,
+            'updateComponents': {
+                'surfaceId': surface_id,
+                'components': components,
+            },
+        },
+    ]
+
+
+def update_surface_messages(
+    surface_id: str,
+    *,
+    data: dict[str, typing.Any] | None = None,
+    components: list[dict[str, typing.Any]] | None = None,
+) -> list[dict[str, typing.Any]]:
+    messages: list[dict[str, typing.Any]] = []
+    if data is not None:
+        messages.append(
+            {
+                'version': A2UI_MESSAGE_VERSION,
+                'updateDataModel': {
+                    'surfaceId': surface_id,
+                    'path': '/',
+                    'value': data,
+                },
+            }
+        )
+    if components is not None:
+        messages.append(
+            {
+                'version': A2UI_MESSAGE_VERSION,
+                'updateComponents': {
+                    'surfaceId': surface_id,
+                    'components': components,
+                },
+            }
+        )
+    if not messages:
+        raise ValueError('DingTalk A2UI update requires data or components')
+    return messages
+
+
+def text_card_components() -> list[dict[str, typing.Any]]:
+    return [
+        {
+            'id': 'root',
+            'component': 'Column',
+            'children': ['content'],
+            'gap': 8,
+        },
+        {
+            'id': 'content',
+            'component': 'Markdown',
+            'content': binding('/content'),
+        },
+    ]
+
+
+def create_text_card_messages(surface_id: str, content: str = '') -> list[dict[str, typing.Any]]:
+    return create_surface_messages(
+        surface_id,
+        data={'content': content},
+        components=text_card_components(),
+    )
+
+
+def update_text_card_messages(surface_id: str, content: str) -> list[dict[str, typing.Any]]:
+    return [
+        {
+            'version': A2UI_MESSAGE_VERSION,
+            'updateDataModel': {
+                'surfaceId': surface_id,
+                'path': '/content',
+                'value': content,
+            },
+        }
+    ]

src/langbot/libs/dingtalk_api/api.py

@@ -14,15 +14,17 @@ import websockets
 from .EchoHandler import EchoTextHandler
 from .card_callback import DingTalkCardActionHandler
 from .dingtalkevent import DingTalkEvent
-from .thinking import card_text_fields, visible_think_markers
+from .a2ui import (
+    A2UI_PROTOCOL_VERSION,
+    create_text_card_messages,
+    serialize_messages,
+    update_text_card_messages,
+)
+from .thinking import visible_think_markers
 import httpx
 import traceback
 from langbot.pkg.utils import httpclient
-from langbot.pkg.utils.outbound_url import (
-    UnsafeOutboundUrlError,
-    assert_safe_local_media_path,
-    assert_safe_outbound_url,
-)
+from langbot.pkg.utils.outbound_url import assert_safe_outbound_url
 
 
 _stdout_logger = logging.getLogger('langbot.dingtalk_api')
@@ -64,34 +66,8 @@ async def _read_httpx_media_limited(response: httpx.Response) -> bytes:
     return bytes(body)
 
 
-def _stringify_card_param_map(card_param_map: Optional[dict]) -> dict:
-    """DingTalk cardParamMap only accepts string values.
-
-    Keep callers free to pass structured values for template variables such
-    as button groups or select options, then encode them once at the API
-    boundary.
-    """
-    if not card_param_map:
-        return {}
-    result = {}
-    for key, value in card_param_map.items():
-        if value is None:
-            result[key] = ''
-        elif isinstance(value, str):
-            result[key] = value
-        else:
-            result[key] = json.dumps(value, ensure_ascii=False)
-    return result
-
-
 def is_stream_card_configured(config: dict) -> bool:
-    if not config.get('markdown_card', True):
-        return False
-    template_id = config.get('card_template_id')
-    # Older manifests stored instructional placeholder text as the default.
-    return (
-        isinstance(template_id, str) and bool(template_id.strip()) and template_id.strip() != '填写你的卡片template_id'
-    )
+    return bool(config.get('markdown_card', True))
 
 
 class DingTalkClient:
@@ -132,10 +108,8 @@ class DingTalkClient:
         self.access_token_expiry_time = ''
         self.markdown_card = markdown_card
         self.logger = logger
-        # Legacy access_token used by the OLD oapi.dingtalk.com endpoints
-        # (e.g. /media/upload, which is the only documented way to get an
-        # `@xxx` media_id usable in card Avatar.imageUrl). The new v1.0
-        # token doesn't work there — different auth domain.
+        # Attachment uploads still use the legacy oapi.dingtalk.com token.
+        # The v1.0 application token does not work for that auth domain.
         self.legacy_access_token = ''
         self.legacy_access_token_expiry_time: typing.Optional[float] = None
         self._stopped = False  # Flag to control the event loop
@@ -731,118 +705,107 @@ class DingTalkClient:
 
     async def create_and_card(
         self,
-        temp_card_id: str,
         incoming_message: dingtalk_stream.ChatbotMessage,
         quote_origin: bool = False,
-        card_auto_layout: bool = True,
     ):
         """Create + deliver the streaming chat card for a chatbot reply.
 
-        Replaces the old `dingtalk_stream.AICardReplier`-based path. Returns
-        `(layout_config, out_track_id)` keeps the adapter's tuple contract
-        while preserving the layout setting on subsequent data updates.
+        Returns ``(card state, out_track_id)`` for subsequent A2UI updates.
         """
         out_track_id = uuid.uuid4().hex
+        surface_id = f'langbot-{out_track_id}'
         is_group = str(incoming_message.conversation_type) == '2'
         if is_group:
-            open_space_id = f'dtv1.card//IM_GROUP.{incoming_message.conversation_id}'
+            open_space_id = f'dtv1.card//im_group.{incoming_message.conversation_id}'
         else:
-            open_space_id = f'dtv1.card//IM_ROBOT.{incoming_message.sender_staff_id}'
+            open_space_id = f'dtv1.card//im_robot.{incoming_message.sender_staff_id}'
 
-        card_param_map = {
-            'content': '',
-            'answer': '',
-            'reasoning': '',
-            'flowStatus': '1',
-            'brand': '',
-            'hasReasoning': '',
-        }
-        if incoming_message.message_type == 'text':
-            card_param_map['query'] = incoming_message.get_text_list()[0]
-        else:
-            card_param_map['query'] = '...'
-
-        await self.create_and_deliver_card(
-            card_template_id=temp_card_id,
+        await self.create_and_deliver_ai_card(
             out_track_id=out_track_id,
             open_space_id=open_space_id,
             is_group=is_group,
-            card_param_map=card_param_map,
-            card_data_config={'autoLayout': card_auto_layout},
+            a2ui_messages=create_text_card_messages(surface_id),
+            flow_status='PROCESSING',
+            summary='LangBot is replying',
         )
-        return {'autoLayout': card_auto_layout}, out_track_id
+        return {'surface_id': surface_id}, out_track_id
 
     async def send_card_message(self, card_instance, card_instance_id: str, content: str, is_final: bool):
-        """Update separate reasoning/answer fields and the legacy content field."""
-        fields = card_text_fields(content)
-        # Old templates continue to render content with explicit think markers;
-        # the supplied split template binds only reasoning and answer.
-        fields['content'] = visible_think_markers(content)
-        fields['flowStatus'] = '3' if is_final else '2'
-        fields['brand'] = ''
-        if isinstance(card_instance, dict):
-            fields['config'] = card_instance
-            fields['autoLayout'] = str(card_instance.get('autoLayout', True)).lower()
-        if not await self.update_card_data(out_track_id=card_instance_id, card_param_map=fields):
-            raise RuntimeError('DingTalk card content update failed; see adapter logs for the API response')
+        """Replace the streamed Markdown content on an A2UI card."""
+        surface_id = (
+            str(card_instance.get('surface_id') or '') if isinstance(card_instance, dict) else ''
+        ) or f'langbot-{card_instance_id}'
+        await self.update_ai_card(
+            out_track_id=card_instance_id,
+            a2ui_messages=update_text_card_messages(surface_id, visible_think_markers(content)),
+            flow_status='FINISH' if is_final else 'PROCESSING',
+        )
 
-    async def create_and_deliver_card(
+    @staticmethod
+    def _check_ai_card_result(response) -> dict:
+        try:
+            data = response.json()
+        except Exception as exc:
+            raise RuntimeError(f'Invalid DingTalk AI card response (HTTP {response.status_code})') from exc
+        if response.status_code != 200 or data.get('success') is not True:
+            raise RuntimeError(
+                f'DingTalk AI card request failed (HTTP {response.status_code}): '
+                f'{data.get("message", data.get("errorMsg", data))}'
+            )
+        result = data.get('result') if isinstance(data.get('result'), dict) else {}
+        failures = [
+            item
+            for item in result.get('deliverResults') or []
+            if isinstance(item, dict) and item.get('success') is not True
+        ]
+        if failures:
+            raise RuntimeError(f'DingTalk AI card delivery failed: {failures}')
+        return result
+
+    async def create_and_deliver_ai_card(
         self,
         *,
-        card_template_id: str,
         out_track_id: str,
         open_space_id: str,
         is_group: bool,
-        card_param_map: Optional[dict] = None,
+        a2ui_messages: list[dict | str],
+        flow_status: str = 'PROCESSING',
+        summary: str = '',
         callback_type: str = 'STREAM',
-        callback_route_key: Optional[str] = None,
-        support_forward: bool = True,
-        dynamic_data_source_configs: Optional[list] = None,
-        card_data_config: Optional[dict] = None,
         at_user_ids: Optional[dict] = None,
         recipients: Optional[list] = None,
-    ) -> bool:
-        """POST /v1.0/card/instances/createAndDeliver.
-
-        Mirrors the SDK's `async_create_and_deliver_card` shape but exposes
-        the dynamic-data-source config slot so we can register a pull URL
-        for variable-length button lists.
-        """
+        biz_context: Optional[dict] = None,
+    ) -> dict:
+        """Create and deliver an A2UI card as the configured robot application."""
         if not await self.check_access_token():
             await self.get_access_token()
 
-        template_params = dict(card_param_map or {})
-        if card_data_config is not None:
-            template_params['config'] = card_data_config
-            template_params['autoLayout'] = str(card_data_config.get('autoLayout', True)).lower()
-        cardData: dict = {'cardParamMap': _stringify_card_param_map(template_params)}
-
+        robot_code = self.robot_code or self.key
         body: dict = {
-            'cardTemplateId': card_template_id,
             'outTrackId': out_track_id,
-            'cardData': cardData,
-            'callbackType': callback_type,
             'openSpaceId': open_space_id,
-            'imGroupOpenSpaceModel': {'supportForward': support_forward},
-            'imRobotOpenSpaceModel': {'supportForward': support_forward},
+            'protocolVersion': A2UI_PROTOCOL_VERSION,
+            'flowStatus': flow_status,
+            'a2uiMessages': serialize_messages(a2ui_messages),
+            'summary': summary,
+            'callbackType': callback_type,
         }
-        if callback_type == 'HTTP' and callback_route_key:
-            body['callbackRouteKey'] = callback_route_key
-
         if is_group:
-            deliver: dict = {'robotCode': self.robot_code or self.key}
-            if at_user_ids:
-                deliver['atUserIds'] = at_user_ids
-            if recipients is not None:
-                deliver['recipients'] = recipients
+            body['imGroupOpenSpaceModel'] = {'supportForward': False}
+            deliver: dict = {'robotCode': robot_code}
             body['imGroupOpenDeliverModel'] = deliver
         else:
-            body['imRobotOpenDeliverModel'] = {'spaceType': 'IM_ROBOT'}
+            body['imRobotOpenSpaceModel'] = {'supportForward': False}
+            deliver = {'robotCode': robot_code}
+            body['imRobotOpenDeliverModel'] = deliver
+        if at_user_ids:
+            deliver['atUserIds'] = at_user_ids
+        if recipients is not None:
+            deliver['recipients'] = recipients
+        if biz_context:
+            body['bizContext'] = biz_context
 
-        if dynamic_data_source_configs:
-            body['openDynamicDataConfig'] = {'dynamicDataSourceConfigs': dynamic_data_source_configs}
-
-        url = f'{DINGTALK_OPENAPI_BASE}/v1.0/card/instances/createAndDeliver'
+        url = f'{DINGTALK_OPENAPI_BASE}/v1.0/card/ai/instances/createAndDeliver'
         headers = {
             'x-acs-dingtalk-access-token': self.access_token,
             'Content-Type': 'application/json',
@@ -850,45 +813,33 @@ class DingTalkClient:
         try:
             async with self._http_client_context() as client:
                 response = await client.post(url, headers=headers, json=body, timeout=30.0)
-            # Propagate business failures as well as transport failures. Callers
-            # must not register or update a card that was never delivered.
-            self._check_send_result(response)
+            result = self._check_ai_card_result(response)
         except Exception as exc:
-            error = RuntimeError(f'DingTalk card creation failed: {exc}')
+            error = RuntimeError(f'DingTalk AI card creation failed: {exc}')
             _stdout_logger.error('%s', error)
             raise error from exc
-        _stdout_logger.info('DingTalk card created: outTrackId=%s', out_track_id)
-        return True
+        _stdout_logger.info('DingTalk AI card created: outTrackId=%s', out_track_id)
+        return result
 
-    async def streaming_update_card(
+    async def update_ai_card(
         self,
         *,
         out_track_id: str,
-        content_key: str,
-        content_value: str,
-        append: bool,
-        finished: bool,
-        failed: bool = False,
-    ) -> bool:
-        """PUT /v1.0/card/streaming.
-
-        Replaces `dingtalk_stream.AICardReplier.async_streaming` — same body
-        shape (outTrackId / guid / key / content / isFull / isFinalize /
-        isError) per the SDK source.
-        """
+        a2ui_messages: list[dict | str],
+        flow_status: str,
+        out_update_id: Optional[str] = None,
+    ) -> dict:
+        """Apply an idempotent A2UI delta to an existing AI card."""
         if not await self.check_access_token():
             await self.get_access_token()
 
         body = {
             'outTrackId': out_track_id,
-            'guid': uuid.uuid4().hex,
-            'key': content_key,
-            'content': content_value,
-            'isFull': not append,
-            'isFinalize': finished,
-            'isError': failed,
+            'outUpdateId': out_update_id or uuid.uuid4().hex,
+            'flowStatus': flow_status,
+            'a2uiMessages': serialize_messages(a2ui_messages),
         }
-        url = f'{DINGTALK_OPENAPI_BASE}/v1.0/card/streaming'
+        url = f'{DINGTALK_OPENAPI_BASE}/v1.0/card/ai/instances/update'
         headers = {
             'x-acs-dingtalk-access-token': self.access_token,
             'Content-Type': 'application/json',
@@ -896,75 +847,17 @@ class DingTalkClient:
         try:
             async with self._http_client_context() as client:
                 response = await client.put(url, headers=headers, json=body, timeout=30.0)
-                if response.status_code == 200:
-                    return True
-                if self.logger:
-                    response_body = await httpclient.response_text(response)
-                    await self.logger.error(
-                        f'DingTalk card streaming failed: status={response.status_code} body={response_body}'
-                    )
-                return False
-        except Exception:
-            if self.logger:
-                await self.logger.error(f'DingTalk card streaming error: {traceback.format_exc()}')
-            return False
-
-    async def update_card_data(
-        self,
-        *,
-        out_track_id: str,
-        card_param_map: Optional[dict] = None,
-        private_data: Optional[dict] = None,
-    ) -> bool:
-        """PUT /v1.0/card/instances — non-streaming card content update."""
-        if not await self.check_access_token():
-            await self.get_access_token()
-
-        body: dict = {
-            'outTrackId': out_track_id,
-            'cardData': {'cardParamMap': _stringify_card_param_map(card_param_map)},
-        }
-        if private_data:
-            body['privateData'] = private_data
-
-        url = f'{DINGTALK_OPENAPI_BASE}/v1.0/card/instances'
-        headers = {
-            'x-acs-dingtalk-access-token': self.access_token,
-            'Content-Type': 'application/json',
-        }
-        try:
-            _stdout_logger.info(
-                'DingTalk update_card_data request: out_track_id=%s body=%s',
-                out_track_id,
-                json.dumps(body, ensure_ascii=False)[:1500],
-            )
-            async with self._http_client_context() as client:
-                response = await client.put(url, headers=headers, json=body, timeout=30.0)
-                response_body = await httpclient.response_text(response, max_chars=300)
-                _stdout_logger.info(
-                    'DingTalk update_card_data response: status=%d body=%s',
-                    response.status_code,
-                    response_body,
-                )
-                if response.status_code == 200:
-                    return True
-                if self.logger:
-                    await self.logger.error(
-                        f'DingTalk update card failed: status={response.status_code} body={response_body}'
-                    )
-                return False
-        except Exception:
-            _stdout_logger.exception('DingTalk update_card_data error')
-            if self.logger:
-                await self.logger.error(f'DingTalk update card error: {traceback.format_exc()}')
-            return False
+            return self._check_ai_card_result(response)
+        except Exception as exc:
+            error = RuntimeError(f'DingTalk AI card update failed: {exc}')
+            _stdout_logger.error('%s', error)
+            raise error from exc
 
     async def get_legacy_access_token(self) -> Optional[str]:
         """Fetch the LEGACY (oapi.dingtalk.com) access_token. This is a
         different auth domain from the v1.0 token cached in
         ``self.access_token`` — only the legacy token authorises the
-        ``/media/upload`` endpoint that returns an ``@xxx`` media_id
-        consumable by card components like Avatar.imageUrl.
+        ``/media/upload`` endpoint used by attachment delivery.
 
         Returns the token string on success, None on failure. Caches
         with a 60s safety margin before the documented 7200s expiry.
@@ -998,67 +891,6 @@ class DingTalkClient:
                 await self.logger.error(f'DingTalk legacy gettoken error: {traceback.format_exc()}')
         return None
 
-    async def upload_image_media(self, file_path: str) -> Optional[str]:
-        """Upload an image file to DingTalk media storage and return the
-        ``@xxx`` media_id, which can be passed straight into card variables
-        like Avatar.imageUrl. Endpoint:
-
-            POST https://oapi.dingtalk.com/media/upload?access_token=…&type=image
-
-        Returns the media_id on success, None on any failure (caller
-        should handle a None gracefully — DingTalk falls back to a
-        default avatar when imageUrl is empty/unknown).
-        """
-        if not os.path.exists(file_path):
-            if self.logger:
-                await self.logger.error(f'DingTalk upload_image_media: file not found {file_path}')
-            return None
-
-        try:
-            # Caller-supplied path: confine it to the media roots so a plugin or
-            # external input cannot make the adapter upload arbitrary files.
-            safe_file_path = assert_safe_local_media_path(file_path)
-        except UnsafeOutboundUrlError as exc:
-            if self.logger:
-                await self.logger.error(f'DingTalk upload_image_media: rejected path: {exc}')
-            return None
-
-        token = await self.get_legacy_access_token()
-        if not token:
-            return None
-
-        url = 'https://oapi.dingtalk.com/media/upload'
-        try:
-            file_bytes = await asyncio.to_thread(_read_local_media_limited, safe_file_path)
-            file_name = os.path.basename(safe_file_path)
-            # Best-effort content-type guess; DingTalk accepts the major image
-            # mime types and otherwise infers from the bytes.
-            ext = os.path.splitext(file_name)[1].lower().lstrip('.')
-            mime = {'png': 'image/png', 'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'gif': 'image/gif'}.get(
-                ext, 'application/octet-stream'
-            )
-            async with self._http_client_context() as client:
-                response = await client.post(
-                    url,
-                    params={'access_token': token, 'type': 'image'},
-                    files={'media': (file_name, file_bytes, mime)},
-                    timeout=30.0,
-                )
-            data = await httpclient.parse_json_response(response) if response.status_code == 200 else {}
-            if data.get('errcode') == 0 and data.get('media_id'):
-                _stdout_logger.info('DingTalk upload_image_media OK: media_id=%s', data['media_id'])
-                return data['media_id']
-            if self.logger:
-                response_body = await httpclient.response_text(response, max_chars=300)
-                await self.logger.error(
-                    f'DingTalk upload_image_media failed: status={response.status_code} body={response_body}'
-                )
-        except Exception:
-            _stdout_logger.exception('DingTalk upload_image_media error')
-            if self.logger:
-                await self.logger.error(f'DingTalk upload_image_media error: {traceback.format_exc()}')
-        return None
-
     async def start(self):
         """启动 WebSocket 连接，监听消息"""
         self._stopped = False

src/langbot/libs/dingtalk_api/card_callback.py

@@ -26,6 +26,10 @@ from dingtalk_stream.card_callback import CardCallbackMessage
 
 
 _PARAM_PATHS = (
+    ('actionData', 'a2uiAction', 'context'),
+    ('a2uiAction', 'context'),
+    ('action', 'context'),
+    ('context',),
     ('params',),
     ('cardPrivateData', 'params'),
     ('userPrivateData', 'params'),
@@ -77,7 +81,7 @@ class DingTalkCardActionHandler(dingtalk_stream.CallbackHandler):
             # there). Pull it from the raw callback.data instead.
             raw = callback.data if isinstance(callback.data, dict) else {}
             params = _merge_params(_extract_params(content), _extract_params(raw))
-            action_id = raw.get('actionId') or ''
+            action_id = raw.get('actionId') or params.get('actionId') or params.get('action_id') or ''
             if not action_id:
                 # Some templates nest it under actionData / cardPrivateData.
                 action_data = raw.get('actionData') or {}

src/langbot/pkg/agent/runner/orchestrator.py

@@ -7,13 +7,14 @@ import asyncio
 import contextlib
 import copy
 import typing
+from uuid import uuid4
 
 from langbot_plugin.api.entities.builtin.provider import message as provider_message
 from langbot_plugin.api.entities.builtin.provider import prompt as provider_prompt
 from langbot_plugin.api.entities.builtin.pipeline import query as pipeline_query
 
 from langbot_plugin.entities.io.actions.enums import PluginToRuntimeAction
-from .reply_stream import ReplyStreamSession
+from .reply_stream import ReplyStreamRequest, ReplyStreamSession
 from ...core import app
 from ...i18n import t
 from ...i18n.workspace import get_workspace_language
@@ -36,7 +37,7 @@ from .interaction_manager import InteractionManager
 from .query_bridge import QueryRunBridge
 from .registry import RunnerRegistry
 from .resource_builder import AgentResourceBuilder
-from .platform_tools import freeze_platform_context
+from .platform_tools import PLATFORM_TOOLS_BY_NAME, freeze_platform_context
 from .result_normalizer import AgentResultNormalizer
 from .run_journal import AgentRunJournal
 from .session_registry import AgentRunSessionRegistry, get_session_registry
@@ -253,6 +254,14 @@ class AgentRunOrchestrator:
         terminal_usage: dict[str, typing.Any] | None = None
         execution_outcome = 'unknown'
         execution_error = ''
+        auto_reply_enabled = (
+            binding.processor_type == 'agent'
+            and binding.delivery_policy.enable_reply
+            and event.event_type == 'message.received'
+            and reply_streams.adapter is not None
+        )
+        auto_reply_stream_id = uuid4()
+        explicit_delivery_completed = False
         # 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
@@ -361,6 +370,15 @@ class AgentRunOrchestrator:
                         ):
                             continue
 
+                        if result_type == 'tool.call.completed':
+                            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
+
                         journal_started = time.monotonic()
                         await self.journal.append_run_result(
                             result_dict=result_dict,
@@ -447,6 +465,24 @@ 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 result_type == 'message.delta' and isinstance(result, provider_message.MessageChunk):
+                                reply_text = result.all_content or result.content
+                                operation = 'update'
+                            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 result is not None:
                             yield result
 

src/langbot/pkg/agent/runner/platform_tools.py

@@ -899,6 +899,27 @@ async def execute_platform_tool(
                     context=context, target_type=target_type, target_id=target_id, message=message
                 )
 
+        reply_streams = session.get('reply_streams')
+        if (
+            definition.name == 'event_reply'
+            and message_chain is None
+            and reply_streams is not None
+            and getattr(reply_streams, 'adapter', None) is bot.adapter
+        ):
+            from uuid import uuid4
+
+            from .reply_stream import ReplyStreamRequest
+
+            async def api_func(target_type, target_id, message):
+                text = ''.join(part.text for part in message if isinstance(part, platform_message.Plain))
+                return await reply_streams.apply(
+                    ReplyStreamRequest(
+                        stream_id=uuid4(),
+                        operation='finish',
+                        text=text,
+                    )
+                )
+
         if not callable(api_func):
             raise ValueError(f'Platform API {definition.api} is declared but not implemented')
         if definition.api == 'send_message':

src/langbot/pkg/api/http/controller/groups/platform/adapters.py

@@ -309,24 +309,6 @@ class AdaptersRouterGroup(group.RouterGroup):
                 importutil.read_resource_file_bytes(icon_path), mimetype=mimetypes.guess_type(icon_path)[0]
             )
 
-        @self.route('/dingtalk/stream-card-template', methods=['GET'], auth_type=group.AuthType.NONE)
-        async def _() -> quart.Response:
-            filename = 'dingtalk_stream_card.json'
-            response = quart.Response(
-                importutil.read_resource_file_bytes(f'templates/{filename}'), mimetype='application/json'
-            )
-            response.headers['Content-Disposition'] = f'attachment; filename={filename}'
-            return response
-
-        @self.route('/dingtalk/human-input-card-template', methods=['GET'], auth_type=group.AuthType.NONE)
-        async def _() -> quart.Response:
-            filename = 'dingtalk_human_input_card.json'
-            response = quart.Response(
-                importutil.read_resource_file_bytes(f'templates/{filename}'), mimetype='application/json'
-            )
-            response.headers['Content-Disposition'] = f'attachment; filename={filename}'
-            return response
-
         # In-memory session store for active registrations
         _create_app_sessions: dict = {}
         _SESSION_TTL = 900  # 15 minutes

src/langbot/pkg/platform/adapters/dingtalk/adapter.py

@@ -130,7 +130,7 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo
             'get_file_url',
             'call_platform_api',
         ]
-        if self.config.get('human_input_card_template_id') and hasattr(self.bot, 'create_and_deliver_card'):
+        if hasattr(self.bot, 'create_and_deliver_ai_card'):
             apis.append('interaction.request')
         return apis
 
@@ -225,14 +225,8 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo
             oldest_id = next(iter(self.card_instance_id_dict))
             self.card_instance_id_dict.pop(oldest_id, None)
             getattr(self, '_card_last_updates', {}).pop(oldest_id, None)
-        card_template_id = self.config['card_template_id'].strip()
         incoming_message = event.source_platform_object.incoming_message
-        card_auto_layout = self.config.get('card_auto_layout', True)
-        card_instance, card_instance_id = await self.bot.create_and_card(
-            card_template_id,
-            incoming_message,
-            card_auto_layout=card_auto_layout,
-        )
+        card_instance, card_instance_id = await self.bot.create_and_card(incoming_message)
         self.card_instance_id_dict[message_id] = (card_instance, card_instance_id)
         return True
 

src/langbot/pkg/platform/adapters/dingtalk/interaction.py

@@ -7,6 +7,7 @@ import time
 import typing
 import uuid
 
+from langbot.libs.dingtalk_api.a2ui import binding, create_surface_messages, update_surface_messages
 from langbot.libs.dingtalk_api.dingtalkevent import DingTalkEvent
 from langbot_plugin.api.entities.builtin.platform import events as platform_events
 
@@ -16,69 +17,141 @@ def interaction_delivery_capabilities() -> dict[str, typing.Any]:
         'field_types': ['text', 'textarea', 'number', 'select'],
         'action_styles': ['default', 'primary', 'danger'],
         'supports_updates': True,
-        'max_fields': 1,
+        'max_fields': 8,
     }
 
 
-def _select_options(options: list[dict[str, typing.Any]]) -> list[dict[str, typing.Any]]:
-    locales = ['zh_CN', 'zh_TW', 'en_US', 'ja_JP', 'vi_VN', 'th_TH', 'id_ID', 'ms_MY', 'ko_KR']
-    result = []
-    for option in options:
-        value = str(option.get('value') or '')
-        label = str(option.get('label') or value)
-        if value:
-            result.append({'value': value, 'text': {locale: label for locale in locales}})
-    return result
+def _json_pointer_segment(value: str) -> str:
+    return value.replace('~', '~0').replace('/', '~1')
 
 
-def _field_card_params(field: dict[str, typing.Any]) -> dict[str, typing.Any] | None:
-    field_type = str(field.get('type') or '')
-    field_id = str(field.get('id') or '')
-    if field_type not in {'text', 'textarea', 'number', 'select'} or not field_id:
-        return None
-    label = str(field.get('label') or field_id)
-    placeholder = str(field.get('placeholder') or label)
-    default = field.get('default')
-    params: dict[str, typing.Any] = {
-        'input_visible': '',
-        'input_title': '',
-        'input_placeholder': '',
-        'input_value': '',
-        'select_visible': '',
-        'select_placeholder': '',
-        'select_options': [],
-        'index_o': [],
-        'select_index': -1,
-    }
-    if field_type == 'select':
-        raw_options = field.get('options') if isinstance(field.get('options'), list) else []
-        options = [option for option in raw_options if isinstance(option, dict)]
-        encoded_options = _select_options(options)
-        if not encoded_options:
-            return None
-        selected_index = next(
-            (index for index, option in enumerate(options) if str(option.get('value')) == str(default)),
-            -1,
-        )
-        params.update(
-            {
-                'select_visible': 'true',
-                'select_placeholder': placeholder,
-                'select_options': [str(option.get('value') or '') for option in options],
-                'index_o': encoded_options,
-                'select_index': selected_index,
+def _build_interaction_messages(
+    surface_id: str,
+    request: dict[str, typing.Any],
+    callback_token: str,
+) -> tuple[dict[str, typing.Any], list[dict[str, typing.Any]], dict[str, str]]:
+    raw_fields = request.get('fields') if isinstance(request.get('fields'), list) else []
+    raw_actions = request.get('actions') if isinstance(request.get('actions'), list) else []
+    fields = [field for field in raw_fields if isinstance(field, dict)][:8]
+    actions = [action for action in raw_actions if isinstance(action, dict)]
+
+    form: dict[str, typing.Any] = {}
+    field_types: dict[str, str] = {}
+    required_fields: list[tuple[str, str]] = []
+    components: list[dict[str, typing.Any]] = []
+    children: list[str] = []
+
+    title = str(request.get('title') or '').strip()
+    description = str(request.get('description') or '').strip()
+    if title:
+        components.append({'id': 'title', 'component': 'Text', 'text': title})
+        children.append('title')
+    if description:
+        components.append({'id': 'description', 'component': 'Markdown', 'content': description})
+        children.append('description')
+
+    for index, field in enumerate(fields):
+        field_id = str(field.get('id') or f'field_{index}')
+        field_type = str(field.get('type') or 'text')
+        if field_type not in {'text', 'textarea', 'number', 'select'}:
+            continue
+        field_types[field_id] = field_type
+        path = f'/form/{_json_pointer_segment(field_id)}'
+        component_id = f'field-{index}'
+        label = str(field.get('label') or field_id)
+        placeholder = str(field.get('placeholder') or '')
+        default = field.get('default')
+
+        if field_type == 'select':
+            options = [
+                {
+                    'label': str(option.get('label') or option.get('value') or ''),
+                    'value': str(option.get('value') or ''),
+                }
+                for option in field.get('options') or []
+                if isinstance(option, dict) and str(option.get('value') or '')
+            ]
+            if not options:
+                continue
+            selected = default if isinstance(default, list) else ([] if default in (None, '') else [str(default)])
+            form[field_id] = [str(value) for value in selected]
+            component = {
+                'id': component_id,
+                'component': 'ChoicePicker',
+                'label': placeholder or label,
+                'variant': 'mutuallyExclusive',
+                'displayStyle': 'dropdown',
+                'options': options,
+                'value': binding(path),
             }
-        )
-    else:
-        params.update(
-            {
-                'input_visible': 'true',
-                'input_title': label,
-                'input_placeholder': placeholder,
-                'input_value': '' if default is None else str(default),
+        elif field_type == 'number':
+            numeric_default = default if isinstance(default, (int, float)) and not isinstance(default, bool) else 0
+            form[field_id] = numeric_default
+            component = {
+                'id': component_id,
+                'component': 'NumberInput',
+                'label': label,
+                'placeholder': placeholder,
+                'value': binding(path),
             }
+        else:
+            form[field_id] = '' if default is None else str(default)
+            component = {
+                'id': component_id,
+                'component': 'TextField',
+                'label': label,
+                'placeholder': placeholder,
+                'variant': 'longText' if field_type == 'textarea' else 'shortText',
+                'value': binding(path),
+            }
+        components.append(component)
+        children.append(component_id)
+        if field.get('required'):
+            required_fields.append((path, label))
+
+    rendered_actions = actions or ([{'label': 'Submit', 'style': 'primary'}] if field_types else [])
+    for index, action in enumerate(rendered_actions):
+        label_id = f'action-label-{index}'
+        button_id = f'action-{index}'
+        callback_data = f'lbi:{callback_token}:a:{index}' if actions else f'lbi:{callback_token}:f:0'
+        style = str(action.get('style') or 'default')
+        button: dict[str, typing.Any] = {
+            'id': button_id,
+            'component': 'Button',
+            'child': label_id,
+            'variant': 'primary' if style == 'primary' else 'default',
+            'action': {
+                'event': {
+                    'name': 'langbot_interaction_submit',
+                    'context': {
+                        'actionId': callback_data,
+                        'values': binding('/form'),
+                    },
+                }
+            },
+        }
+        if required_fields:
+            button['checks'] = [
+                {
+                    'condition': {'call': 'required', 'args': {'value': binding(path)}},
+                    'message': f'{label} is required',
+                }
+                for path, label in required_fields
+            ]
+        components.extend(
+            [
+                button,
+                {
+                    'id': label_id,
+                    'component': 'Text',
+                    'text': str(action.get('label') or action.get('id') or 'Submit'),
+                },
+            ]
         )
-    return params
+        children.append(button_id)
+
+    components.insert(0, {'id': 'root', 'component': 'Column', 'children': children, 'gap': 12})
+    return {'form': form}, components, field_types
 
 
 def _prune_callback_contexts(adapter: typing.Any, now: float | None = None) -> None:
@@ -88,48 +161,6 @@ def _prune_callback_contexts(adapter: typing.Any, now: float | None = None) -> N
         contexts.pop(key, None)
 
 
-def _buttons(request: dict[str, typing.Any], callback_token: str) -> list[dict[str, typing.Any]]:
-    fields = request.get('fields') if isinstance(request.get('fields'), list) else []
-    actions = request.get('actions') if isinstance(request.get('actions'), list) else []
-    items: list[tuple[str, str, str]] = []
-    if actions and not fields:
-        items = [
-            (
-                str(action.get('label') or action.get('id') or index + 1),
-                f'lbi:{callback_token}:a:{index}',
-                str(action.get('style') or 'default'),
-            )
-            for index, action in enumerate(actions)
-            if isinstance(action, dict)
-        ]
-    elif len(fields) == 1 and not actions and isinstance(fields[0], dict):
-        field = fields[0]
-        if field.get('type') != 'select':
-            return []
-        options = field.get('options') if isinstance(field.get('options'), list) else []
-        items = [
-            (
-                str(option.get('label') or option.get('value') or index + 1),
-                f'lbi:{callback_token}:f:0:{index}',
-                'default',
-            )
-            for index, option in enumerate(options)
-            if isinstance(option, dict)
-        ]
-    return [
-        {
-            'text': label,
-            'color': 'blue' if style == 'primary' else 'red' if style == 'danger' else 'gray',
-            'status': 'normal',
-            'event': {
-                'type': 'sendCardRequest',
-                'params': {'actionId': callback_data, 'params': {'action_id': callback_data}},
-            },
-        }
-        for label, callback_data, style in items
-    ]
-
-
 async def send_interaction(adapter: typing.Any, params: dict[str, typing.Any]) -> dict[str, typing.Any]:
     request = params.get('request')
     reply_target = params.get('reply_target')
@@ -140,11 +171,7 @@ async def send_interaction(adapter: typing.Any, params: dict[str, typing.Any]) -
     target_id = str(reply_target.get('target_id') or '')
     fields = request.get('fields') if isinstance(request.get('fields'), list) else []
     actions = request.get('actions') if isinstance(request.get('actions'), list) else []
-    buttons = _buttons(request, callback_token)
-    field_params = (
-        _field_card_params(fields[0]) if len(fields) == 1 and not actions and isinstance(fields[0], dict) else None
-    )
-    if not buttons and field_params is None:
+    if not fields and not actions:
         fallback = '\n\n'.join(
             part
             for part in (
@@ -158,48 +185,53 @@ async def send_interaction(adapter: typing.Any, params: dict[str, typing.Any]) -
         return {'ok': True, 'message_id': result.message_id, 'rich': False}
     if target_type not in {'group', 'person'} or not target_id:
         raise ValueError('interaction.request has an invalid DingTalk target')
-    field_label = str(fields[0].get('label') or '').strip() if fields and isinstance(fields[0], dict) else ''
-    content = '\n\n'.join(
-        part
-        for part in (
-            f'## {str(request.get("title") or "").strip()}',
-            str(request.get('description') or '').strip(),
-            field_label,
+    update_target = params.get('update_target')
+    if isinstance(update_target, dict):
+        if update_target.get('rich') is not True or not update_target.get('message_id'):
+            raise ValueError('DingTalk interaction update target is not an A2UI card')
+        out_track_id = str(update_target['message_id'])
+    else:
+        out_track_id = uuid.uuid4().hex
+    surface_id = f'langbot-interaction-{out_track_id}'
+    data, components, field_types = _build_interaction_messages(surface_id, request, callback_token)
+    if isinstance(update_target, dict):
+        await adapter.bot.update_ai_card(
+            out_track_id=out_track_id,
+            a2ui_messages=update_surface_messages(surface_id, data=data, components=components),
+            flow_status='CONFIRMING',
         )
-        if part
-    )
-    out_track_id = uuid.uuid4().hex
-    card_param_map = {'content': content, 'btns': buttons}
-    if field_params is not None:
-        card_param_map.update(field_params)
-    await adapter.bot.create_and_deliver_card(
-        card_template_id=adapter.config['human_input_card_template_id'],
-        out_track_id=out_track_id,
-        open_space_id=(
-            f'dtv1.card//IM_GROUP.{target_id}' if target_type == 'group' else f'dtv1.card//IM_ROBOT.{target_id}'
-        ),
-        is_group=target_type == 'group',
-        card_param_map=card_param_map,
-    )
-    if field_params is not None:
-        _prune_callback_contexts(adapter)
-        field = fields[0]
-        adapter.interaction_callback_contexts[out_track_id] = {
-            'callback_token': callback_token,
-            'field_id': str(field.get('id') or ''),
-            'field_type': str(field.get('type') or ''),
-            'options': [
-                str(option.get('value') or '') for option in field.get('options') or [] if isinstance(option, dict)
-            ],
-            'expires_at': time.monotonic() + 30 * 60,
-        }
-    return {'ok': True, 'message_id': out_track_id, 'rich': True}
+    else:
+        await adapter.bot.create_and_deliver_ai_card(
+            out_track_id=out_track_id,
+            open_space_id=(
+                f'dtv1.card//im_group.{target_id}' if target_type == 'group' else f'dtv1.card//im_robot.{target_id}'
+            ),
+            is_group=target_type == 'group',
+            a2ui_messages=create_surface_messages(surface_id, data=data, components=components),
+            flow_status='CONFIRMING',
+            summary=str(request.get('title') or request.get('fallback_text') or 'Interaction required'),
+            biz_context={'callbackToken': callback_token},
+        )
+    _prune_callback_contexts(adapter)
+    adapter.interaction_callback_contexts[out_track_id] = {
+        'callback_token': callback_token,
+        'field_types': field_types,
+        'expires_at': time.monotonic() + 30 * 60,
+    }
+    return {
+        'ok': True,
+        'message_id': out_track_id,
+        'rich': True,
+        'updated': isinstance(update_target, dict),
+    }
 
 
 def parse_callback_data(value: str) -> dict[str, typing.Any]:
     parts = value.split(':')
     if len(parts) == 4 and parts[0] == 'lbi' and parts[1] and parts[2] == 'a' and parts[3].isdigit():
         return {'callback_token': parts[1], 'action_ref': int(parts[3])}
+    if len(parts) == 4 and parts[0] == 'lbi' and parts[1] and parts[2] == 'f' and parts[3].isdigit():
+        return {'callback_token': parts[1], 'field_ref': int(parts[3])}
     if (
         len(parts) == 5
         and parts[0] == 'lbi'
@@ -254,7 +286,7 @@ def _find_named_value(value: typing.Any, names: set[str]) -> typing.Any:
 
 
 def _callback_track_id(callback: dict[str, typing.Any]) -> str:
-    value = _find_named_value(callback, {'outTrackId', 'out_track_id', 'outtrackid'})
+    value = _find_named_value(callback, {'outTrackId', 'out_track_id', 'outtrackid', 'card_instance_id'})
     return str(value or '')
 
 
@@ -307,6 +339,24 @@ def interaction_event_from_native(
     callback_data = _find_callback_data(callback)
     if callback_data:
         parsed = parse_callback_data(callback_data)
+        raw_values = _mapping(_find_named_value(callback, {'values'}))
+        context: dict[str, typing.Any] = {}
+        if callback_contexts is not None:
+            track_id = _callback_track_id(callback)
+            context = callback_contexts.get(track_id)
+            if context is None:
+                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 = {}
+            for field_id, value in raw_values.items():
+                if field_types.get(field_id) == 'select' and isinstance(value, list):
+                    value = value[0] if value else ''
+                values[str(field_id)] = value
+            parsed['values'] = values
     else:
         contexts = callback_contexts if callback_contexts is not None else {}
         track_id = _callback_track_id(callback)
@@ -322,12 +372,13 @@ def interaction_event_from_native(
         }
         contexts.pop(track_id, None)
     space_id = str(callback.get('space_id') or '')
-    if 'IM_GROUP.' in space_id:
+    normalized_space_id = space_id.lower()
+    if 'im_group.' in normalized_space_id:
         target_type = 'group'
-        target_id = space_id.split('IM_GROUP.', 1)[1]
-    elif 'IM_ROBOT.' in space_id:
+        target_id = space_id[normalized_space_id.index('im_group.') + len('im_group.') :]
+    elif 'im_robot.' in normalized_space_id:
         target_type = 'person'
-        target_id = space_id.split('IM_ROBOT.', 1)[1]
+        target_id = space_id[normalized_space_id.index('im_robot.') + len('im_robot.') :]
     else:
         raise ValueError('DingTalk interaction callback has no delivery space')
     return platform_events.PlatformSpecificEvent(

src/langbot/pkg/platform/adapters/dingtalk/manifest.yaml

@@ -56,97 +56,12 @@ spec:
         zh_Hans: 使用卡片回复
         zh_Hant: 使用卡片回覆
       description:
-        en_US: Controls Markdown and streaming card replies. Disable to use plain text while keeping your template settings. When enabled, a configured template automatically enables streaming.
-        zh_Hans: 控制 Markdown 和流式卡片回复。关闭后使用纯文本，保留模板配置；开启且已配置模板时自动使用流式回复。
-        zh_Hant: 控制 Markdown 和串流卡片回覆。關閉後使用純文字，保留範本設定；開啟且已設定範本時自動使用串流回覆。
+        en_US: Uses built-in A2UI cards for Markdown, streaming replies, and interactions. No card template is required. The app must have AICard.Instance.Write permission.
+        zh_Hans: 使用内置 A2UI 卡片提供 Markdown、流式回复和交互能力，无需配置卡片模板。应用必须开通 AICard.Instance.Write 权限。
+        zh_Hant: 使用內建 A2UI 卡片提供 Markdown、串流回覆和互動能力，無需設定卡片範本。應用必須開通 AICard.Instance.Write 權限。
       type: boolean
       required: false
       default: true
-    - name: card_template_id
-      show_if:
-        field: markdown_card
-        operator: eq
-        value: true
-      label:
-        en_US: Card Template ID
-        zh_Hans: 卡片模板ID
-        zh_Hant: 卡片範本ID
-      description:
-        en_US: Template for streaming text replies. Save and publish it in DingTalk Card Platform, then paste its ID here to automatically enable streaming while card replies are enabled. Leave empty for ordinary replies. Separate from the human input template.
-        zh_Hans: 用于流式文字回复。在钉钉卡片平台保存并发布模板后，开启卡片回复后，填写 ID 即自动启用流式回复，留空使用普通回复；与人工输入卡片模板分开配置。
-        zh_Hant: 用於串流文字回覆。在釘釘卡片平台儲存並發布範本後，開啟卡片回覆後，填寫 ID 即自動啟用串流回覆，留空使用一般回覆；與人工輸入卡片範本分開設定。
-      type: string
-      required: false
-      default: ""
-    - name: stream_card_template_download
-      show_if:
-        field: markdown_card
-        operator: eq
-        value: true
-      label:
-        en_US: Download Streaming Reply Card Template
-        zh_Hans: 下载流式回复卡片模板
-        zh_Hant: 下載串流回覆卡片範本
-      description:
-        en_US: Displays reasoning and the answer separately. Import through More > Import Card Template in the DingTalk AI card editor, save and publish, then paste its ID into Card Template ID above. Respects the pipeline thinking removal setting.
-        zh_Hans: 分开展示思考与正文。在钉钉 AI 卡片编辑器中选择「更多 → 导入卡片模板」，保存并发布后，将 ID 填入上方「卡片模板 ID」。是否保留思考仍由流水线设置控制。
-        zh_Hant: 分開展示思考與正文。在釘釘 AI 卡片編輯器中選擇「更多 → 匯入卡片範本」，儲存並發布後，將 ID 填入上方「卡片範本 ID」。是否保留思考仍由流水線設定控制。
-      type: download-link
-      required: false
-      default: ''
-      url: /api/v1/platform/adapters/dingtalk/stream-card-template
-      download_filename: dingtalk_stream_card.json
-    - name: card_auto_layout
-      show_if:
-        field: markdown_card
-        operator: eq
-        value: true
-      label:
-        en_US: Card Auto Layout
-        zh_Hans: 卡片宽屏自动布局
-        zh_Hant: 卡片寬螢幕自動佈局
-      description:
-        en_US: Enabled by default. Allows supported cards to adapt to wider windows. The LangBot streaming reply template includes the required variables; you can disable this for a narrower layout.
-        zh_Hans: 默认开启，让支持的卡片适应较宽的聊天窗口。LangBot 流式回复模板已包含所需变量；如需较窄布局，可关闭此项。
-        zh_Hant: 預設開啟，讓支援的卡片適應較寬的聊天視窗。LangBot 串流回覆範本已包含所需變數；如需較窄版面，可關閉此項。
-      type: boolean
-      required: false
-      default: true
-    - name: human_input_card_template_id
-      label:
-        en_US: Human Input Card Template ID
-        zh_Hans: 人工输入卡片模板 ID
-        zh_Hant: 人工輸入卡片範本 ID
-      description:
-        en_US: Optional. Used when a workflow or plugin requests user input, a choice, or confirmation before continuing (for example, a Dify Human Input node). Ordinary chat does not require it. Import the template below and enter its published ID here; streaming text replies use the separate Card Template ID.
-        zh_Hans: 可选。当工作流或插件需要用户填写信息、选择选项或确认后再继续执行时使用，例如 Dify 的人工输入节点。普通聊天无需配置。请导入下方模板，将发布后的 ID 填在这里；流式文字回复使用单独的「卡片模板 ID」。
-        zh_Hant: 可選。當工作流或外掛需要使用者填寫資訊、選擇選項或確認後再繼續執行時使用，例如 Dify 的人工輸入節點。一般聊天無需設定。請匯入下方範本，將發布後的 ID 填在這裡；串流文字回覆使用獨立的「卡片範本 ID」。
-      type: string
-      required: false
-      default: ""
-    - name: human_input_card_template_download
-      label:
-        en_US: Download Human Input Card Template
-        zh_Hans: 下载人工输入卡片模板
-        zh_Hant: 下載人工輸入卡片範本
-      description:
-        en_US: Download the human input template. Create an AI card in DingTalk Card Platform, choose More > Import Card Template in the editor, then save and publish. Paste its ID into the field above.
-        zh_Hans: 下载人工输入模板后，在钉钉卡片平台新建 AI 卡片，在编辑器中选择「更多 → 导入卡片模板」，保存并发布，将 ID 填入上方字段。
-        zh_Hant: 下載人工輸入範本後，在釘釘卡片平台新建 AI 卡片，在編輯器中選擇「更多 → 匯入卡片範本」，儲存並發布，將 ID 填入上方欄位。
-      type: download-link
-      required: false
-      default: ''
-      url: /api/v1/platform/adapters/dingtalk/human-input-card-template
-      download_filename: dingtalk_human_input_card.json
-      help_links:
-        zh: https://langbot.app/docs/zh/usage/platforms/dingtalk#导入人工输入卡片模板
-        en: https://open-dev.dingtalk.com/fe/card
-        ja: https://open-dev.dingtalk.com/fe/card
-      help_label:
-        en_US: Import Guide
-        zh_Hans: 导入指引
-        zh_Hant: 匯入指引
-        ja_JP: インポート手順
 
   supported_events:
     - message.received

src/langbot/pkg/platform/botmgr.py

@@ -898,8 +898,8 @@ class RuntimeBot:
             ),
             state_policy=StatePolicy(state_scopes=['conversation', 'actor', 'subject', 'runner']),
             delivery_policy=DeliveryPolicy(
-                enable_streaming=False,
-                enable_reply=False,
+                enable_streaming=agent.get('kind') == 'agent' and event_type == 'message.received',
+                enable_reply=agent.get('kind') == 'agent' and event_type == 'message.received',
                 enable_interactions=agent.get('kind') != 'event_processor',
             ),
             agent_id=agent.get('uuid'),
@@ -1291,6 +1291,8 @@ class RuntimeBot:
             )
 
         envelope = self._eba_event_to_agent_envelope(event, adapter, execution_id)
+        if target_type == 'agent' and isinstance(event, platform_events.MessageReceivedEvent):
+            envelope.delivery.supports_streaming = True
         if target_type == 'event_processor':
             envelope.data = event.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'})
             if isinstance(event, platform_events.MessageReceivedEvent):

src/langbot/pkg/platform/sources/dingtalk.py

@@ -1,7 +1,10 @@
-from langbot.libs.dingtalk_api.thinking import visible_think_markers
+from langbot.libs.dingtalk_api.a2ui import (
+    binding,
+    create_surface_messages,
+    create_text_card_messages,
+    update_surface_messages,
+)
 import asyncio
-import json
-import pathlib
 import re
 import traceback
 import typing
@@ -317,17 +320,6 @@ def _dingtalk_missing_completed_input_lines(form_data: dict, form_content: str)
     return [line for line in _dingtalk_completed_input_lines(form_data) if line.strip() not in rendered_lines]
 
 
-def _dingtalk_supports_native_field(form_data: dict) -> bool:
-    current_name = str(form_data.get('_current_input_field') or '').strip()
-    if not current_name or form_data.get('_action_select_only'):
-        return False
-    for field in _dingtalk_form_input_defs(form_data):
-        if str(field.get('output_variable_name') or '').strip() != current_name:
-            continue
-        return str(field.get('type') or 'text').strip().lower() not in {'file', 'file-list'}
-    return False
-
-
 def _dingtalk_field_name(field: dict) -> str:
     return str(field.get('output_variable_name') or field.get('name') or field.get('id') or '').strip()
 
@@ -355,28 +347,6 @@ def _dingtalk_select_options(field: dict) -> list[str]:
     return []
 
 
-def _dingtalk_select_block_options(options: list[str]) -> list[dict]:
-    """Build the option shape consumed by DingTalk SelectBlock templates."""
-    locales = (
-        'zh_CN',
-        'zh_TW',
-        'en_US',
-        'ja_JP',
-        'vi_VN',
-        'th_TH',
-        'id_ID',
-        'ne_NP',
-        'ms_MY',
-        'ko_KR',
-        'ru_RU',
-        'es_EA',
-        'tr_TR',
-        'fr_FR',
-        'pt_BR',
-    )
-    return [{'value': option, 'text': {locale: option for locale in locales}} for option in options]
-
-
 def _dingtalk_current_input_field(form_data: dict) -> dict | None:
     current_name = str(form_data.get('_current_input_field') or '').strip()
     if not current_name or form_data.get('_action_select_only'):
@@ -387,62 +357,8 @@ def _dingtalk_current_input_field(form_data: dict) -> dict | None:
     return None
 
 
-def _dingtalk_form_component_params(form_data: dict) -> dict:
-    field = _dingtalk_current_input_field(form_data)
-    params = {
-        'input_visible': '',
-        'input_title': '',
-        'input_placeholder': '',
-        'input_value': '',
-        'select_visible': '',
-        'select_placeholder': '',
-        'select_options': [],
-        'index_o': [],
-        'test_index': [],
-        'select_index': -1,
-    }
-    if not field:
-        return params
-
-    field_name = _dingtalk_field_name(field)
-    field_type = _dingtalk_field_type(field)
-    value = form_data.get('inputs', {}).get(field_name, '')
-    if field_type == 'select':
-        options = _dingtalk_select_options(field)
-        selected_index = -1
-        if value not in (None, ''):
-            try:
-                selected_index = options.index(str(value))
-            except ValueError:
-                selected_index = -1
-        params.update(
-            {
-                'select_visible': 'true',
-                'select_placeholder': field_name or 'Select',
-                'select_options': options,
-                'index_o': _dingtalk_select_block_options(options),
-                'test_index': _dingtalk_select_block_options(options),
-                'select_index': selected_index,
-            }
-        )
-    elif field_type not in {'file', 'file-list'}:
-        params.update(
-            {
-                'input_visible': 'true',
-                'input_title': field_name or 'Input',
-                'input_placeholder': field_name or 'Input',
-                'input_value': '' if value is None else str(value),
-            }
-        )
-    return params
-
-
-def _dingtalk_empty_form_component_params() -> dict:
-    return _dingtalk_form_component_params({})
-
-
 def _dingtalk_extract_component_inputs(params: dict) -> dict:
-    """Normalize DingTalk native Input/SelectBlock callback payloads."""
+    """Normalize legacy template and A2UI input callback payloads."""
     if not isinstance(params, dict):
         return {}
 
@@ -468,6 +384,8 @@ def _dingtalk_extract_component_inputs(params: dict) -> dict:
                 break
     if isinstance(select_value, dict):
         select_value = select_value.get('value') or select_value.get('label') or select_value.get('index')
+    if isinstance(select_value, list):
+        select_value = select_value[0] if select_value else ''
     if select_value not in (None, ''):
         if isinstance(select_value, str):
             select_value = select_value.strip()
@@ -476,6 +394,134 @@ def _dingtalk_extract_component_inputs(params: dict) -> dict:
     return result
 
 
+def _dingtalk_a2ui_form(
+    form_data: dict,
+    out_track_id: str,
+    *,
+    prior_content: str = '',
+) -> tuple[dict, list[dict]]:
+    node_title = str(form_data.get('node_title') or 'Human Input Required')
+    form_content = _dingtalk_clean_form_content(form_data)
+    content_parts = [part for part in (prior_content.strip(), f'## {node_title}', form_content) if part]
+    data: dict[str, typing.Any] = {'content': '\n\n'.join(content_parts), 'form': {}}
+    components: list[dict] = [
+        {'id': 'root', 'component': 'Column', 'children': ['content'], 'gap': 12},
+        {'id': 'content', 'component': 'Markdown', 'content': binding('/content')},
+    ]
+    children = components[0]['children']
+
+    field = _dingtalk_current_input_field(form_data)
+    if field is not None:
+        field_name = _dingtalk_field_name(field) or 'value'
+        field_type = _dingtalk_field_type(field)
+        default = (form_data.get('inputs') or {}).get(field_name, field.get('default'))
+        label = str(field.get('label') or field_name)
+        placeholder = str(field.get('placeholder') or '')
+        if field_type == 'select':
+            options = [{'label': option, 'value': option} for option in _dingtalk_select_options(field)]
+            data['form']['value'] = [] if default in (None, '') else [str(default)]
+            components.append(
+                {
+                    'id': 'field',
+                    'component': 'ChoicePicker',
+                    'label': placeholder or label,
+                    'variant': 'mutuallyExclusive',
+                    'displayStyle': 'dropdown',
+                    'options': options,
+                    'value': binding('/form/value'),
+                }
+            )
+            value_key = 'select'
+        elif field_type == 'number':
+            data['form']['value'] = default if isinstance(default, (int, float)) else 0
+            components.append(
+                {
+                    'id': 'field',
+                    'component': 'NumberInput',
+                    'label': label,
+                    'placeholder': placeholder,
+                    'value': binding('/form/value'),
+                }
+            )
+            value_key = 'input'
+        else:
+            data['form']['value'] = '' if default is None else str(default)
+            components.append(
+                {
+                    'id': 'field',
+                    'component': 'TextField',
+                    'label': label,
+                    'placeholder': placeholder,
+                    'variant': 'longText' if field_type == 'textarea' else 'shortText',
+                    'value': binding('/form/value'),
+                }
+            )
+            value_key = 'input'
+        components.extend(
+            [
+                {
+                    'id': 'submit-input',
+                    'component': 'Button',
+                    'child': 'submit-input-label',
+                    'variant': 'primary',
+                    'action': {
+                        'event': {
+                            'name': 'langbot_form_input',
+                            'context': {
+                                'actionId': '__langbot_input__',
+                                'out_track_id': out_track_id,
+                                value_key: binding('/form/value'),
+                            },
+                        }
+                    },
+                    'checks': [
+                        {
+                            'condition': {
+                                'call': 'required',
+                                'args': {'value': binding('/form/value')},
+                            },
+                            'message': f'{label} is required',
+                        }
+                    ],
+                },
+                {'id': 'submit-input-label', 'component': 'Text', 'text': 'Submit'},
+            ]
+        )
+        children.extend(['field', 'submit-input'])
+    else:
+        for index, action in enumerate(form_data.get('actions') or []):
+            action_id = str(action.get('id') or '')
+            label_id = f'action-label-{index}'
+            button_id = f'action-{index}'
+            style = str(action.get('button_style') or '')
+            components.extend(
+                [
+                    {
+                        'id': button_id,
+                        'component': 'Button',
+                        'child': label_id,
+                        'variant': 'primary' if style == 'primary' or (not style and index == 0) else 'default',
+                        'action': {
+                            'event': {
+                                'name': 'langbot_form_action',
+                                'context': {
+                                    'actionId': action_id,
+                                    'out_track_id': out_track_id,
+                                },
+                            }
+                        },
+                    },
+                    {
+                        'id': label_id,
+                        'component': 'Text',
+                        'text': str(action.get('title') or action_id or f'Option {index + 1}'),
+                    },
+                ]
+            )
+            children.append(button_id)
+    return data, components
+
+
 class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     bot: DingTalkClient
     bot_account_id: str
@@ -508,17 +554,6 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     ]
     ap: typing.Any = None
     bot_uuid: str = ''
-    # DingTalk media_id (`@xxx` format) for the bot avatar image, fetched
-    # on adapter startup by uploading the bundled LangBot logo via the
-    # legacy /media/upload endpoint. Empty string when the upload hasn't
-    # run yet or failed — the template's Avatar then falls back to its
-    # default (initials of `name`).
-    bot_avatar_media_id: str = ''
-
-    # Path to the LangBot logo bundled in the repo (`res/logo-blue.png`),
-    # resolved relative to this file. Updated to find the file even when
-    # LangBot is installed as a package or run from a different cwd.
-    _LOGO_PATH: typing.ClassVar[pathlib.Path] = pathlib.Path(__file__).resolve().parents[5] / 'res' / 'logo-blue.png'
 
     def __init__(self, config: dict, logger: EventLogger):
         required_keys = [
@@ -633,12 +668,10 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         message_id = bot_message.resp_message_id
         msg_seq = bot_message.msg_sequence
 
-        form_template_id = (self.config.get('human_input_card_template_id') or '').strip()
         form_data = getattr(bot_message, '_form_data', None)
         if is_final and self.ap is not None:
             self.ap.logger.info(
-                f'DingTalk reply_message_chunk final: form_data_present={form_data is not None}, '
-                f'form_template_configured={bool(form_template_id)}'
+                f'DingTalk reply_message_chunk final: form_data_present={form_data is not None}, a2ui_enabled=True'
             )
 
         if form_data and is_final:
@@ -672,51 +705,16 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                     return
 
             card_instance, card_instance_id = chat_card_entry
-            # btns is reserved exclusively for Dify form-action buttons.
-            # The template renders an Avatar header above the markdown
-            # content; no feedback buttons get injected here.
             if content:
-                if form_template_id:
-                    # The card content has already been written via
-                    # update_card_data (in _paint_form_on_card and the
-                    # initial card creation). The streaming endpoint
-                    # (PUT /v1.0/card/streaming) does not propagate
-                    # updates on cards whose content was last set via
-                    # update_card_data — they take different code paths
-                    # on the DingTalk client. Stick with update_card_data
-                    # for the whole turn for consistency.
-                    try:
-                        await self.bot.update_card_data(
-                            out_track_id=card_instance_id,
-                            card_param_map=self._card_params(
-                                content=content,
-                                btns='[]',
-                                flowStatus='3' if is_final else '1',
-                                **_dingtalk_empty_form_component_params(),
-                            ),
-                        )
-                        session_key = self._session_key_from_event(message_source)
-                        if session_key:
-                            self.active_turn_text[session_key] = content
-                    except Exception:
-                        if self.ap is not None:
-                            self.ap.logger.exception('DingTalk: update card content failed')
-                else:
-                    await self.bot.send_card_message(card_instance, card_instance_id, content, is_final)
+                await self.bot.send_card_message(card_instance, card_instance_id, content, is_final)
+                session_key = self._session_key_from_event(message_source)
+                if session_key:
+                    self.active_turn_text[session_key] = content
             if is_final:
-                if form_template_id and not content:
-                    # Empty final chunk still needs to leave the card with
-                    # flowStatus=3 so the spinner stops.
-                    try:
-                        await self.bot.update_card_data(
-                            out_track_id=card_instance_id,
-                            card_param_map=self._card_params(
-                                flowStatus='3',
-                                **_dingtalk_empty_form_component_params(),
-                            ),
-                        )
-                    except Exception:
-                        pass
+                if not content:
+                    session_key = self._session_key_from_event(message_source)
+                    final_content = self.active_turn_text.get(session_key, '') if session_key else ''
+                    await self.bot.send_card_message(card_instance, card_instance_id, final_content, True)
                 if bot_message.tool_calls is None:
                     self.card_instance_id_dict.pop(message_id, None)
                     event = await DingTalkEventConverter.yiri2target(message_source)
@@ -737,63 +735,24 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
 
     async def create_message_card(self, message_id, event):
         self._prune_card_state()
-        form_template_id = (self.config.get('human_input_card_template_id') or '').strip()
-        legacy_template_id = (self.config.get('card_template_id') or '').strip()
 
         # Synthetic events (button clicks): look up the card already in
         # active_turn_card so reply_message_chunk can stream to it.
         if event is None or event.source_platform_object is None:
-            if form_template_id:
-                session_key = self._session_key_from_event(event) if event is not None else ''
-                carry = self.active_turn_card.get(session_key, '') if session_key else ''
-                if carry:
-                    self.card_instance_id_dict[message_id] = (None, carry)
-                    return True
+            session_key = self._session_key_from_event(event) if event is not None else ''
+            carry = self.active_turn_card.get(session_key, '') if session_key else ''
+            if carry:
+                self.card_instance_id_dict[message_id] = ({'surface_id': f'langbot-{carry}'}, carry)
+                return True
             return False
 
-        if form_template_id:
-            # Create one card with the form template, empty buttons,
-            # pending state. Streaming writes content to it; form pause
-            # paints buttons onto it. One card per turn, no duplication.
-            incoming_message = event.source_platform_object.incoming_message
-            out_track_id = uuid.uuid4().hex
-            is_group = str(incoming_message.conversation_type) == '2'
-            if is_group:
-                open_space_id = f'dtv1.card//IM_GROUP.{incoming_message.conversation_id}'
-            else:
-                open_space_id = f'dtv1.card//IM_ROBOT.{incoming_message.sender_staff_id}'
-            try:
-                await self.bot.create_and_deliver_card(
-                    card_template_id=form_template_id,
-                    out_track_id=out_track_id,
-                    open_space_id=open_space_id,
-                    is_group=is_group,
-                    card_param_map=self._card_params(
-                        content='',
-                        btns='[]',
-                        flowStatus='1',
-                        **_dingtalk_empty_form_component_params(),
-                    ),
-                    callback_type='STREAM',
-                )
-            except Exception:
-                if self.ap is not None:
-                    self.ap.logger.exception('DingTalk: create form-template card failed')
-                return False
-            self.card_instance_id_dict[message_id] = (None, out_track_id)
-            session_key = self._session_key_from_event(event)
-            if session_key:
-                self.active_turn_card[session_key] = out_track_id
-                self.active_turn_text[session_key] = ''
-            return True
-
-        # Legacy chat-card path (no form template).
         incoming_message = event.source_platform_object.incoming_message
-        card_auto_layout = self.config.get('card_auto_layout', True)
-        card_instance, card_instance_id = await self.bot.create_and_card(
-            legacy_template_id, incoming_message, card_auto_layout=card_auto_layout
-        )
+        card_instance, card_instance_id = await self.bot.create_and_card(incoming_message)
         self.card_instance_id_dict[message_id] = (card_instance, card_instance_id)
+        session_key = self._session_key_from_event(event)
+        if session_key:
+            self.active_turn_card[session_key] = card_instance_id
+            self.active_turn_text[session_key] = ''
         return True
 
     def _session_key_from_event(self, event) -> str:
@@ -839,38 +798,8 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             self.bot.on_message('GroupMessage')(on_message)
 
     async def run_async(self):
-        # Upload the bundled LangBot logo so the card Avatar can render
-        # via DingTalk's media CDN — external URLs (e.g. raw.githubusercontent)
-        # are blocked by DingTalk's Avatar.imageUrl resolver. Non-fatal if
-        # the upload fails: cards still render without an avatar image.
-        if self._LOGO_PATH.exists():
-            media_id = await self.bot.upload_image_media(str(self._LOGO_PATH))
-            if media_id:
-                self.bot_avatar_media_id = media_id
-                if self.ap is not None:
-                    self.ap.logger.info(f'DingTalk bot avatar uploaded: media_id={media_id}')
-            else:
-                if self.ap is not None:
-                    self.ap.logger.warning('DingTalk bot avatar upload failed; card will use default')
-        else:
-            if self.ap is not None:
-                self.ap.logger.warning(f'DingTalk bot avatar source not found: {self._LOGO_PATH}')
         await self.bot.start()
 
-    def _card_params(self, **extra) -> dict:
-        """Build a cardParamMap dict that always carries `bot_avatar`
-        (when uploaded) alongside whatever caller-specific params. The
-        bot_avatar key gets dropped on every update_card_data call —
-        DingTalk wipes unspecified template variables, so re-sending it
-        on each update is mandatory."""
-        params = {}
-        if self.bot_avatar_media_id:
-            params['bot_avatar'] = self.bot_avatar_media_id
-        params.update(extra)
-        if isinstance(params.get('content'), str):
-            params['content'] = visible_think_markers(params['content'])
-        return params
-
     async def kill(self) -> bool:
         task_set = getattr(self, '_background_tasks', set())
         background_tasks = list(task_set)
@@ -916,8 +845,8 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     def _derive_open_space(self, message_source: platform_events.MessageEvent) -> tuple[str, bool]:
         """Return (openSpaceId, is_group) for the given inbound event."""
         if isinstance(message_source, platform_events.GroupMessage):
-            return f'dtv1.card//IM_GROUP.{message_source.group.id}', True
-        return f'dtv1.card//IM_ROBOT.{message_source.sender.id}', False
+            return f'dtv1.card//im_group.{message_source.group.id}', True
+        return f'dtv1.card//im_robot.{message_source.sender.id}', False
 
     def _derive_session_descriptor(
         self, message_source: platform_events.MessageEvent
@@ -942,53 +871,23 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         message: platform_message.MessageChain,
         form_data: dict,
     ) -> None:
-        """Surface human-input prompt + buttons on the active card.
-
-        In single-card mode (form_template_id configured): update the
-        EXISTING card with form buttons so it transitions from streaming
-        output to prompt+buttons on the same card. In legacy mode:
-        finalize the chat card and deliver a separate form card.
-        """
+        """Surface a human-input prompt on the active A2UI card."""
         if self.ap is not None:
             self.ap.logger.info(
                 f'DingTalk _handle_form_chunk: actions={len(form_data.get("actions") or [])}, '
                 f'node_title={form_data.get("node_title", "")!r}'
             )
         message_id = bot_message.resp_message_id
-        template_id = (self.config.get('human_input_card_template_id') or '').strip()
-
-        if template_id:
-            # Single-card mode: paint prompt + buttons onto the existing card.
-            session_key = self._session_key_from_event(message_source)
-            entry = self.card_instance_id_dict.get(message_id)
-            out_track_id = entry[1] if entry else None
-            if not out_track_id and session_key:
-                out_track_id = self.active_turn_card.get(session_key, '')
-            if out_track_id:
-                await self._paint_form_on_card(message_source, out_track_id, form_data, session_key)
-                self.card_instance_id_dict.pop(message_id, None)
-                return
-
-            # No existing card (e.g. Dify paused immediately with no LLM
-            # output before the pause). Create a form card directly.
-            await self._send_form_card(message_source, form_data, template_id)
-            self.card_instance_id_dict.pop(message_id, None)
-            return
-
-        # Legacy mode: finalize the streaming card with text fallback.
-        chat_card_entry = self.card_instance_id_dict.pop(message_id, None)
-        if chat_card_entry is not None:
-            _, chat_out_track_id = chat_card_entry
-            markdown_enabled = self.config.get('markdown_card', False)
-            text_content, _ = await DingTalkMessageConverter.yiri2target(message, markdown_enabled)
-            if not text_content and bot_message.content:
-                text_content = bot_message.content
-            try:
-                await self.bot.send_card_message(None, chat_out_track_id, text_content or '​', True)
-            except Exception:
-                await self.logger.error(f'DingTalk: finalize chat card before form failed: {traceback.format_exc()}')
-
-        await self.send_message_text_form(message_source, form_data)
+        session_key = self._session_key_from_event(message_source)
+        entry = self.card_instance_id_dict.get(message_id)
+        out_track_id = entry[1] if entry else None
+        if not out_track_id and session_key:
+            out_track_id = self.active_turn_card.get(session_key, '')
+        if out_track_id:
+            await self._paint_form_on_card(message_source, out_track_id, form_data, session_key)
+        else:
+            await self._send_form_card(message_source, form_data)
+        self.card_instance_id_dict.pop(message_id, None)
 
     async def _paint_form_on_card(
         self,
@@ -1002,13 +901,6 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         node_title = form_data.get('node_title', '') or 'Human Input Required'
         form_content = _dingtalk_clean_form_content(form_data)
         should_show_actions = not _dingtalk_pending_input_defs(form_data)
-        component_params = _dingtalk_form_component_params(form_data)
-        native_field = _dingtalk_supports_native_field(form_data)
-        if self.ap is not None and component_params.get('select_visible'):
-            self.ap.logger.info(
-                f'DingTalk form select params: field={form_data.get("_current_input_field", "")!r} '
-                f'options={len(component_params.get("select_options") or [])}'
-            )
 
         # Record form state for the click-handler.
         self._prune_card_state()
@@ -1031,37 +923,18 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             'created_at': time.monotonic(),
         }
 
-        btns = self._build_btns(actions if should_show_actions else [], out_track_id)
-        parts: list[str] = []
         prior = self.active_turn_text.get(session_key, '') if session_key else ''
-        if prior.strip():
-            parts.append(prior.rstrip())
-            parts.append('<hr>')
-        # DingTalk's card markdown widget strips `\n\n` paragraph breaks in
-        # template content slots, fusing inline siblings into a single line.
-        # Force visual line breaks with explicit HTML `<br>` tags so the
-        # title sits on its own line above form_content.
-        if node_title:
-            parts.append(f'**{node_title}**')
-        if form_content:
-            parts.append(_dingtalk_card_markdown(form_content))
-        missing_completed_lines = _dingtalk_missing_completed_input_lines(form_data, form_content)
-        if missing_completed_lines:
-            parts.append('<hr>' + '<br>'.join(missing_completed_lines))
-        input_hint_lines = [] if native_field else _dingtalk_input_hint_lines(form_data)
-        if input_hint_lines:
-            parts.append('Fill these fields in chat before choosing an action:<br>' + '<br>'.join(input_hint_lines))
-        display_content = '<br><br>'.join(parts)
+        data, components = _dingtalk_a2ui_form(form_data, out_track_id, prior_content=prior)
 
         try:
-            await self.bot.update_card_data(
+            await self.bot.update_ai_card(
                 out_track_id=out_track_id,
-                card_param_map=self._card_params(
-                    content=display_content,
-                    btns=json.dumps(btns, ensure_ascii=False),
-                    flowStatus='3',
-                    **component_params,
+                a2ui_messages=update_surface_messages(
+                    f'langbot-{out_track_id}',
+                    data=data,
+                    components=components,
                 ),
+                flow_status='CONFIRMING',
             )
         except Exception:
             if self.ap is not None:
@@ -1069,40 +942,10 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             await self.send_message_text_form(message_source, form_data)
             return
 
-    @staticmethod
-    def _build_btns(actions: list, out_track_id: str) -> list:
-        btns = []
-        for idx, action in enumerate(actions):
-            action_id = str(action.get('id') or '')
-            title = str(action.get('title') or action_id or f'选项 {idx + 1}')
-            style = (action.get('button_style') or '').lower()
-            if style == 'primary' or (style == '' and idx == 0):
-                color = 'blue'
-            elif style == 'danger':
-                color = 'red'
-            else:
-                color = 'gray'
-            btns.append(
-                {
-                    'text': title,
-                    'color': color,
-                    'status': 'normal',
-                    'event': {
-                        'type': 'sendCardRequest',
-                        'params': {
-                            'actionId': action_id,
-                            'params': {'action_id': action_id, 'out_track_id': out_track_id},
-                        },
-                    },
-                }
-            )
-        return btns
-
     async def _send_form_card(
         self,
         message_source: platform_events.MessageEvent,
         form_data: dict,
-        template_id: str,
     ) -> None:
         """Deliver a new card pre-loaded with the human-input prompt + buttons."""
         out_track_id = uuid.uuid4().hex
@@ -1114,13 +957,6 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         node_title = form_data.get('node_title', '') or 'Human Input Required'
         form_content = _dingtalk_clean_form_content(form_data)
         should_show_actions = not _dingtalk_pending_input_defs(form_data)
-        component_params = _dingtalk_form_component_params(form_data)
-        native_field = _dingtalk_supports_native_field(form_data)
-        if self.ap is not None and component_params.get('select_visible'):
-            self.ap.logger.info(
-                f'DingTalk form select params: field={form_data.get("_current_input_field", "")!r} '
-                f'options={len(component_params.get("select_options") or [])}'
-            )
 
         self._prune_card_state()
         self.card_state[out_track_id] = {
@@ -1143,38 +979,25 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             'created_at': time.monotonic(),
         }
 
-        parts = []
-        if node_title:
-            parts.append(f'**{node_title}**')
-        if form_content:
-            parts.append(_dingtalk_card_markdown(form_content))
-        missing_completed_lines = _dingtalk_missing_completed_input_lines(form_data, form_content)
-        if missing_completed_lines:
-            parts.append('<hr>' + '<br>'.join(missing_completed_lines))
-        input_hint_lines = [] if native_field else _dingtalk_input_hint_lines(form_data)
-        if input_hint_lines:
-            parts.append('Fill these fields in chat before choosing an action:<br>' + '<br>'.join(input_hint_lines))
-        display_content = '<br><br>'.join(parts)
-
-        btns = self._build_btns(actions if should_show_actions else [], out_track_id)
+        data, components = _dingtalk_a2ui_form(form_data, out_track_id)
 
         try:
             if self.ap is not None:
                 self.ap.logger.info(
-                    f'DingTalk _send_form_card: out_track_id={out_track_id} template_id={template_id} '
-                    f'open_space_id={open_space_id} is_group={is_group} btns={len(btns)}'
+                    f'DingTalk _send_form_card: out_track_id={out_track_id} '
+                    f'open_space_id={open_space_id} is_group={is_group} actions={len(actions)}'
                 )
-            await self.bot.create_and_deliver_card(
-                card_template_id=template_id,
+            await self.bot.create_and_deliver_ai_card(
                 out_track_id=out_track_id,
                 open_space_id=open_space_id,
                 is_group=is_group,
-                card_param_map=self._card_params(
-                    content=display_content,
-                    btns=json.dumps(btns, ensure_ascii=False),
-                    flowStatus='3',
-                    **component_params,
+                a2ui_messages=create_surface_messages(
+                    f'langbot-{out_track_id}',
+                    data=data,
+                    components=components,
                 ),
+                flow_status='CONFIRMING',
+                summary=node_title,
                 callback_type='STREAM',
             )
         except Exception:
@@ -1193,43 +1016,24 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         card stays put with the selection notice, and a fresh card is
         spawned here for the LLM reply to stream into.
         """
-        form_template_id = (self.config.get('human_input_card_template_id') or '').strip()
-        legacy_template_id = (self.config.get('card_template_id') or '').strip()
-        template_id = form_template_id or legacy_template_id
-        if not template_id:
-            return None
         out_track_id = uuid.uuid4().hex
         open_space_id, is_group = self._derive_open_space(message_source)
-        if form_template_id:
-            card_param_map = self._card_params(
-                content='',
-                btns='[]',
-                flowStatus='1',
-                **_dingtalk_empty_form_component_params(),
-            )
-            card_data_config = None
-        else:
-            # Legacy chat-card template doesn't carry a `bot_avatar`
-            # variable, so don't decorate the param map here.
-            card_param_map = {'content': '', 'query': '...'}
-            card_data_config = {'autoLayout': self.config.get('card_auto_layout', True)}
+        surface_id = f'langbot-{out_track_id}'
         try:
-            success = await self.bot.create_and_deliver_card(
-                card_template_id=template_id,
+            await self.bot.create_and_deliver_ai_card(
                 out_track_id=out_track_id,
                 open_space_id=open_space_id,
                 is_group=is_group,
-                card_param_map=card_param_map,
-                card_data_config=card_data_config,
+                a2ui_messages=create_text_card_messages(surface_id),
+                flow_status='PROCESSING',
+                summary='LangBot is replying',
                 callback_type='STREAM',
             )
         except Exception:
             if self.ap is not None:
                 self.ap.logger.exception('DingTalk: lazy create resume chat card failed')
             return None
-        if not success:
-            return None
-        entry = (None, out_track_id)
+        entry = ({'surface_id': surface_id}, out_track_id)
         self.card_instance_id_dict[message_id] = entry
         # Register as the active card so any further chunks on this turn
         # (and a subsequent re-pause) land on the same new card.
@@ -1601,14 +1405,17 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         if self.ap is not None:
             self.ap.logger.info(f'DingTalk _mark_card_resolved: out_track_id={out_track_id} action={action_title!r}')
         try:
-            await self.bot.update_card_data(
+            await self.bot.update_ai_card(
                 out_track_id=out_track_id,
-                card_param_map=self._card_params(
-                    content=content,
-                    btns='[]',
-                    flowStatus='3',
-                    **_dingtalk_empty_form_component_params(),
+                a2ui_messages=update_surface_messages(
+                    f'langbot-{out_track_id}',
+                    data={'content': content},
+                    components=[
+                        {'id': 'root', 'component': 'Column', 'children': ['content'], 'gap': 8},
+                        {'id': 'content', 'component': 'Markdown', 'content': binding('/content')},
+                    ],
                 ),
+                flow_status='CONFIRMED',
             )
         except Exception:
             if self.ap is not None:

src/langbot/pkg/platform/sources/dingtalk.yaml

@@ -55,97 +55,12 @@ spec:
         zh_Hans: 使用卡片回复
         zh_Hant: 使用卡片回覆
       description:
-        en_US: Controls Markdown and streaming card replies. Disable to use plain text while keeping your template settings. When enabled, a configured template automatically enables streaming.
-        zh_Hans: 控制 Markdown 和流式卡片回复。关闭后使用纯文本，保留模板配置；开启且已配置模板时自动使用流式回复。
-        zh_Hant: 控制 Markdown 和串流卡片回覆。關閉後使用純文字，保留範本設定；開啟且已設定範本時自動使用串流回覆。
+        en_US: Uses built-in A2UI cards for Markdown, streaming replies, and interactions. No card template is required. The app must have AICard.Instance.Write permission.
+        zh_Hans: 使用内置 A2UI 卡片提供 Markdown、流式回复和交互能力，无需配置卡片模板。应用必须开通 AICard.Instance.Write 权限。
+        zh_Hant: 使用內建 A2UI 卡片提供 Markdown、串流回覆和互動能力，無需設定卡片範本。應用必須開通 AICard.Instance.Write 權限。
       type: boolean
       required: false
       default: true
-    - name: card_template_id
-      show_if:
-        field: markdown_card
-        operator: eq
-        value: true
-      label:
-        en_US: card template id
-        zh_Hans: 卡片模板ID
-        zh_Hant: 卡片範本ID
-      description:
-        en_US: Template for streaming text replies. Save and publish it in DingTalk Card Platform, then paste its ID here to automatically enable streaming while card replies are enabled. Leave empty for ordinary replies. Separate from the human input template.
-        zh_Hans: 用于流式文字回复。在钉钉卡片平台保存并发布模板后，开启卡片回复后，填写 ID 即自动启用流式回复，留空使用普通回复；与人工输入卡片模板分开配置。
-        zh_Hant: 用於串流文字回覆。在釘釘卡片平台儲存並發布範本後，開啟卡片回覆後，填寫 ID 即自動啟用串流回覆，留空使用一般回覆；與人工輸入卡片範本分開設定。
-      type: string
-      required: false
-      default: ""
-    - name: stream_card_template_download
-      show_if:
-        field: markdown_card
-        operator: eq
-        value: true
-      label:
-        en_US: Download Streaming Reply Card Template
-        zh_Hans: 下载流式回复卡片模板
-        zh_Hant: 下載串流回覆卡片範本
-      description:
-        en_US: Displays reasoning and the answer separately. Import through More > Import Card Template in the DingTalk AI card editor, save and publish, then paste its ID into Card Template ID above. Respects the pipeline thinking removal setting.
-        zh_Hans: 分开展示思考与正文。在钉钉 AI 卡片编辑器中选择「更多 → 导入卡片模板」，保存并发布后，将 ID 填入上方「卡片模板 ID」。是否保留思考仍由流水线设置控制。
-        zh_Hant: 分開展示思考與正文。在釘釘 AI 卡片編輯器中選擇「更多 → 匯入卡片範本」，儲存並發布後，將 ID 填入上方「卡片範本 ID」。是否保留思考仍由流水線設定控制。
-      type: download-link
-      required: false
-      default: ''
-      url: /api/v1/platform/adapters/dingtalk/stream-card-template
-      download_filename: dingtalk_stream_card.json
-    - name: card_auto_layout
-      show_if:
-        field: markdown_card
-        operator: eq
-        value: true
-      label:
-        en_US: Card Auto Layout
-        zh_Hans: 卡片宽屏自动布局
-        zh_Hant: 卡片寬螢幕自動佈局
-      description:
-        en_US: Enabled by default. Allows supported cards to adapt to wider windows. The LangBot streaming reply template includes the required variables; you can disable this for a narrower layout.
-        zh_Hans: 默认开启，让支持的卡片适应较宽的聊天窗口。LangBot 流式回复模板已包含所需变量；如需较窄布局，可关闭此项。
-        zh_Hant: 預設開啟，讓支援的卡片適應較寬的聊天視窗。LangBot 串流回覆範本已包含所需變數；如需較窄版面，可關閉此項。
-      type: boolean
-      required: false
-      default: true
-    - name: human_input_card_template_id
-      label:
-        en_US: Human Input Card Template ID
-        zh_Hans: 人工输入卡片模板ID
-        zh_Hant: 人工輸入卡片範本ID
-      description:
-        en_US: Optional. Used when a workflow or plugin requests user input, a choice, or confirmation before continuing (for example, a Dify Human Input node). Ordinary chat does not require it. Import the template below and enter its published ID here; streaming text replies use the separate Card Template ID.
-        zh_Hans: 可选。当工作流或插件需要用户填写信息、选择选项或确认后再继续执行时使用，例如 Dify 的人工输入节点。普通聊天无需配置。请导入下方模板，将发布后的 ID 填在这里；流式文字回复使用单独的「卡片模板 ID」。
-        zh_Hant: 可選。當工作流或外掛需要使用者填寫資訊、選擇選項或確認後再繼續執行時使用，例如 Dify 的人工輸入節點。一般聊天無需設定。請匯入下方範本，將發布後的 ID 填在這裡；串流文字回覆使用獨立的「卡片範本 ID」。
-      type: string
-      required: false
-      default: ""
-    - name: human_input_card_template_download
-      label:
-        en_US: Download Human Input Card Template
-        zh_Hans: 下载人工输入卡片模板
-        zh_Hant: 下載人工輸入卡片範本
-      description:
-        en_US: Download the human input template. Create an AI card in DingTalk Card Platform, choose More > Import Card Template in the editor, then save and publish. Paste its ID into the field above.
-        zh_Hans: 下载人工输入模板后，在钉钉卡片平台新建 AI 卡片，在编辑器中选择「更多 → 导入卡片模板」，保存并发布，将 ID 填入上方字段。
-        zh_Hant: 下載人工輸入範本後，在釘釘卡片平台新建 AI 卡片，在編輯器中選擇「更多 → 匯入卡片範本」，儲存並發布，將 ID 填入上方欄位。
-      type: download-link
-      required: false
-      default: ""
-      url: /api/v1/platform/adapters/dingtalk/human-input-card-template
-      download_filename: dingtalk_human_input_card.json
-      help_links:
-        zh: https://langbot.app/docs/zh/usage/platforms/dingtalk#导入人工输入卡片模板
-        en: https://open-dev.dingtalk.com/fe/card
-        ja: https://open-dev.dingtalk.com/fe/card
-      help_label:
-        en_US: Import Guide
-        zh_Hans: 导入指引
-        zh_Hant: 匯入指引
-        ja_JP: インポート手順
 execution:
   python:
     path: ./dingtalk.py

tests/unit_tests/agent/test_orchestrator_integration.py

@@ -1269,6 +1269,103 @@ class TestQueryEntrySessionQueryId:
         mcp_loader.build_resource_context_for_query.assert_awaited_once_with(execution_query)
 
 
+@pytest.mark.asyncio
+async def test_direct_agent_streams_output_to_the_originating_platform(clean_agent_state):
+    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
+
+    connector = FakePluginConnector(
+        results=[
+            {
+                'type': 'message.delta',
+                'data': {
+                    'chunk': {
+                        'role': 'assistant',
+                        'content': 'hello',
+                        'all_content': 'hello',
+                    }
+                },
+            },
+            {
+                'type': 'message.completed',
+                'data': {'message': {'role': 'assistant', 'content': 'hello world'}},
+            },
+            {'type': 'run.completed', 'data': {'finish_reason': 'stop'}},
+        ]
+    )
+    ap = FakeApplication(connector, clean_agent_state)
+    orchestrator = AgentRunOrchestrator(ap, 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',
+        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=True,
+        ),
+    )
+    binding = AgentBinding(
+        binding_id='binding-1',
+        scope=BindingScope(scope_type='bot', scope_id='bot_001'),
+        event_types=['message.received'],
+        runner_id=RUNNER_ID,
+        resource_policy=ResourcePolicy(),
+        state_policy=StatePolicy(),
+        delivery_policy=DeliveryPolicy(enable_streaming=True, 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,
+                '_execution_context': TEST_CONTEXT,
+            },
+        )
+    ]
+
+    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()
+
+
 class TestQueryEntryAdapterParams:
     """Tests for params handling in Query entry adapter."""
 

tests/unit_tests/agent/test_platform_tools.py

@@ -364,6 +364,36 @@ async def test_event_reply_builds_message_chain_for_the_frozen_target() -> None:
     assert call.kwargs['message'][0].text == 'hello'
 
 
+@pytest.mark.asyncio
+async def test_event_reply_uses_run_reply_stream_when_available() -> None:
+    adapter = SimpleNamespace(
+        get_supported_apis=lambda: ['send_message'],
+        send_message=AsyncMock(return_value=None),
+    )
+    reply_streams = SimpleNamespace(
+        adapter=adapter,
+        apply=AsyncMock(return_value={'status': 'completed', 'delivery': 'streaming'}),
+    )
+    ap = SimpleNamespace(
+        platform_mgr=SimpleNamespace(get_bot_by_uuid=AsyncMock(return_value=SimpleNamespace(adapter=adapter)))
+    )
+    session = {
+        'reply_streams': reply_streams,
+        'authorization': {
+            'bot_id': 'bot-1',
+            'platform_context': freeze_platform_context(_event()),
+        },
+    }
+
+    result = await execute_platform_tool(ap, object(), session, 'event_reply', {'text': 'hello'})
+
+    request = reply_streams.apply.await_args.args[0]
+    assert request.operation == 'finish'
+    assert request.text == 'hello'
+    assert result == {'status': 'completed', 'delivery': 'streaming'}
+    adapter.send_message.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_present_interactive_card_uses_host_interaction_manager() -> None:
     adapter = SimpleNamespace(get_supported_apis=lambda: ['interaction.request'])

tests/unit_tests/platform/test_dingtalk_adapter.py

@@ -8,11 +8,11 @@ import pytest
 from langbot.pkg.platform import botmgr as _botmgr  # noqa: F401
 from langbot.pkg.platform.sources.dingtalk import (
     DingTalkAdapter,
+    _dingtalk_a2ui_form,
     _dingtalk_card_markdown,
     _dingtalk_clean_form_content,
     _dingtalk_completed_input_lines,
     _dingtalk_extract_component_inputs,
-    _dingtalk_form_component_params,
     _dingtalk_missing_completed_input_lines,
     _dingtalk_pending_input_defs,
 )
@@ -29,29 +29,30 @@ def test_dingtalk_auxiliary_tasks_are_bounded():
     assert len(adapter._background_tasks) == 100
 
 
-def test_dingtalk_select_component_params_expose_options():
-    params = _dingtalk_form_component_params(
+def test_dingtalk_dify_input_card_uses_a2ui_binding_and_required_check():
+    data, components = _dingtalk_a2ui_form(
         {
-            '_current_input_field': 'choice',
+            'node_title': 'Review',
+            'form_content': 'Add a comment',
+            '_current_input_field': 'comment',
             'input_defs': [
                 {
-                    'output_variable_name': 'choice',
-                    'type': 'select',
-                    'option_source': {'type': 'constant', 'value': ['A', 'B']},
+                    'output_variable_name': 'comment',
+                    'type': 'textarea',
+                    'label': 'Comment',
                 }
             ],
             'inputs': {},
-        }
+        },
+        'track-id',
     )
 
-    assert params['select_visible'] == 'true'
-    assert params['select_placeholder'] == 'choice'
-    assert params['select_options'] == ['A', 'B']
-    assert [option['value'] for option in params['index_o']] == ['A', 'B']
-    assert [option['value'] for option in params['test_index']] == ['A', 'B']
-    assert params['index_o'][0]['text']['zh_CN'] == 'A'
-    assert params['index_o'][0]['text']['en_US'] == 'A'
-    assert params['select_index'] == -1
+    field = next(component for component in components if component['component'] == 'TextField')
+    button = next(component for component in components if component['component'] == 'Button')
+    assert data['form']['value'] == ''
+    assert field['value'] == {'path': '/form/value'}
+    assert button['action']['event']['context']['input'] == {'path': '/form/value'}
+    assert button['checks'][0]['condition']['call'] == 'required'
 
 
 def test_dingtalk_extract_select_from_builtin_result_dict():

tests/unit_tests/platform/test_dingtalk_api.py

@@ -1,57 +1,19 @@
-"""Tests for DingTalk API payload helpers."""
+"""Tests for DingTalk A2UI API payloads."""
 
 import json
 from contextlib import asynccontextmanager
 from unittest.mock import AsyncMock
 
-from langbot.libs.dingtalk_api.api import DingTalkClient, _stringify_card_param_map
-from langbot.pkg.utils import httpclient
+import pytest
+
+from langbot.libs.dingtalk_api.a2ui import create_text_card_messages, update_text_card_messages
+from langbot.libs.dingtalk_api.api import DingTalkClient
 
 
-def test_dingtalk_card_param_map_stringifies_select_component_arrays():
-    params = _stringify_card_param_map(
-        {
-            'content': 'Pick one',
-            'btns': json.dumps([{'text': 'OK'}], ensure_ascii=False),
-            'select_options': ['A', 'B'],
-            'index_o': [
-                {
-                    'value': 'A',
-                    'text': {'zh_CN': 'A', 'en_US': 'A'},
-                }
-            ],
-            'test_index': [
-                {
-                    'value': 'A',
-                    'text': {'zh_CN': 'A', 'en_US': 'A'},
-                }
-            ],
-            'select_index': -1,
-        }
-    )
-
-    assert params['content'] == 'Pick one'
-    assert params['btns'] == '[{"text": "OK"}]'
-    assert params['select_options'] == '["A", "B"]'
-    assert json.loads(params['index_o'])[0]['value'] == 'A'
-    assert json.loads(params['test_index'])[0]['value'] == 'A'
-    assert params['select_index'] == '-1'
-
-
-def test_dingtalk_card_param_map_stringifies_unregistered_structures():
-    params = _stringify_card_param_map({'other': ['A'], 'empty': None})
-
-    assert params['other'] == '["A"]'
-    assert params['empty'] == ''
-
-
-async def test_create_card_embeds_layout_config_as_template_parameter(monkeypatch):
-    response = type('Response', (), {'status_code': 200, 'json': lambda self: {}})()
-    post = AsyncMock(return_value=response)
-
+def _client(http_client):
     @asynccontextmanager
     async def client_context():
-        yield type('HttpClient', (), {'post': post})()
+        yield http_client
 
     client = object.__new__(DingTalkClient)
     client.access_token = 'access-token'
@@ -59,26 +21,100 @@ async def test_create_card_embeds_layout_config_as_template_parameter(monkeypatc
     client.key = 'client-id'
     client.logger = None
     client.check_access_token = AsyncMock(return_value=True)
+    client.get_access_token = AsyncMock()
     client._http_client_context = client_context
-    monkeypatch.setattr(httpclient, 'response_text', AsyncMock(return_value='{}'))
+    return client
 
-    original_params = {'content': 'hello'}
-    delivered = await client.create_and_deliver_card(
-        card_template_id='template-id',
+
+def _response(result=None):
+    payload = {'success': True, 'result': result or {}}
+    return type('Response', (), {'status_code': 200, 'json': lambda self: payload})()
+
+
+@pytest.mark.asyncio
+async def test_create_ai_card_uses_group_robot_delivery_model():
+    post = AsyncMock(
+        return_value=_response(
+            {
+                'outTrackId': 'track-id',
+                'deliverResults': [{'spaceType': 'IM_GROUP', 'success': True}],
+            }
+        )
+    )
+    client = _client(type('HttpClient', (), {'post': post})())
+
+    result = await client.create_and_deliver_ai_card(
         out_track_id='track-id',
-        open_space_id='dtv1.card//IM_ROBOT.user-id',
-        is_group=False,
-        card_param_map=original_params,
-        card_data_config={'autoLayout': True},
+        open_space_id='dtv1.card//im_group.group-id',
+        is_group=True,
+        a2ui_messages=create_text_card_messages('surface-id', 'hello'),
+        summary='hello',
     )
 
-    request_body = post.await_args.kwargs['json']
-    assert delivered is True
-    assert request_body['cardData'] == {
-        'cardParamMap': {
-            'content': 'hello',
-            'config': '{"autoLayout": true}',
-            'autoLayout': 'true',
-        }
-    }
-    assert original_params == {'content': 'hello'}
+    request = post.await_args
+    body = request.kwargs['json']
+    assert request.args[0].endswith('/v1.0/card/ai/instances/createAndDeliver')
+    assert result['outTrackId'] == 'track-id'
+    assert body['imGroupOpenDeliverModel'] == {'robotCode': 'robot-code'}
+    assert body['imGroupOpenSpaceModel'] == {'supportForward': False}
+    assert 'imRobotOpenDeliverModel' not in body
+    assert 'cardTemplateId' not in body
+    assert body['protocolVersion'] == '1.0'
+    assert body['flowStatus'] == 'PROCESSING'
+    assert all(isinstance(message, str) for message in body['a2uiMessages'])
+    assert json.loads(body['a2uiMessages'][0])['createSurface']['surfaceId'] == 'surface-id'
+
+
+@pytest.mark.asyncio
+async def test_create_ai_card_uses_single_chat_robot_delivery_model():
+    post = AsyncMock(return_value=_response({'deliverResults': [{'spaceType': 'IM_ROBOT', 'success': True}]}))
+    client = _client(type('HttpClient', (), {'post': post})())
+
+    await client.create_and_deliver_ai_card(
+        out_track_id='track-id',
+        open_space_id='dtv1.card//im_robot.user-id',
+        is_group=False,
+        a2ui_messages=create_text_card_messages('surface-id'),
+    )
+
+    body = post.await_args.kwargs['json']
+    assert body['imRobotOpenDeliverModel'] == {'robotCode': 'robot-code'}
+    assert body['imRobotOpenSpaceModel'] == {'supportForward': False}
+    assert 'imGroupOpenDeliverModel' not in body
+
+
+@pytest.mark.asyncio
+async def test_update_ai_card_uses_a2ui_delta_and_idempotency_key():
+    put = AsyncMock(return_value=_response({'outTrackId': 'track-id', 'flowStatus': 'FINISH'}))
+    client = _client(type('HttpClient', (), {'put': put})())
+
+    await client.update_ai_card(
+        out_track_id='track-id',
+        out_update_id='update-id',
+        flow_status='FINISH',
+        a2ui_messages=update_text_card_messages('surface-id', 'done'),
+    )
+
+    request = put.await_args
+    body = request.kwargs['json']
+    assert request.args[0].endswith('/v1.0/card/ai/instances/update')
+    assert body['outTrackId'] == 'track-id'
+    assert body['outUpdateId'] == 'update-id'
+    assert body['flowStatus'] == 'FINISH'
+    assert json.loads(body['a2uiMessages'][0])['updateDataModel']['path'] == '/content'
+
+
+@pytest.mark.asyncio
+async def test_create_ai_card_rejects_failed_delivery_result():
+    post = AsyncMock(
+        return_value=_response({'deliverResults': [{'spaceType': 'IM_GROUP', 'success': False, 'errorMsg': 'denied'}]})
+    )
+    client = _client(type('HttpClient', (), {'post': post})())
+
+    with pytest.raises(RuntimeError, match='delivery failed'):
+        await client.create_and_deliver_ai_card(
+            out_track_id='track-id',
+            open_space_id='dtv1.card//im_group.group-id',
+            is_group=True,
+            a2ui_messages=create_text_card_messages('surface-id'),
+        )

tests/unit_tests/platform/test_dingtalk_eba_adapter.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import json
 import pathlib
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock, patch
@@ -51,7 +50,8 @@ class DummyDingTalkClient(DingTalkClient):
         self.download_image = AsyncMock(return_value='data:image/png;base64,BBBB')
         self.create_and_card = AsyncMock(return_value=('card', 'card-id'))
         self.send_card_message = AsyncMock()
-        self.create_and_deliver_card = AsyncMock(return_value=True)
+        self.create_and_deliver_ai_card = AsyncMock(return_value={})
+        self.update_ai_card = AsyncMock(return_value={})
         self.start = AsyncMock()
         self.stop = AsyncMock()
 
@@ -85,9 +85,6 @@ def make_adapter() -> DingTalkAdapter:
         'robot_code': 'robot-code',
         'markdown_card': True,
         'enable-stream-reply': False,
-        'card_auto_layout': False,
-        'card_template_id': 'template-id',
-        'human_input_card_template_id': 'human-input-template-id',
     }
     with patch('langbot.pkg.platform.adapters.dingtalk.adapter.DingTalkClient', DummyDingTalkClient):
         return DingTalkAdapter(config, DummyLogger())
@@ -161,36 +158,6 @@ def test_dingtalk_platform_api_map_matches_manifest():
     assert set(PLATFORM_API_MAP) == manifest_actions
 
 
-def test_dingtalk_human_input_card_template_matches_interaction_contract():
-    template_path = (
-        pathlib.Path(__file__).parents[3] / 'src' / 'langbot' / 'templates' / 'dingtalk_human_input_card.json'
-    )
-    exported_template = json.loads(template_path.read_text(encoding='utf-8'))
-    editor_data = json.loads(exported_template['editorData'])
-    variable_ids = {item['id'] for item in editor_data['variableList']}
-    serialized_editor = json.dumps(editor_data, ensure_ascii=False)
-
-    assert exported_template['type'] == 'im'
-    assert exported_template['mode'] == 'card'
-    assert {
-        'content',
-        'btns',
-        'input_visible',
-        'input_title',
-        'input_placeholder',
-        'input_value',
-        'select_visible',
-        'select_placeholder',
-        'select_options',
-        'select_index',
-        'index_o',
-    } <= variable_ids
-    assert '"componentName": "Input"' in serialized_editor
-    assert '"componentName": "SelectBlock"' in serialized_editor
-    assert '__built_in_inputResult__' in serialized_editor
-    assert '__built_in_selectResult__' in serialized_editor
-
-
 @pytest.mark.asyncio
 async def test_dingtalk_message_converter_maps_outbound_components():
     content, at = await DingTalkMessageConverter.yiri2target(
@@ -367,10 +334,12 @@ async def test_dingtalk_interaction_delivery_and_callback_event():
     )
 
     assert result['rich'] is True
-    kwargs = adapter.bot.create_and_deliver_card.await_args.kwargs
-    assert kwargs['card_template_id'] == 'human-input-template-id'
-    assert kwargs['open_space_id'] == 'dtv1.card//IM_GROUP.group-1'
-    assert kwargs['card_param_map']['btns'][0]['event']['params']['actionId'] == 'lbi:callback-token:a:0'
+    kwargs = adapter.bot.create_and_deliver_ai_card.await_args.kwargs
+    assert kwargs['open_space_id'] == 'dtv1.card//im_group.group-1'
+    assert kwargs['flow_status'] == 'CONFIRMING'
+    components = kwargs['a2ui_messages'][2]['updateComponents']['components']
+    button = next(component for component in components if component['component'] == 'Button')
+    assert button['action']['event']['context']['actionId'] == 'lbi:callback-token:a:0'
 
     native = dingtalk_card_callback_event(
         content={'actionId': 'lbi:callback-token:a:0'},
@@ -409,20 +378,34 @@ async def test_dingtalk_native_input_uses_template_variables_and_submits_value()
         },
     )
 
-    params = adapter.bot.create_and_deliver_card.await_args.kwargs['card_param_map']
-    assert params['input_visible'] == 'true'
-    assert params['input_title'] == 'Comment'
-    assert params['input_placeholder'] == 'Explain the decision'
-    assert params['select_visible'] == ''
-    assert params['btns'] == []
+    kwargs = adapter.bot.create_and_deliver_ai_card.await_args.kwargs
+    components = kwargs['a2ui_messages'][2]['updateComponents']['components']
+    field = next(component for component in components if component['component'] == 'TextField')
+    button = next(component for component in components if component['component'] == 'Button')
+    assert field['label'] == 'Comment'
+    assert field['placeholder'] == 'Explain the decision'
+    assert field['variant'] == 'longText'
+    assert button['checks'] == [
+        {
+            'condition': {'call': 'required', 'args': {'value': {'path': '/form/comment'}}},
+            'message': 'Comment is required',
+        }
+    ]
     assert result['message_id'] in adapter.interaction_callback_contexts
 
     native = dingtalk_card_callback_event(
         content={
-            'outTrackId': result['message_id'],
-            'params': {'input': 'Looks good'},
+            'actionData': {
+                'a2uiAction': {
+                    'context': {
+                        'actionId': 'lbi:callback-token:f:0',
+                        'values': {'comment': 'Looks good'},
+                    }
+                }
+            }
         },
-        space_id='dtv1.card//IM_GROUP.group-1',
+        card_instance_id=result['message_id'],
+        space_id='dtv1.card//im_group.group-1',
     )
     event = interaction_event_from_native(native, adapter.interaction_callback_contexts)
 
@@ -459,17 +442,25 @@ async def test_dingtalk_native_select_uses_select_block_and_normalizes_callback(
         },
     )
 
-    params = adapter.bot.create_and_deliver_card.await_args.kwargs['card_param_map']
-    assert params['select_visible'] == 'true'
-    assert [option['value'] for option in params['index_o']] == ['normal', 'urgent']
-    assert params['input_visible'] == ''
+    kwargs = adapter.bot.create_and_deliver_ai_card.await_args.kwargs
+    components = kwargs['a2ui_messages'][2]['updateComponents']['components']
+    field = next(component for component in components if component['component'] == 'ChoicePicker')
+    assert field['displayStyle'] == 'dropdown'
+    assert [option['value'] for option in field['options']] == ['normal', 'urgent']
 
     native = dingtalk_card_callback_event(
         content={
-            'out_track_id': result['message_id'],
-            'params': {'select': '{"index": 1, "value": "urgent"}'},
+            'actionData': {
+                'a2uiAction': {
+                    'context': {
+                        'actionId': 'lbi:callback-token:f:0',
+                        'values': {'priority': ['urgent']},
+                    }
+                }
+            }
         },
-        space_id='dtv1.card//IM_ROBOT.user-1',
+        card_instance_id=result['message_id'],
+        space_id='dtv1.card//im_robot.user-1',
     )
     event = interaction_event_from_native(native, adapter.interaction_callback_contexts)
 
@@ -478,6 +469,58 @@ async def test_dingtalk_native_select_uses_select_block_and_normalizes_callback(
     assert event.data['values'] == {'priority': 'urgent'}
 
 
+@pytest.mark.asyncio
+async def test_dingtalk_interaction_update_reuses_existing_a2ui_card():
+    adapter = make_adapter()
+
+    result = await adapter.call_platform_api(
+        'interaction.request',
+        {
+            'callback_token': 'new-token',
+            'reply_target': {'target_type': 'group', 'target_id': 'group-1'},
+            'update_target': {'message_id': 'existing-track-id', 'rich': True},
+            'request': {
+                'interaction_id': 'form-2',
+                'title': 'Updated approval',
+                'actions': [{'id': 'approve', 'label': 'Approve', 'style': 'primary'}],
+                'fallback_text': 'Reply approve.',
+            },
+        },
+    )
+
+    adapter.bot.create_and_deliver_ai_card.assert_not_awaited()
+    kwargs = adapter.bot.update_ai_card.await_args.kwargs
+    assert kwargs['out_track_id'] == 'existing-track-id'
+    assert kwargs['flow_status'] == 'CONFIRMING'
+    assert 'updateDataModel' in kwargs['a2ui_messages'][0]
+    assert 'updateComponents' in kwargs['a2ui_messages'][1]
+    assert result == {
+        'ok': True,
+        'message_id': 'existing-track-id',
+        'rich': True,
+        'updated': True,
+    }
+    assert adapter.interaction_callback_contexts['existing-track-id']['callback_token'] == 'new-token'
+
+
+def test_dingtalk_stale_callback_does_not_consume_replacement_context():
+    contexts = {
+        'card-1': {
+            'callback_token': 'new-token',
+            'field_types': {},
+            'expires_at': float('inf'),
+        }
+    }
+    native = dingtalk_card_callback_event(
+        content={'actionId': 'lbi:old-token:a:0'},
+        card_instance_id='card-1',
+        space_id='dtv1.card//im_group.group-1',
+    )
+
+    assert interaction_event_from_native(native, contexts) is None
+    assert contexts['card-1']['callback_token'] == 'new-token'
+
+
 @pytest.mark.asyncio
 async def test_dingtalk_adapter_dispatches_native_input_submission():
     adapter = make_adapter()
@@ -509,10 +552,17 @@ async def test_dingtalk_adapter_dispatches_native_input_submission():
     )
     native = dingtalk_card_callback_event(
         content={
-            'outTrackId': result['message_id'],
-            'params': {'inputResult': '42.5'},
+            'actionData': {
+                'a2uiAction': {
+                    'context': {
+                        'actionId': 'lbi:callback-token:f:0',
+                        'values': {'score': 42.5},
+                    }
+                }
+            }
         },
-        space_id='dtv1.card//IM_GROUP.group-1',
+        card_instance_id=result['message_id'],
+        space_id='dtv1.card//im_group.group-1',
     )
 
     await adapter._handle_native_event(native)

tests/unit_tests/platform/test_routing_rules.py

@@ -787,8 +787,8 @@ async def test_bound_event_processor_receives_one_complete_typed_event():
 @pytest.mark.parametrize('output_kind', ['message', 'chunks', 'tool_rounds'])
 @pytest.mark.parametrize('explicit_reply', [False, True])
 @pytest.mark.parametrize('runner_fails', [False, True])
-async def test_processor_outputs_require_explicit_platform_actions(kind, output_kind, explicit_reply, runner_fails):
-    """Draining runner results must not send text, duplicate replies, or stream cards."""
+async def test_processor_output_delivery_policy(kind, output_kind, explicit_reply, runner_fails):
+    """Direct Agents may auto-deliver output; event processors remain explicit-only."""
     from langbot_plugin.api.entities.builtin.platform import entities, events, message
     from langbot_plugin.api.entities.builtin.provider import message as provider_message
     from datetime import datetime
@@ -812,9 +812,9 @@ async def test_processor_outputs_require_explicit_platform_actions(kind, output_
             restored = message.MessageChain.model_validate(envelope.data['message_chain'])
             assert restored.source.id == 'source-1'
             assert restored.source.time == datetime(2026, 10, 8, 12)
-        assert envelope.delivery.supports_streaming is False
-        assert binding.delivery_policy.enable_streaming is False
-        assert binding.delivery_policy.enable_reply is False
+        assert envelope.delivery.supports_streaming is (kind == 'agent')
+        assert binding.delivery_policy.enable_streaming is (kind == 'agent')
+        assert binding.delivery_policy.enable_reply is (kind == 'agent')
         assert adapter_context['_delivery_adapter'] is adapter
         if explicit_reply:
             session = {'authorization': {'bot_id': 'bot-1', 'platform_context': freeze_platform_context(envelope)}}