From 983e12acdfbe46870bd228188de4592d2eae99bb Mon Sep 17 00:00:00 2001
From: fdc310 <2213070223@qq.com>
Date: Fri, 30 Oct 2026 01:06:10 +0800
Subject: [PATCH] fix(lark): stream replies after HITL submissions
---
.../pkg/platform/adapters/lark/adapter.py | 75 +++++++++++++++----
.../platform/test_lark_eba_adapter.py | 51 ++++++++++++-
2 files changed, 109 insertions(+), 17 deletions(-)
diff --git a/src/langbot/pkg/platform/adapters/lark/adapter.py b/src/langbot/pkg/platform/adapters/lark/adapter.py
index ec0635968..84d0e313a 100644
--- a/src/langbot/pkg/platform/adapters/lark/adapter.py
+++ b/src/langbot/pkg/platform/adapters/lark/adapter.py
@@ -490,6 +490,31 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte
self.reply_to_monitoring_msg.pop(next(iter(self.reply_to_monitoring_msg)), None)
return True
+ async def create_message_card_for_target(self, message_id: str, target_type: str, target_id: str) -> bool:
+ if target_type not in {'group', 'person'} or not target_id:
+ raise ValueError('Lark streaming card requires a group or person target')
+ card_id = await self.create_card_id(message_id)
+ content = {'type': 'card', 'data': {'card_id': card_id, 'template_variable': {'content': 'Thinking...'}}}
+ request = (
+ CreateMessageRequest.builder()
+ .receive_id_type('chat_id' if target_type == 'group' else 'open_id')
+ .request_body(
+ CreateMessageRequestBody.builder()
+ .receive_id(target_id)
+ .content(json.dumps(content))
+ .msg_type('interactive')
+ .uuid(str(uuid.uuid4()))
+ .build()
+ )
+ .build()
+ )
+ response: CreateMessageResponse = await self.api_client.im.v1.message.acreate(request)
+ if not response.success():
+ self.card_id_dict.pop(message_id, None)
+ self.card_sequence_dict.pop(card_id, None)
+ raise RuntimeError(f'Lark create_message_card_for_target failed: {response.code} {response.msg}')
+ return True
+
async def create_card_id(self, message_id) -> str:
while len(self.card_id_dict) >= 1000:
old_key = next(iter(self.card_id_dict))
@@ -552,13 +577,6 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte
quote_origin: bool = False,
is_final: bool = False,
):
- card_id = self.card_id_dict[bot_message.resp_message_id]
- has_sent_update = self.card_sequence_dict.get(card_id, 0) > 0
- now = time.monotonic()
- last_update = self.card_last_update_dict.get(card_id, 0.0)
- is_high_frequency_chunk = now - last_update < 1.0
- if has_sent_update and is_high_frequency_chunk and bot_message.msg_sequence % 8 != 0 and not is_final:
- return
cumulative_content = getattr(bot_message, 'all_content', None)
if isinstance(cumulative_content, str) and cumulative_content:
content = cumulative_content
@@ -568,8 +586,37 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte
''.join(ele.get('text', '') for ele in paragraph if ele.get('tag') in {'text', 'md'})
for paragraph in text_elements
)
- if (is_final and not bot_message.tool_calls) or card_id in self.closed_streaming_cards:
- await self._replace_streaming_card(card_id, content, finished=is_final and not bot_message.tool_calls)
+ await self._update_streaming_card(
+ bot_message.resp_message_id,
+ content,
+ is_final=is_final,
+ tool_calls=bot_message.tool_calls,
+ msg_sequence=bot_message.msg_sequence,
+ tenant_key=self._tenant_key_from_source(message_source),
+ )
+
+ async def reply_message_chunk_for_target(self, *, message_id: str, content: str, is_final: bool) -> None:
+ await self._update_streaming_card(message_id, content, is_final=is_final, msg_sequence=1)
+
+ async def _update_streaming_card(
+ self,
+ message_id: str,
+ content: str,
+ *,
+ is_final: bool,
+ msg_sequence: int,
+ tool_calls: typing.Any = None,
+ tenant_key: str | None = None,
+ ) -> None:
+ card_id = self.card_id_dict[message_id]
+ has_sent_update = self.card_sequence_dict.get(card_id, 0) > 0
+ now = time.monotonic()
+ last_update = self.card_last_update_dict.get(card_id, 0.0)
+ is_high_frequency_chunk = now - last_update < 1.0
+ if has_sent_update and is_high_frequency_chunk and msg_sequence % 8 != 0 and not is_final:
+ return
+ if (is_final and not tool_calls) or card_id in self.closed_streaming_cards:
+ await self._replace_streaming_card(card_id, content, finished=is_final and not tool_calls)
else:
sequence = self._next_card_sequence(card_id)
request = (
@@ -582,18 +629,16 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte
.build()
)
response: ContentCardElementResponse = await self.api_client.cardkit.v1.card_element.acontent(
- request, self.request_option(self._tenant_key_from_source(message_source))
+ request, self.request_option(tenant_key)
)
if not response.success():
if self._streaming_mode_closed(response):
- await self._replace_streaming_card(
- card_id, content, finished=is_final and not bot_message.tool_calls
- )
+ await self._replace_streaming_card(card_id, content, finished=is_final and not tool_calls)
else:
raise RuntimeError(f'Lark card_element update failed: {response.code} {response.msg}')
self.card_last_update_dict[card_id] = now
- if is_final and bot_message.tool_calls is None:
- self.card_id_dict.pop(bot_message.resp_message_id, None)
+ if is_final and tool_calls is None:
+ self.card_id_dict.pop(message_id, None)
self.card_sequence_dict.pop(card_id, None)
self.card_last_update_dict.pop(card_id, None)
self.closed_streaming_cards.discard(card_id)
diff --git a/tests/unit_tests/platform/test_lark_eba_adapter.py b/tests/unit_tests/platform/test_lark_eba_adapter.py
index fc07b24f8..d20d5830a 100644
--- a/tests/unit_tests/platform/test_lark_eba_adapter.py
+++ b/tests/unit_tests/platform/test_lark_eba_adapter.py
@@ -7,6 +7,7 @@ import asyncio
import time
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
import pytest
import yaml
@@ -20,6 +21,7 @@ from langbot.pkg.platform.adapters.lark.interaction import (
interaction_delivery_capabilities,
interaction_event_from_callback,
)
+from langbot.pkg.agent.runner.reply_stream import ReplyStreamRequest, ReplyStreamSession
from langbot_plugin.api.definition.abstract.platform.event_logger import AbstractEventLogger
from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
from langbot_plugin.api.entities.builtin.platform import events as platform_events
@@ -729,7 +731,10 @@ async def test_lark_streaming_card_uses_strictly_increasing_sequences():
final_card = json.loads(final_request.request_body.card.data)
assert final_card['body']['elements'][0] == {'tag': 'markdown', 'content': 'answer'}
assert 'header' not in final_card
- assert final_card['body']['elements'][-1]['columns'][1]['elements'][0]['content'] == '[LangBot](https://langbot.app?utm_source=feishu&utm_medium=bot_card&utm_campaign=langbot)'
+ assert (
+ final_card['body']['elements'][-1]['columns'][1]['elements'][0]['content']
+ == '[LangBot](https://langbot.app?utm_source=feishu&utm_medium=bot_card&utm_campaign=langbot)'
+ )
assert not final_card['config'].get('streaming_mode', False)
assert 'response-1' not in adapter.card_id_dict
assert 'stream-card-1' not in adapter.card_sequence_dict
@@ -828,6 +833,43 @@ async def test_lark_streaming_card_falls_back_to_full_card_update_when_stream_cl
assert 'stream-card-1' not in adapter.closed_streaming_cards
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ ('target_type', 'target_id', 'receive_id_type'),
+ [
+ ('person', 'user-1', 'open_id'),
+ ('group', 'chat-1', 'chat_id'),
+ ],
+)
+async def test_lark_hitl_reply_stream_creates_and_updates_new_card(target_type, target_id, receive_id_type):
+ adapter = make_adapter({'enable-stream-reply': True})
+ event = SimpleNamespace(
+ event_type='interaction.submitted',
+ delivery=SimpleNamespace(
+ surface='platform',
+ platform_capabilities={},
+ reply_target={'target_type': target_type, 'target_id': target_id, 'message_id': 'interaction-card'},
+ ),
+ )
+ session = ReplyStreamSession(event, adapter)
+ stream_id = uuid4()
+
+ first = await session.apply(ReplyStreamRequest(stream_id=stream_id, operation='update', text='partial'))
+ final = await session.apply(ReplyStreamRequest(stream_id=stream_id, operation='finish', text='complete'))
+
+ assert first['delivery'] == final['delivery'] == 'streaming'
+ request = adapter.api_client.im.v1.message.acreate.await_args.args[0]
+ assert request.receive_id_type == receive_id_type
+ assert request.request_body.receive_id == target_id
+ assert request.request_body.msg_type == 'interactive'
+ assert adapter.api_client.im.v1.message.areply.await_count == 0
+ assert adapter.api_client.cardkit.v1.card_element.acontent.await_count == 1
+ final_card = json.loads(adapter.api_client.cardkit.v1.card.aupdate.await_args.args[0].request_body.card.data)
+ assert final_card['body']['elements'][0]['content'] == 'complete'
+ assert not adapter.card_id_dict
+ await session.close()
+
+
def test_lark_interaction_callback_builds_scoped_host_event():
callback = SimpleNamespace(
event=SimpleNamespace(
@@ -991,6 +1033,7 @@ async def test_lark_webhook_rejects_wrong_signature():
def test_lark_stream_card_separates_multiple_reasoning_rounds():
from langbot.pkg.platform.adapters.lark.stream_card import build_card, streaming_text
+
content = 'first thoughtfirst answersecond thoughtfinal answer'
preview = streaming_text(content)
assert '' not in preview and '' not in preview
@@ -1006,6 +1049,7 @@ def test_lark_stream_card_separates_multiple_reasoning_rounds():
def test_lark_live_preview_is_bounded_but_final_reasoning_is_complete():
from langbot.pkg.platform.adapters.lark.stream_card import build_card, streaming_text
+
thought = 'reasoning ' * 200
content = '' + thought + '