refactor: introduce typed host runtime

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

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:

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

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

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,

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