From 7afabeda0280991060c3d1f0e5d367cee64bb4e0 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 23 Aug 2026 15:13:05 +0800 Subject: [PATCH] refactor: register protocol stream tasks --- app/api/context.py | 8 ++ app/api/endpoints/anthropic.py | 15 ++- app/api/endpoints/openai.py | 23 ++++- .../backend-architecture-next-stage.md | 5 +- .../architecture/dependency-baseline.json | 10 +- tests/test_api_background_task_registry.py | 98 ++++++++++++++++++- 6 files changed, 150 insertions(+), 9 deletions(-) diff --git a/app/api/context.py b/app/api/context.py index d700adf09..247d2cf1b 100644 --- a/app/api/context.py +++ b/app/api/context.py @@ -40,6 +40,14 @@ def get_background_task_registry( return runtime.tasks +def get_background_task_registry_compat(request: Request) -> TaskRegistry: + """返回协议兼容端点使用的任务登记器，允许未启动 lifespan 的旧调用回退。""" + runtime = getattr(request.app.state, "host_runtime", None) + if isinstance(runtime, HostRuntime): + return runtime.tasks + return get_task_registry() + + def resolve_background_task_registry(value: object) -> TaskRegistry: """兼容直接调用 endpoint 的旧入口，并优先使用注入的任务登记器。""" if isinstance(value, TaskRegistry): diff --git a/app/api/endpoints/anthropic.py b/app/api/endpoints/anthropic.py index 4335d2421..f7faec133 100644 --- a/app/api/endpoints/anthropic.py +++ b/app/api/endpoints/anthropic.py @@ -2,7 +2,7 @@ import asyncio import uuid from typing import AsyncIterator, List, Optional -from fastapi import APIRouter, Header, Security +from fastapi import APIRouter, Depends, Header, Security from fastapi.responses import JSONResponse from app.schemas.openai import AnthropicErrorDetail as _SchemaAnthropicErrorDetail @@ -25,6 +25,11 @@ from app.api.presentation.sse import build_sse_response, encode_named_event from app.agent.runtime_loader import get_running_agent_manager from app.application.configuration import get_api_runtime_config_snapshot from app.adapters.web.security.access import anthropic_api_key_header +from app.api.context import ( + get_background_task_registry_compat, + resolve_background_task_registry, +) +from app.runtime.tasks import TaskRegistry ANTHROPIC_ERROR_RESPONSES = { 400: {"model": _SchemaAnthropicErrorResponse, "description": "请求格式错误"}, @@ -88,6 +93,7 @@ async def _stream_anthropic_response( user_id: str, prompt: str, images: List[str], + task_registry: TaskRegistry | None = None, ) -> AsyncIterator[str]: event_queue: asyncio.Queue = asyncio.Queue() @@ -113,7 +119,10 @@ async def _stream_anthropic_response( finally: await event_queue.put(None) - task = asyncio.create_task(_run_agent()) + task = resolve_background_task_registry(task_registry).create( + _run_agent(), + owner="api.anthropic.stream", + ) try: yield encode_named_event( "message_start", @@ -207,6 +216,7 @@ async def messages( payload: _SchemaAnthropicMessagesRequest, x_api_key: Optional[str] = Security(anthropic_api_key_header), anthropic_version: Optional[str] = Header(default=None, alias="anthropic-version"), + task_registry: TaskRegistry = Depends(get_background_task_registry_compat), ): auth_error = _check_auth(x_api_key) if auth_error: @@ -242,6 +252,7 @@ async def messages( user_id=session_id, prompt=prompt, images=images, + task_registry=task_registry, ), ) diff --git a/app/api/endpoints/openai.py b/app/api/endpoints/openai.py index 2a65b0a3e..66ae296e8 100644 --- a/app/api/endpoints/openai.py +++ b/app/api/endpoints/openai.py @@ -4,7 +4,7 @@ import uuid from threading import Lock from typing import AsyncIterator, List, Optional, Tuple -from fastapi import APIRouter, Request, Security +from fastapi import APIRouter, Depends, Request, Security from fastapi.responses import JSONResponse from fastapi.security import HTTPAuthorizationCredentials @@ -34,6 +34,11 @@ from app.agent.contracts import ReplyMode from app.application.configuration import get_api_runtime_config_snapshot from app.adapters.web.security.access import openai_bearer_scheme from app.schemas.types import NotificationChannel +from app.api.context import ( + get_background_task_registry_compat, + resolve_background_task_registry, +) +from app.runtime.tasks import TaskRegistry OPENAI_ERROR_RESPONSES = { 400: {"model": _SchemaOpenAIErrorResponse, "description": "请求格式错误"}, @@ -228,6 +233,7 @@ async def _stream_response( prompt: str, images: List[str], cleanup_session: bool, + task_registry: TaskRegistry | None = None, ) -> AsyncIterator[str]: event_queue: asyncio.Queue = asyncio.Queue() @@ -255,7 +261,10 @@ async def _stream_response( finally: await event_queue.put(None) - task = asyncio.create_task(_run_agent()) + task = resolve_background_task_registry(task_registry).create( + _run_agent(), + owner="api.openai.stream", + ) try: yield _sse_payload( @@ -488,6 +497,7 @@ async def _chat_completions_impl( credentials: Optional[HTTPAuthorizationCredentials] = Security( openai_bearer_scheme ), + task_registry: TaskRegistry | None = None, ): auth_error = _check_auth(credentials) if auth_error: @@ -545,6 +555,7 @@ async def _chat_completions_impl( prompt=prompt, images=images, cleanup_session=not use_server_session, + task_registry=task_registry, ), ) @@ -684,9 +695,15 @@ async def chat_completions( payload: _SchemaOpenAIChatCompletionsRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Security(openai_bearer_scheme), + task_registry: TaskRegistry = Depends(get_background_task_registry_compat), ): """OpenAI Chat Completions 兼容公开入口。""" - return await _chat_completions_impl(payload, request, credentials) + return await _chat_completions_impl( + payload, + request, + credentials, + task_registry=task_registry, + ) @router.post( diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 1dc1ab82c..fef49ebcc 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -26,7 +26,7 @@ ### P1：需要优先治理的真实债务 -1. **后台任务的统一所有权已覆盖 API 入口，但仍有更深层任务机制待分级。** `app/runtime/tasks.py` 已建立 lifespan 级 TaskRegistry，启动收尾、插件 Release 刷新、Webhook E0 广播、CookieCloud E1 手工调度、消息入口、Seerr 订阅、整理历史 AI 重做和 WebAgent 断线后执行/快照保存均不再维护端点模块级任务集合或 Starlette 回调，shutdown 会停止接收、取消并有限等待，且生命周期清单明确登记其顺序。主仓 `app/` 已无裸 FastAPI `BackgroundTasks`；当前仍有约 `50` 个更底层 `create_task`/等价任务创建点，与线程池和 APScheduler 并存，后续需逐项确认 owner、取消、等待、重试、幂等和是否 durable，关键业务副作用优先接入已有 Outbox/恢复表。 +1. **后台任务的统一所有权已覆盖 API 入口，但仍有更深层任务机制待分级。** `app/runtime/tasks.py` 已建立 lifespan 级 TaskRegistry，启动收尾、插件 Release 刷新、Webhook E0 广播、CookieCloud E1 手工调度、消息入口、Seerr 订阅、整理历史 AI 重做、OpenAI/Anthropic 协议流和 WebAgent 断线后执行/快照保存均不再维护端点模块级任务集合或 Starlette 回调，shutdown 会停止接收、取消并有限等待，且生命周期清单明确登记其顺序。主仓 `app/` 已无裸 FastAPI `BackgroundTasks`；当前仍有约 `50` 个更底层 `create_task`/等价任务创建点，与线程池和 APScheduler 并存，后续需逐项确认 owner、取消、等待、重试、幂等和是否 durable，关键业务副作用优先接入已有 Outbox/恢复表。 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. **查询侧数据库兼容 ABI 已完成正式装饰器清零。** 写事务装饰器和正式 `db_query/async_db_query` 均为 `0`。站点、消息、用户、订阅、下载/整理历史、工作流、MediaServer、SiteUserData、AgentChat、AgentTaskRun、TransferPending、SystemConfig、PassKey 和 SubscribeHistory 的宿主查询已迁到显式 Session 路径；对应旧插件 Model 调用由独立 `legacy_*` 外壳保留，可同时接受显式 Session 与无 Session 的位置/关键字参数。后续重点转为减少 ORM 对象跨层流转，并保持正式装饰器零回退。 4. **组合根和全局状态仍形成复杂的隐式运行时图。** Singleton 实例、模块级 provider、`configure_*` 注册函数和兼容 Facade 同时存在；它们解决了旧 ABI 和启动顺序问题，但增加测试污染、重复装配、实例身份和初始化顺序风险。`app/startup/lifecycle/__init__.py:161-376` 已有声明式生命周期，`app/startup/modules_initializer.py:505-530` 也有分阶段关闭，但尚未做到所有进程级资源都只通过 typed HostRuntime 访问。后续应以“新代码禁止新增 Service Locator/Singleton 依赖、旧入口有命中观测”为 ratchet。 @@ -833,6 +833,9 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas - 整理历史单条与批量 AI 重做分别登记为 `api.history.ai_redo` 和 `api.history.ai_redo_batch`；请求响应、进度键、Agent prompt、输出回调与旧直接调用入口保持不变。 两类任务随 lifespan shutdown 取消并有限等待，但仍属于进程内 E1 工作，不宣称崩溃后自动恢复。 +- OpenAI Chat Completions 与 Anthropic Messages 的流式 Agent 执行分别登记为 + `api.openai.stream` 和 `api.anthropic.stream`。SSE payload、断线取消、临时会话清理和非流式入口保持 + 原语义；未启动完整 lifespan 的协议校验和旧直接调用通过兼容依赖回退到默认登记器。 #### ARCH-251：用现有数据库做首个 durable side-effect pilot diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index ff5ca46ed..2eef8683f 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6471, - "edge_sha256": "30a14c4218ffd8e2798ab93231e7fa0d4ed3e368a5f1cdf75b09336e863e0a50", + "edge_count": 6477, + "edge_sha256": "41e51256026afbd3abc9684dd1beb653026be927cab14384df6ff6d248806972", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -1685,6 +1685,7 @@ "app.api.endpoints.anthropic -> app.agent", "app.api.endpoints.anthropic -> app.agent.runtime_loader", "app.api.endpoints.anthropic -> app.api", + "app.api.endpoints.anthropic -> app.api.context", "app.api.endpoints.anthropic -> app.api.endpoints", "app.api.endpoints.anthropic -> app.api.endpoints.openai", "app.api.endpoints.anthropic -> app.api.openai_utils", @@ -1692,6 +1693,8 @@ "app.api.endpoints.anthropic -> app.api.presentation.sse", "app.api.endpoints.anthropic -> app.application", "app.api.endpoints.anthropic -> app.application.configuration", + "app.api.endpoints.anthropic -> app.runtime", + "app.api.endpoints.anthropic -> app.runtime.tasks", "app.api.endpoints.anthropic -> app.schemas", "app.api.endpoints.anthropic -> app.schemas.openai", "app.api.endpoints.auth -> app.api", @@ -2035,11 +2038,14 @@ "app.api.endpoints.openai -> app.agent.contracts", "app.api.endpoints.openai -> app.agent.runtime_loader", "app.api.endpoints.openai -> app.api", + "app.api.endpoints.openai -> app.api.context", "app.api.endpoints.openai -> app.api.openai_utils", "app.api.endpoints.openai -> app.api.presentation", "app.api.endpoints.openai -> app.api.presentation.sse", "app.api.endpoints.openai -> app.application", "app.api.endpoints.openai -> app.application.configuration", + "app.api.endpoints.openai -> app.runtime", + "app.api.endpoints.openai -> app.runtime.tasks", "app.api.endpoints.openai -> app.schemas", "app.api.endpoints.openai -> app.schemas.openai", "app.api.endpoints.openai -> app.schemas.types", diff --git a/tests/test_api_background_task_registry.py b/tests/test_api_background_task_registry.py index 868d74809..7fc0c6a85 100644 --- a/tests/test_api_background_task_registry.py +++ b/tests/test_api_background_task_registry.py @@ -3,7 +3,7 @@ import asyncio from types import SimpleNamespace -from app.api.endpoints import history, message, site, subscribe, webhook +from app.api.endpoints import anthropic, history, message, openai, site, subscribe, webhook from app.runtime.tasks import TaskRegistry @@ -31,6 +31,40 @@ class _TaskRegistry(TaskRegistry): self.calls.append((None, (), {"cancel_on_shutdown": cancel_on_shutdown}, owner)) +class _RunningTaskRegistry(TaskRegistry): + """执行协议流任务并保留 owner，验证真实 TaskRegistry 行为。""" + + def __init__(self) -> None: + """初始化 owner 调用记录。""" + super().__init__() + self.owners: list[str] = [] + + def create( + self, + coroutine, + *, + owner: str, + cancel_on_shutdown: bool = True, + ) -> asyncio.Task: + """记录 owner 后委托真实登记器创建任务。""" + self.owners.append(owner) + return super().create( + coroutine, + owner=owner, + cancel_on_shutdown=cancel_on_shutdown, + ) + + +class _ProtocolManager: + """提供兼容协议流结束时需要的最小 AgentManager 接口。""" + + async def clear_session(self, **_kwargs) -> None: + """模拟清理临时协议会话。""" + + async def stop_current_task(self, _session_id: str) -> None: + """模拟停止保留会话的当前任务。""" + + class _WebhookRequest: """提供 webhook 端点读取的最小请求接口。""" @@ -182,3 +216,65 @@ def test_history_batch_ai_redo_uses_task_registry() -> None: assert registry.calls == [ (None, (), {"cancel_on_shutdown": True}, "api.history.ai_redo_batch") ] + + +def test_openai_stream_uses_task_registry(monkeypatch) -> None: + """OpenAI SSE Agent 执行应登记为请求级后台任务。""" + + async def run_agent(**kwargs): + """向协议队列写入一个增量后结束。""" + await kwargs["event_queue"].put("reply") + return "", [] + + monkeypatch.setattr(openai, "_run_managed_agent", run_agent) + + async def scenario() -> None: + registry = _RunningTaskRegistry() + events = [ + event + async for event in openai._stream_response( + manager=_ProtocolManager(), + session_id="session", + user_id="user", + username="tester", + prompt="hello", + images=[], + cleanup_session=True, + task_registry=registry, + ) + ] + + assert events[-1] == "data: [DONE]\n\n" + assert registry.owners == ["api.openai.stream"] + + asyncio.run(scenario()) + + +def test_anthropic_stream_uses_task_registry(monkeypatch) -> None: + """Anthropic SSE Agent 执行应登记为请求级后台任务。""" + + async def run_agent(**kwargs): + """向协议队列写入一个增量后结束。""" + await kwargs["event_queue"].put("reply") + return "", [] + + monkeypatch.setattr(anthropic, "_run_managed_agent", run_agent) + + async def scenario() -> None: + registry = _RunningTaskRegistry() + events = [ + event + async for event in anthropic._stream_anthropic_response( + manager=_ProtocolManager(), + session_id="session", + user_id="user", + prompt="hello", + images=[], + task_registry=registry, + ) + ] + + assert "event: message_stop" in events[-1] + assert registry.owners == ["api.anthropic.stream"] + + asyncio.run(scenario())