fix(wecombot): stream replies after HITL cards

src/langbot/libs/wecom_ai_bot_api/ws_client.py

@@ -60,6 +60,7 @@ _UPLOAD_CHUNK_SIZE = 512 * 1024
 
 _DEDUP_CACHE_MAX = 4096
 _STREAM_CACHE_MAX = 1024
+_INTERACTION_STREAM_MAX_AGE_SECONDS = 570
 _FEEDBACK_CACHE_MAX = 4096
 _PENDING_FORM_MAX = 1024
 _PENDING_FORM_TTL_SECONDS = 1800
@@ -140,6 +141,7 @@ class WecomBotWsClient:
         self._completed_stream_content: dict[str, str] = {}
         # Stream session info for feedback tracking
         self._stream_sessions: dict[str, dict] = {}  # msg_id -> session info
+        self._interaction_stream_origins: dict[str, dict] = {}  # msg_id -> original callback context
         # Feedback tracking: feedback_id -> session info
         self._feedback_sessions: dict[str, dict] = {}  # feedback_id -> {msg_id, user_id, chat_id, stream_id, req_id}
         # msg_id -> feedback_id (for associating feedback with message)
@@ -177,6 +179,72 @@ class WecomBotWsClient:
             if task_id:
                 self._pending_forms_by_task.pop(task_id, None)
 
+    async def finish_stream_for_interaction(self, msg_id: str) -> bool:
+        """Release an unused stream or finish text already shown before a card."""
+        key = self._stream_ids.get(msg_id)
+        if not key:
+            return True
+        session = self._stream_sessions.get(msg_id, {})
+        if not self._stream_sessions.get(msg_id, {}).get('stream_started'):
+            self._remember_interaction_stream_origin(msg_id, session)
+            self._stream_ids.pop(msg_id, None)
+            self._stream_last_content.pop(msg_id, None)
+            self._stream_sessions.pop(msg_id, None)
+            return True
+        req_id, stream_id = key.split('|', 1)
+        content = self._stream_last_content.get(msg_id, '')
+        try:
+            ack = await self.reply_stream(req_id, stream_id, content, finish=True)
+        except Exception:
+            await self.logger.warning(f'Failed to finish WeCom loading stream: {traceback.format_exc()}')
+            return False
+        if not isinstance(ack, dict) or ack.get('errcode', 0) != 0:
+            detail = ack.get('errmsg', ack.get('errcode')) if isinstance(ack, dict) else 'no acknowledgement'
+            await self.logger.warning(f'Failed to finish WeCom loading stream: {detail}')
+            return False
+        self._remember_interaction_stream_origin(msg_id, session)
+        self._stream_ids.pop(msg_id, None)
+        self._stream_last_content.pop(msg_id, None)
+        self._stream_sessions.pop(msg_id, None)
+        return True
+
+    def _remember_interaction_stream_origin(self, msg_id: str, session: dict) -> None:
+        if session.get('req_id'):
+            self._interaction_stream_origins[msg_id] = {
+                key: session.get(key) for key in ('req_id', 'user_id', 'chat_id', 'created_at')
+            }
+            self._cap_mapping(self._interaction_stream_origins, _STREAM_CACHE_MAX)
+
+    def begin_interaction_stream(self, msg_id: str, target_type: str, target_id: str) -> bool:
+        """Use a fresh stream ID so the resumed answer appears after the card."""
+        origin = self._interaction_stream_origins.get(msg_id)
+        if not origin or msg_id in self._stream_ids:
+            return False
+        age = time.monotonic() - float(origin.get('created_at') or 0)
+        if age >= _INTERACTION_STREAM_MAX_AGE_SECONDS:
+            self._interaction_stream_origins.pop(msg_id, None)
+            return False
+        expected = origin.get('chat_id') if target_type == 'group' else origin.get('user_id')
+        if target_type not in {'group', 'person', 'private'} or str(expected or '') != target_id:
+            return False
+        stream_id = _generate_req_id('stream')
+        self._stream_ids[msg_id] = f'{origin["req_id"]}|{stream_id}'
+        self._stream_sessions[msg_id] = {**origin, 'stream_id': stream_id, 'stream_started': False}
+        self._prune_stream_state()
+        return True
+
+    async def push_interaction_stream_chunk(self, msg_id: str, content: str, is_final: bool) -> bool:
+        first_frame = not self._stream_sessions.get(msg_id, {}).get('stream_started')
+        if await self.push_stream_chunk(msg_id, content, is_final=is_final):
+            if first_frame and (is_final or self._stream_sessions.get(msg_id, {}).get('stream_started')):
+                await self.logger.info(f'WeCom interaction reply stream accepted for message {msg_id}')
+            return True
+        await self.logger.warning(f'WeCom interaction reply stream unavailable for message {msg_id}')
+        self._stream_ids.pop(msg_id, None)
+        self._stream_last_content.pop(msg_id, None)
+        self._stream_sessions.pop(msg_id, None)
+        return False
+
     def _prune_pending_forms(self) -> None:
         cutoff = time.monotonic() - _PENDING_FORM_TTL_SECONDS
         for task_id, pending in tuple(self._pending_forms_by_task.items()):
@@ -254,6 +322,7 @@ class WecomBotWsClient:
         self._stream_ids.clear()
         self._stream_last_content.clear()
         self._stream_sessions.clear()
+        self._interaction_stream_origins.clear()
         self._feedback_sessions.clear()
         self._msg_feedback_ids.clear()
         self._pending_forms_by_task.clear()
@@ -727,8 +796,13 @@ class WecomBotWsClient:
 
             # WeCom replaces the displayed stream content on each refresh, so
             # every frame must contain the complete snapshot, not only a delta.
-            await self.reply_stream(req_id, stream_id, next_content, finish=is_final, feedback_id=feedback_id)
+            ack = await self.reply_stream(req_id, stream_id, next_content, finish=is_final, feedback_id=feedback_id)
+            if not isinstance(ack, dict) or ack.get('errcode', 0) != 0:
+                detail = ack.get('errmsg', ack.get('errcode')) if isinstance(ack, dict) else 'no acknowledgement'
+                raise RuntimeError(f'WeCom stream frame rejected: {detail}')
             self._stream_last_content[msg_id] = next_content
+            if msg_id in self._stream_sessions:
+                self._stream_sessions[msg_id]['stream_started'] = True
             if is_final:
                 self._completed_stream_content[msg_id] = hashlib.sha256(next_content.encode()).hexdigest()
                 self._cap_mapping(self._completed_stream_content, _STREAM_CACHE_MAX)
@@ -962,15 +1036,11 @@ class WecomBotWsClient:
                     'user_id': message_data.get('userid', ''),
                     'chat_id': message_data.get('chatid', ''),
                     'chat_type': message_data.get('type', 'single'),
+                    'stream_started': False,
+                    'created_at': time.monotonic(),
                 }
+                self._remember_interaction_stream_origin(msg_id, self._stream_sessions[msg_id])
                 self._prune_stream_state()
-                # Send an initial empty stream frame so the WeCom client
-                # shows its built-in loading spinner while the pipeline
-                # processes the message (e.g. RAG retrieval).
-                try:
-                    await self.reply_stream(req_id, stream_id, '', finish=False)
-                except Exception:
-                    await self.logger.warning(f'Failed to send initial stream frame: {traceback.format_exc()}')
             message_data['stream_id'] = stream_id
             message_data['req_id'] = req_id
 

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

