fix(wechat-woc): 修复本账号消息循环拉取导致的死循环问题
1. 为微信企业微信适配器添加本账号发送消息过滤逻辑,分别在拉取适配器和流连接器适配器中实现 2. 修复拉取 worker 空消息场景下游标卡死问题,当有新游标且无有效消息时持久化新游标 3. 增加过滤日志便于排查消息丢失情况,统一本账号消息判断逻辑
This commit is contained in:
parent
713fcdcd8e
commit
3c26a68fa6
@ -176,6 +176,26 @@ class PullerWorker(BaseTransportWorker):
|
|||||||
continue
|
continue
|
||||||
|
|
||||||
if not result.messages:
|
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.backoff_attempt = 0
|
||||||
task_info.consecutive_successes += 1
|
task_info.consecutive_successes += 1
|
||||||
if task_info.consecutive_successes == 1:
|
if task_info.consecutive_successes == 1:
|
||||||
|
|||||||
@ -180,6 +180,23 @@ class WeChatWocPullerAdapter:
|
|||||||
# create_time"严格区分,后者必须前进)
|
# create_time"严格区分,后者必须前进)
|
||||||
next_cursor = cursor
|
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(
|
await self._logger.debug(
|
||||||
"woc poll completed",
|
"woc poll completed",
|
||||||
account_id=account_id,
|
account_id=account_id,
|
||||||
@ -286,4 +303,26 @@ class WeChatWocPullerAdapter:
|
|||||||
raise DependencyError(dep="config", cause=exc) from exc
|
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"]
|
__all__ = ["WeChatWocPullerAdapter"]
|
||||||
|
|||||||
@ -180,7 +180,14 @@ class WeChatWocStreamConnectorAdapter:
|
|||||||
while True:
|
while True:
|
||||||
# 缓冲队列有消息时优先返回,避免阻塞在读事件上
|
# 缓冲队列有消息时优先返回,避免阻塞在读事件上
|
||||||
if not buffer.empty():
|
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:
|
try:
|
||||||
event = await anext(events_iter)
|
event = await anext(events_iter)
|
||||||
@ -512,4 +519,24 @@ class WeChatWocStreamConnectorAdapter:
|
|||||||
raise DependencyError(dep="config", cause=exc) from exc
|
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"]
|
__all__ = ["WeChatWocStreamConnectorAdapter"]
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user