fix: avoid webchat stream result crash on queue errors (#6123)

Co-authored-by: stablegenius49 <185121704+stablegenius49@users.noreply.github.com>

astrbot/dashboard/routes/chat.py

@@ -36,6 +36,20 @@ async def track_conversation(convs: dict, conv_id: str):
         convs.pop(conv_id, None)
 
 
+async def _poll_webchat_stream_result(back_queue, username: str):
+    try:
+        result = await asyncio.wait_for(back_queue.get(), timeout=1)
+    except asyncio.TimeoutError:
+        return None, False
+    except asyncio.CancelledError:
+        logger.debug(f"[WebChat] 用户 {username} 断开聊天长连接。")
+        return None, True
+    except Exception as e:
+        logger.error(f"WebChat stream error: {e}")
+        return None, False
+    return result, False
+
+
 class ChatRoute(Route):
     def __init__(
         self,
@@ -343,16 +357,12 @@ class ChatRoute(Route):
 
                 async with track_conversation(self.running_convs, webchat_conv_id):
                     while True:
-                        try:
+                        result, should_break = await _poll_webchat_stream_result(
-                            result = await asyncio.wait_for(back_queue.get(), timeout=1)
+                            back_queue, username
-                        except asyncio.TimeoutError:
+                        )
-                            continue
+                        if should_break:
-                        except asyncio.CancelledError:
-                            logger.debug(f"[WebChat] 用户 {username} 断开聊天长连接。")
                             client_disconnected = True
-                        except Exception as e:
+                            break
-                            logger.error(f"WebChat stream error: {e}")
-
                         if not result:
                             continue
 

tests/test_chat_route.py

@@ -0,0 +1,56 @@
+import asyncio
+
+import pytest
+
+from astrbot.dashboard.routes.chat import _poll_webchat_stream_result
+
+
+class _QueueThatRaises:
+    def __init__(self, exc: BaseException):
+        self._exc = exc
+
+    async def get(self):
+        raise self._exc
+
+
+class _QueueWithResult:
+    def __init__(self, result):
+        self._result = result
+
+    async def get(self):
+        return self._result
+
+
+@pytest.mark.asyncio
+async def test_poll_webchat_stream_result_breaks_on_cancelled_error():
+    result, should_break = await _poll_webchat_stream_result(
+        _QueueThatRaises(asyncio.CancelledError()),
+        "alice",
+    )
+
+    assert result is None
+    assert should_break is True
+
+
+@pytest.mark.asyncio
+async def test_poll_webchat_stream_result_continues_on_generic_exception():
+    result, should_break = await _poll_webchat_stream_result(
+        _QueueThatRaises(RuntimeError("boom")),
+        "alice",
+    )
+
+    assert result is None
+    assert should_break is False
+
+
+@pytest.mark.asyncio
+async def test_poll_webchat_stream_result_returns_queue_payload():
+    payload = {"type": "end", "data": ""}
+
+    result, should_break = await _poll_webchat_stream_result(
+        _QueueWithResult(payload),
+        "alice",
+    )
+
+    assert result == payload
+    assert should_break is False