@@ -402,7 +402,12 @@ class InteractionManager:
         except pydantic.ValidationError:
             return
 
-    async def acknowledge_submission(self, record: dict[str, typing.Any], adapter: typing.Any) -> None:
+    async def acknowledge_submission(
+        self,
+        record: dict[str, typing.Any],
+        adapter: typing.Any,
+        callback_context: dict[str, typing.Any] | None = None,
+    ) -> None:
         """Best-effort transition of submitted controls into a read-only state."""
         delivery_result = record.get('delivery_result')
         if not isinstance(delivery_result, dict) or not self._supports_platform_api(
@@ -416,6 +421,7 @@ class InteractionManager:
                     'request': record.get('request') or {},
                     'submission': record.get('submission') or {},
                     'update_target': delivery_result,
+                    **({'callback_context': callback_context} if callback_context else {}),
                 },
             )
             if isinstance(delivery_result, dict):

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

@@ -375,6 +375,103 @@ class InteractionStore:
                 raise InteractionNotFoundError('Interaction disappeared after submission')
             return self._to_dict(refreshed)
 
+    async def consume_text_reply(
+        self,
+        *,
+        workspace_id: str,
+        bot_id: str,
+        conversation_id: str,
+        actor_id: str,
+        binding_id: str,
+        runner_id: str,
+        processor_id: str,
+        text: str,
+        message_id: str,
+        message_time: int | float | None = None,
+    ) -> dict[str, typing.Any] | None:
+        """Consume the latest scoped, plain-text form waiting for this user."""
+        answer = text.strip()
+        if (
+            not answer
+            or len(answer) > 20000
+            or not all(
+                (workspace_id, bot_id, conversation_id, actor_id, binding_id, runner_id, processor_id, message_id)
+            )
+        ):
+            return None
+        now = _utc_now()
+        source_marker = 'text-reply:' + _token_hash(message_id)
+        scope = (
+            AgentInteraction.workspace_id == workspace_id,
+            AgentInteraction.bot_id == bot_id,
+            AgentInteraction.conversation_id == conversation_id,
+            AgentInteraction.actor_id == actor_id,
+            AgentInteraction.binding_id == binding_id,
+            AgentInteraction.runner_id == runner_id,
+            AgentInteraction.processor_type == 'agent',
+            AgentInteraction.processor_id == processor_id,
+        )
+        async with self._session_factory() as session:
+            duplicate = await session.execute(
+                sqlalchemy.select(AgentInteraction.id)
+                .where(
+                    *scope,
+                    AgentInteraction.status == 'submitted',
+                    AgentInteraction.status_reason == source_marker,
+                )
+                .limit(1)
+            )
+            if duplicate.scalar_one_or_none() is not None:
+                return {'duplicate_text_reply': True}
+            result = await session.execute(
+                sqlalchemy.select(AgentInteraction)
+                .where(*scope)
+                .order_by(AgentInteraction.created_at.desc(), AgentInteraction.id.desc())
+                .limit(1)
+            )
+            row = result.scalar_one_or_none()
+            if row is None or row.status != 'pending' or (row.expires_at and _as_utc(row.expires_at) <= now):
+                return None
+            if message_time is not None and float(message_time) < _as_utc(row.created_at).timestamp() - 1:
+                return None
+            delivery = _json_loads(row.delivery_result_json, None)
+            request = _json_loads(row.request_json, {})
+            fields = request.get('fields') if isinstance(request, dict) else None
+            if not (isinstance(delivery, dict) and delivery.get('ok') is True and delivery.get('rich') is False):
+                return None
+            if not (
+                isinstance(fields, list)
+                and len(fields) == 1
+                and isinstance(fields[0], dict)
+                and fields[0].get('type') in {'text', 'textarea'}
+                and fields[0].get('id')
+                and not request.get('actions')
+            ):
+                return None
+            submission = {
+                'interaction_id': row.interaction_id,
+                'action_id': None,
+                'values': {fields[0]['id']: answer},
+                'submitted_at': int(now.timestamp()),
+            }
+            updated = await session.execute(
+                sqlalchemy.update(AgentInteraction)
+                .where(AgentInteraction.id == row.id, AgentInteraction.status == 'pending')
+                .values(
+                    status='submitted',
+                    status_reason=source_marker,
+                    submitted_at=_db_datetime(now),
+                    submission_json=_json_dumps(submission),
+                    updated_at=_db_datetime(now),
+                )
+            )
+            if updated.rowcount != 1:
+                await session.rollback()
+                return None
+            await session.commit()
+            refreshed = await session.get(AgentInteraction, row.id)
+            return self._to_dict(refreshed)
+
     async def mark_delivery_failed(self, run_id: str, interaction_id: str, reason: str) -> bool:
         """Mark an undelivered pending interaction terminal."""
         now = _utc_now()

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

@@ -539,6 +539,16 @@ class AgentRunOrchestrator:
                                 auto_reply_deferred = False
                                 explicit_reply_text = None
 
+                        if auto_reply_enabled and result_type == 'run.completed' and auto_reply_stream_open:
+                            await reply_streams.apply(
+                                ReplyStreamRequest(
+                                    stream_id=auto_reply_stream_id,
+                                    operation='finish',
+                                    text=auto_reply_stream_text,
+                                )
+                            )
+                            auto_reply_stream_open = False
+
                         if result is not None:
                             yield result
 

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

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import asyncio
+import logging
+import time
 from dataclasses import dataclass, field
 from typing import Literal
 from uuid import UUID, uuid4
@@ -12,6 +14,9 @@ from langbot_plugin.api.entities.builtin.platform import events, message
 from langbot_plugin.api.entities.builtin.provider import message as provider_message
 
 
+logger = logging.getLogger(__name__)
+
+
 class ReplyStreamRequest(pydantic.BaseModel):
     model_config = pydantic.ConfigDict(extra='forbid')
 
@@ -26,6 +31,8 @@ class _Reply:
     text: str = ''
     started: bool = False
     native: bool = False
+    interaction_origin: str = ''
+    last_sent_at: float = 0.0
     final_attempted: bool = False
     sequence: int = 0
     status: str = 'open'
@@ -41,6 +48,11 @@ class ReplyStreamSession:
         if not isinstance(self.source, events.MessageEvent):
             self.source = None
         self.target = dict(event.delivery.reply_target or {})
+        self.interaction_origin = (
+            str(self.target.get('message_id') or '')
+            if getattr(event, 'event_type', None) == 'interaction.submitted'
+            else ''
+        )
         self.mock = (
             event.delivery.surface == 'webui' and (event.delivery.platform_capabilities or {}).get('debug_mock') is True
         )
@@ -91,12 +103,24 @@ class ReplyStreamSession:
                         and callable(getattr(self.adapter, 'create_message_card_for_target', None))
                         and callable(getattr(self.adapter, 'reply_message_chunk_for_target', None))
                     )
+                    interaction_streaming = (
+                        self.source is None
+                        and bool(self.interaction_origin)
+                        and callable(getattr(self.adapter, 'begin_interaction_stream', None))
+                        and callable(getattr(self.adapter, 'reply_interaction_stream_chunk', None))
+                    )
                     reply.native = (
-                        self.source is not None or target_streaming
+                        self.source is not None or target_streaming or interaction_streaming
                     ) and await self.adapter.is_stream_output_supported()
                     if reply.native:
                         if self.source is not None:
                             await self.adapter.create_message_card(reply.delivery_id, self.source)
