diff --git a/api/v1/endpoints/agent.py b/api/v1/endpoints/agent.py index 2265765bf..9e20d1fed 100644 --- a/api/v1/endpoints/agent.py +++ b/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(), diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 7423542fc..756924836 100644 --- a/docs/CHANGELOG.md +++ b/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 diff --git a/tests/test_agent_sse_cleanup.py b/tests/test_agent_sse_cleanup.py new file mode 100644 index 000000000..25eb4c03c --- /dev/null +++ b/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()