perf: optimize webchat and wecom ai queue lifecycle

[Soulter]

fixes: #4815

Summary

• replace polling queue listeners with event-driven listener registration for webchat and wecom ai bot
• add bounded queue sizes to provide backpressure and avoid unbounded memory growth
• isolate webchat back queues by message_id and clean them in request finally blocks
• improve wecom ai stream cleanup, including complete/end handling and finished-stream fallback for repeated polling

Validation

• ruff format .
• ruff check .

[dosubot]

Related Documentation

Checked 1 published document(s) in 1 knowledge base(s). No updates required.

^{How did I do? Any feedback?}  

[sourcery-ai]

你好，我发现了 5 个问题，并补充了一些整体性的反馈：

• 在 WecomAIQueueMgr 中，is_stream_finished 和 cleanup_expired_responses 都在管理 completed_streams 的 TTL，但一个用的是参数（默认值），另一个则是硬编码的 60 秒字面量；建议将这个保留时长集中到一个地方进行配置，以避免随着时间推移在行为上产生微妙差异。
• 两个队列管理器的 _listen_to_queue 方法在监听回调抛出异常时，只记录了简短的日志信息并吞掉异常；建议在日志中加入 exc_info=True 或完整的 traceback，方便在生产环境中排查事件驱动回调中的失败。

给 AI 代理的提示词

Please address the comments from this code review:

## Overall Comments
- In `WecomAIQueueMgr`, `is_stream_finished` and `cleanup_expired_responses` both manage `completed_streams` TTLs but use different hard-coded/default values (function arg vs. literal 60s); consider centralizing this retention period in one place to avoid subtle discrepancies in behavior over time.
- Both queue managers’ `_listen_to_queue` methods swallow listener callback exceptions with only a short log message; it may be helpful to include `exc_info=True` or the full traceback in these logs so that failures in the event-driven callbacks are easier to diagnose in production.

## Individual Comments

