diff --git a/backend/package/yuxi/channels/application/transport/puller_worker.py b/backend/package/yuxi/channels/application/transport/puller_worker.py index 76982a2c..17dc64a2 100644 --- a/backend/package/yuxi/channels/application/transport/puller_worker.py +++ b/backend/package/yuxi/channels/application/transport/puller_worker.py @@ -176,6 +176,26 @@ class PullerWorker(BaseTransportWorker): continue if not result.messages: + # 适配器可能过滤了消息(如 wechat_woc 过滤本账号发送的回复), + # 此时 next_cursor 已基于全量消息前进。若不持久化,下次仍用旧 + # cursor 拉取被过滤的消息,导致 cursor 卡死形成静默死循环 + # (不触发 AgentRun 但无限轮询,且阻塞后续用户消息)。仅当 + # next_cursor 非空且前进时持久化;无消息且 cursor 不变时保持 + # 原行为(不写 DB),避免空轮询频繁持久化。 + if result.next_cursor is not None and result.next_cursor != cursor: + persisted = await self._saveCursor( + channel_type, + account_id, + result.next_cursor, + last_persisted_cursor, + error_trace_id, + expected_version=account_version, + ) + if persisted: + cursor = result.next_cursor + last_persisted_cursor = result.next_cursor + # 持久化失败时不进入退避重试(无消息非错误),用旧 cursor + # 下次重新拉取,保证 at-least-once 语义。 task_info.backoff_attempt = 0 task_info.consecutive_successes += 1 if task_info.consecutive_successes == 1: diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/puller_adapter.py b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/puller_adapter.py index a3a35a4d..b5ff1a3e 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/puller_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/puller_adapter.py @@ -180,6 +180,23 @@ class WeChatWocPullerAdapter: # create_time"严格区分,后者必须前进) next_cursor = cursor + # 过滤本账号发送的消息(is_sender 真值)。bridge /api/messages/since + # 返回会话内全部消息,含系统经 outbound 发送的回复;若不过滤,这些 + # 回复会被当作用户消息重新拉入,触发 AgentRun → 新回复 → 再被回拉, + # 形成无限循环。游标已基于全量消息计算,过滤不影响前进(C-1)。 + original_count = len(messages) + messages = [m for m in messages if not _is_self_sent(m)] + skipped_self = original_count - len(messages) + if skipped_self > 0: + await self._logger.debug( + "woc poll filtered self-sent messages", + account_id=account_id, + trace_id=trace_id, + original_count=original_count, + filtered_count=skipped_self, + remaining_count=len(messages), + ) + await self._logger.debug( "woc poll completed", account_id=account_id, @@ -286,4 +303,26 @@ class WeChatWocPullerAdapter: raise DependencyError(dep="config", cause=exc) from exc +def _is_self_sent(msg: dict) -> bool: + """判断消息是否为本账号发送(``is_sender`` 真值)。 + + bridge 契约返回 int ``0``/``1``(``0``=对方发送,``1``=本账号发送)。 + 兼容 bool 与 string 真值,显式排除 ``0``/``"0"``/``False`` 避免字符串 + 误判导致用户消息被丢弃(消息丢失比循环更严重)。 + + 非 dict 输入返回 False(透传下游保持原有单条失败语义,避免在 poll + 列表推导中抛异常导致整批消息丢失)。 + + @consistency: 无状态,纯函数。 + """ + if not isinstance(msg, dict): + return False + is_sender = msg.get("is_sender") + if is_sender is None or is_sender is False or is_sender == 0: + return False + if isinstance(is_sender, str): + return is_sender.strip().lower() in {"1", "true", "yes"} + return bool(is_sender) + + __all__ = ["WeChatWocPullerAdapter"] diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/stream_connector_adapter.py b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/stream_connector_adapter.py index d8115ba6..5aa61f6b 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/stream_connector_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/stream_connector_adapter.py @@ -180,7 +180,14 @@ class WeChatWocStreamConnectorAdapter: while True: # 缓冲队列有消息时优先返回,避免阻塞在读事件上 if not buffer.empty(): - return await buffer.get() + msg = await buffer.get() + # 过滤本账号发送的消息(is_sender 真值)。bridge SSE 推送 + # 含系统经 outbound 发送的回复;若不过滤,这些回复会被 + # 当作用户消息重新处理,触发 AgentRun → 新回复 → 再被 + # 推送,形成无限循环。游标由 sync 全局游标驱动,不受影响。 + if _is_self_sent(msg): + continue + return msg try: event = await anext(events_iter) @@ -512,4 +519,24 @@ class WeChatWocStreamConnectorAdapter: raise DependencyError(dep="config", cause=exc) from exc +def _is_self_sent(msg: dict) -> bool: + """判断消息是否为本账号发送(``is_sender`` 真值)。 + + bridge 契约返回 int ``0``/``1``(``0``=对方发送,``1``=本账号发送)。 + 兼容 bool 与 string 真值,显式排除 ``0``/``"0"``/``False`` 避免字符串 + 误判导致用户消息被丢弃。与 ``puller_adapter._is_self_sent`` 同逻辑 + (跨模块不 import 私有函数,各自内联)。 + + @consistency: 无状态,纯函数。 + """ + if not isinstance(msg, dict): + return False + is_sender = msg.get("is_sender") + if is_sender is None or is_sender is False or is_sender == 0: + return False + if isinstance(is_sender, str): + return is_sender.strip().lower() in {"1", "true", "yes"} + return bool(is_sender) + + __all__ = ["WeChatWocStreamConnectorAdapter"]