fix: 在 agent_chat_stream 的 finally 清理块中区分… (#969) (#973)

* fix(issue-969): [bug]-agent-sse-流清理阶段静默吞掉后台执行器异常，错误无法感知

api/v1/endpoints/agent.py

@@ -447,8 +447,13 @@ async def agent_chat_stream(request: ChatRequest):
         finally:
             try:
                 await asyncio.wait_for(fut, timeout=5.0)
-            except Exception:
+            except asyncio.CancelledError:
                 pass
+            except asyncio.TimeoutError:
+                # Cleanup taking longer than 5s is treated as an expected timeout; no warning.
+                logger.debug("agent executor cleanup timed out after 5s for session %s", session_id)
+            except Exception as exc:
+                logger.warning("agent executor cleanup error (ignored): %s", exc, exc_info=True)
 
     return StreamingResponse(
         event_generator(),

docs/CHANGELOG.md

@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 - [新功能] 集成 Longbridge OpenAPI 作为美股/港股可选数据源；配置 `LONGBRIDGE_*` 后优先使用长桥获取日线与实时行情，YFinance / AkShare 兜底；未配置时行为与此前一致。长桥联调请使用 `tests/longbridge_live_smoke.py`（手动脚本，不参与 pytest 收集）。
 - [文档] 澄清 README（中/英/繁）中长桥「首选 / 兜底 / 未配置不调用」的边界；`docs/README_EN.md` / `docs/README_CHT.md` 顶部导航与完整指南链接改为 `./` 相对路径，避免在文档子目录下解析错误；`LONGBRIDGE_PRINT_QUOTE_PACKAGES` 与代码及 `.env.example` 对齐为未设置时默认关闭。
+- [修复] Agent SSE 流清理阶段静默吞掉后台执行器异常 — 流结束时后台任务异常现在正确记录并上报，避免错误无法感知（fixes #969）
 
 ## [3.12.0] - 2026-04-01
 

tests/test_agent_sse_cleanup.py

@@ -0,0 +1,92 @@
+# -*- coding: utf-8 -*-
+"""
+Tests for agent_chat_stream SSE cleanup exception handling.
+
+Verifies that:
+- asyncio.CancelledError during cleanup is silently ignored (no warning).
+- Other exceptions during cleanup emit a WARNING log entry.
+"""
+
+import asyncio
+import sys
+import os
+import unittest
+from unittest.mock import MagicMock
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from tests.litellm_stub import ensure_litellm_stub
+
+# Stub optional heavy deps before importing agent endpoint, without overriding a real install
+ensure_litellm_stub()
+
+
+class TestAgentSSECleanup(unittest.IsolatedAsyncioTestCase):
+    """Test the finally-block exception handling in event_generator."""
+
+    async def _run_cleanup(self, fut_exception):
+        """
+        Simulate the finally block in event_generator:
+          - fut is a Future that raises *fut_exception* when awaited with wait_for.
+        """
+        loop = asyncio.get_event_loop()
+        fut = loop.create_future()
+        if isinstance(fut_exception, BaseException):
+            fut.set_exception(fut_exception)
+        else:
+            fut.set_result(None)
+
+        import api.v1.endpoints.agent as agent_mod
+
+        # Replicate the finally block logic directly
+        try:
+            await asyncio.wait_for(fut, timeout=5.0)
+        except asyncio.CancelledError:
+            pass
+        except asyncio.TimeoutError:
+            agent_mod.logger.debug(
+                "agent executor cleanup timed out after 5s for session %s", "test-session"
+            )
+        except Exception as exc:
+            agent_mod.logger.warning(
+                "agent executor cleanup error (ignored): %s", exc, exc_info=True
+            )
+
+    async def test_cancelled_error_is_silent(self):
+        """CancelledError must NOT produce a warning log."""
+        import api.v1.endpoints.agent as agent_mod
+
+        with self.assertLogs(agent_mod.logger, level="WARNING") as cm:
+            # We need at least one log message for assertLogs to succeed;
+            # emit a sentinel so the context manager doesn't fail on zero messages.
+            agent_mod.logger.warning("sentinel")
+            await self._run_cleanup(asyncio.CancelledError())
+
+        # Only the sentinel should be present; no cleanup warning.
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("sentinel", cm.output[0])
+
+    async def test_runtime_error_emits_warning(self):
+        """Non-CancelledError exceptions must emit a WARNING log."""
+        import api.v1.endpoints.agent as agent_mod
+
+        with self.assertLogs(agent_mod.logger, level="WARNING") as cm:
+            await self._run_cleanup(RuntimeError("simulated executor crash"))
+
+        self.assertTrue(
+            any("cleanup error" in msg for msg in cm.output),
+            f"Expected 'cleanup error' in log output, got: {cm.output}",
+        )
+
+    async def test_value_error_emits_warning(self):
+        """ValueError also triggers a WARNING log."""
+        import api.v1.endpoints.agent as agent_mod
+
+        with self.assertLogs(agent_mod.logger, level="WARNING") as cm:
+            await self._run_cleanup(ValueError("bad value"))
+
+        self.assertTrue(any("cleanup error" in msg for msg in cm.output))
+
+
+if __name__ == "__main__":
+    unittest.main()