### Comment 1
<location> `astrbot/core/platform/sources/wecom_ai_bot/wecomai_queue_mgr.py:183-187` </location>
<code_context>
+            return False
+        return True
+
     def cleanup_expired_responses(self, max_age_seconds: int = 300):
         """清理过期的待处理响应

</code_context>

<issue_to_address>
**建议（suggestion）：** 对齐 `pending_responses` 与 `completed_streams` 的 TTL 处理方式，以避免它们的存活时间出现出人意料的差异。

`cleanup_expired_responses` 使用 `max_age_seconds` 来处理 `pending_responses`，但 `completed_streams` 始终使用硬编码的 60 秒进行清理：

```python
expired_finished = [
    session_id
    for session_id, finished_at in self.completed_streams.items()
    if current_time - finished_at > 60
]
```
这意味着调用方修改 `max_age_seconds` 时只会影响 `pending_responses`，从 API 上并不直观。建议将 `completed_streams` 的 TTL 绑定到 `max_age_seconds`（或由其推导出的值），或者把 `60` 提取为具名常量/参数，使 TTL 行为更加显式且可配置。

```suggestion
        expired_finished = [
            session_id
            for session_id, finished_at in self.completed_streams.items()
            if current_time - finished_at > max_age_seconds
        ]
```
</issue_to_address>

### Comment 2
<location> `astrbot/core/platform/sources/webchat/webchat_queue_mgr.py:39-41` </location>
<code_context>
+        """Remove back queue for the given request ID"""
+        self.back_queues.pop(request_id, None)

     def remove_queues(self, conversation_id: str):
         """Remove queues for the given conversation ID"""
-        if conversation_id in self.queues:
-            del self.queues[conversation_id]
-        if conversation_id in self.back_queues:
-            del self.back_queues[conversation_id]
+        self.remove_queue(conversation_id)
+
+    def remove_queue(self, conversation_id: str):
</code_context>

<issue_to_address>
**问题（bug_risk，待确认）：** 请再次确认与某个会话关联的 back 队列是否也需要被清理，以避免内存泄漏。

之前 `remove_queues(conversation_id)` 会删除以该 `conversation_id` 为 key 的输入队列和 back 队列。重构之后，back 队列改为使用 `request_id` 作为 key，而 `remove_queues` 现在只调用 `remove_queue`，它只会移除输入队列/监听器。

如果仍有现有调用方依赖 `remove_queues(conversation_id)` 来完整清理会话相关资源，那么对应的 back 队列现在可能会被无限期保留。请检查所有调用点（或者添加一个能够根据会话名义/显式映射来移除该会话所有队列的辅助方法），以确保在需要的情况下 back 队列同样会被清理。
</issue_to_address>

### Comment 3
<location> `astrbot/core/platform/sources/wecom_ai_bot/wecomai_queue_mgr.py:227-236` </location>
<code_context>
-        """Listen to a specific conversation queue"""
-        queue = self.webchat_queue_mgr.get_or_create_queue(conversation_id)
-        while True:
-            try:
-                data = await queue.get()
-                await self.callback(data)
</code_context>

<issue_to_address>
**建议（suggestion）：** 对 WeCom 监听器回调中的异常也记录 traceback 详情。

当前 `_listen_to_queue` 里，在回调报错时只记录了异常消息：

```python
logger.error(f"处理会话 {session_id} 消息时发生错误: {e}")
```
由于这是在后台任务中运行，捕获完整的堆栈信息会让问题更易于调试。建议改为使用 `logger.exception`，或者 `logger.error(..., exc_info=True)`，以便将完整 traceback 打到日志中。
</issue_to_address>

### Comment 4
<location> `astrbot/core/platform/sources/wecom_ai_bot/wecomai_queue_mgr.py:16` </location>
<code_context>
 class WebChatQueueMgr:
-    def __init__(self) -> None:
-        self.queues = {}
+    def __init__(self, queue_maxsize: int = 128, back_queue_maxsize: int = 512) -> None:
+        self.queues: dict[str, asyncio.Queue] = {}
         """Conversation ID to asyncio.Queue mapping"""
</code_context>

<issue_to_address>
**问题（complexity）：** 建议重构新增的监听器与生命周期管理逻辑，以拆分职责、集中清理逻辑，并封装异步等待/取消模式，从而提升可读性与可维护性。

当前在该类中添加的监听器/生命周期逻辑比较密集。可以在保持功能不变的前提下，通过以下方式降低复杂度：

### 1. 分离生命周期职责

目前 `remove_queues`、`remove_queue`、`cleanup_expired_responses` 和 `mark_finished` 的逻辑交织在一起。建议将“标记完成”和“清理”分离，并把清理逻辑统一到一个地方。

```python
def mark_stream_finished(self, session_id: str) -> None:
    """单独负责标记流结束"""
    self.completed_streams[session_id] = asyncio.get_event_loop().time()
    logger.debug(f"[WecomAI] 标记流已结束: {session_id}")

def _cleanup_session(self, session_id: str) -> None:
    """集中清理所有与 session 相关的状态"""
    # 输入队列 + 监听
    if session_id in self.queues:
        del self.queues[session_id]
        logger.debug(f"[WecomAI] 移除输入队列: {session_id}")
    close_event = self._queue_close_events.pop(session_id, None)
    if close_event is not None:
        close_event.set()
    task = self._listener_tasks.pop(session_id, None)
    if task is not None:
        task.cancel()

    # 输出队列
    if session_id in self.back_queues:
        del self.back_queues[session_id]
        logger.debug(f"[WecomAI] 移除输出队列: {session_id}")

    # 待处理响应
    if session_id in self.pending_responses:
        del self.pending_responses[session_id]
        logger.debug(f"[WecomAI] 移除待处理响应: {session_id}")
```

然后让对外的 API 更小、更清晰：

```python
def remove_queues(self, session_id: str, mark_finished: bool = False) -> None:
    self._cleanup_session(session_id)
    if mark_finished:
        self.mark_stream_finished(session_id)

def remove_queue(self, session_id: str) -> None:
    # 保留向后兼容，但委托给统一清理逻辑
    self._cleanup_session(session_id)
```

`cleanup_expired_responses` 可以调用 `_cleanup_session`，避免部分清理逻辑重复。

---

### 2. 封装 wait/cancel 模式

`_listen_to_queue` 目前在多个地方内联了一个带手动取消的 `asyncio.wait` 模式。建议提取到一个辅助方法中，这样取消行为更容易审计和复用：

```python
async def _wait_queue_or_close(
    self,
    queue: asyncio.Queue,
    close_event: asyncio.Event,
) -> dict | None:
    """返回队列数据, 或在关闭时返回 None"""
    get_task = asyncio.create_task(queue.get())
    close_task = asyncio.create_task(close_event.wait())
    try:
        done, pending = await asyncio.wait(
            {get_task, close_task},
            return_when=asyncio.FIRST_COMPLETED,
        )
        for task in pending:
            task.cancel()
        if close_task in done:
            return None
        return get_task.result()
    finally:
        if not get_task.done():
            get_task.cancel()
        if not close_task.done():
            close_task.cancel()
```

这样 `_listen_to_queue` 会清晰很多：

```python
async def _listen_to_queue(
    self,
    session_id: str,
    queue: asyncio.Queue,
    close_event: asyncio.Event,
):
    while True:
        try:
            data = await self._wait_queue_or_close(queue, close_event)
            if data is None:  # close_event set
                break
            if self._listener_callback is None:
                continue
            try:
                await self._listener_callback(data)
            except Exception as e:
                logger.error(f"处理会话 {session_id} 消息时发生错误: {e}")
        except asyncio.CancelledError:
            break
```

这样既保留了原有的行为（队列或关闭二选一、取消未完成任务），又把复杂的 `asyncio.wait` 逻辑集中到一个地方。

---

### 3. 让监听器职责更加明确

为减少“推模式 vs 拉模式”的双重 API 混淆，可以让管理器的对外接口更小、更明确：

- 在文档中说明 `set_listener` 是可选的；当设置后，管理器会自动为每个 session 启动监听任务。
- 可以考虑为每个 session 显式开启监听，而不是默认在队列创建时就绑定监听任务，这样可以减弱“存储 + 编排”混在一起的感觉。例如：

```python
def enable_listener_for_session(self, session_id: str) -> None:
    if self._listener_callback is None:
        raise RuntimeError("Listener callback must be set before enabling listener.")
    self._start_listener_if_needed(session_id)
```

这样 `get_or_create_queue` 只负责创建队列，而监听编排是显式的、按需开启的，不再是隐式行为。

这些调整在保留现有功能的前提下，让生命周期边界更加清晰，也让异步编排逻辑更容易理解。
</issue_to_address>

### Comment 5
<location> `astrbot/core/platform/sources/webchat/webchat_queue_mgr.py:13` </location>
<code_context>
-        """Conversation ID to asyncio.Queue mapping for responses"""
+        self.back_queues: dict[str, asyncio.Queue] = {}
+        """Request ID to asyncio.Queue mapping for responses"""
+        self._queue_close_events: dict[str, asyncio.Event] = {}
+        self._listener_tasks: dict[str, asyncio.Task] = {}
+        self._listener_callback: Callable[[tuple], Awaitable[None]] | None = None
</code_context>

<issue_to_address>
**问题（complexity）：** 建议通过移除显式的关闭事件、仅依赖任务取消和简单的队列读取循环来简化监听器生命周期管理。

你可以只依赖任务取消来简化监听编排逻辑，而不再维护 `_queue_close_events` 与 `_listener_tasks` 两套状态。这样可以去掉一整个状态维度，也不用在 `_listen_to_queue` 内部进行复杂的 `asyncio.wait` 控制，同时仍能保持原有行为。

具体建议：

- 完全移除 `_queue_close_events`。
- 在 `get_or_create_queue` 中，只负责创建队列并在有回调时启动监听器。
- 在 `remove_queue` 中，取消监听任务；此时 `queue.get()` 会被取消，循环可以在捕获 `CancelledError` 后退出。
- 将 `_listen_to_queue` 简化为一个直接 `while True: data = await queue.get()` 的循环。

示例修改（截取相关部分）：

```python
class WebChatQueueMgr:
    def __init__(self, queue_maxsize: int = 128, back_queue_maxsize: int = 512) -> None:
        self.queues: dict[str, asyncio.Queue] = {}
        self.back_queues: dict[str, asyncio.Queue] = {}
        self._listener_tasks: dict[str, asyncio.Task] = {}
        self._listener_callback: Callable[[tuple], Awaitable[None]] | None = None
        self.queue_maxsize = queue_maxsize
        self.back_queue_maxsize = back_queue_maxsize

    def get_or_create_queue(self, conversation_id: str) -> asyncio.Queue:
        if conversation_id not in self.queues:
            self.queues[conversation_id] = asyncio.Queue(maxsize=self.queue_maxsize)
            self._start_listener_if_needed(conversation_id)
        return self.queues[conversation_id]

    def remove_queue(self, conversation_id: str):
        self.queues.pop(conversation_id, None)

        task = self._listener_tasks.pop(conversation_id, None)
        if task is not None:
            task.cancel()
```

监听器启动基本保持不变，只是不再需要 close event：

```python
    def _start_listener_if_needed(self, conversation_id: str):
        if self._listener_callback is None:
            return
        task = self._listener_tasks.get(conversation_id)
        if task is not None and not task.done():
            return

        queue = self.queues.get(conversation_id)
        if queue is None:
            return

        task = asyncio.create_task(
            self._listen_to_queue(conversation_id, queue),
            name=f"webchat_listener_{conversation_id}",
        )
        self._listener_tasks[conversation_id] = task
        task.add_done_callback(lambda _: self._listener_tasks.pop(conversation_id, None))
        logger.debug(f"Started listener for conversation: {conversation_id}")
```

监听循环也会更直观：

```python
    async def _listen_to_queue(
        self,
        conversation_id: str,
        queue: asyncio.Queue,
    ):
        while True:
            try:
                data = await queue.get()
            except asyncio.CancelledError:
                break

            if self._listener_callback is None:
                continue

            try:
                await self._listener_callback(data)
            except Exception as e:
                logger.error(
                    f"Error processing message from conversation {conversation_id}: {e}"
                )
```

这样可以：

- 去掉 `_queue_close_events` 以及每轮循环中创建/取消 `get_task`/`close_task` 的逻辑；
- 让生命周期更清晰：`remove_queue` → 取消任务 → 监听器退出；
- 同时保留“每个会话一个监听器”和 back 队列的现有行为与 API。
</issue_to_address>

---

Sourcery 对开源项目免费 —— 如果你觉得这些评论有帮助，欢迎分享 ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{帮我变得更有用！请在每条评论上点击 👍 或 👎，我会根据你的反馈改进后续的审查建议。}

Original comment in English

Hey - I've found 5 issues, and left some high level feedback:

• In WecomAIQueueMgr, is_stream_finished and cleanup_expired_responses both manage completed_streams TTLs but use different hard-coded/default values (function arg vs. literal 60s); consider centralizing this retention period in one place to avoid subtle discrepancies in behavior over time.
• Both queue managers’ _listen_to_queue methods swallow listener callback exceptions with only a short log message; it may be helpful to include exc_info=True or the full traceback in these logs so that failures in the event-driven callbacks are easier to diagnose in production.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- In `WecomAIQueueMgr`, `is_stream_finished` and `cleanup_expired_responses` both manage `completed_streams` TTLs but use different hard-coded/default values (function arg vs. literal 60s); consider centralizing this retention period in one place to avoid subtle discrepancies in behavior over time.
- Both queue managers’ `_listen_to_queue` methods swallow listener callback exceptions with only a short log message; it may be helpful to include `exc_info=True` or the full traceback in these logs so that failures in the event-driven callbacks are easier to diagnose in production.

## Individual Comments

### Comment 1
<location> `astrbot/core/platform/sources/wecom_ai_bot/wecomai_queue_mgr.py:183-187` </location>
<code_context>
+            return False
+        return True
+
     def cleanup_expired_responses(self, max_age_seconds: int = 300):
         """清理过期的待处理响应

