fix(wecomcs): handle service events and prevent duplicate replies

src/langbot/libs/wecom_customer_service_api/api.py

@@ -5,6 +5,7 @@ import base64
 import binascii
 import contextvars
 import functools
+import hashlib
 import httpx
 import json
 import os
@@ -119,6 +120,10 @@ class WecomCSClient:
             'example': [],
         }
         self._http_client: httpx.AsyncClient | None = None
+        self._delivery_tasks: set[asyncio.Task] = set()
+        self._received_messages: dict[str, float] = {}
+        self._sync_lock = asyncio.Lock()
+        self._sync_cursors: dict[str, str] = {}
 
     @asynccontextmanager
     async def _http_client_context(self):
@@ -127,6 +132,10 @@ class WecomCSClient:
         yield self._http_client
 
     async def close(self) -> None:
+        for task in self._delivery_tasks:
+            task.cancel()
+        if self._delivery_tasks:
+            await asyncio.gather(*self._delivery_tasks, return_exceptions=True)
         if self._http_client is not None:
             await self._http_client.aclose()
             self._http_client = None
@@ -170,47 +179,44 @@ class WecomCSClient:
             else:
                 raise Exception(f'未获取access token: {data}')
 
-    @_bounded_token_retry
     async def get_detailed_message_list(self, xml_msg: str):
-        # 在本方法中解析消息，并且获得消息的具体内容
-        if isinstance(xml_msg, bytes):
-            xml_msg = xml_msg.decode('utf-8')
+        """Drain all callback pages, including non-message events and empty pages."""
         root = await asyncio.to_thread(ET.fromstring, xml_msg)
-        token = root.find('Token').text
-        open_kfid = root.find('OpenKfId').text
-
-        # if open_kfid in self.openkfid_list:
-        #     return None
-        # else:
-        #     self.openkfid_list.append(open_kfid)
-
-        if not await self.check_access_token():
-            self.access_token = await self.get_access_token(self.secret)
-
-        url = self.base_url + '/kf/sync_msg?access_token=' + self.access_token
-        async with self._http_client_context() as client:
-            params = {
-                'token': token,
-                'voice_format': 0,
-                'open_kfid': open_kfid,
-            }
-            response = await client.post(url, json=params)
-            data = await httpclient.parse_json_response(response)
-            if data['errcode'] == 40014 or data['errcode'] == 42001:
-                self.access_token = await self.get_access_token(self.secret)
-                return await self.get_detailed_message_list(xml_msg)
-            if data['errcode'] != 0:
-                raise Exception('Failed to get message')
-
-            last_msg_data = data['msg_list'][-1]
-            open_kfid = last_msg_data.get('open_kfid')
-            # 进行获取图片操作
-            if last_msg_data.get('msgtype') == 'image':
-                media_id = last_msg_data.get('image').get('media_id')
-                picurl = await self.get_pic_url(media_id)
-                last_msg_data['picurl'] = picurl
-            # await self.change_service_status(userid=external_userid,openkfid=open_kfid,servicer=servicer)
-            return last_msg_data
+        token = root.findtext('Token')
+        open_kfid = root.findtext('OpenKfId')
+        if not token or not open_kfid:
+            raise ValueError('WeCom customer-service callback is missing Token or OpenKfId')
+        async with self._sync_lock:
+            cursor = self._sync_cursors.get(open_kfid)
+            messages = []
+            refreshed = False
+            while True:
+                if not await self.check_access_token():
+                    self.access_token = await self.get_access_token(self.secret)
+                params = {'token': token, 'voice_format': 0, 'open_kfid': open_kfid, 'limit': 1000}
+                if cursor:
+                    params['cursor'] = cursor
+                async with self._http_client_context() as client:
+                    response = await client.post(
+                        self.base_url + '/kf/sync_msg', params={'access_token': self.access_token}, json=params,
+                    )
+                    data = await httpclient.parse_json_response(response)
+                if data.get('errcode') in (40014, 42001) and not refreshed:
+                    self.access_token = await self.get_access_token(self.secret)
+                    refreshed = True
+                    continue
+                if data.get('errcode') != 0:
+                    raise RuntimeError(f"WeCom sync_msg failed (errcode={data.get('errcode')})")
+                messages.extend(data.get('msg_list') or [])
+                next_cursor = data.get('next_cursor')
+                if data.get('has_more'):
+                    if not next_cursor or next_cursor == cursor:
+                        raise RuntimeError('WeCom sync_msg returned a non-advancing cursor')
+                    cursor = next_cursor
+                    continue
+                if next_cursor:
+                    self._sync_cursors[open_kfid] = next_cursor
+                return messages
 
     @_bounded_token_retry
     async def change_service_status(self, userid: str, openkfid: str, servicer: str):