+                        elif interaction_streaming:
+                            reply.native = await self.adapter.begin_interaction_stream(
+                                self.interaction_origin, self.target
+                            )
+                            if reply.native:
+                                reply.interaction_origin = self.interaction_origin
                         else:
                             target_type, target_id = self.target.get('target_type'), self.target.get('target_id')
                             if not target_type or not target_id:
@@ -105,8 +129,30 @@ class ReplyStreamSession:
                                 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:
+                    should_refresh = (
+                        request.operation != 'update'
+                        or not reply.interaction_origin
+                        or reply.sequence == 0
+                        or time.monotonic() - reply.last_sent_at >= 0.5
+                    )
+                    if should_refresh:
+                        try:
+                            await self._update_native(key, reply, final=request.operation != 'update')
+                            reply.last_sent_at = time.monotonic()
+                        except Exception:
+                            if not reply.interaction_origin:
+                                raise
+                            logger.warning(
+                                'WeCom interaction stream failed; sending the final reply proactively', exc_info=True
+                            )
+                            reply.native = False
+                if (
+                    reply.started
+                    and not reply.native
+                    and request.operation == 'finish'
+                    and reply.text
+                    and not self.mock
+                ):
                     if self.source is not None:
                         await self.adapter.reply_message(
                             message_source=self.source, message=self._message(reply.text), quote_origin=False
@@ -154,6 +200,12 @@ class ReplyStreamSession:
                 quote_origin=False,
                 is_final=final,
             )