</code_context>

<issue_to_address>
**suggestion:** Align the TTL handling for `pending_responses` and `completed_streams` to avoid surprising lifetime differences.

`cleanup_expired_responses` uses `max_age_seconds` for `pending_responses`, but `completed_streams` is always pruned with a hard‑coded 60 seconds:

```python
expired_finished = [
    session_id
    for session_id, finished_at in self.completed_streams.items()
    if current_time - finished_at > 60
]
```
This means callers changing `max_age_seconds` affect only `pending_responses`, which is not obvious from the API. Consider tying `completed_streams` to `max_age_seconds` (or a derived value) or extracting the `60` into a named constant/parameter so TTL behavior is explicit and configurable.

```suggestion
        expired_finished = [
            session_id
            for session_id, finished_at in self.completed_streams.items()
            if current_time - finished_at > max_age_seconds
        ]
```
</issue_to_address>

### Comment 2
<location> `astrbot/core/platform/sources/webchat/webchat_queue_mgr.py:39-41` </location>
<code_context>
+        """Remove back queue for the given request ID"""
+        self.back_queues.pop(request_id, None)

     def remove_queues(self, conversation_id: str):
         """Remove queues for the given conversation ID"""
-        if conversation_id in self.queues:
-            del self.queues[conversation_id]
-        if conversation_id in self.back_queues:
-            del self.back_queues[conversation_id]
+        self.remove_queue(conversation_id)
+
+    def remove_queue(self, conversation_id: str):
</code_context>