@@ -292,9 +298,36 @@ class WecomCSClient:
                 return await self.send_text_msg(open_kfid, external_userid, msgid, content)
             if data['errcode'] != 0:
                 await self.logger.error(f'发送消息失败：{data}')
+                if data['errcode'] == 95002:
+                    raise ValueError(
+                        'WeCom CS rejected the reply (95002): the customer must send a message first, '
+                        'and ordinary replies must be sent within 48 hours. Entering a session alone '
+                        'does not open this window; welcome replies require the event reply API.'
+                    )
                 raise Exception(f'Failed to send message: {data}')
             return data
 
+    @_bounded_token_retry
+    async def send_event_text(self, code: str, msgid: str, content: str):
+        """Send a single event response using its short-lived, one-time code."""
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+        url = f'{self.base_url}/kf/send_msg_on_event?access_token={self.access_token}'
+        payload = {'code': code, 'msgid': msgid, 'msgtype': 'text', 'text': {'content': content}}
+        async with self._http_client_context() as client:
+            response = await client.post(url, json=payload)
+            data = await httpclient.parse_json_response(response)
+        if data.get('errcode') in (40014, 42001):
+            self.access_token = await self.get_access_token(self.secret)
+            return await self.send_event_text(code, msgid, content)
+        if data.get('errcode') != 0:
+            raise ValueError(
+                f'WeCom CS event reply failed (errcode={data.get("errcode")}): {data.get("errmsg")}. '
+                'The event code is single-use and must match the current session state; '
+                'welcome and session-end codes expire after 20 seconds. Do not retry an expired or used code.'
+            )
+        return data
+
     @_bounded_token_retry
     async def send_image_msg(self, open_kfid: str, external_userid: str, msgid: str, media_id: str):
         if not await self.check_access_token():
@@ -375,12 +408,11 @@ class WecomCSClient:
                 if ret != 0:
                     raise Exception(f'消息解密失败，错误码: {ret}')
 
-                # 解析消息并处理
-                message_data = await self.get_detailed_message_list(xml_msg)
-                if message_data is not None:
-                    event = WecomCSEvent.from_payload(message_data)
-                    if event:
-                        await self._handle_message(event)
+                # Acknowledge after verification; API fetching and processing must
+                # survive a callback disconnect.
+                task = asyncio.create_task(self._process_callback(xml_msg))
+                self._delivery_tasks.add(task)
+                task.add_done_callback(self._delivery_tasks.discard)
 
                 return 'success'
         except Exception as e:
@@ -390,6 +422,46 @@ class WecomCSClient:
                 traceback.print_exc()
             return f'Error processing request: {str(e)}', 400
 
