Stream DingTalk replies without a source message

src/langbot/libs/dingtalk_api/api.py

@@ -712,13 +712,20 @@ class DingTalkClient:
 
         Returns ``(card state, out_track_id)`` for subsequent A2UI updates.
         """
+        is_group = str(incoming_message.conversation_type) == '2'
+        return await self.create_and_card_for_target(
+            'group' if is_group else 'person',
+            incoming_message.conversation_id if is_group else incoming_message.sender_staff_id,
+        )
+
+    async def create_and_card_for_target(self, target_type: str, target_id: str):
+        """Create a streaming card when the reply has no original chat message."""
+        if target_type not in {'group', 'person'} or not target_id:
+            raise ValueError('DingTalk streaming card requires a group or person target')
         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}'
-        else:
-            open_space_id = f'dtv1.card//im_robot.{incoming_message.sender_staff_id}'
+        is_group = target_type == 'group'
+        open_space_id = f'dtv1.card//im_group.{target_id}' if is_group else f'dtv1.card//im_robot.{target_id}'
 
         await self.create_and_deliver_ai_card(
             out_track_id=out_track_id,

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

@@ -86,9 +86,24 @@ class ReplyStreamSession:
                         raise ValueError('This run has no platform delivery adapter')
                     # Set before I/O: an uncertain create must never be retried as another send.
                     reply.started = True
-                    reply.native = self.source is not None and await self.adapter.is_stream_output_supported()
+                    target_streaming = (
+                        self.source is None
+                        and callable(getattr(self.adapter, 'create_message_card_for_target', None))
+                        and callable(getattr(self.adapter, 'reply_message_chunk_for_target', None))
+                    )
+                    reply.native = (
+                        self.source is not None or target_streaming
+                    ) and await self.adapter.is_stream_output_supported()
                     if reply.native:
-                        await self.adapter.create_message_card(reply.delivery_id, self.source)
+                        if self.source is not None:
+                            await self.adapter.create_message_card(reply.delivery_id, self.source)
+                        else:
+                            target_type, target_id = self.target.get('target_type'), self.target.get('target_id')
+                            if not target_type or not target_id:
+                                raise ValueError('This event has no reply target')
+                            await self.adapter.create_message_card_for_target(
+                                reply.delivery_id, str(target_type), str(target_id)
+                            )
                 if reply.started and reply.native:
                     await self._update_native(key, reply, final=request.operation != 'update')
                 elif request.operation == 'finish' and reply.text and not self.mock:
@@ -131,13 +146,18 @@ class ReplyStreamSession:
             msg_sequence=reply.sequence,
             is_final=final,
         )
-        await self.adapter.reply_message_chunk(
-            message_source=self.source,
-            bot_message=chunk,
-            message=self._message(reply.text),
-            quote_origin=False,
-            is_final=final,
-        )
+        if self.source is not None:
+            await self.adapter.reply_message_chunk(
+                message_source=self.source,
+                bot_message=chunk,
+                message=self._message(reply.text),
+                quote_origin=False,
+                is_final=final,
+            )
+        else:
+            await self.adapter.reply_message_chunk_for_target(
+                message_id=reply.delivery_id, content=reply.text, is_final=final
+            )
 
     async def close(self):
         """Close visible streams on cancellation; never send a buffered partial reply."""

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

@@ -263,6 +263,25 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo
         self.card_instance_id_dict[message_id] = (card_instance, card_instance_id)
         return True
 
+    async def create_message_card_for_target(self, message_id: str, target_type: str, target_id: str) -> bool:
+        card_instance, card_instance_id = await self.bot.create_and_card_for_target(target_type, target_id)
+        self.card_instance_id_dict[message_id] = (card_instance, card_instance_id)
+        return True
+
+    async def reply_message_chunk_for_target(self, *, message_id: str, content: str, is_final: bool) -> None:
+        now = time.monotonic()
+        last_updates = getattr(self, '_card_last_updates', None)
+        if last_updates is None:
+            self._card_last_updates = last_updates = {}
+        if not is_final and now - last_updates.get(message_id, 0) < 0.35:
+            return
+        card_instance, card_instance_id = self.card_instance_id_dict[message_id]
+        await self.bot.send_card_message(card_instance, card_instance_id, content, is_final)
+        last_updates[message_id] = now
+        if is_final:
+            self.card_instance_id_dict.pop(message_id, None)
+            last_updates.pop(message_id, None)
+
     async def is_stream_output_supported(self) -> bool:
         return is_stream_card_configured(self.config)
 

tests/unit_tests/agent/test_reply_stream.py

@@ -120,6 +120,28 @@ async def test_sdk_stream_reuses_adapter_or_sends_one_final_message(native, sour
     await session.close()
 
 
+async def test_target_streaming_adapter_updates_card_without_source_message():
+    session, adapter, _ = make_session(source=False)
+    adapter.create_message_card_for_target = AsyncMock(return_value=True)
+    adapter.reply_message_chunk_for_target = AsyncMock()
+    key = uuid4()
+
+    await session.apply(request(key, text='hello'))
+    result = await session.apply(request(key, 'finish', 'hello world'))
+
+    assert result['delivery'] == 'streaming'
+    adapter.create_message_card_for_target.assert_awaited_once_with(
+        adapter.reply_message_chunk_for_target.await_args.kwargs['message_id'], 'person', 'user-1'
+    )
+    assert [call.kwargs['content'] for call in adapter.reply_message_chunk_for_target.await_args_list] == [
+        'hello',
+        'hello world',
+    ]
+    assert [call.kwargs['is_final'] for call in adapter.reply_message_chunk_for_target.await_args_list] == [False, True]
+    adapter.send_message.assert_not_awaited()
+    await session.close()
+
+
 @pytest.mark.parametrize('native', [True, False])
 @pytest.mark.parametrize('error', [RuntimeError, asyncio.CancelledError])
 async def test_exception_finalizes_visible_card_without_sending_buffered_partial(native, error):

tests/unit_tests/platform/test_dingtalk_api.py

@@ -46,6 +46,20 @@ def test_text_card_keeps_brand_when_content_updates():
     assert update_text_card_messages('surface-id', 'done')[0]['updateDataModel']['path'] == '/content'
 
 
+@pytest.mark.asyncio
+async def test_create_card_for_target_uses_robot_a2ui_delivery():
+    client = _client(None)
+    client.create_and_deliver_ai_card = AsyncMock(return_value={})
+
+    card, track_id = await client.create_and_card_for_target('person', 'user-1')
+
+    assert card == {'surface_id': f'langbot-{track_id}'}
+    request = client.create_and_deliver_ai_card.await_args.kwargs
+    assert request['open_space_id'] == 'dtv1.card//im_robot.user-1'
+    assert request['is_group'] is False
+    assert request['a2ui_messages'][2]['updateComponents']['components'][0]['children'][0] == 'langbot-brand'
+
+
 @pytest.mark.asyncio
 async def test_create_ai_card_uses_group_robot_delivery_model():
     post = AsyncMock(

tests/unit_tests/platform/test_dingtalk_eba_adapter.py

@@ -51,6 +51,7 @@ class DummyDingTalkClient(DingTalkClient):
         self.get_audio_url = AsyncMock(return_value='data:audio/ogg;base64,AAAA')
         self.download_image = AsyncMock(return_value='data:image/png;base64,BBBB')
         self.create_and_card = AsyncMock(return_value=('card', 'card-id'))
+        self.create_and_card_for_target = AsyncMock(return_value=({'surface_id': 'surface-id'}, 'card-id'))
         self.send_card_message = AsyncMock()
         self.create_and_deliver_ai_card = AsyncMock(return_value={})
         self.update_ai_card = AsyncMock(return_value={})
@@ -318,6 +319,20 @@ async def test_dingtalk_send_reply_and_platform_api_use_underlying_client():
     assert file_url == {'url': 'https://example.test/file'}
 
 
+@pytest.mark.asyncio
+async def test_dingtalk_target_reply_stream_updates_one_card():
+    adapter = make_adapter()
+
+    assert await adapter.create_message_card_for_target('reply-id', 'person', 'user-1') is True
+    await adapter.reply_message_chunk_for_target(message_id='reply-id', content='hello', is_final=False)
+    await adapter.reply_message_chunk_for_target(message_id='reply-id', content='hello world', is_final=True)
+
+    adapter.bot.create_and_card_for_target.assert_awaited_once_with('person', 'user-1')
+    assert [call.args[2] for call in adapter.bot.send_card_message.await_args_list] == ['hello', 'hello world']
+    assert [call.args[3] for call in adapter.bot.send_card_message.await_args_list] == [False, True]
+    assert 'reply-id' not in adapter.card_instance_id_dict
+
+
 @pytest.mark.asyncio
 async def test_dingtalk_interaction_delivery_and_callback_event():
     adapter = make_adapter()