<issue_to_address>
**question (bug_risk):** Double‑check whether back queues associated with a conversation need to be cleaned up to avoid leaks.

Previously `remove_queues(conversation_id)` removed both input and back queues keyed by that `conversation_id`. After the refactor, back queues are keyed by `request_id`, and `remove_queues` now only calls `remove_queue`, which only removes the input queue/listener.

If any existing callers still rely on `remove_queues(conversation_id)` to fully clean up conversation resources, back queues may now be left behind indefinitely. Please review call sites (or add a helper that removes all queues for a conversation, e.g., via naming or an explicit mapping) to ensure back queues are also cleaned up where needed.
</issue_to_address>

### Comment 3
<location> `astrbot/core/platform/sources/wecom_ai_bot/wecomai_queue_mgr.py:227-236` </location>
<code_context>
-        """Listen to a specific conversation queue"""
-        queue = self.webchat_queue_mgr.get_or_create_queue(conversation_id)
-        while True:
-            try:
-                data = await queue.get()
-                await self.callback(data)
</code_context>

<issue_to_address>
**suggestion:** Also log traceback details for WeCom listener callback failures.

Here, `_listen_to_queue` logs callback errors with only the exception message:

```python
logger.error(f"处理会话 {session_id} 消息时发生错误: {e}")
```
Since this runs in a background task, capturing the stack trace will make failures debuggable. Please switch to `logger.exception` or `logger.error(..., exc_info=True)` so the full traceback is logged.
</issue_to_address>

### Comment 4
<location> `astrbot/core/platform/sources/wecom_ai_bot/wecomai_queue_mgr.py:16` </location>
<code_context>
 class WebChatQueueMgr:
-    def __init__(self) -> None:
-        self.queues = {}
+    def __init__(self, queue_maxsize: int = 128, back_queue_maxsize: int = 512) -> None:
+        self.queues: dict[str, asyncio.Queue] = {}
         """Conversation ID to asyncio.Queue mapping"""
</code_context>

<issue_to_address>
**issue (complexity):** Consider refactoring the new listener and lifecycle management logic to separate concerns, centralize cleanup, and encapsulate the async wait/cancel pattern for better clarity and maintainability.

The added listener/lifecycle logic is quite dense in this class. You can keep the feature but reduce complexity by:

### 1. Separate lifecycle responsibilities

Right now `remove_queues`, `remove_queue`, `cleanup_expired_responses`, and `mark_finished` logic are mixed. Split “mark finished” from “cleanup” and unify cleanup in one place.

```python
def mark_stream_finished(self, session_id: str) -> None:
    """单独负责标记流结束"""
    self.completed_streams[session_id] = asyncio.get_event_loop().time()
    logger.debug(f"[WecomAI] 标记流已结束: {session_id}")

def _cleanup_session(self, session_id: str) -> None:
    """集中清理所有与 session 相关的状态"""
    # 输入队列 + 监听
    if session_id in self.queues:
        del self.queues[session_id]
        logger.debug(f"[WecomAI] 移除输入队列: {session_id}")
    close_event = self._queue_close_events.pop(session_id, None)
    if close_event is not None:
        close_event.set()
    task = self._listener_tasks.pop(session_id, None)
    if task is not None:
        task.cancel()

    # 输出队列
    if session_id in self.back_queues:
        del self.back_queues[session_id]
        logger.debug(f"[WecomAI] 移除输出队列: {session_id}")

    # 待处理响应
    if session_id in self.pending_responses:
        del self.pending_responses[session_id]
        logger.debug(f"[WecomAI] 移除待处理响应: {session_id}")
```

Then keep public APIs small and clear:

```python
def remove_queues(self, session_id: str, mark_finished: bool = False) -> None:
    self._cleanup_session(session_id)
    if mark_finished:
        self.mark_stream_finished(session_id)

def remove_queue(self, session_id: str) -> None:
    # 保留向后兼容，但委托给统一清理逻辑
    self._cleanup_session(session_id)
```

`cleanup_expired_responses` can call `_cleanup_session` instead of partially duplicating cleanup logic.

---

### 2. Encapsulate the wait/cancel pattern

`_listen_to_queue` currently inlines an `asyncio.wait` pattern with manual cancellation in multiple places. Extract that into a helper so the cancellation behavior is easy to audit and reuse:

```python
async def _wait_queue_or_close(
    self,
    queue: asyncio.Queue,
    close_event: asyncio.Event,
) -> dict | None:
    """返回队列数据, 或在关闭时返回 None"""
    get_task = asyncio.create_task(queue.get())
    close_task = asyncio.create_task(close_event.wait())
    try:
        done, pending = await asyncio.wait(
            {get_task, close_task},
            return_when=asyncio.FIRST_COMPLETED,
        )
        for task in pending:
            task.cancel()
        if close_task in done:
            return None
        return get_task.result()
    finally:
        if not get_task.done():
            get_task.cancel()
        if not close_task.done():
            close_task.cancel()
```

Then `_listen_to_queue` becomes much easier to follow:

