From 773ea8cb9bce055842ec8e6b5bbb4886f5974e46 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 21 Aug 2026 20:56:25 +0800 Subject: [PATCH] refactor: introduce typed host runtime --- app/api/context.py | 48 +++++++ app/api/data.py | 11 +- app/api/deps.py | 15 ++- app/startup/context.py | 78 ++++++++++++ app/startup/lifecycle/__init__.py | 9 +- app/startup/modules_initializer.py | 20 ++- docs/architecture-overview.md | 6 +- .../backend-architecture-next-stage.md | 13 +- docs/rules/05-architecture.md | 5 + tests/conftest.py | 1 - .../architecture/dependency-baseline.json | 18 ++- tests/test_host_runtime_context.py | 119 ++++++++++++++++++ 12 files changed, 325 insertions(+), 18 deletions(-) create mode 100644 app/api/context.py create mode 100644 app/startup/context.py create mode 100644 tests/test_host_runtime_context.py diff --git a/app/api/context.py b/app/api/context.py new file mode 100644 index 000000000..168e17d7d --- /dev/null +++ b/app/api/context.py @@ -0,0 +1,48 @@ +"""从 FastAPI AppState 读取类型化宿主能力。""" + +from collections.abc import AsyncGenerator +from typing import cast + +from fastapi import Depends, Request + +from app.application.messaging.chat import AsyncAgentChatRepository, AsyncUnitOfWork +from app.startup.context import AgentChatRuntime, HostRuntime + + +def get_host_runtime(request: Request) -> HostRuntime: + """返回当前 lifespan 挂载的宿主运行时。""" + runtime = getattr(request.app.state, "host_runtime", None) + if not isinstance(runtime, HostRuntime): + raise RuntimeError("HostRuntime 尚未由启动组合根装配") + return runtime + + +def get_agent_chat_runtime( + runtime: HostRuntime = Depends(get_host_runtime), +) -> AgentChatRuntime: + """从完整宿主运行时收窄到 Agent 会话能力。""" + return runtime.agent_chat + + +async def get_agent_chat_session( + runtime: AgentChatRuntime = Depends(get_agent_chat_runtime), +) -> AsyncGenerator[object, None]: + """从类型化 Agent 会话运行时生成请求独占会话。""" + async for session in runtime.async_session(): + yield session + + +def get_agent_chat_repository( + session: object = Depends(get_agent_chat_session), + runtime: AgentChatRuntime = Depends(get_agent_chat_runtime), +) -> AsyncAgentChatRepository: + """构造绑定当前请求会话的 Agent 会话仓储。""" + return cast(AsyncAgentChatRepository, runtime.repository(session)) + + +def get_agent_chat_transaction( + session: object = Depends(get_agent_chat_session), + runtime: AgentChatRuntime = Depends(get_agent_chat_runtime), +) -> AsyncUnitOfWork: + """构造绑定当前请求会话的 Agent 会话事务端口。""" + return cast(AsyncUnitOfWork, runtime.transaction(session)) diff --git a/app/api/data.py b/app/api/data.py index b6a879883..930a6a78a 100644 --- a/app/api/data.py +++ b/app/api/data.py @@ -48,6 +48,12 @@ class ApiDataPorts: _ports: ApiDataPorts | None = None +def configure_api_data_runtime(ports: ApiDataPorts) -> None: + """让旧全局 Facade 委托启动组合根创建的同一个端口实例。""" + global _ports + _ports = ports + + def configure_api_data_ports( *, sync_session: SessionProvider, @@ -57,14 +63,13 @@ def configure_api_data_ports( unit_of_work: dict[str, UnitOfWorkFactory], ) -> None: """由启动组合根登记 API 数据实现，切断 API 对数据库实现包的直接导入。""" - global _ports - _ports = ApiDataPorts( + configure_api_data_runtime(ApiDataPorts( sync_session=sync_session, async_session=async_session, repositories=repositories, standalone=standalone, unit_of_work=unit_of_work, - ) + )) def get_api_data_ports() -> ApiDataPorts: diff --git a/app/api/deps.py b/app/api/deps.py index e3cc96526..90fb04be0 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -29,6 +29,14 @@ from app.application.workflow import ( ) from app.application.messaging.message import MessageQueryService from app.application.messaging.chat import AgentChatService +from app.api.context import ( + get_agent_chat_repository, + get_agent_chat_transaction, +) +from app.application.messaging.chat import ( + AsyncAgentChatRepository, + AsyncUnitOfWork as AgentChatUnitOfWork, +) from app.application.mediaserver import MediaServerQueryService from app.application.servarr import ServarrSubscriptionService from app.application.dashboard import DashboardQueryService @@ -304,12 +312,13 @@ def get_message_query_service( def get_agent_chat_service( - db: AsyncSession = Depends(get_async_db), + repository: AsyncAgentChatRepository = Depends(get_agent_chat_repository), + unit_of_work: AgentChatUnitOfWork = Depends(get_agent_chat_transaction), ) -> AgentChatService: """组装 Agent 会话历史查询和删除服务。""" return AgentChatService( - repository=_repository("agent_chat", db), - unit_of_work=_transaction("async", db), + repository=repository, + unit_of_work=unit_of_work, ) diff --git a/app/startup/context.py b/app/startup/context.py new file mode 100644 index 000000000..e516761ee --- /dev/null +++ b/app/startup/context.py @@ -0,0 +1,78 @@ +"""宿主启动阶段构建的类型化运行时上下文。""" + +from collections.abc import AsyncGenerator, Generator +from dataclasses import dataclass +from typing import Protocol + +from app.application.messaging.chat import ( + AsyncAgentChatRepository, + AsyncUnitOfWork, +) + + +class AgentChatRepositoryFactory(Protocol): + """由请求会话构造 Agent 会话仓储的工厂端口。""" + + def __call__(self, session: object) -> AsyncAgentChatRepository: + """绑定请求会话并返回 Agent 会话仓储。""" + ... + + +class AsyncUnitOfWorkFactory(Protocol): + """由请求会话构造异步事务端口的工厂。""" + + def __call__(self, session: object) -> AsyncUnitOfWork: + """绑定请求会话并返回异步事务端口。""" + ... + + +class AsyncSessionProvider(Protocol): + """FastAPI 请求级异步会话提供器。""" + + def __call__(self) -> AsyncGenerator[object, None]: + """生成一个请求独占的异步数据库会话。""" + ... + + +class SyncSessionProvider(Protocol): + """兼容 API Facade 使用的同步会话提供器。""" + + def __call__(self) -> Generator[object, None, None]: + """生成一个请求独占的同步数据库会话。""" + ... + + +class CompatibilityApiData(Protocol): + """未迁移 API 领域继续使用的结构化兼容 Facade。""" + + sync_session: SyncSessionProvider + async_session: AsyncSessionProvider + + def repository(self, name: str, session: object) -> object: + """按旧能力名构造请求级仓储。""" + ... + + def standalone_repository(self, name: str) -> object: + """按旧能力名构造独立仓储。""" + ... + + def transaction(self, name: str, session: object) -> object: + """按旧能力名构造事务端口。""" + ... + + +@dataclass(frozen=True, slots=True) +class AgentChatRuntime: + """Agent 会话 API 可见的最小数据运行时。""" + + async_session: AsyncSessionProvider + repository: AgentChatRepositoryFactory + transaction: AsyncUnitOfWorkFactory + + +@dataclass(frozen=True, slots=True) +class HostRuntime: + """宿主组合根构建且在一个 FastAPI lifespan 内共享的运行时对象。""" + + agent_chat: AgentChatRuntime + compatibility_api_data: CompatibilityApiData diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index a6b42f9cf..75ed6f5f8 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -128,6 +128,13 @@ async def run_startup_step( logger.info("启动%s完成，耗时=%.2fms", name, elapsed_ms) +async def initialize_modules_component(app: FastAPI) -> None: + """启动模块并把其类型化运行时发布到当前 FastAPI AppState。""" + runtime = await init_modules() + if runtime is not None: + app.state.host_runtime = runtime + + def prepare_plugin_restore() -> None: """先装配插件外部系统服务，再恢复插件及其依赖。""" configure_plugin_services() @@ -197,7 +204,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]: LifecycleComponent( name="模块服务", dependencies=("路由",), - start=init_modules, + start=lambda: initialize_modules_component(app), stop=stop_modules, start_order=70, stop_order=70, diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index 7076226b2..59ea27b08 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -53,7 +53,7 @@ from app.application.site.query import SiteQueryService, configure_site_query_se from app.application.site.health import SiteHealthService, configure_site_health_service from app.application.workflow import WorkflowQueryService, configure_workflow_query from app.application.agentdata import configure_agent_data_ports -from app.api.data import configure_api_data_ports +from app.api.data import ApiDataPorts, configure_api_data_runtime from app.application.subscription.write import configure_subscribe_writer from app.adapters.external.server import ( MoviePilotServerHelper, @@ -97,6 +97,7 @@ from app.startup.managed_resources_initializer import ( stop_managed_resources, ) from app.startup.subscription import TransactionalSubscribeWriter +from app.startup.context import AgentChatRuntime, HostRuntime from app.adapters.web.security.access import set_superuser_token_payload_provider from app.application.security.auth import build_superuser_token_payload from app.application.image import configure_wallpaper_providers @@ -396,16 +397,15 @@ async def stop_modules(): await run_step("临时文件", clear_temp) -async def init_modules(): +async def init_modules() -> HostRuntime: """ - 启动模块 + 启动模块并返回本次 lifespan 唯一的类型化 HostRuntime。 """ # 数据访问能力统一在启动组合根注入，Runtime 和 Adapter 不再直接依赖 Oper。 - configure_api_data_ports( + api_data = ApiDataPorts( sync_session=get_db, async_session=get_async_db, repositories={ - "agent_chat": AgentChatOper, "download_history": DownloadHistoryOper, "media_server": MediaServerOper, "message": MessageOper, @@ -427,6 +427,15 @@ async def init_modules(): "sync": SqlAlchemyUnitOfWork, }, ) + host_runtime = HostRuntime( + agent_chat=AgentChatRuntime( + async_session=get_async_db, + repository=AgentChatOper, + transaction=SqlAlchemyAsyncUnitOfWork, + ), + compatibility_api_data=api_data, + ) + configure_api_data_runtime(host_runtime.compatibility_api_data) configure_runtime_data_providers() configure_chain_data_ports( site=lambda: SiteOper(), @@ -513,3 +522,4 @@ async def init_modules(): start_frontend() # 检查认证状态 check_auth() + return host_runtime diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 9fd051335..3503fe267 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -219,7 +219,8 @@ sequenceDiagram Life->>Init: get_engine() / get_global_async_engine() 预热 + fail-fast Life->>Init: check_connection_budget() 连接预算核算 Life->>Init: init_routers(app) 注册 API 路由 - Life->>Init: init_modules() 发现并初始化模块 + Life->>Init: init_modules() 发现并初始化模块，返回 HostRuntime + Life->>FastAPI: app.state.host_runtime = HostRuntime Life->>Init: init_plugins() / init_scheduler() / init_monitor() Life->>Init: init_command() / init_workflow() Life->>Init: replay_pending_transfers()（后台回放未整理文件） @@ -244,6 +245,9 @@ sequenceDiagram 和 TestClient 因而共享同一 fail-fast 语义。 - **引擎预热 fail-fast**：同步/异步数据库引擎在单线程期完成首次创建， 避免调度器放出大量线程后再创建引擎导致连接锁竞争。 +- **类型化请求装配**：`startup/context.py` 的 frozen slots `HostRuntime` 是 lifespan 内唯一宿主 + 上下文，`api/context.py` 从 `app.state` 收窄到具体领域能力。Agent 会话已迁移，不再通过 + 字符串仓储键定位；`ApiDataPorts` 暂作未迁移领域的同实例兼容 Facade。 - **安全模式**：`MOVIEPILOT_SAFE_MODE` 会跳过插件、定时器、监控器、命令与工作流，用于故障自救。 - **进程拓扑**：全功能 V3 强制 `API_WORKERS=1`，避免每个 worker 重复启动插件和后台控制面；安全模式可临时使用多 worker 诊断，但不是正式扩容方案。 - **健康语义**：`/health/live` 只确认进程和事件循环可响应；`/health/ready` 仅在数据库 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 35574e020..ffe24d607 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0（ARCH-201～203）、阶段 1（ARCH-210～212）与阶段 2（ARCH-220～222）已完成，后续任务按 ID 独立提交和回滚 +> 实施进度：阶段 0（ARCH-201～203）、阶段 1（ARCH-210～212）、阶段 2（ARCH-220～222）与 ARCH-230 已完成，后续任务按 ID 独立提交和回滚 ## 1. 结论先行 @@ -448,6 +448,17 @@ app/api/dependencies/ # 按领域拆分依赖工厂 - 不创建一个更大的全局 `services: dict[str, Any]`； - 不把完整 HostRuntime 传入 Domain 或每个小函数。 +**实施记录（2026-08-21）**： + +- `app/startup/context.py` 定义 frozen slots `HostRuntime` 与首个窄能力 + `AgentChatRuntime`，仓储、Session、UoW 字段均为具体 Protocol 工厂，不是字符串字典。 +- `init_modules()` 保留零参数兼容签名并返回本次 lifespan 唯一 Runtime；生命周期组件把结果挂到 + `app.state.host_runtime`。`app/api/context.py` 只向 Depends 暴露 Agent chat 的最小能力。 +- `get_agent_chat_service` 不再读取全局 `_ports` 或 `"agent_chat"` key；该 key 已从宿主和测试 + `ApiDataPorts.repositories` 删除。未迁移领域仍通过 `compatibility_api_data` 使用同一个实例。 +- fake Runtime 请求测试证明仓储与 UoW 共享同一请求会话，且无需加载真实 DB engine、 + PluginManager 或其他运行时服务；旧 `configure_api_data_ports()` 调用形态继续可用。 + #### ARCH-231：按领域拆分 API dependency 与 presentation **目标**：`app/api/deps.py` 从 512 行集中装配点变成兼容聚合入口，端点只负责 HTTP 解析、鉴权依赖和结果映射。 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 64e3891c6..7fd359971 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -94,6 +94,11 @@ create additional top-level directory categories. `app/startup/` remains the established composition root and is not nested under runtime. It injects providers and callbacks, orders initialization/shutdown and decides restart policy. Lower-level runtime modules must not import startup. +Startup publishes its frozen, slotted `HostRuntime` through FastAPI `app.state`. +API dependencies must narrow that object to a domain runtime (for example, +`AgentChatRuntime`) instead of adding a string key to a global service map. +Legacy registries may delegate the same object while domains migrate, but they +must not construct a second set of service instances. `app.schemas` and `app.db` are compatibility facades, not implementation dependency hubs. Host code imports concrete schema submodules; the schema root diff --git a/tests/conftest.py b/tests/conftest.py index f0e5f9dcd..a64639ea9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -85,7 +85,6 @@ def configure_plugin_system_services(): sync_session=get_db, async_session=get_async_db, repositories={ - "agent_chat": AgentChatOper, "download_history": DownloadHistoryOper, "media_server": MediaServerOper, "message": MessageOper, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index b1df6c7e7..889a41e4d 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": 6095, - "edge_sha256": "06c09c175ac007c7ef891e2a25f5036c8bc3993a817837c03824bb591168ea73", + "edge_count": 6105, + "edge_sha256": "47fd6792530be771c9854d3f2951097d3b110cd46b1cab11643a070f3b11299c", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -1475,6 +1475,11 @@ "app.agent.tools.manager -> app.runtime.log", "app.api.apiv1 -> app.api", "app.api.apiv1 -> app.api.router_specs", + "app.api.context -> app.application", + "app.api.context -> app.application.messaging", + "app.api.context -> app.application.messaging.chat", + "app.api.context -> app.startup", + "app.api.context -> app.startup.context", "app.api.deps -> app.adapters", "app.api.deps -> app.adapters.external", "app.api.deps -> app.adapters.external.server", @@ -1482,6 +1487,7 @@ "app.api.deps -> app.adapters.web.security", "app.api.deps -> app.adapters.web.security.access", "app.api.deps -> app.api", + "app.api.deps -> app.api.context", "app.api.deps -> app.api.data", "app.api.deps -> app.application", "app.api.deps -> app.application.commands", @@ -5698,6 +5704,9 @@ "app.startup.command_initializer -> app.application", "app.startup.command_initializer -> app.application.commands", "app.startup.command_initializer -> app.command", + "app.startup.context -> app.application", + "app.startup.context -> app.application.messaging", + "app.startup.context -> app.application.messaging.chat", "app.startup.database -> app.adapters", "app.startup.database -> app.adapters.system", "app.startup.database -> app.adapters.system.backup", @@ -5871,6 +5880,7 @@ "app.startup.modules_initializer -> app.schemas.types", "app.startup.modules_initializer -> app.startup", "app.startup.modules_initializer -> app.startup.agent_initializer", + "app.startup.modules_initializer -> app.startup.context", "app.startup.modules_initializer -> app.startup.database", "app.startup.modules_initializer -> app.startup.managed_resources_initializer", "app.startup.modules_initializer -> app.startup.subscription", @@ -6112,7 +6122,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 758, + "module_count": 760, "modules": [ "app", "app.adapters", @@ -6300,6 +6310,7 @@ "app.agent.tools.tags", "app.api", "app.api.apiv1", + "app.api.context", "app.api.data", "app.api.deps", "app.api.endpoints", @@ -6837,6 +6848,7 @@ "app.startup.agent_initializer", "app.startup.cache_initializer", "app.startup.command_initializer", + "app.startup.context", "app.startup.database", "app.startup.database_initializer", "app.startup.domain_initializer", diff --git a/tests/test_host_runtime_context.py b/tests/test_host_runtime_context.py new file mode 100644 index 000000000..23b204fc4 --- /dev/null +++ b/tests/test_host_runtime_context.py @@ -0,0 +1,119 @@ +"""类型化 HostRuntime 与 FastAPI AppState 注入测试。""" + +from dataclasses import FrozenInstanceError +from types import SimpleNamespace + +import pytest +from fastapi import Depends, FastAPI +from fastapi.testclient import TestClient + +from app.api.context import ( + get_agent_chat_repository, + get_agent_chat_transaction, +) +from app.api.data import ( + ApiDataPorts, + configure_api_data_runtime, + get_api_data_ports, +) +from app.startup import lifecycle +from app.startup.context import AgentChatRuntime, HostRuntime + + +class _Repository: + """记录绑定会话的 Agent 会话仓储替身。""" + + def __init__(self, session: object) -> None: + """保存由类型化运行时提供的请求会话。""" + self.session = session + + +class _UnitOfWork: + """记录绑定会话的异步事务替身。""" + + def __init__(self, session: object) -> None: + """保存与仓储相同的请求会话。""" + self.session = session + + async def commit(self) -> None: + """模拟提交。""" + + async def rollback(self) -> None: + """模拟回滚。""" + + +def _runtime() -> HostRuntime: + """构造不加载数据库引擎或 PluginManager 的假宿主运行时。""" + async def async_session(): + """生成一个可被 FastAPI 依赖缓存的会话标记。""" + yield object() + + def sync_session(): + """提供兼容 ApiDataPorts 所需的空同步生成器。""" + if False: + yield object() + + compatibility = ApiDataPorts( + sync_session=sync_session, + async_session=async_session, + repositories={}, + standalone={}, + unit_of_work={}, + ) + return HostRuntime( + agent_chat=AgentChatRuntime( + async_session=async_session, + repository=_Repository, + transaction=_UnitOfWork, + ), + compatibility_api_data=compatibility, + ) + + +def test_host_runtime_is_frozen_slotted_and_reuses_compatibility_facade() -> None: + """运行时不可动态扩字段，旧 Facade 必须指向同一个端口实例。""" + runtime = _runtime() + + configure_api_data_runtime(runtime.compatibility_api_data) + + assert not hasattr(runtime, "__dict__") + assert get_api_data_ports() is runtime.compatibility_api_data + with pytest.raises(FrozenInstanceError): + runtime.agent_chat = runtime.agent_chat + + +def test_fastapi_dependencies_use_fake_runtime_without_real_services() -> None: + """请求依赖可只注入假 Runtime，且仓储与 UoW 共享同一请求会话。""" + app = FastAPI() + app.state.host_runtime = _runtime() + + @app.get("/probe") + async def probe( + repository=Depends(get_agent_chat_repository), + unit_of_work=Depends(get_agent_chat_transaction), + ) -> dict[str, bool]: + """返回两个类型化能力是否绑定同一请求会话。""" + return {"same_session": repository.session is unit_of_work.session} + + with TestClient(app) as client: + response = client.get("/probe") + + assert response.status_code == 200 + assert response.json() == {"same_session": True} + + +@pytest.mark.asyncio +async def test_lifecycle_component_attaches_init_modules_result(monkeypatch) -> None: + """模块组件把 init_modules 的构建结果发布到当前 AppState。""" + runtime = _runtime() + app = FastAPI() + + async def init_modules() -> HostRuntime: + """返回不触发真实启动副作用的假运行时。""" + return runtime + + monkeypatch.setattr(lifecycle, "init_modules", init_modules) + + await lifecycle.initialize_modules_component(app) + + assert app.state.host_runtime is runtime