refactor: drain traditional agent streams

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)

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 隐式事务零回退。
 

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")