```python
async def _listen_to_queue(
    self,
    session_id: str,
    queue: asyncio.Queue,
    close_event: asyncio.Event,
):
    while True:
        try:
            data = await self._wait_queue_or_close(queue, close_event)
            if data is None:  # close_event set
                break
            if self._listener_callback is None:
                continue
            try:
                await self._listener_callback(data)
            except Exception as e:
                logger.error(f"处理会话 {session_id} 消息时发生错误: {e}")
        except asyncio.CancelledError:
            break
```

This keeps the same behavior (queue-or-close, cancel pending tasks) but localizes the tricky `asyncio.wait` logic into a single helper.

---

### 3. Make listener responsibility more explicit

To reduce the “dual API” confusion (push vs pull), keep the manager surface small and explicit:

- Document that `set_listener` is optional and that, when set, the manager will start per-session listener tasks automatically.
- Consider requiring explicit opt-in per session, so the queue manager doesn’t always mix “storage + orchestration”. For example:

```python
def enable_listener_for_session(self, session_id: str) -> None:
    if self._listener_callback is None:
        raise RuntimeError("Listener callback must be set before enabling listener.")
    self._start_listener_if_needed(session_id)
```

Then `get_or_create_queue` only creates queues, and listener orchestration is clearly separated and opt-in instead of implicit.

These changes keep all current functionality but make lifecycle boundaries clearer and the async orchestration logic easier to reason about.
</issue_to_address>

### Comment 5
<location> `astrbot/core/platform/sources/webchat/webchat_queue_mgr.py:13` </location>
<code_context>
-        """Conversation ID to asyncio.Queue mapping for responses"""
+        self.back_queues: dict[str, asyncio.Queue] = {}
+        """Request ID to asyncio.Queue mapping for responses"""
+        self._queue_close_events: dict[str, asyncio.Event] = {}
+        self._listener_tasks: dict[str, asyncio.Task] = {}
+        self._listener_callback: Callable[[tuple], Awaitable[None]] | None = None
</code_context>

<issue_to_address>
**issue (complexity):** Consider simplifying the listener lifecycle by removing the explicit close events and relying solely on task cancellation with a straightforward queue-read loop.

You can simplify the listener orchestration by relying solely on task cancellation instead of maintaining both `_queue_close_events` and `_listener_tasks`. That removes an entire axis of state and the `asyncio.wait` juggling inside `_listen_to_queue`, while preserving behavior.

Concretely:

- Drop `_queue_close_events` entirely.
- In `get_or_create_queue`, just create the queue and start the listener if a callback is set.
- In `remove_queue`, cancel the listener task; `queue.get()` will be cancelled and the loop can exit on `CancelledError`.
- Simplify `_listen_to_queue` to a straightforward `while True: data = await queue.get()` loop.

Example changes (trimmed to the relevant parts):

```python
class WebChatQueueMgr:
    def __init__(self, queue_maxsize: int = 128, back_queue_maxsize: int = 512) -> None:
        self.queues: dict[str, asyncio.Queue] = {}
        self.back_queues: dict[str, asyncio.Queue] = {}
        self._listener_tasks: dict[str, asyncio.Task] = {}
        self._listener_callback: Callable[[tuple], Awaitable[None]] | None = None
        self.queue_maxsize = queue_maxsize
        self.back_queue_maxsize = back_queue_maxsize

    def get_or_create_queue(self, conversation_id: str) -> asyncio.Queue:
        if conversation_id not in self.queues:
            self.queues[conversation_id] = asyncio.Queue(maxsize=self.queue_maxsize)
            self._start_listener_if_needed(conversation_id)
        return self.queues[conversation_id]

    def remove_queue(self, conversation_id: str):
        self.queues.pop(conversation_id, None)

        task = self._listener_tasks.pop(conversation_id, None)
        if task is not None:
            task.cancel()
```

Listener startup stays almost the same, but no close event:

```python
    def _start_listener_if_needed(self, conversation_id: str):
        if self._listener_callback is None:
            return
        task = self._listener_tasks.get(conversation_id)
        if task is not None and not task.done():
            return

        queue = self.queues.get(conversation_id)
        if queue is None:
            return

        task = asyncio.create_task(
            self._listen_to_queue(conversation_id, queue),
            name=f"webchat_listener_{conversation_id}",
        )
        self._listener_tasks[conversation_id] = task
        task.add_done_callback(lambda _: self._listener_tasks.pop(conversation_id, None))
        logger.debug(f"Started listener for conversation: {conversation_id}")
```

The listener loop becomes much easier to reason about:

```python
    async def _listen_to_queue(
        self,
        conversation_id: str,
        queue: asyncio.Queue,
    ):
        while True:
            try:
                data = await queue.get()
            except asyncio.CancelledError:
                break

            if self._listener_callback is None:
                continue

            try:
                await self._listener_callback(data)
            except Exception as e:
                logger.error(
                    f"Error processing message from conversation {conversation_id}: {e}"
                )
```

This:

- Removes `_queue_close_events` and the per-iteration creation/cancellation of `get_task`/`close_task`.
- Makes the lifecycle clearer: `remove_queue` → cancel task → listener exits.
- Keeps the listener-per-conversation behavior and back-queue API intact.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

建议（suggestion）： 对齐 pending_responses 与 completed_streams 的 TTL 处理方式，以避免它们的存活时间出现出人意料的差异。

cleanup_expired_responses 使用 max_age_seconds 来处理 pending_responses，但 completed_streams 始终使用硬编码的 60 秒进行清理：

expired_finished = [
    session_id
    for session_id, finished_at in self.completed_streams.items()
    if current_time - finished_at > 60
]

这意味着调用方修改 max_age_seconds 时只会影响 pending_responses，从 API 上并不直观。建议将 completed_streams 的 TTL 绑定到 max_age_seconds（或由其推导出的值），或者把 60 提取为具名常量/参数，使 TTL 行为更加显式且可配置。