+        elif reply.interaction_origin:
+            await self.adapter.reply_interaction_stream_chunk(
+                source_message_id=reply.interaction_origin,
+                content=reply.text,
+                is_final=final,
+            )
         else:
             await self.adapter.reply_message_chunk_for_target(
                 message_id=reply.delivery_id, content=reply.text, is_final=final

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

@@ -18,6 +18,7 @@ from langbot.pkg.platform.adapters.wecombot.event_converter import WecomBotEvent
 from langbot.pkg.platform.adapters.wecombot.message_converter import WecomBotMessageConverter
 from langbot.pkg.platform.adapters.wecombot.platform_api import PLATFORM_API_MAP
 from langbot.pkg.platform.adapters.wecombot.interaction import (
+    acknowledge_interaction,
     interaction_delivery_capabilities,
     interaction_event_from_native,
     send_interaction,
@@ -123,7 +124,7 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
             'call_platform_api',
         ]
         if not self.config.get('enable-webhook', False) and hasattr(self.bot, 'send_template_card'):
-            apis.append('interaction.request')
+            apis.extend(['interaction.request', 'interaction.acknowledge'])
         return apis
 
     def get_interaction_capabilities(self) -> dict[str, typing.Any]:
@@ -174,6 +175,9 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
         if list(self._iter_media_components(items)):
             raise NotSupportedError('send_message:media requires an originating message callback')
         raw = await self.bot.send_message(str(target_id), content)
+        if not isinstance(raw, dict) or raw.get('errcode', 0) != 0:
+            detail = raw.get('errmsg', raw.get('errcode')) if isinstance(raw, dict) else 'no acknowledgement'
+            raise RuntimeError(f'WeCom message delivery failed: {detail}')
         return platform_events.MessageResult(raw={'result': raw})
 
     async def reply_to_event(
@@ -222,8 +226,8 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
         if not self.config.get('enable-webhook', False) and event.get('req_id'):
             if content:
                 finished = await self.bot.push_stream_chunk(event.message_id, content, is_final=True)
-                if not finished:
-                    raw = await self.bot.reply_text(event.get('req_id'), content)
+                if not finished and content.strip(' \t\r\n\u200b\u200c\u200d\ufeff'):
+                    raw = await self.bot.send_message(str(event.get('chatid') or event.get('userid') or ''), content)
             for item in self._iter_media_components(items):
                 await self._send_media(self.bot, event.get('req_id'), item)
         else:
@@ -248,8 +252,14 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
 
         content = format_stream_text(self._join_text_components(items))
         success = await self.bot.push_stream_chunk(event.message_id, content, is_final=is_final)
-        if not success and is_final and not self.config.get('enable-webhook', False) and event.get('req_id'):
-            await self.bot.reply_text(event.get('req_id'), content)
+        if (
+            not success
+            and is_final
+            and content.strip(' \t\r\n\u200b\u200c\u200d\ufeff')
+            and not self.config.get('enable-webhook', False)
+            and event.get('req_id')
+        ):
+            await self.bot.send_message(str(event.get('chatid') or event.get('userid') or ''), content)
         if is_final:
             if not self.config.get('enable-webhook', False) and event.get('req_id'):
                 for item in self._iter_media_components(items):
@@ -261,9 +271,29 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
     async def is_stream_output_supported(self) -> bool:
         return self.config.get('enable-stream-reply', True)
 
+    async def begin_interaction_stream(self, source_message_id: str, target: dict) -> bool:
+        if self.config.get('enable-webhook', False):
+            return False
+        started = self.bot.begin_interaction_stream(
+            source_message_id,
+            str(target.get('target_type') or ''),
+            str(target.get('target_id') or ''),
+        )
+        if not started:
+            await self.logger.warning('WeCom interaction stream context unavailable; buffering final reply')
+        return started
+
+    async def reply_interaction_stream_chunk(self, *, source_message_id: str, content: str, is_final: bool) -> None:
+        from .stream_text import format_stream_text
+
+        if not await self.bot.push_interaction_stream_chunk(source_message_id, format_stream_text(content), is_final):
+            raise RuntimeError('WeCom rejected the new interaction reply stream')
+
     async def call_platform_api(self, action: str, params: dict = {}) -> dict:
         if action == 'interaction.request' and 'interaction.request' in self.get_supported_apis():
             return await send_interaction(self, params)
+        if action == 'interaction.acknowledge' and 'interaction.acknowledge' in self.get_supported_apis():
+            return await acknowledge_interaction(self, params)
         handler = PLATFORM_API_MAP.get(action)
         if handler is None:
             raise NotSupportedError(f'call_platform_api:{action}')

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

@@ -7,7 +7,7 @@ import time
 import typing
 import uuid
 
-from langbot.libs.wecom_ai_bot_api.api import extract_template_card_event_payload
+from langbot.libs.wecom_ai_bot_api.api import extract_template_card_action, extract_template_card_event_payload
 from langbot.libs.wecom_ai_bot_api.wecombotevent import WecomBotEvent
 from langbot_plugin.api.entities.builtin.platform import events as platform_events
 
@@ -126,7 +126,59 @@ async def send_interaction(adapter: typing.Any, params: dict[str, typing.Any]) -
     if not isinstance(raw, dict) or raw.get('errcode', 0) != 0:
         detail = raw.get('errmsg', raw.get('errcode')) if isinstance(raw, dict) else 'no acknowledgement'
         raise RuntimeError(f'WeCom template card delivery failed: {detail}')
-    return {'ok': True, 'message_id': None, 'rich': True, 'raw': raw}
+    source_message_id = str(reply_target.get('message_id') or '')
+    if source_message_id and hasattr(adapter.bot, 'finish_stream_for_interaction'):
+        await adapter.bot.finish_stream_for_interaction(source_message_id)
+    return {
+        'ok': True,
+        'message_id': None,
+        'rich': True,
+        'task_id': card['template_card']['task_id'],
+        'raw': raw,
+    }
+
+
+async def acknowledge_interaction(adapter: typing.Any, params: dict[str, typing.Any]) -> dict[str, typing.Any]:
+    request = params.get('request')
+    submission = params.get('submission')
+    target = params.get('update_target')
+    callback = params.get('callback_context')
+    if not all(isinstance(value, dict) for value in (request, submission, target, callback)):
+        raise ValueError('interaction.acknowledge requires request, submission, update_target, and callback_context')
+    task_id = str(callback.get('task_id') or '')
+    req_id = str(callback.get('req_id') or '')
+    if not req_id or not task_id or task_id != target.get('task_id') or target.get('rich') is not True:
+        raise ValueError('WeCom interaction callback does not match the delivered card')
+
+    card = build_interaction_card(request, 'submitted')
+    if card is None:
+        raise ValueError('WeCom interaction has no card to update')
+    updated = card['template_card']
+    updated['task_id'] = task_id
+    action_id = submission.get('action_id')
+    if updated['card_type'] == 'button_interaction':
+        for index, action in enumerate((request.get('actions') or [])[:6]):
+            updated['button_list'][index]['key'] = 'submitted'
+            if isinstance(action, dict) and action.get('id') == action_id:
+                button = updated['button_list'][index]
+                button['text'] = f'✓ {button["text"]}'
+                button['replace_text'] = button['text']
+                break
+    else:
+        field = request['fields'][0]
+        selected = (submission.get('values') or {}).get(field['id'])
+        for index, option in enumerate(field.get('options') or []):
+            if option.get('value') == selected:
+                updated['select_list'][0]['selected_id'] = f'opt_{index + 1}'
+                break
+        updated['select_list'][0]['disable'] = True
+        updated['submit_button']['text'] = '✓ Submitted'
+        updated['submit_button']['key'] = 'submitted'
+    raw = await adapter.bot.update_template_card(req_id, updated)
+    if not isinstance(raw, dict) or raw.get('errcode', 0) != 0:
+        detail = raw.get('errmsg', raw.get('errcode')) if isinstance(raw, dict) else 'no acknowledgement'
+        raise RuntimeError(f'WeCom template card update failed: {detail}')
+    return {**target, 'updated': True, 'raw': raw}
 
 
 def _template_card_event(event: WecomBotEvent) -> dict[str, typing.Any]:
@@ -227,6 +279,7 @@ def interaction_event_from_native(
     if not callback_key.startswith('lbi:'):
         return None
     parsed = parse_callback_key(callback_key)
+    task_id, _, _ = extract_template_card_action(_template_card_event(event))
     if parsed.pop('select_submit', False):
         parsed['field_ref'] = 0
         parsed['option_ref'] = _selected_option_ref(_template_card_event(event))
@@ -242,6 +295,8 @@ def interaction_event_from_native(
             'target_type': target_type,
             'target_id': target_id,
             'display_text': 'submitted',
+            'callback_req_id': str(event.get('req_id') or ''),
+            'callback_task_id': task_id,
         },
         timestamp=time.time(),
         source_platform_object=event,

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

@@ -191,6 +191,7 @@ spec:
       - get_group_member_list
       - call_platform_api
       - interaction.request
+      - interaction.acknowledge
 
   platform_specific_apis:
     - action: is_websocket_mode

src/langbot/pkg/platform/botmgr.py

@@ -1304,6 +1304,58 @@ class RuntimeBot:
 
         try:
             with trace_mod.scope(route_ref=self._route_ref(event_binding, target_type, target_uuid)):
+                if (
+                    target_type == 'agent'
+                    and isinstance(event, platform_events.MessageReceivedEvent)
+                    and event.adapter_name == 'wecombot-omni'
+                    and event.message_chain
+                    and all(
+                        isinstance(
+                            component,
+                            (
+                                platform_message.Source,
+                                platform_message.Plain,
+                                platform_message.At,
+                                platform_message.AtAll,
+                            ),
+                        )
+                        for component in event.message_chain
+                    )
+                ):
+                    answer = ''.join(
+                        component.text
+                        for component in event.message_chain
+                        if isinstance(component, platform_message.Plain)
+                    )
+                    record = await self.ap.agent_run_orchestrator.interaction_manager.store.consume_text_reply(
+                        workspace_id=self.execution_context.workspace_uuid,
+                        bot_id=self.bot_entity.uuid,
+                        conversation_id=envelope.conversation_id or '',
+                        actor_id=envelope.actor.actor_id if envelope.actor else '',
+                        binding_id=binding.binding_id,
+                        runner_id=binding.runner_id,
+                        processor_id=target_uuid,
+                        text=answer,
+                        message_id=str(event.message_id or self._extract_message_id(event.message_chain)),
+                        message_time=event.timestamp,
+                    )
+                    if record is not None:
+                        if not record.get('duplicate_text_reply'):
+                            await self._resume_agent_interaction(
+                                record,
+                                event,
+                                adapter,
+                                envelope.actor.actor_id if envelope.actor else None,
+                                execution_id,
+                            )
+                        return await self._record_event_route_trace(
+                            event_type=event_type,
+                            status='delivered',
+                            binding=event_binding,
+                            target_type=target_type,
+                            target_uuid=target_uuid,
+                            text=f'Event {event_type} handled text interaction reply',
+                        )
                 from .processing_indicator import processing_indicator
 
                 reply_type, reply_id = (None, None)
@@ -1535,7 +1587,15 @@ class RuntimeBot:
         # Only a validated callback resolving an original processor is eligible.
         # Admit before acknowledgement and target lookup to preserve failures.
         trace_mod.admit()
-        await self.ap.agent_run_orchestrator.interaction_manager.acknowledge_submission(record, adapter)
+        callback_context = None
+        if getattr(event, 'adapter_name', None) == 'wecombot-omni':
+            callback_context = {
+                'req_id': data.get('callback_req_id'),
+                'task_id': data.get('callback_task_id'),
+            }
+        await self.ap.agent_run_orchestrator.interaction_manager.acknowledge_submission(
+            record, adapter, callback_context
+        )
 
         if record['processor_type'] == 'agent':
             await self._resume_agent_interaction(record, event, adapter, actor_id, execution_id)
@@ -1615,7 +1675,7 @@ class RuntimeBot:
     async def _resume_agent_interaction(
         self,
         record: dict[str, typing.Any],
-        event: platform_events.PlatformSpecificEvent,
+        event: platform_events.EBAEvent,
         adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
         actor_id: str | None,
         execution_id: str | None = None,
@@ -1639,9 +1699,13 @@ class RuntimeBot:
         binding.binding_id = record['binding_id']
 
         submission = record['submission']
+        event_data = getattr(event, 'data', None)
+        values = (submission or {}).get('values') or {}
+        text_answer = next(iter(values.values()), None) if len(values) == 1 else None
         display_text = str(
             (submission or {}).get('action_id')
-            or (event.data if isinstance(event.data, dict) else {}).get('display_text')
+            or (event_data if isinstance(event_data, dict) else {}).get('display_text')
+            or text_answer
             or 'submitted'
         )
         input_data: dict[str, typing.Any] = {

tests/unit_tests/agent/test_interaction_manager.py

@@ -190,6 +190,63 @@ async def test_structured_delivery_uses_frozen_target_and_persists_request(store
     assert record['delivery_result']['message_id'] == 'message-1'
 
 
+@pytest.mark.asyncio
+async def test_text_reply_consumes_latest_scoped_plain_form_once(store):
+    async def create(interaction_id, *, field_type='textarea', rich=False):
+        await store.create_request(
+            interaction_id=interaction_id,
+            run_id=f'run-{interaction_id}',
+            binding_id='binding-1',
+            runner_id='runner-1',
+            processor_type='agent',
+            processor_id='agent-1',
+            request={
+                'fields': [{'id': 'question', 'type': field_type}],
+                'actions': [],
+            },
+            delivery_target={'target_type': 'person', 'target_id': 'user-1'},
+            workspace_id='workspace-1',
+            bot_id='bot-1',
+            conversation_id='person_user-1',
+            actor_id='user-1',
+            expires_at=time.time() + 600,
+        )
+        await store.record_delivery_success(f'run-{interaction_id}', interaction_id, {'ok': True, 'rich': rich})
+
+    async def reply(**overrides):
+        scope = dict(
+            workspace_id='workspace-1',
+            bot_id='bot-1',
+            conversation_id='person_user-1',
+            actor_id='user-1',
+            binding_id='binding-1',
+            runner_id='runner-1',
+            processor_id='agent-1',
+            text='  answer  ',
+            message_id='message-1',
+        )
+        scope.update(overrides)
+        return await store.consume_text_reply(**scope)
+
+    await create('old')
+    await create('new')
+    assert await reply(actor_id='user-2') is None
+    assert await reply(conversation_id='person_user-2') is None
+    assert await reply(workspace_id='workspace-2') is None
+    assert await reply(text='   ') is None
+    submitted = await reply()
+    assert submitted['interaction_id'] == 'new'
+    assert submitted['submission']['values'] == {'question': 'answer'}
+    assert (await reply())['duplicate_text_reply'] is True
+    assert await reply(message_id='message-2') is None
+    assert (await store.get_request('run-old', 'old'))['status'] == 'pending'
+
+    await create('rich', rich=True)
+    assert await reply(message_id='message-3') is None
+    await create('select', field_type='select')
+    assert await reply(message_id='message-4') is None
+
+
 @pytest.mark.asyncio
 async def test_agent_tool_card_uses_host_token_and_persists_resume_target(store, event):
     adapter = FakeAdapter(supports_interactions=True)

tests/unit_tests/agent/test_orchestrator_integration.py

@@ -1271,7 +1271,10 @@ class TestQueryEntrySessionQueryId:
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize('event_type', ['message.received', 'interaction.submitted'])
-async def test_direct_agent_streams_output_to_the_originating_platform(clean_agent_state, event_type):
+@pytest.mark.parametrize('completed_message', [False, True])
+async def test_direct_agent_streams_output_to_the_originating_platform(
+    clean_agent_state, event_type, completed_message
+):
     from langbot.pkg.agent.runner.host_models import (
         AgentBinding,
         AgentEventEnvelope,
@@ -1283,25 +1286,27 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age
     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',
-                    }
-                },
+    runner_results = [
+        {
+            'type': 'message.delta',
+            'data': {
+                'chunk': {
+                    'role': 'assistant',
+                    'content': 'hello',
+                    'all_content': 'hello',
+                }
             },
+        },
+    ]
+    if completed_message:
+        runner_results.append(
             {
                 'type': 'message.completed',
                 'data': {'message': {'role': 'assistant', 'content': 'hello world'}},
-            },
-            {'type': 'run.completed', 'data': {'finish_reason': 'stop'}},
-        ]
-    )
+            }
+        )
+    runner_results.append({'type': 'run.completed', 'data': {'finish_reason': 'stop'}})
+    connector = FakePluginConnector(results=runner_results)
     ap = FakeApplication(connector, clean_agent_state)
     orchestrator = AgentRunOrchestrator(ap, FakeRegistry(make_descriptor()))
     adapter = types.SimpleNamespace(
@@ -1358,11 +1363,13 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age
         )
     ]
 
-    assert [item.content for item in results] == ['hello', 'hello world']
+    assert [item.content for item in results] == (['hello', 'hello world'] if completed_message else ['hello'])
     if event_type == 'message.received':
         adapter.create_message_card.assert_awaited_once()
         chunks = adapter.reply_message_chunk.await_args_list
-        assert [call.kwargs['bot_message'].all_content for call in chunks] == ['hello', 'hello world']
+        assert [call.kwargs['bot_message'].all_content for call in chunks] == (
+            ['hello', 'hello world'] if completed_message else ['hello', 'hello']
+        )
         assert [call.kwargs['is_final'] for call in chunks] == [False, True]
         adapter.reply_message.assert_not_awaited()
         adapter.send_message.assert_not_awaited()
@@ -1371,7 +1378,7 @@ async def test_direct_agent_streams_output_to_the_originating_platform(clean_age
         adapter.reply_message_chunk.assert_not_awaited()
         adapter.reply_message.assert_not_awaited()
         adapter.send_message.assert_awaited_once()
-        assert adapter.send_message.await_args.args[2][0].text == 'hello world'
+        assert adapter.send_message.await_args.args[2][0].text == ('hello world' if completed_message else 'hello')
 
 
 @pytest.mark.asyncio

tests/unit_tests/agent/test_reply_stream.py

@@ -142,6 +142,97 @@ async def test_target_streaming_adapter_updates_card_without_source_message():
     await session.close()
 
 
+async def test_interaction_callback_buffers_one_final_message_after_card_update():
+    event = SimpleNamespace(
+        event_type='interaction.submitted',
+        delivery=SimpleNamespace(
+            surface='platform',
+            platform_capabilities={},
+            reply_target={'target_type': 'person', 'target_id': 'user-1'},
+        ),
+    )
+    source = events.PlatformSpecificEvent(
+        type='platform.specific',
+        adapter_name='wecombot-omni',
+        action='interaction.submitted',
+        data={'callback_req_id': 'callback-1'},
+    )
+    adapter = SimpleNamespace(
+        is_stream_output_supported=AsyncMock(return_value=True),
+        send_message=AsyncMock(),
+    )
+    session = ReplyStreamSession(event, adapter, source)
+    key = uuid4()
+    await session.apply(request(key, text='partial'))
+    result = await session.apply(request(key, 'finish', 'complete'))
+
+    assert result['delivery'] == 'buffered'
+    adapter.send_message.assert_awaited_once()
+    assert adapter.send_message.await_args.args[2][0].text == 'complete'
+    await session.close()
+
+
+async def test_interaction_reply_is_sent_after_card_as_new_message():
+    event = SimpleNamespace(
+        event_type='interaction.submitted',
+        delivery=SimpleNamespace(
+            surface='platform',
+            platform_capabilities={},
+            reply_target={'target_type': 'person', 'target_id': 'user-1', 'message_id': 'original-1'},
+        ),
+    )
+    adapter = SimpleNamespace(
+        is_stream_output_supported=AsyncMock(return_value=True),
+        send_message=AsyncMock(),
+    )
+    session = ReplyStreamSession(event, adapter, source=None)
+    key = uuid4()
+    await session.apply(request(key, text='partial'))
+    result = await session.apply(request(key, 'finish', 'complete'))
+
+    assert result['delivery'] == 'buffered'
+    adapter.send_message.assert_awaited_once()
+    assert adapter.send_message.await_args.args[2][0].text == 'complete'
+    await session.close()
+
+
+@pytest.mark.parametrize('stream_fails', [False, True])
+async def test_interaction_uses_new_stream_with_buffered_fallback(stream_fails):
+    event = SimpleNamespace(
+        event_type='interaction.submitted',
+        delivery=SimpleNamespace(
+            surface='platform',
+            platform_capabilities={},
+            reply_target={'target_type': 'person', 'target_id': 'user-1', 'message_id': 'original-1'},
+        ),
+    )
+    adapter = SimpleNamespace(
+        is_stream_output_supported=AsyncMock(return_value=True),
+        begin_interaction_stream=AsyncMock(return_value=True),
+        reply_interaction_stream_chunk=AsyncMock(side_effect=RuntimeError('invalid req_id') if stream_fails else None),
+        send_message=AsyncMock(),
+    )
+    session = ReplyStreamSession(event, adapter)
+    key = uuid4()
+    await session.apply(request(key, text='part'))
+    result = await session.apply(request(key, 'finish', 'part done'))
+
+    adapter.begin_interaction_stream.assert_awaited_once_with('original-1', event.delivery.reply_target)
+    if stream_fails:
+        assert result['delivery'] == 'buffered'
+        adapter.send_message.assert_awaited_once()
+        assert adapter.send_message.await_args.args[2][0].text == 'part done'
+    else:
+        assert result['delivery'] == 'streaming'
+        assert [call.kwargs['content'] for call in adapter.reply_interaction_stream_chunk.await_args_list] == [
+            'part',
+            'part done',
+        ]
+        assert adapter.reply_interaction_stream_chunk.await_args.kwargs['is_final'] is 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_routing_rules.py

@@ -589,6 +589,21 @@ class TestInteractionResumeRouting:
             actor_id='user-1',
         )
 
+    @pytest.mark.asyncio
+    async def test_wecom_callback_passes_native_card_update_context(self):
+        record = self._record('pipeline')
+        bot = self._make_bot(record)
+        adapter = SimpleNamespace()
+        event = self._event()
+        event.adapter_name = 'wecombot-omni'
+        event.data.update({'callback_req_id': 'callback-req', 'callback_task_id': 'task-1'})
+
+        await bot._handle_interaction_submission(event, adapter)
+
+        bot.ap.agent_run_orchestrator.interaction_manager.acknowledge_submission.assert_awaited_once_with(
+            record, adapter, {'req_id': 'callback-req', 'task_id': 'task-1'}
+        )
+
     @pytest.mark.asyncio
     async def test_callback_forwards_compact_platform_references(self):
         bot = self._make_bot(self._record('pipeline'))
@@ -676,6 +691,27 @@ class TestInteractionResumeRouting:
         assert binding.delivery_policy.enable_reply is True
         adapter.send_message.assert_not_awaited()
 
+        from langbot_plugin.api.entities.builtin.platform import entities, events, message
+
+        text_record = self._record('agent')
+        text_record['submission'] = {
+            'interaction_id': 'form-1',
+            'action_id': None,
+            'values': {'name': 'My question'},
+            'submitted_at': 1235,
+        }
+        text_event = events.MessageReceivedEvent(
+            timestamp=1235,
+            message_id='message-2',
+            message_chain=message.MessageChain([message.Plain(text='My question')]),
+            sender=entities.User(id='user-1'),
+            chat_type=entities.ChatType.GROUP,
+            chat_id='chat-1',
+        )
+        await bot._resume_agent_interaction(text_record, text_event, adapter, 'user-1')
+        assert captured[1][0].input.text == 'My question'
+        assert captured[1][0].input.interaction.values == {'name': 'My question'}
+
     def test_agent_product_to_binding_does_not_fallback_to_component_ref(self):
         """An empty config runner stays unconfigured even if component_ref is stale."""
         from langbot.pkg.platform.botmgr import RuntimeBot
@@ -782,6 +818,53 @@ async def test_bound_event_processor_receives_one_complete_typed_event():
     bot.ap.plugin_connector.emit_event.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_wecom_plain_reply_resumes_pending_agent_interaction():
+    from datetime import datetime
+
+    from langbot_plugin.api.entities.builtin.platform import entities, events, message
+
+    bot = TestEventRouteTrace._make_bot(
+        [{'id': 'binding-1', 'event_pattern': 'message.received', 'target_type': 'agent', 'target_uuid': 'agent-1'}]
+    )
+    run = Mock()
+    consume = AsyncMock(return_value={'interaction_id': 'form-1'})
+    bot.ap = SimpleNamespace(
+        agent_service=SimpleNamespace(
+            get_agent=AsyncMock(
+                return_value={
+                    'uuid': 'agent-1',
+                    'kind': 'agent',
+                    'supported_event_patterns': ['message.received'],
+                    'config': {'runner': {'id': 'runner-1'}, 'runner_config': {'runner-1': {}}},
+                }
+            )
+        ),
+        agent_run_orchestrator=SimpleNamespace(
+            run=run, interaction_manager=SimpleNamespace(store=SimpleNamespace(consume_text_reply=consume))
+        ),
+    )
+    bot._resume_agent_interaction = AsyncMock()
+    event = events.MessageReceivedEvent(
+        adapter_name='wecombot-omni',
+        message_id='message-1',
+        message_chain=message.MessageChain(
+            [message.Source(id='message-1', time=datetime.now()), message.Plain(text='answer')]
+        ),
+        sender=entities.User(id='user-1'),
+        chat_type=entities.ChatType.PRIVATE,
+        chat_id='user-1',
+    )
+
+    result = await bot._dispatch_eba_event_to_processor(event, SimpleNamespace(get_supported_apis=lambda: []))
+
+    assert result['status'] == 'delivered'
+    assert consume.await_args.kwargs['text'] == 'answer'
+    assert consume.await_args.kwargs['conversation_id'] == 'person_user-1'
+    bot._resume_agent_interaction.assert_awaited_once()
+    run.assert_not_called()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize('kind', ['agent', 'event_processor'])
 @pytest.mark.parametrize('output_kind', ['message', 'chunks', 'tool_rounds'])

tests/unit_tests/platform/test_wecombot_eba_adapter.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import pathlib
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 import yaml
@@ -48,8 +48,12 @@ class DummyWecomBotWsClient:
         self.send_message = AsyncMock(return_value={'ok': True})
         self.reply_text = AsyncMock(return_value={'reply': True})
         self.push_stream_chunk = AsyncMock(return_value=True)
+        self.finish_stream_for_interaction = AsyncMock(return_value=True)
+        self.begin_interaction_stream = Mock(return_value=True)
+        self.push_interaction_stream_chunk = AsyncMock(return_value=True)
         self.set_message = AsyncMock(return_value={'set': True})
         self.send_template_card = AsyncMock(return_value={'sent': True})
+        self.update_template_card = AsyncMock(return_value={'errcode': 0})
 
     def on_message(self, msg_type: str):
         def decorator(func):
@@ -109,6 +113,30 @@ def make_adapter(enable_webhook: bool = False) -> WecomBotAdapter:
         return WecomBotAdapter(config, DummyLogger())
 
 
+@pytest.mark.asyncio
+async def test_interaction_card_finishes_original_loading_stream():
+    adapter = make_adapter()
+    await adapter.call_platform_api(
+        'interaction.request',
+        {
+            'request': {'title': 'Continue', 'actions': [{'id': 'ok', 'label': 'OK'}]},
+            'reply_target': {'target_type': 'person', 'target_id': 'user-1', 'message_id': 'original-1'},
+            'callback_token': 'token-1',
+        },
+    )
+    adapter.bot.finish_stream_for_interaction.assert_awaited_once_with('original-1')
+
+
+@pytest.mark.asyncio
+async def test_interaction_reply_uses_new_wecom_stream():
+    adapter = make_adapter()
+    target = {'target_type': 'person', 'target_id': 'user-1', 'message_id': 'original-1'}
+    assert await adapter.begin_interaction_stream('original-1', target)
+    adapter.bot.begin_interaction_stream.assert_called_once_with('original-1', 'person', 'user-1')
+    await adapter.reply_interaction_stream_chunk(source_message_id='original-1', content='part', is_final=False)
+    adapter.bot.push_interaction_stream_chunk.assert_awaited_once_with('original-1', 'part', False)
+
+
 def wecombot_event(**overrides) -> WecomBotEvent:
     event_type = overrides.get('type', 'single')
     payload = {
@@ -216,6 +244,7 @@ async def test_wecombot_select_card_submits_native_dropdown_value():
         'eventtype': 'template_card_event',
         'template_card_event': {
             'card_type': 'multiple_interaction',
+            'task_id': result['task_id'],
             'event_key': card['submit_button']['key'],
             'selected_items': {
                 'selected_item': [
@@ -226,6 +255,8 @@ async def test_wecombot_select_card_submits_native_dropdown_value():
     }
     event = interaction_event_from_native(native)
     assert event.data['callback_token'] == 'callback-token'
+    assert event.data['callback_req_id'] == 'req-1'
+    assert event.data['callback_task_id'] == result['task_id']
     resolved = InteractionManager._resolve_submission_refs(
         {'interaction_id': request['interaction_id'], 'request': request}, event.data
     )
@@ -238,6 +269,70 @@ async def test_wecombot_select_card_submits_native_dropdown_value():
     assert listener.await_args.args[0].data['option_ref'] == 1
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize('kind', ['button', 'select'])
+async def test_wecombot_acknowledges_submitted_card(kind):
+    adapter = make_adapter()
+    request = {'title': 'Choose', 'actions': [{'id': 'yes', 'label': 'Yes'}, {'id': 'no', 'label': 'No'}]}
+    submission = {'action_id': 'no', 'values': {}}
+    if kind == 'select':
+        request = {
+            'title': 'Choose',
+            'actions': [],
+            'fields': [
+                {
+                    'id': 'choice',
+                    'type': 'select',
+                    'label': 'Choice',
+                    'options': [{'value': 'a', 'label': 'A'}, {'value': 'b', 'label': 'B'}],
+                }
+            ],
+        }
+        submission = {'action_id': None, 'values': {'choice': 'b'}}
+    delivered = await adapter.call_platform_api(
+        'interaction.request',
+        {
+            'callback_token': 'token-1',
+            'reply_target': {'target_type': 'person', 'target_id': 'user-1'},
+            'request': request,
+        },
+    )
+
+    result = await adapter.call_platform_api(
+        'interaction.acknowledge',
+        {
+            'request': request,
+            'submission': submission,
+            'update_target': delivered,
+            'callback_context': {'req_id': 'callback-req', 'task_id': delivered['task_id']},
+        },
+    )
+
+    assert result['updated'] is True
+    req_id, card = adapter.bot.update_template_card.await_args.args
+    assert req_id == 'callback-req'
+    assert card['task_id'] == delivered['task_id']
+    if kind == 'select':
+        assert card['select_list'][0]['selected_id'] == 'opt_2'
+        assert card['select_list'][0]['disable'] is True
+        assert card['submit_button']['key'] == 'submitted'
+    else:
+        assert card['button_list'][1]['replace_text'].endswith('No')
+        assert all(button['key'] == 'submitted' for button in card['button_list'])
+
+    with pytest.raises(ValueError, match='does not match'):
+        await adapter.call_platform_api(
+            'interaction.acknowledge',
+            {
+                'request': request,
+                'submission': submission,
+                'update_target': delivered,
+                'callback_context': {'req_id': 'callback-req', 'task_id': 'another-card'},
+            },
+        )
+    assert adapter.bot.update_template_card.await_count == 1
+
+
 def test_wecombot_select_rejects_more_than_ten_options():
     with pytest.raises(ValueError, match='at most 10 options'):
         build_interaction_card(
@@ -475,6 +570,29 @@ async def test_wecombot_send_reply_feedback_and_platform_api_use_underlying_clie
     assert feedback_calls[0].inaccurate_reasons == ['1', '2']
 
 
+@pytest.mark.asyncio
+async def test_wecombot_expired_stream_falls_back_to_proactive_reply():
+    adapter = make_adapter()
+    adapter.bot.push_stream_chunk.return_value = False
+    source_event = await WecomBotEventConverter().target2yiri(wecombot_event())
+    message = platform_message.MessageChain([platform_message.Plain(text='late answer')])
+
+    await adapter.reply_message_chunk(source_event, None, message, is_final=True)
+
+    adapter.bot.send_message.assert_awaited_once_with('user-1', 'late answer')
+    adapter.bot.reply_text.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_wecombot_send_message_rejects_failed_ack():
+    adapter = make_adapter()
+    adapter.bot.send_message.return_value = {'errcode': 400, 'errmsg': 'invalid message'}
+    with pytest.raises(RuntimeError, match='invalid message'):
+        await adapter.send_message(
+            'person', 'user-1', platform_message.MessageChain([platform_message.Plain(text='hello')])
+        )
+
+
 @pytest.mark.asyncio
 async def test_wecombot_webhook_mode_rejects_proactive_send():
     adapter = make_adapter(enable_webhook=True)

tests/unit_tests/platform/test_wecombot_template_card.py

@@ -1,14 +1,10 @@
-import sys
+import time
 import types
-from unittest.mock import Mock
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 
 
-logger_module = types.ModuleType('langbot.pkg.platform.logger')
-logger_module.EventLogger = object
-sys.modules.setdefault('langbot.pkg.platform.logger', logger_module)
-
 from langbot.libs.wecom_ai_bot_api.api import (  # noqa: E402
     WecomBotClient,
     build_button_interaction_payload,
@@ -45,7 +41,7 @@ def test_webhook_dispatch_tasks_are_bounded():
 
 
 @pytest.mark.asyncio
-async def test_ws_initial_stream_frame_precedes_pipeline_dispatch(monkeypatch):
+async def test_ws_message_waits_for_real_content_before_creating_stream(monkeypatch):
     from langbot.libs.wecom_ai_bot_api import ws_client as ws_client_module
 
     order = []
@@ -61,9 +57,7 @@ async def test_ws_initial_stream_frame_precedes_pipeline_dispatch(monkeypatch):
         return {'msgid': 'msg-1', 'type': 'single', 'userid': 'user-1'}
 
     async def reply_stream(*args, **kwargs):
-        del args, kwargs
-        order.append('initial-frame')
-        return {}
+        raise AssertionError('An empty stream must not be created')
 
     async def dispatch_event(event):
         del event
@@ -76,11 +70,126 @@ async def test_ws_initial_stream_frame_precedes_pipeline_dispatch(monkeypatch):
 
     await client._handle_message_callback({'headers': {'req_id': 'req-1'}, 'body': {}})
 
-    assert order == ['initial-frame', 'pipeline-dispatch']
+    assert order == ['pipeline-dispatch']
+    assert client._stream_sessions['msg-1']['stream_started'] is False
 
 
 @pytest.mark.asyncio
-async def test_ws_initial_stream_failure_still_dispatches_message(monkeypatch):
+async def test_ws_interaction_releases_unused_stream_without_visible_message():
+    client = WecomBotWsClient('bot-id', 'secret', object())
+    client._stream_ids['msg-1'] = 'req-1|stream-1'
+    client._stream_sessions['msg-1'] = {'user_id': 'user-1'}
+    sent = []
+
+    async def reply_stream(req_id, stream_id, content, finish=False):
+        sent.append((req_id, stream_id, content, finish))
+        return {'errcode': 0}
+
+    client.reply_stream = reply_stream
+    assert await client.finish_stream_for_interaction('msg-1')
+    assert sent == []
+    assert 'msg-1' not in client._stream_ids
+    assert 'msg-1' not in client._stream_sessions
+
+
+@pytest.mark.asyncio
+async def test_ws_interaction_starts_a_new_stream_after_card():
+    client = WecomBotWsClient('bot-id', 'secret', types.SimpleNamespace(info=AsyncMock()))
+    client._stream_ids['msg-1'] = 'req-1|old-stream'
+    client._stream_sessions['msg-1'] = {
+        'req_id': 'req-1',
+        'user_id': 'user-1',
+        'chat_id': '',
+        'created_at': time.monotonic(),
+        'stream_started': False,
+    }
+    sent = []
+
+    async def reply_stream(req_id, stream_id, content, finish=False, feedback_id=''):
+        sent.append((req_id, stream_id, content, finish))
+        return {'errcode': 0}
+
+    client.reply_stream = reply_stream
+    assert await client.finish_stream_for_interaction('msg-1')
+    assert sent == []
+    assert client.begin_interaction_stream('msg-1', 'person', 'user-1')
+    assert not client.begin_interaction_stream('msg-1', 'person', 'user-1')
+    assert await client.push_interaction_stream_chunk('msg-1', 'part', False)
+    assert await client.push_interaction_stream_chunk('msg-1', 'part done', True)
+    assert [item[2:] for item in sent] == [('part', False), ('part done', True)]
+    assert sent[0][0] == sent[1][0] == 'req-1'
+    assert sent[0][1] == sent[1][1]
+    assert sent[0][1] != 'old-stream'
+
+
+def test_ws_interaction_stream_checks_target_and_expiry():
+    client = WecomBotWsClient('bot-id', 'secret', object())
+    client._interaction_stream_origins['msg-1'] = {
+        'req_id': 'req-1',
+        'user_id': 'user-1',
+        'chat_id': 'group-1',
+        'created_at': time.monotonic(),
+    }
+    assert not client.begin_interaction_stream('msg-1', 'person', 'other-user')
+    assert not client.begin_interaction_stream('msg-1', 'group', 'other-group')
+    client._interaction_stream_origins['msg-1']['created_at'] -= 571
+    assert not client.begin_interaction_stream('msg-1', 'person', 'user-1')
+    assert 'msg-1' not in client._interaction_stream_origins
+
+
+@pytest.mark.asyncio
+async def test_ws_interaction_preserves_existing_stream_text_when_finishing():
+    client = WecomBotWsClient('bot-id', 'secret', object())
+    client._stream_ids['msg-1'] = 'req-1|stream-1'
+    client._stream_sessions['msg-1'] = {'stream_started': True}
+    client._stream_last_content['msg-1'] = 'Earlier answer'
+    sent = []
+
+    async def reply_stream(req_id, stream_id, content, finish=False):
+        sent.append((content, finish))
+        return {'errcode': 0}
+
+    client.reply_stream = reply_stream
+    assert await client.finish_stream_for_interaction('msg-1')
+    assert sent == [('Earlier answer', True)]
+
+
+@pytest.mark.asyncio
+async def test_ws_interaction_finish_failure_keeps_stream_state():
+    logger = types.SimpleNamespace(warning=AsyncMock())
+    client = WecomBotWsClient('bot-id', 'secret', logger)
+    client._stream_ids['msg-1'] = 'req-1|stream-1'
+    client._stream_sessions['msg-1'] = {'stream_started': True}
+
+    async def reject(*args, **kwargs):
+        raise ConnectionError('disconnected')
+
+    client.reply_stream = reject
+    assert not await client.finish_stream_for_interaction('msg-1')
+    assert client._stream_ids['msg-1'] == 'req-1|stream-1'
+
+
+@pytest.mark.asyncio
+async def test_ws_rejected_stream_refresh_does_not_complete_session():
+    logger = types.SimpleNamespace(error=Mock())
+    client = WecomBotWsClient('bot-id', 'secret', logger)
+    client._stream_ids['msg-1'] = 'req-1|stream-1'
+    client._stream_sessions['msg-1'] = {'user_id': 'user-1'}
+
+    async def reject(*args, **kwargs):
+        return {'errcode': 400, 'errmsg': 'invalid req_id'}
+
+    async def record_error(message):
+        assert 'invalid req_id' in message
+
+    client.reply_stream = reject
+    logger.error = record_error
+    assert not await client.push_stream_chunk('msg-1', 'answer', is_final=True)
+    assert 'msg-1' in client._stream_ids
+
+
+@pytest.mark.asyncio
+async def test_ws_stream_starts_with_first_real_content(monkeypatch):
     from langbot.libs.wecom_ai_bot_api import ws_client as ws_client_module
 
     dispatched = []
@@ -105,9 +214,11 @@ async def test_ws_initial_stream_failure_still_dispatches_message(monkeypatch):
         del args, kwargs
         return {'msgid': 'msg-1', 'type': 'single', 'userid': 'user-1'}
 
-    async def reply_stream(*args, **kwargs):
-        del args, kwargs
-        raise ConnectionError('simulated reply failure')
+    sent = []
+
+    async def reply_stream(req_id, stream_id, content, finish=False, feedback_id=''):
+        sent.append((req_id, content, finish))
+        return {'errcode': 0}
 
     async def dispatch_event(event):
         dispatched.append(event)
@@ -118,10 +229,17 @@ async def test_ws_initial_stream_failure_still_dispatches_message(monkeypatch):
     client._dispatch_event = dispatch_event
 
     await client._handle_message_callback({'headers': {'req_id': 'req-1'}, 'body': {}})
+    assert sent == []
+    assert await client.push_stream_chunk('msg-1', 'hello')
 
     assert len(dispatched) == 1
-    assert len(logger.warnings) == 1
-    assert 'simulated reply failure' in logger.warnings[0]
+    assert sent == [('req-1', 'hello', False)]
+    assert client._stream_sessions['msg-1']['stream_started'] is True
+    assert logger.warnings == []
+
+    assert await client.push_stream_chunk('msg-1', 'hello', is_final=True)
+    assert 'msg-1' not in client._stream_ids
+    assert client.begin_interaction_stream('msg-1', 'person', 'user-1')
 
 
 def test_extract_template_card_action_supports_nested_button_key():