+    async def _process_callback(self, xml_msg):
+        try:
+            messages = await self.get_detailed_message_list(xml_msg)
+            for message_data in messages:
+                try:
+                    await self._process_synced_message(message_data)
+                except Exception:
+                    if self.logger:
+                        await self.logger.error(f'WeCom customer-service event failed: {traceback.format_exc()}')
+
+        except Exception:
+            if self.logger:
+                await self.logger.error(f'WeCom customer-service delivery failed: {traceback.format_exc()}')
+
+    async def _process_synced_message(self, message_data):
+        # Agent/API and human replies must not become new customer input.
+        if message_data.get('msgtype') != 'event' and message_data.get('origin', 3) != 3:
+            return
+        if message_data.get('msgtype') == 'image':
+            media_id = (message_data.get('image') or {}).get('media_id')
+            if media_id:
+                message_data['picurl'] = await self.get_pic_url(media_id)
+        event = WecomCSEvent.from_payload(message_data)
+        if event:
+            # A callback disconnect must not cancel processor execution.
+            # Retries can return the same message from sync_msg.
+            now = time.monotonic()
+            self._received_messages = {
+                key: seen for key, seen in self._received_messages.items() if now - seen < 4 * 86400
+            }
+            message_id = message_data.get('msgid') or hashlib.sha256(
+                json.dumps(message_data, sort_keys=True, ensure_ascii=False, separators=(',', ':')).encode()
+            ).hexdigest()
+            if not message_id or message_id not in self._received_messages:
+                if message_id:
+                    if len(self._received_messages) >= 4096:
+                        self._received_messages.pop(next(iter(self._received_messages)))
+                    self._received_messages[message_id] = now
+                await self._handle_message(event)
+
     async def run_task(self, host: str, port: int, *args, **kwargs):
         """
         启动 Quart 应用。

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

@@ -749,6 +749,14 @@ async def execute_platform_tool(
                     message_id=delivery['reply_target']['message_id'],
                 )
                 return await bot.adapter.reply_message(source, message)
+        # Some platforms require a one-time event credential instead of an
+        # ordinary conversation send. Only pass the Host-frozen event context.
+        event_reply_handler = getattr(bot.adapter, 'reply_to_event', None)
+        if definition.name == 'event_reply' and callable(event_reply_handler):
+            async def api_func(target_type, target_id, message):
+                return await event_reply_handler(
+                    context=context, target_type=target_type, target_id=target_id, message=message
+                )
         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/service/monitoring_execution_details.py

@@ -37,6 +37,22 @@ def event_content(content, event_data):
     return {'text': event_summary(event_data), 'event': event_data, 'input': content}
 
 
+def event_preview(payload, fallback='', event_type=''):
+    """Never use serialized event metadata as a human-readable list preview."""
+    if event_type.startswith('wecomcs.'):
+        return ''
+    if isinstance(payload, dict):
+        if payload.get('type'):
+            if str(payload['type']).startswith('wecomcs.'):
+                return ''
+            return event_summary(payload)[:1000]
+        if isinstance(payload.get('text'), str):
+            return payload['text'][:1000]
+    if fallback and not isinstance(parsed(fallback), (dict, list)):
+        return fallback[:1000]
+    return ''
+
+
 def event_item(row):
     metadata = parsed(row.metadata_json, {})
     return {
@@ -240,14 +256,13 @@ class ExecutionDetailsMixin:
     ):
         """Persist entry facts even when routing never creates a processor run."""
         from ....agent.runner.event_log_store import EventLogStore
-        from .monitoring import _message_preview
 
         workspace = self._require_write_context(context)
         store = EventLogStore(self.ap.persistence_mgr.get_db_engine())
         # Reuse the message sanitizer (including inline-media handling) instead
         # of storing an unbounded adapter object in the event journal.
         content = parsed(self._sanitize_message_content(json.dumps(payload, ensure_ascii=False, default=str)), {})
-        summary = _message_preview(json.dumps(content, ensure_ascii=False), 1000) if content else event_type
+        summary = event_preview(content, event_type=event_type)
         await store.append_event(
             event_id=event_id,
             event_type=event_type,
@@ -266,7 +281,7 @@ class ExecutionDetailsMixin:
             safe_routes = []
             for route in routes:
                 if isinstance(route, BaseException):
-                    safe_routes.append({'status': 'failed', 'reason': str(route)})
+                    safe_routes.append({'status': 'failed', 'reason': str(route) or type(route).__name__})
                 elif isinstance(route, dict):
                     safe_routes.append(
                         {
@@ -362,7 +377,7 @@ class ExecutionDetailsMixin:
             'status': status,
             'status_group': status,
             'title': row.event_type,
-            'input_preview': row.input_summary or '',
+            'input_preview': event_preview(parsed(row.input_json), row.input_summary, row.event_type),
             'target_kind': 'event',
             'target_id': None,
             'target_name': None,
@@ -385,7 +400,11 @@ class ExecutionDetailsMixin:
             'runner_id': None,
             'debug': row.source == 'debug',
             'has_error': status == 'failed',
-            'status_reason': metadata.get('reason'),
+            'status_reason': metadata.get('reason') or '; '.join(
+                route.get('reason') or route.get('failure_code') or 'failure_reason_not_recorded'
+                for route in metadata.get('routes', [])
+                if isinstance(route, dict) and route.get('status') == 'failed'
+            ) or None,
         }
 
     async def enrich_execution_rows(self, workspace, items):
@@ -449,7 +468,7 @@ class ExecutionDetailsMixin:
                 meta = parsed(event.metadata_json, {})
                 item.update(
                     event_type=event.event_type,
-                    input_preview=event.input_summary or item.get('input_preview') or '',
+                    input_preview=event_preview(parsed(event.input_json), event.input_summary, event.event_type),
                     user_id=event.actor_id,
                     user_name=event.actor_name,
                     platform=meta.get('platform') or event.source,

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

@@ -79,6 +79,12 @@ class WecomCSAdapter(WecomCSAPIMixin, abstract_platform_adapter.AbstractPlatform
     def get_supported_events(self) -> list[str]:
         return [
             'message.received',
+            'message.deleted',
+            'wecomcs.enter_session',
+            'wecomcs.msg_send_fail',
+            'wecomcs.servicer_status_change',
+            'wecomcs.session_status_change',
+            'wecomcs.reject_customer_msg_switch_change',
             'platform.specific',
         ]
 
@@ -110,6 +116,36 @@ class WecomCSAdapter(WecomCSAPIMixin, abstract_platform_adapter.AbstractPlatform
             )
         return platform_events.MessageResult(raw={'results': raw_results})
 
+    async def reply_to_event(
+        self, *, context: dict, target_type: str, target_id: str, message: platform_message.MessageChain
+    ) -> platform_events.MessageResult:
+        """Use event credentials for welcome and session-status replies."""
+        event_type = context.get('event_type')
+        if event_type not in ('wecomcs.enter_session', 'wecomcs.session_status_change'):
+            return await self.send_message(target_type, target_id, message)
+        data = context.get('data') or {}
+        code_key = 'welcome_code' if event_type == 'wecomcs.enter_session' else 'msg_code'
+        code = data.get(code_key) or (data.get('data') or {}).get(code_key)
+        if not code:
+            raise ValueError(
+                f'WeCom CS did not provide {code_key} for this event; an event response is not available. '
+                'Entering a session does not open the ordinary messaging window. '
+                'Ask the customer to send a message before sending an ordinary reply.'
+            )
+        timestamp = data.get('timestamp')
+        if event_type == 'wecomcs.enter_session' and timestamp and time.time() - float(timestamp) >= 20:
+            raise ValueError(
+                'WeCom CS welcome reply expired: welcome_code is valid for only 20 seconds after entry. '
+                'Use a fast, fixed welcome response; do not wait for a long AI run or retry this event.'
+            )
+        if not message.root or any(not isinstance(item, platform_message.Plain) for item in message.root):
+            raise ValueError('WeCom CS event replies currently support a single text message only')
+        text = ''.join(item.text for item in message.root)
+        if not text.strip():
+            raise ValueError('WeCom CS event reply text must not be empty')
+        result = await self.bot.send_event_text(str(code), self._make_outbound_msgid(), text)
+        return platform_events.MessageResult(message_id=result.get('msgid'), raw=result)
+
     async def reply_message(
         self,
         message_source: platform_events.MessageEvent,
@@ -183,7 +219,7 @@ class WecomCSAdapter(WecomCSAPIMixin, abstract_platform_adapter.AbstractPlatform
         async def on_message(event: WecomCSEvent):
             await self._handle_native_event(event)
 
-        for msg_type in ('text', 'image', 'file', 'voice'):
+        for msg_type in ('text', 'image', 'file', 'voice', 'event'):
             self.bot.on_message(msg_type)(on_message)
 
     async def _handle_native_event(self, event: WecomCSEvent):

src/langbot/pkg/platform/adapters/wecomcs/event_converter.py

@@ -29,6 +29,39 @@ class WecomCSEventConverter(abstract_platform_adapter.AbstractEventConverter):
     async def target2yiri(event: WecomCSEvent, bot: WecomCSClient | None = None) -> platform_events.Event | None:
         if event.type in {'text', 'image', 'file', 'voice'}:
             return await WecomCSEventConverter.message_to_eba(event, bot)
+        if event.type == 'event':
+            data = event.get('event') or {}
+            event_type = data.get('event_type', 'unknown')
+            customer_id = data.get('external_userid') or ''
+            common = {
+                'adapter_name': ADAPTER_NAME,
+                'timestamp': float(event.timestamp or 0),
+                'source_platform_object': event,
+                'chat_id': make_private_chat_id(customer_id, data.get('open_kfid')) if customer_id else '',
+            }
+            if event_type in {'user_recall_msg', 'servicer_recall_msg'}:
+                operator_id = data.get('servicer_userid') if event_type == 'servicer_recall_msg' else customer_id
+                return platform_events.MessageDeletedEvent(
+                    message_id=data.get('recall_msgid') or '',
+                    operator=platform_entities.User(id=operator_id) if operator_id else None,
+                    **common,
+                )
+            event_classes = {
+                'enter_session': platform_events.WecomCSEnterSessionEvent,
+                'msg_send_fail': platform_events.WecomCSMessageSendFailedEvent,
+                'servicer_status_change': platform_events.WecomCSServicerStatusChangedEvent,
+                'session_status_change': platform_events.WecomCSSessionStatusChangedEvent,
+                'reject_customer_msg_switch_change': platform_events.WecomCSRejectCustomerMessageChangedEvent,
+            }
+            event_class = event_classes.get(event_type)
+            if event_class:
+                fields = {key: value for key, value in data.items() if key in event_class.model_fields and value is not None}
+                fields.update(common)
+                fields['data'] = dict(data)
+                actor_id = data.get('servicer_userid') or customer_id
+                fields['user'] = platform_entities.User(id=actor_id) if actor_id else None
+                return event_class(**fields)
+            return WecomCSEventConverter.platform_specific(event, f'wecomcs.{event_type}')
         return WecomCSEventConverter.platform_specific(event, f'wecomcs.{event.type or "unknown"}')
 
     @staticmethod

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

@@ -93,6 +93,12 @@ spec:
 
   supported_events:
     - message.received
+    - message.deleted
+    - wecomcs.enter_session
+    - wecomcs.msg_send_fail
+    - wecomcs.servicer_status_change
+    - wecomcs.session_status_change
+    - wecomcs.reject_customer_msg_switch_change
     - platform.specific
 
   supported_apis:

src/langbot/pkg/platform/botmgr.py

@@ -4,6 +4,7 @@ import asyncio
 import contextlib
 import dataclasses
 import functools
+import hashlib
 import json
 import re
 import time
@@ -85,6 +86,7 @@ class RuntimeBot:
     ):
         if not isinstance(execution_context, ExecutionContext):
             raise WorkspaceRequiredError('RuntimeBot requires an ExecutionContext')
+        self._ingress_dedup_lock = asyncio.Lock()
         if not execution_context.instance_uuid.strip() or not execution_context.workspace_uuid.strip():
             raise WorkspaceRequiredError('RuntimeBot requires an instance and Workspace')
         if execution_context.placement_generation <= 0:
@@ -888,6 +890,18 @@ class RuntimeBot:
     @staticmethod
     def _platform_event_raw_id(event: platform_events.EBAEvent) -> str:
         event_type = getattr(event, 'type', None) or event.__class__.__name__
+        if isinstance(event, platform_events.MessageDeletedEvent) and event.message_id:
+            # A recall is distinct from the original message.
+            return f'{event_type}:{event.message_id}'
+        if getattr(event, 'adapter_name', None) == 'wecomcs-omni':
+            source = getattr(event, 'source_platform_object', None)
+            if isinstance(source, dict):
+                if source.get('msgid'):
+                    return str(source['msgid'])
+                # Native service events can lack msgid. Preserve their identity
+                # across sync retries and process restarts instead of using UUIDs.
+                payload = json.dumps(source, sort_keys=True, ensure_ascii=False, separators=(',', ':'))
+                return f'{event_type}:{hashlib.sha256(payload.encode()).hexdigest()}'
         return str(
             getattr(event, 'message_id', None) or getattr(event, 'feedback_id', None) or f'{event_type}:{uuid.uuid4()}'
         )
@@ -904,6 +918,16 @@ class RuntimeBot:
 
         event.bot_uuid = self.bot_entity.uuid
         execution_id = self._platform_event_execution_id(event)
+        if getattr(event, 'adapter_name', None) == 'wecomcs-omni':
+            from ..agent.runner.event_log_store import EventLogStore
+
+            # The sync API can replay history after an adapter restart. Reuse
+            # the durable ingress journal, including records from earlier runs.
+            async with self._ingress_dedup_lock:
+                store = EventLogStore(self.ap.persistence_mgr.get_db_engine())
+                if await store.get_event(execution_id):
+                    return
+                await self._persist_monitoring_ingress(event, adapter, execution_id)
         with ingress(
             getattr(self, 'ap', None),
             'event_done',
@@ -945,7 +969,7 @@ class RuntimeBot:
         if service is None or not execution_id:
             return
         try:
-            sender = getattr(event, 'sender', None)
+            sender = getattr(event, 'sender', None) or getattr(event, 'user', None) or getattr(event, 'operator', None)
             target_type, target_id, _ = self._infer_reply_target(event)
             await service.record_ingress_event(
                 self.execution_context,