Suggested change

	expired_finished = [
	session_id
	for session_id, finished_at in self.completed_streams.items()
	if current_time - finished_at > 60
	]
	expired_finished = [
	session_id
	for session_id, finished_at in self.completed_streams.items()
	if current_time - finished_at > max_age_seconds
	]

Original comment in English

suggestion: Align the TTL handling for pending_responses and completed_streams to avoid surprising lifetime differences.

cleanup_expired_responses uses max_age_seconds for pending_responses, but completed_streams is always pruned with a hard‑coded 60 seconds:

expired_finished = [
    session_id
    for session_id, finished_at in self.completed_streams.items()
    if current_time - finished_at > 60
]

This means callers changing max_age_seconds affect only pending_responses, which is not obvious from the API. Consider tying completed_streams to max_age_seconds (or a derived value) or extracting the 60 into a named constant/parameter so TTL behavior is explicit and configurable.

Suggested change

	expired_finished = [
	session_id
	for session_id, finished_at in self.completed_streams.items()
	if current_time - finished_at > 60
	]
	expired_finished = [
	session_id
	for session_id, finished_at in self.completed_streams.items()
	if current_time - finished_at > max_age_seconds
	]

[sourcery-ai]

建议（suggestion）： 对 WeCom 监听器回调中的异常也记录 traceback 详情。

当前 _listen_to_queue 里，在回调报错时只记录了异常消息：

logger.error(f"处理会话 {session_id} 消息时发生错误: {e}")

由于这是在后台任务中运行，捕获完整的堆栈信息会让问题更易于调试。建议改为使用 logger.exception，或者 logger.error(..., exc_info=True)，以便将完整 traceback 打到日志中。

Original comment in English

suggestion: Also log traceback details for WeCom listener callback failures.

Here, _listen_to_queue logs callback errors with only the exception message:

logger.error(f"处理会话 {session_id} 消息时发生错误: {e}")

Since this runs in a background task, capturing the stack trace will make failures debuggable. Please switch to logger.exception or logger.error(..., exc_info=True) so the full traceback is logged.

[sourcery-ai]

问题（complexity）： 建议重构新增的监听器与生命周期管理逻辑，以拆分职责、集中清理逻辑，并封装异步等待/取消模式，从而提升可读性与可维护性。

当前在该类中添加的监听器/生命周期逻辑比较密集。可以在保持功能不变的前提下，通过以下方式降低复杂度：

1. 分离生命周期职责

目前 remove_queues、remove_queue、cleanup_expired_responses 和 mark_finished 的逻辑交织在一起。建议将“标记完成”和“清理”分离，并把清理逻辑统一到一个地方。

def mark_stream_finished(self, session_id: str) -> None:
    """单独负责标记流结束"""
    self.completed_streams[session_id] = asyncio.get_event_loop().time()
    logger.debug(f"[WecomAI] 标记流已结束: {session_id}")

def _cleanup_session(self, session_id: str) -> None:
    """集中清理所有与 session 相关的状态"""
    # 输入队列 + 监听
    if session_id in self.queues:
        del self.queues[session_id]
        logger.debug(f"[WecomAI] 移除输入队列: {session_id}")
    close_event = self._queue_close_events.pop(session_id, None)
    if close_event is not None:
        close_event.set()
    task = self._listener_tasks.pop(session_id, None)
    if task is not None:
        task.cancel()

    # 输出队列
    if session_id in self.back_queues:
        del self.back_queues[session_id]
        logger.debug(f"[WecomAI] 移除输出队列: {session_id}")

    # 待处理响应
    if session_id in self.pending_responses:
        del self.pending_responses[session_id]
        logger.debug(f"[WecomAI] 移除待处理响应: {session_id}")

然后让对外的 API 更小、更清晰：

def remove_queues(self, session_id: str, mark_finished: bool = False) -> None:
    self._cleanup_session(session_id)
    if mark_finished:
        self.mark_stream_finished(session_id)

def remove_queue(self, session_id: str) -> None:
    # 保留向后兼容，但委托给统一清理逻辑
    self._cleanup_session(session_id)

cleanup_expired_responses 可以调用 _cleanup_session，避免部分清理逻辑重复。

---

2. 封装 wait/cancel 模式

_listen_to_queue 目前在多个地方内联了一个带手动取消的 asyncio.wait 模式。建议提取到一个辅助方法中，这样取消行为更容易审计和复用：

async def _wait_queue_or_close(
    self,
    queue: asyncio.Queue,
    close_event: asyncio.Event,
) -> dict | None:
    """返回队列数据, 或在关闭时返回 None"""
    get_task = asyncio.create_task(queue.get())
    close_task = asyncio.create_task(close_event.wait())
    try:
        done, pending = await asyncio.wait(
            {get_task, close_task},
            return_when=asyncio.FIRST_COMPLETED,
        )
        for task in pending:
            task.cancel()
        if close_task in done:
            return None
        return get_task.result()
    finally:
        if not get_task.done():
            get_task.cancel()
        if not close_task.done():
            close_task.cancel()

这样 _listen_to_queue 会清晰很多：

async def _listen_to_queue(
    self,
    session_id: str,
    queue: asyncio.Queue,
    close_event: asyncio.Event,
):
    while True:
        try:
            data = await self._wait_queue_or_close(queue, close_event)
            if data is None:  # close_event set
                break
            if self._listener_callback is None:
                continue
            try:
                await self._listener_callback(data)
            except Exception as e:
                logger.error(f"处理会话 {session_id} 消息时发生错误: {e}")
        except asyncio.CancelledError:
            break

这样既保留了原有的行为（队列或关闭二选一、取消未完成任务），又把复杂的 asyncio.wait 逻辑集中到一个地方。

---

3. 让监听器职责更加明确

