diff --git a/app/api/endpoints/agent.py b/app/api/endpoints/agent.py index 1fb0595f8..766cd9f25 100644 --- a/app/api/endpoints/agent.py +++ b/app/api/endpoints/agent.py @@ -2221,9 +2221,11 @@ async def _web_agent_stream_impl( return yield ": heartbeat\n\n" except asyncio.CancelledError: + return + finally: if not collection_task.done(): collection_task.cancel() - return + await asyncio.gather(collection_task, return_exceptions=True) assistant_message = _build_web_agent_display_message_from_events(events) display_messages.append(assistant_message) diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index c4cd42d47..d6664d3d8 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -89,6 +89,8 @@ 稳定 owner,并继续保留无事件循环时 `asyncio.run` 和原同步 ABI。 插件市场 Release 合并层的普通读取和强刷子任务也已登记稳定 owner;API 外层任务被关停取消时,内部 `shield` 不再让网络请求逃逸生命周期预算,仓库级并发合并、缓存键和 V1/V2/V3 返回兼容保持不变。 + 请求作用域的结构化并发不进入全局登记器:传统 WebAgent SSE 的 collection 子任务改由生成器 + `finally` 取消并等待清理,断线和 ASGI 取消均不会留下请求级 task。 2. **动态模块契约仍以 legacy 聚合语义为主。** 当前登记 `212` 个模块方法,其中 `194` 个仍使用 `legacy` aggregation,只有 `14` 个 `first_non_empty`、`4` 个 `ordered_list_merge`。`app/runtime/extensions/module/contracts.py:422-455` 已能登记 family、输入/结果标签和基础签名诊断,但 `193` 个方法没有 required parameters,调度器 `app/runtime/extensions/module/dispatcher.py:109-260` 仍主要依赖运行时反射、返回值形状和短路规则。未知第三方方法保留 legacy fallback 是兼容要求,不应删除;宿主高频能力则应逐族补齐可执行的输入校验、结果校验、超时和错误语义。 3. **Model/Base 的数据库装饰器和隐式会话 ABI 已全部清零。** 查询、写事务和 `legacy_*` 装饰器均为 `0`;所有 Model `db` 参数要求显式 Session,Base CRUD 仅在调用方事务内查询或 stage。可无会话构造的入口统一留在 Oper,经组合根事务执行器运行;插件 SDK 不再导出宿主 Model。后续重点转为减少 ORM 对象跨层流转,并保持 Model 隐式事务零回退。 diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index fbed1abb4..6606e46b1 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -312,7 +312,12 @@ def test_build_web_agent_command_items_returns_slash_commands(): def test_build_web_agent_command_items_includes_sites_command(): """WebAgent 命令建议应包含内建站点管理命令。""" - with patch("app.command.Scheduler"), patch("app.command.ThreadHelper"): + with patch( + "app.api.endpoints.agent.get_commands", + return_value={ + "/sites": {"description": "管理站点", "category": "站点"}, + }, + ): commands = _build_web_agent_command_items() assert any(command["command"] == "/sites" for command in commands) @@ -1651,6 +1656,64 @@ def test_web_agent_traditional_stream_keeps_alive_and_saves_after_done(): assert snapshot_finished.wait(timeout=1) +def test_web_agent_traditional_stream_drains_collection_on_cancellation(): + """传统 SSE 被取消时必须等待请求级 collection 子任务完成清理。""" + payload = schemas.AgentWebChatRequest(text="/状态", session_id="traditional-cancel") + request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False)) + user = SimpleNamespace(id=1, name="admin", is_superuser=True) + + async def scenario(): + """取消正在等待的 SSE 读取,并观察 collection 的清理时序。""" + started = asyncio.Event() + cancelling = asyncio.Event() + release_cleanup = asyncio.Event() + cleanup_finished = asyncio.Event() + + async def blocked_collect(**_kwargs): + """阻塞传统消息收集,并在取消后等待测试释放清理。""" + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelling.set() + await release_cleanup.wait() + cleanup_finished.set() + raise + + with patch( + "app.api.endpoints.agent._is_web_agent_traditional_message", + return_value=True, + ), patch( + "app.api.endpoints.agent._ensure_web_agent_command_allowed", + return_value=None, + ), patch( + "app.api.endpoints.agent._get_web_agent_unknown_command_message", + return_value=None, + ), patch( + "app.api.endpoints.agent._build_web_agent_session_id_async", + return_value="web-agent:traditional-cancel", + ), patch( + "app.api.endpoints.agent._collect_web_agent_traditional_events", + side_effect=blocked_collect, + ): + response = await web_agent_stream(payload, request, user) + iterator = response.body_iterator.__aiter__() + await asyncio.wait_for(anext(iterator), timeout=1) + pending_chunk = asyncio.create_task(anext(iterator)) + await asyncio.wait_for(started.wait(), timeout=1) + pending_chunk.cancel() + await asyncio.wait_for(cancelling.wait(), timeout=1) + assert pending_chunk.done() is False + assert cleanup_finished.is_set() is False + + release_cleanup.set() + result = await asyncio.gather(pending_chunk, return_exceptions=True) + assert isinstance(result[0], StopAsyncIteration) + assert cleanup_finished.is_set() is True + + asyncio.run(scenario()) + + def test_web_agent_stream_sends_done_before_snapshot_persistence_finishes(): """展示快照落库缓慢时,前端终态不应被数据库操作阻塞。""" payload = schemas.AgentWebChatRequest(text="检查系统", session_id="browser-snapshot")