为减少“推模式 vs 拉模式”的双重 API 混淆，可以让管理器的对外接口更小、更明确：

• 在文档中说明 set_listener 是可选的；当设置后，管理器会自动为每个 session 启动监听任务。
• 可以考虑为每个 session 显式开启监听，而不是默认在队列创建时就绑定监听任务，这样可以减弱“存储 + 编排”混在一起的感觉。例如：

def enable_listener_for_session(self, session_id: str) -> None:
    if self._listener_callback is None:
        raise RuntimeError("Listener callback must be set before enabling listener.")
    self._start_listener_if_needed(session_id)

这样 get_or_create_queue 只负责创建队列，而监听编排是显式的、按需开启的，不再是隐式行为。

这些调整在保留现有功能的前提下，让生命周期边界更加清晰，也让异步编排逻辑更容易理解。

Original comment in English

issue (complexity): Consider refactoring the new listener and lifecycle management logic to separate concerns, centralize cleanup, and encapsulate the async wait/cancel pattern for better clarity and maintainability.

The added listener/lifecycle logic is quite dense in this class. You can keep the feature but reduce complexity by:

1. Separate lifecycle responsibilities

Right now remove_queues, remove_queue, cleanup_expired_responses, and mark_finished logic are mixed. Split “mark finished” from “cleanup” and unify cleanup in one place.

def mark_stream_finished(self, session_id: str) -> None:
    """单独负责标记流结束"""
    self.completed_streams[session_id] = asyncio.get_event_loop().time()
    logger.debug(f"[WecomAI] 标记流已结束: {session_id}")

def _cleanup_session(self, session_id: str) -> None:
    """集中清理所有与 session 相关的状态"""
    # 输入队列 + 监听
    if session_id in self.queues:
        del self.queues[session_id]
        logger.debug(f"[WecomAI] 移除输入队列: {session_id}")
    close_event = self._queue_close_events.pop(session_id, None)
    if close_event is not None:
        close_event.set()
    task = self._listener_tasks.pop(session_id, None)
    if task is not None:
        task.cancel()

    # 输出队列
    if session_id in self.back_queues:
        del self.back_queues[session_id]
        logger.debug(f"[WecomAI] 移除输出队列: {session_id}")

    # 待处理响应
    if session_id in self.pending_responses:
        del self.pending_responses[session_id]
        logger.debug(f"[WecomAI] 移除待处理响应: {session_id}")

Then keep public APIs small and clear:

def remove_queues(self, session_id: str, mark_finished: bool = False) -> None:
    self._cleanup_session(session_id)
    if mark_finished:
        self.mark_stream_finished(session_id)

def remove_queue(self, session_id: str) -> None:
    # 保留向后兼容，但委托给统一清理逻辑
    self._cleanup_session(session_id)

cleanup_expired_responses can call _cleanup_session instead of partially duplicating cleanup logic.

---

2. Encapsulate the wait/cancel pattern

_listen_to_queue currently inlines an asyncio.wait pattern with manual cancellation in multiple places. Extract that into a helper so the cancellation behavior is easy to audit and reuse:

async def _wait_queue_or_close(
    self,
    queue: asyncio.Queue,
    close_event: asyncio.Event,
) -> dict | None:
    """返回队列数据, 或在关闭时返回 None"""
    get_task = asyncio.create_task(queue.get())
    close_task = asyncio.create_task(close_event.wait())
    try:
        done, pending = await asyncio.wait(
            {get_task, close_task},
            return_when=asyncio.FIRST_COMPLETED,
        )
        for task in pending:
            task.cancel()
        if close_task in done:
            return None
        return get_task.result()
    finally:
        if not get_task.done():
            get_task.cancel()
        if not close_task.done():
            close_task.cancel()

Then _listen_to_queue becomes much easier to follow:

async def _listen_to_queue(
    self,
    session_id: str,
    queue: asyncio.Queue,
    close_event: asyncio.Event,
):
    while True:
        try:
            data = await self._wait_queue_or_close(queue, close_event)
            if data is None:  # close_event set
                break
            if self._listener_callback is None:
                continue
            try:
                await self._listener_callback(data)
            except Exception as e:
                logger.error(f"处理会话 {session_id} 消息时发生错误: {e}")
        except asyncio.CancelledError:
            break

This keeps the same behavior (queue-or-close, cancel pending tasks) but localizes the tricky asyncio.wait logic into a single helper.

---

3. Make listener responsibility more explicit

To reduce the “dual API” confusion (push vs pull), keep the manager surface small and explicit:

• Document that set_listener is optional and that, when set, the manager will start per-session listener tasks automatically.
• Consider requiring explicit opt-in per session, so the queue manager doesn’t always mix “storage + orchestration”. For example:

def enable_listener_for_session(self, session_id: str) -> None:
    if self._listener_callback is None:
        raise RuntimeError("Listener callback must be set before enabling listener.")
    self._start_listener_if_needed(session_id)

Then get_or_create_queue only creates queues, and listener orchestration is clearly separated and opt-in instead of implicit.

These changes keep all current functionality but make lifecycle boundaries clearer and the async orchestration logic easier to reason about.

[sourcery-ai]

问题（complexity）： 建议通过移除显式的关闭事件、仅依赖任务取消和简单的队列读取循环来简化监听器生命周期管理。

你可以只依赖任务取消来简化监听编排逻辑，而不再维护 _queue_close_events 与 _listener_tasks 两套状态。这样可以去掉一整个状态维度，也不用在 _listen_to_queue 内部进行复杂的 asyncio.wait 控制，同时仍能保持原有行为。

具体建议：

• 完全移除 _queue_close_events。
• 在 get_or_create_queue 中，只负责创建队列并在有回调时启动监听器。
• 在 remove_queue 中，取消监听任务；此时 queue.get() 会被取消，循环可以在捕获 CancelledError 后退出。
• 将 _listen_to_queue 简化为一个直接 while True: data = await queue.get() 的循环。

示例修改（截取相关部分）：

class WebChatQueueMgr:
    def __init__(self, queue_maxsize: int = 128, back_queue_maxsize: int = 512) -> None:
        self.queues: dict[str, asyncio.Queue] = {}
        self.back_queues: dict[str, asyncio.Queue] = {}
        self._listener_tasks: dict[str, asyncio.Task] = {}
        self._listener_callback: Callable[[tuple], Awaitable[None]] | None = None
        self.queue_maxsize = queue_maxsize
        self.back_queue_maxsize = back_queue_maxsize

    def get_or_create_queue(self, conversation_id: str) -> asyncio.Queue:
        if conversation_id not in self.queues:
            self.queues[conversation_id] = asyncio.Queue(maxsize=self.queue_maxsize)
            self._start_listener_if_needed(conversation_id)
        return self.queues[conversation_id]

    def remove_queue(self, conversation_id: str):
        self.queues.pop(conversation_id, None)

        task = self._listener_tasks.pop(conversation_id, None)
        if task is not None:
            task.cancel()

监听器启动基本保持不变，只是不再需要 close event：

    def _start_listener_if_needed(self, conversation_id: str):
        if self._listener_callback is None:
            return
        task = self._listener_tasks.get(conversation_id)
        if task is not None and not task.done():
            return

        queue = self.queues.get(conversation_id)
        if queue is None:
            return

        task = asyncio.create_task(
            self._listen_to_queue(conversation_id, queue),
            name=f"webchat_listener_{conversation_id}",
        )
        self._listener_tasks[conversation_id] = task
        task.add_done_callback(lambda _: self._listener_tasks.pop(conversation_id, None))
        logger.debug(f"Started listener for conversation: {conversation_id}")

监听循环也会更直观：

    async def _listen_to_queue(
        self,
        conversation_id: str,
        queue: asyncio.Queue,
    ):
        while True:
            try:
                data = await queue.get()
            except asyncio.CancelledError:
                break

            if self._listener_callback is None:
                continue

            try:
                await self._listener_callback(data)
            except Exception as e:
                logger.error(
                    f"Error processing message from conversation {conversation_id}: {e}"
                )

这样可以：

• 去掉 _queue_close_events 以及每轮循环中创建/取消 get_task/close_task 的逻辑；
• 让生命周期更清晰：remove_queue → 取消任务 → 监听器退出；
• 同时保留“每个会话一个监听器”和 back 队列的现有行为与 API。

Original comment in English

issue (complexity): Consider simplifying the listener lifecycle by removing the explicit close events and relying solely on task cancellation with a straightforward queue-read loop.

You can simplify the listener orchestration by relying solely on task cancellation instead of maintaining both _queue_close_events and _listener_tasks. That removes an entire axis of state and the asyncio.wait juggling inside _listen_to_queue, while preserving behavior.

Concretely:

• Drop _queue_close_events entirely.
• In get_or_create_queue, just create the queue and start the listener if a callback is set.
• In remove_queue, cancel the listener task; queue.get() will be cancelled and the loop can exit on CancelledError.
• Simplify _listen_to_queue to a straightforward while True: data = await queue.get() loop.

Example changes (trimmed to the relevant parts):

class WebChatQueueMgr:
    def __init__(self, queue_maxsize: int = 128, back_queue_maxsize: int = 512) -> None:
        self.queues: dict[str, asyncio.Queue] = {}
        self.back_queues: dict[str, asyncio.Queue] = {}
        self._listener_tasks: dict[str, asyncio.Task] = {}
        self._listener_callback: Callable[[tuple], Awaitable[None]] | None = None
        self.queue_maxsize = queue_maxsize
        self.back_queue_maxsize = back_queue_maxsize

    def get_or_create_queue(self, conversation_id: str) -> asyncio.Queue:
        if conversation_id not in self.queues:
            self.queues[conversation_id] = asyncio.Queue(maxsize=self.queue_maxsize)
            self._start_listener_if_needed(conversation_id)
        return self.queues[conversation_id]

    def remove_queue(self, conversation_id: str):
        self.queues.pop(conversation_id, None)

        task = self._listener_tasks.pop(conversation_id, None)
        if task is not None:
            task.cancel()

Listener startup stays almost the same, but no close event:

    def _start_listener_if_needed(self, conversation_id: str):
        if self._listener_callback is None:
            return
        task = self._listener_tasks.get(conversation_id)
        if task is not None and not task.done():
            return

        queue = self.queues.get(conversation_id)
        if queue is None:
            return

        task = asyncio.create_task(
            self._listen_to_queue(conversation_id, queue),
            name=f"webchat_listener_{conversation_id}",
        )
        self._listener_tasks[conversation_id] = task
        task.add_done_callback(lambda _: self._listener_tasks.pop(conversation_id, None))
        logger.debug(f"Started listener for conversation: {conversation_id}")

The listener loop becomes much easier to reason about:

    async def _listen_to_queue(
        self,
        conversation_id: str,
        queue: asyncio.Queue,
    ):
        while True:
            try:
                data = await queue.get()
            except asyncio.CancelledError:
                break

            if self._listener_callback is None:
                continue

            try:
                await self._listener_callback(data)
            except Exception as e:
                logger.error(
                    f"Error processing message from conversation {conversation_id}: {e}"
                )

This:

• Removes _queue_close_events and the per-iteration creation/cancellation of get_task/close_task.
• Makes the lifecycle clearer: remove_queue → cancel task → listener exits.
• Keeps the listener-per-conversation behavior and back-queue API intact.