diff --git a/app/startup/composition/configuration.py b/app/startup/composition/configuration.py index 7382c8d7f..d7e7baaa1 100644 --- a/app/startup/composition/configuration.py +++ b/app/startup/composition/configuration.py @@ -1,17 +1,92 @@ """把可变部署设置转换成宿主各领域使用的类型化配置快照。""" +from dataclasses import dataclass from pathlib import Path +from typing import Callable, cast from app.application.configuration import ( ApiRuntimeConfig, ChainRuntimeConfig, + RuntimeConfiguration, + RuntimeSettingsService, SchedulerRuntimeConfig, + SystemConfigService, TokenRuntimeConfig, + configure_runtime_configuration, + configure_runtime_settings, + configure_system_config, + configure_token_runtime_config, ) +from app.application.database import AsyncDatabaseExecutor +from app.application.security.userconfig import ( + UserConfigurationService, + configure_user_configuration, +) +from app.db.adapters.configuration import TransactionalUserConfigurationRepository +from app.db.oper.systemconfig import SystemConfigOper +from app.db.session import SessionFactory from app.runtime.config import Settings +from app.runtime.settings import ( + configure_runtime_setting_provider, + configure_runtime_setting_updater, +) from app.schemas.types import MediaType +@dataclass(frozen=True, slots=True) +class ConfigurationComposition: + """保存启动阶段加载完成的配置服务与类型化快照门面。""" + + system_config: SystemConfigOper + runtime: RuntimeConfiguration + settings: RuntimeSettingsService + + +async def compose_configuration( + *, + executor: AsyncDatabaseExecutor, + settings: Settings, +) -> ConfigurationComposition: + """加载系统与用户配置快照，并在全部成功后发布配置服务。""" + system_config = cast(Callable[[], SystemConfigOper], SystemConfigOper)() + user_config = TransactionalUserConfigurationRepository(SessionFactory) + await executor.run(system_config.load_snapshot) + await executor.run(user_config.load_snapshot) + configure_system_config( + SystemConfigService( + repository=system_config, + async_executor=executor, + ) + ) + configure_user_configuration( + UserConfigurationService( + repository=user_config, + async_executor=executor, + ) + ) + return ConfigurationComposition( + system_config=system_config, + runtime=RuntimeConfiguration( + api=lambda: build_api_runtime_config(settings), + scheduler=lambda: build_scheduler_runtime_config(settings), + chain=lambda: build_chain_runtime_config(settings), + ), + settings=RuntimeSettingsService(settings), + ) + + +def publish_configuration( + composition: ConfigurationComposition, + settings: Settings, +) -> None: + """发布 HostRuntime 使用的同一配置对象及兼容设置入口。""" + configure_runtime_configuration(composition.runtime) + configure_runtime_settings(composition.settings) + configure_runtime_setting_provider(lambda key: getattr(settings, key)) + configure_runtime_setting_updater(composition.settings.update) + configure_token_runtime_config(lambda: build_token_runtime_config(settings)) + + def normalize_subscribe_rss_interval(value: object) -> int: """把无效或过小的 RSS 间隔收敛为兼容的安全值。""" try: diff --git a/app/startup/composition/database.py b/app/startup/composition/database.py index 1fc3685ed..9dec64f6f 100644 --- a/app/startup/composition/database.py +++ b/app/startup/composition/database.py @@ -2,11 +2,14 @@ from __future__ import annotations +from dataclasses import dataclass + from app.adapters.system.backup.database import ( PostgreSQLBackupBackend, SQLiteBackupBackend, ) from app.adapters.system.backup.files import BackupFiles +from app.api.data import ApiDataPorts from app.application.backup import BackupPolicy, DatabaseBackupService from app.application.database import ( DatabaseGovernance, @@ -17,13 +20,166 @@ from app.application.maintenance import ( DataCleanupService, read_cleanup_policy, ) +from app.application.plugin.transaction import ( + PluginPersistenceService, + configure_plugin_persistence, +) +from app.application.query import DataQueryService, configure_data_query_service +from app.application.workflow import WorkflowQueryService, configure_workflow_query +from app.db.adapters.history.download import SessionDownloadHistoryRepository +from app.db.adapters.pluginidentity import TransactionalPluginIdentityStore +from app.db.adapters.plugininstallation import TransactionalPluginInstallationStore +from app.db.adapters.query import SqlAlchemyDataQueryAdapter +from app.db.adapters.site import SessionSiteRepository +from app.db.adapters.subscription import ( + SessionSubscriptionHistoryRepository, + SessionSubscriptionRepository, +) +from app.db.adapters.transaction import TransactionalWriteRunner +from app.db.adapters.user import SqlAlchemyUserRepository, TransactionalUserRepository +from app.db.adapters.workflow import TransactionalWorkflowQueryRepository from app.db.engine import get_engine from app.db.health import probe_database from app.db.maintenance import DatabaseCleanupRepository -from app.db.session import SessionFactory +from app.db.oper.mediaserver import MediaServerOper +from app.db.oper.message import MessageOper +from app.db.oper.passkey import PassKeyOper +from app.db.oper.systemconfig import SystemConfigOper +from app.db.oper.workflow import WorkflowOper +from app.db.session import ( + SessionFactory, + async_session_scope, + get_async_db, + get_db, +) +from app.db.uow import ( + SqlAlchemyAsyncUnitOfWork, + SqlAlchemyUnitOfWork, + configure_transaction_runners, +) +from app.db.worker import DatabaseWorker from app.runtime.settings import get_runtime_setting +@dataclass(frozen=True, slots=True) +class DatabaseRuntime: + """保存数据库 worker 与兼容写入口共享的事务执行器。""" + + worker: DatabaseWorker + transaction: TransactionalWriteRunner + + +@dataclass(frozen=True, slots=True) +class DatabaseComposition: + """保存初始化器后续装配所需的数据库查询能力。""" + + api_data: ApiDataPorts + workflow_query: WorkflowQueryService + + +_database_runtime: DatabaseRuntime | None = None + + +async def start_database_runtime() -> DatabaseRuntime: + """启动进程唯一数据库 worker，并登记兼容写事务 runner。""" + global _database_runtime + transaction = TransactionalWriteRunner( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + configure_transaction_runners( + sync=transaction.sync, + async_=transaction.async_, + ) + worker = DatabaseWorker() + await worker.start() + runtime = DatabaseRuntime(worker=worker, transaction=transaction) + _database_runtime = runtime + return runtime + + +async def stop_database_runtime() -> None: + """关闭数据库 worker；失败时保留 owner 供诊断和重试。""" + global _database_runtime + runtime = _database_runtime + if runtime is not None: + await runtime.worker.shutdown() + _database_runtime = None + + +def database_runtime_active() -> bool: + """返回数据库 worker 是否仍由当前进程持有。""" + return _database_runtime is not None + + +def build_transactional_user_repository() -> TransactionalUserRepository: + """构造供 Chain、Agent 与进程级认证共享的短会话用户仓储。""" + return TransactionalUserRepository( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + + +def compose_database_services( + *, + runtime: DatabaseRuntime, + system_config: SystemConfigOper, +) -> DatabaseComposition: + """组合查询、插件持久化与旧 API 数据 Facade 所需实现。""" + data_query_adapter = SqlAlchemyDataQueryAdapter(SessionFactory) + configure_data_query_service( + DataQueryService( + subscriptions=data_query_adapter, + histories=data_query_adapter, + async_executor=runtime.worker, + ) + ) + configure_plugin_persistence( + PluginPersistenceService( + executor=runtime.worker, + identities=TransactionalPluginIdentityStore(SessionFactory), + installations=TransactionalPluginInstallationStore( + SessionFactory, + system_config.update_atomically, + ), + ) + ) + workflow_query = WorkflowQueryService( + repository=TransactionalWorkflowQueryRepository( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + ) + configure_workflow_query(workflow_query) + return DatabaseComposition( + api_data=ApiDataPorts( + sync_session=get_db, + async_session=get_async_db, + repositories={ + "download_history": SessionDownloadHistoryRepository, + "media_server": MediaServerOper, + "message": MessageOper, + "passkey": PassKeyOper, + "site": SessionSiteRepository, + "subscribe": SessionSubscriptionRepository, + "subscribe_history": SessionSubscriptionHistoryRepository, + "user": SqlAlchemyUserRepository, + "workflow": WorkflowOper, + }, + standalone={ + "passkey": PassKeyOper, + "system_config": SystemConfigOper, + "user": build_transactional_user_repository, + }, + unit_of_work={ + "async": SqlAlchemyAsyncUnitOfWork, + "sync": SqlAlchemyUnitOfWork, + }, + ), + workflow_query=workflow_query, + ) + + def build_database_governance() -> DatabaseGovernance: """以缓存同步引擎为事实源构造一个完整数据库治理门面。""" engine = get_engine() diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index fda8ee4b2..0941ed65a 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -8,14 +8,8 @@ from typing import Any, Callable, Optional, cast from app.adapters.cache.redis import AsyncRedisHelper, RedisHelper from app.adapters.network.http import AsyncRequestUtils, RequestUtils from app.adapters.network.ip import IpUtils -from app.application.plugin.transaction import ( - PluginPersistenceService, - configure_plugin_persistence, -) from app.chain.mediaserver import MediaServerChain from app.chain.tmdb import TmdbChain -from app.db.adapters.pluginidentity import TransactionalPluginIdentityStore -from app.db.adapters.plugininstallation import TransactionalPluginInstallationStore # SitesHelper涉及资源包拉取，提前引入并容错提示 try: @@ -37,7 +31,7 @@ from app.adapters.system.resource import ( configure_resource_version_provider, ) from app.adapters.web.security.access import set_superuser_token_payload_provider -from app.api.data import ApiDataPorts, configure_api_data_runtime +from app.api.data import configure_api_data_runtime from app.application.agent import AgentDataContext from app.application.agenttask import ( AgentTaskExecutionService, @@ -52,19 +46,11 @@ from app.application.chain.events import ( restore_transfer_result, ) from app.application.configuration import ( - RuntimeConfiguration, - RuntimeSettingsService, - SystemConfigService, TransferRetryConfig, - configure_runtime_configuration, - configure_runtime_settings, - configure_system_config, - configure_token_runtime_config, configure_transfer_retry_config, get_configured_system_config, get_runtime_settings, ) -from app.application.database import configure_database_governance from app.application.history import configure_transfer_history_repository from app.application.image import ( ImageResponsePort, @@ -108,10 +94,6 @@ from app.application.outbox import ( durable_event_topic, validate_durable_event_handlers, ) -from app.application.query import ( - DataQueryService, - configure_data_query_service, -) from app.application.security.auth import AuthService, build_superuser_token_payload, configure_auth_service from app.application.security.passkey import ( PASSKEY_CHALLENGE_TTL_SECONDS, @@ -121,10 +103,6 @@ from app.application.security.passkey import ( ) from app.application.security.url import close_image_proxy_block_log_coalescer from app.application.security.user import configure_user_lookups -from app.application.security.userconfig import ( - UserConfigurationService, - configure_user_configuration, -) from app.application.server.report import ServerReportService from app.application.server.share import ServerSharingService from app.application.service import configure_service_directory @@ -134,7 +112,6 @@ from app.application.subscription.contract import SubscriptionRepository from app.application.workflow import ( WorkflowQueryService, configure_workflow_execution, - configure_workflow_query, ) from app.command import CommandChain from app.db.adapters.agent import ( @@ -143,7 +120,6 @@ from app.db.adapters.agent import ( TransactionalPluginDataRepository, ) from app.db.adapters.chain import TransactionalChainDurableEventWriter -from app.db.adapters.configuration import TransactionalUserConfigurationRepository from app.db.adapters.download import TransactionalDownloadFailureRepository from app.db.adapters.history.download import ( SessionDownloadHistoryRepository, @@ -159,7 +135,6 @@ from app.db.adapters.outbox import ( SqlAlchemyAsyncOutboxStager, SqlAlchemyOutboxDispatchStore, ) -from app.db.adapters.query import SqlAlchemyDataQueryAdapter from app.db.adapters.site import SessionSiteRepository, TransactionalSiteRepository from app.db.adapters.subscription import ( SessionSubscriptionHistoryRepository, @@ -167,18 +142,13 @@ from app.db.adapters.subscription import ( TransactionalSubscriptionHistoryRepository, TransactionalSubscriptionRepository, ) -from app.db.adapters.transaction import TransactionalWriteRunner from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository from app.db.adapters.transfer.execution import ( TransactionalTransferExecutionRepository, ) -from app.db.adapters.user import ( - SqlAlchemyUserRepository, - TransactionalUserRepository, -) +from app.db.adapters.user import SqlAlchemyUserRepository from app.db.adapters.workflow import ( TransactionalWorkflowExecutionService, - TransactionalWorkflowQueryRepository, ) from app.db.oper.agentchat import AgentChatOper from app.db.oper.mediaserver import MediaServerOper @@ -196,9 +166,7 @@ from app.db.session import ( from app.db.uow import ( SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork, - configure_transaction_runners, ) -from app.db.worker import DatabaseWorker from app.runtime.cache import AsyncFileCache, FileCache, TTLCache from app.runtime.config import settings as legacy_settings from app.runtime.events import EventHandlerBinding, EventManager @@ -213,8 +181,6 @@ from app.runtime.extensions.service_config import ( from app.runtime.log import logger from app.runtime.observability import record_metric from app.runtime.settings import ( - configure_runtime_setting_provider, - configure_runtime_setting_updater, get_runtime_setting, ) from app.runtime.state import SystemHelper @@ -224,10 +190,9 @@ from app.runtime.thread import ThreadHelper from app.schemas.message import Message, MessageType from app.schemas.types import EventType, SystemConfigKey from app.startup.composition.configuration import ( - build_api_runtime_config, build_chain_runtime_config, - build_scheduler_runtime_config, - build_token_runtime_config, + compose_configuration, + publish_configuration, ) from app.startup.composition.context import ( AgentChatRuntime, @@ -240,7 +205,14 @@ from app.startup.composition.context import ( SubscriptionRuntime, WorkflowRuntime, ) -from app.startup.composition.database import build_database_governance +from app.startup.composition.database import ( + build_transactional_user_repository, + compose_database_services, + configure_database, + database_runtime_active, + start_database_runtime, + stop_database_runtime, +) from app.startup.composition.subscription import ( async_rule_group_mutation_scope, build_subscription_batch_writer, @@ -259,53 +231,6 @@ from app.startup.initializers.resources import ( ) from app.startup.initializers.scheduler import configure_scheduler_agent_tasks -_database_worker: DatabaseWorker | None = None - - -async def stop_database_worker() -> None: - """停止当前进程的数据库短事务 worker。""" - global _database_worker - worker = _database_worker - if worker is not None: - await worker.shutdown() - _database_worker = None - - -async def _initialize_configuration_services( - database_worker: DatabaseWorker, -) -> SystemConfigOper: - """加载完整配置快照后发布系统与用户配置服务。""" - system_config = SystemConfigOper() - user_config = TransactionalUserConfigurationRepository(SessionFactory) - await database_worker.run(system_config.load_snapshot) - await database_worker.run(user_config.load_snapshot) - configure_system_config( - SystemConfigService( - repository=system_config, - async_executor=database_worker, - ) - ) - configure_user_configuration( - UserConfigurationService( - repository=user_config, - async_executor=database_worker, - ) - ) - return system_config - - -def _build_runtime_settings_service() -> RuntimeSettingsService: - """将唯一可变部署配置实现注入管理服务。""" - return RuntimeSettingsService(legacy_settings) - - -def _build_transactional_user_repository() -> TransactionalUserRepository: - """构造供 Chain、Agent 与进程级认证共享的短会话用户仓储。""" - return TransactionalUserRepository( - sync_session=SessionFactory, - async_session=async_session_scope, - ) - def _execute_legacy_transfer_command(**kwargs: Any) -> Any: """把旧 Chain ABI 延迟转入唯一 TransferChain durable command。""" @@ -356,7 +281,7 @@ def _build_chain_runtime_context( transfer_execution_repository=TransactionalTransferExecutionRepository(SessionFactory), media_server_repository=TransactionalMediaServerRepository(SessionFactory), download_failure_repository=TransactionalDownloadFailureRepository(SessionFactory), - user_repository=_build_transactional_user_repository(), + user_repository=build_transactional_user_repository(), legacy_transfer_command=_execute_legacy_transfer_command, configuration=build_chain_runtime_config(legacy_settings), durable_event_writer=TransactionalChainDurableEventWriter(SessionFactory), @@ -935,8 +860,8 @@ async def stop_modules() -> bool: all_converged = False logger.error("Web Agent任务未完成收尾，跳过持久化和数据库关闭以保护活动事务") if persistence_drained: - await run_step("数据库任务", stop_database_worker) - if _database_worker is None: + await run_step("数据库任务", stop_database_runtime) + if not database_runtime_active(): await run_step("数据库连接", close_database) else: all_converged = False @@ -950,84 +875,27 @@ async def _initialize_modules() -> HostRuntime: """ 构造模块服务并返回本次 lifespan 唯一的类型化 HostRuntime。 """ - global _database_worker configure_application_network_ports() - # 兼容 Oper 的无 Session 写入口仍由组合根持有事务，避免模型恢复自动提交。 - transaction_runner = TransactionalWriteRunner( - sync_session=SessionFactory, - async_session=async_session_scope, - ) - configure_transaction_runners( - sync=transaction_runner.sync, - async_=transaction_runner.async_, - ) - database_worker = DatabaseWorker() - await database_worker.start() - _database_worker = database_worker + database_runtime = await start_database_runtime() try: - system_config = await _initialize_configuration_services(database_worker) + configuration = await compose_configuration( + executor=database_runtime.worker, + settings=legacy_settings, + ) except BaseException: try: - await stop_database_worker() + await stop_database_runtime() except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常 logger.error(f"启动失败后的数据库任务清理失败：{cleanup_error}") raise - data_query_adapter = SqlAlchemyDataQueryAdapter(SessionFactory) - configure_data_query_service( - DataQueryService( - subscriptions=data_query_adapter, - histories=data_query_adapter, - async_executor=database_worker, - ) + database_services = compose_database_services( + runtime=database_runtime, + system_config=configuration.system_config, ) - configure_plugin_persistence( - PluginPersistenceService( - executor=database_worker, - identities=TransactionalPluginIdentityStore(SessionFactory), - installations=TransactionalPluginInstallationStore( - SessionFactory, - system_config.update_atomically, - ), - ) - ) - # 数据访问能力统一在启动组合根注入，Runtime 和 Adapter 不再直接依赖 Oper。 - api_data = ApiDataPorts( - sync_session=get_db, - async_session=get_async_db, - repositories={ - "download_history": SessionDownloadHistoryRepository, - "media_server": MediaServerOper, - "message": MessageOper, - "passkey": PassKeyOper, - "site": SessionSiteRepository, - "subscribe": SessionSubscriptionRepository, - "subscribe_history": SessionSubscriptionHistoryRepository, - "user": SqlAlchemyUserRepository, - "workflow": WorkflowOper, - }, - standalone={ - "passkey": PassKeyOper, - "system_config": SystemConfigOper, - "user": _build_transactional_user_repository, - }, - unit_of_work={ - "async": SqlAlchemyAsyncUnitOfWork, - "sync": SqlAlchemyUnitOfWork, - }, - ) - runtime_configuration = RuntimeConfiguration( - api=lambda: build_api_runtime_config(legacy_settings), - scheduler=lambda: build_scheduler_runtime_config(legacy_settings), - chain=lambda: build_chain_runtime_config(legacy_settings), - ) - runtime_settings = _build_runtime_settings_service() - workflow_query = WorkflowQueryService( - repository=TransactionalWorkflowQueryRepository( - sync_session=SessionFactory, - async_session=async_session_scope, - ) - ) - configure_workflow_query(workflow_query) + system_config = configuration.system_config + database_worker = database_runtime.worker + transaction_runner = database_runtime.transaction + workflow_query = database_services.workflow_query download_history_repository = TransactionalDownloadHistoryRepository( sync_session=SessionFactory, async_session=async_session_scope, @@ -1065,7 +933,7 @@ async def _initialize_modules() -> HostRuntime: chat=agent_chat_service, chat_persistence=agent_chat_persistence, tasks=agent_task_repository, - users=_build_transactional_user_repository(), + users=build_transactional_user_repository(), sites=site_repository, subscriptions=subscription_repository, subscription_mutation_scope=subscription_mutation_scope, @@ -1096,7 +964,7 @@ async def _initialize_modules() -> HostRuntime: ), authentication=AuthenticationRuntime( user_repository=SqlAlchemyUserRepository, - standalone_user=_build_transactional_user_repository, + standalone_user=build_transactional_user_repository, system_config=SystemConfigOper, passkey=PassKeyOper, ), @@ -1142,17 +1010,13 @@ async def _initialize_modules() -> HostRuntime: repository=WorkflowOper, system_config=get_configured_system_config, ), - configuration=runtime_configuration, - settings=runtime_settings, + configuration=configuration.runtime, + settings=configuration.settings, tasks=get_task_registry(), ) - configure_runtime_configuration(host_runtime.configuration) - configure_runtime_settings(host_runtime.settings) - configure_runtime_setting_provider(lambda key: getattr(legacy_settings, key)) - configure_runtime_setting_updater(host_runtime.settings.update) - configure_token_runtime_config(lambda: build_token_runtime_config(legacy_settings)) + publish_configuration(configuration, legacy_settings) # 旧 app.api.data 导入只保留 ABI 转发，正式 API 依赖全部读取 HostRuntime。 - configure_api_data_runtime(api_data) + configure_api_data_runtime(database_services.api_data) configure_runtime_data_providers(workflow_query, subscription_repository) workflow_execution = TransactionalWorkflowExecutionService(SessionFactory) configure_workflow_execution(workflow_execution) @@ -1162,17 +1026,17 @@ async def _initialize_modules() -> HostRuntime: max_failed_retries=get_runtime_setting("TRANSFER_MAX_FAILED_RETRIES"), ) ) - configure_database_governance(build_database_governance()) + configure_database() configure_agent_chat_service(agent_chat_service) configure_agent_chat_persistence(agent_chat_persistence) configure_user_lookups( - by_id=lambda user_id: _build_transactional_user_repository().get_by_id(user_id), - by_name=lambda username: _build_transactional_user_repository().get_by_name(username), - by_channel=lambda **bindings: _build_transactional_user_repository().find_name_by_bindings(bindings), + by_id=lambda user_id: build_transactional_user_repository().get_by_id(user_id), + by_name=lambda username: build_transactional_user_repository().get_by_name(username), + by_channel=lambda **bindings: build_transactional_user_repository().find_name_by_bindings(bindings), ) configure_auth_service( AuthService( - users=_build_transactional_user_repository(), + users=build_transactional_user_repository(), config=get_configured_system_config(), passkeys=PassKeyOper(), ) @@ -1264,7 +1128,7 @@ async def init_modules() -> HostRuntime: except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常 logger.error(f"模块启动失败后的消息资源清理失败：{cleanup_error}") try: - await stop_database_worker() + await stop_database_runtime() except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常 logger.error(f"模块启动失败后的数据库任务清理失败：{cleanup_error}") raise diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index eecf8c6dc..aac4b2f70 100644 --- a/docs/architecture-optimization-checklist.md +++ b/docs/architecture-optimization-checklist.md @@ -70,7 +70,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 965 / 8,185 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 965 / 8,209 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -78,7 +78,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | | Python 源码量 | 约 298,700 行 | 60 个文件超过 1,000 行，10 个超过 2,000 行 | | 长方法 | 299 个超过 80 行 | 70 个超过 150 行，23 个超过 250 行；大量是私有方法 | -| 全量 mypy 历史债务 | 10,348 / 593 文件 | strict frontier 当前覆盖 41 个文件，Domain projection 与既有 owner 拆分的受益调用方低水位已固化 | +| 全量 mypy 历史债务 | 10,347 / 593 文件 | strict frontier 当前覆盖 41 个文件，Domain projection 与既有 owner 拆分的受益调用方低水位已固化 | | Ruff 历史诊断 | 648 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率低水位 | Application 80.92%，Domain 79.89% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | @@ -668,8 +668,10 @@ Pylint 10/10、Ruff、mypy/复杂度 ratchet、宿主与最新官方插件基线 - [x] `MediaInfo`/`domain/context.py` 保留 canonical 类型路径；TMDB、豆瓣、Bangumi、AniList 规则已整体迁入 `domain/projection/` 单词 owner，setter 只薄委托，输入不可变、多来源顺序、字段 golden 与宿主直接导入门禁覆盖； 既有 `app.core.context` 精确兼容映射继续指向 canonical 类型，未新增 SDK/Compat。 -- [ ] 将 `startup/initializers/modules.py` 的对象构造按领域移到 `startup/composition/*`；initializer - 只负责顺序、注册和是否重启的决策。 +- [x] L4.1 已将配置快照、`RuntimeSettingsService`、`DatabaseWorker`、兼容事务 runner、查询服务与 + 插件持久化迁入既有 `startup/composition/{configuration,database}.py`；initializer 只调用组合 API。 +- [ ] 继续将 Chain、network、server、outbox、security、agent 与 HostRuntime 构造按领域移到 + `startup/composition/*`，最终让 initializer 只负责顺序、注册和是否重启的决策。 - [ ] 保留生命周期 manifest 和顺序快照；高扇出在组合根是允许的，但业务实现不能继续沉积其中。 ### ARCH-204 收敛 sync/async 重复与测试存量 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 13e062e3b..e56eac5fd 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -283,7 +283,9 @@ sequenceDiagram 订阅、工作流和请求事务均使用命名 runtime 字段，不再通过字符串仓储键定位；API、Scheduler、Chain 从 `HostRuntime.configuration` 获取 frozen 配置快照。系统设置管理 API 通过 `HostRuntime.settings` 的窄服务读写可变部署设置，业务域不接触 Settings 实例；生产与测试组合根统一 - 复用 `startup/composition/configuration.py` 的映射。`ApiDataPorts` 仅保留旧导入 ABI，不参与正式请求链路。 + 复用 `startup/composition/configuration.py` 的映射、快照加载与发布。`startup/composition/database.py` + 持有数据库 worker、兼容事务 runner、查询/插件持久化和旧 `ApiDataPorts` 的具体装配；initializer + 只保留顺序调用。`ApiDataPorts` 仅保留旧导入 ABI，不参与正式请求链路。 - **安全模式**：`MOVIEPILOT_SAFE_MODE` 会跳过插件、定时器、监控器、命令与工作流，用于故障自救。 - **Scheduler 同名职责包**：旧 `app/scheduler.py` 单体已退役；`catalog.py` 负责作业目录和计划投影， `execution.py`、`bridge.py`、`progress.py` 分别负责执行、跨循环句柄和进度终态，`registry.py` @@ -742,7 +744,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 965 | -| 内部导入边 | 8,185 | +| 内部导入边 | 8,209 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture-refactor-roadmap.md b/docs/architecture-refactor-roadmap.md index be1b40a3b..7e7925d8a 100644 --- a/docs/architecture-refactor-roadmap.md +++ b/docs/architecture-refactor-roadmap.md @@ -326,11 +326,27 @@ TMDB、豆瓣、Bangumi、AniList 详情到统一字段的规则一次迁入 `ap **Local verification (2026-08-29)** -- 基于 `origin/v3@d17d62e95` 锁定全量 `7374 passed, 9 skipped`；投影/响应/来源/Compat/架构专项全绿。 +- 基于 `origin/v3@d17d62e95` 锁定全量 `7375 passed, 9 skipped`；投影/响应/来源/Compat/架构专项全绿。 - scoped Pylint `10.00/10` 且零重复；新 `domain/projection` owner mypy 零错误，strict frontier、 Ruff、mypy、复杂度和宿主依赖低水位均通过。 - 最新官方插件仓 `2d6946eb1` 兼容基线通过；`app/plugins/**` 零改动，包根只有说明且无重复导出。 +### S5-L4 Startup composition + +**Status:** `IN_PROGRESS` + +**Delivered batch: configuration and database composition** + +- `startup/composition/configuration.py` 唯一负责系统/用户配置快照加载、`RuntimeConfiguration` 与 + `RuntimeSettingsService` 构造，以及同一对象向正式运行时和兼容设置入口的发布。 +- `startup/composition/database.py` 唯一持有 `DatabaseWorker`、兼容事务 runner、查询服务、 + 插件持久化、工作流查询和旧 `ApiDataPorts` 的具体数据库装配；worker 关闭失败时保留 owner。 +- `startup/initializers/modules.py` 只按原顺序调用 composition API，并继续构造尚未迁移的领域运行时； + 本批未改变 HostRuntime、插件 ABI、SDK/Compat 或生命周期 manifest，`app/plugins/**` 未改动。 +- 基于 `origin/v3@a1e351aa9` 锁定全量 `7377 passed, 9 skipped`；真实正常模式启动和优雅关闭通过。 +- 后续批次继续迁移 Chain/network/server/outbox/security/agent 与最终 HostRuntime 构造；本批不宣称 + S5-L4 整体完成。 + ### S1-L1.1 Durable admission **Status:** `VERIFIED` diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 70d68b91c..74c0eb001 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -136,6 +136,11 @@ composition constructs and injects cross-layer dependencies, initializers expose domain-scoped startup/shutdown hooks, and lifecycle orders those hooks and decides restart policy. Reusable persistence implementations belong in `app/db/adapters/`, not startup. Lower-level runtime modules must not import startup. +`startup/composition/configuration.py` owns configuration snapshot loading, typed +runtime/settings construction and publication. `startup/composition/database.py` +owns the process database worker, compatibility transaction runner, query services, +plugin persistence and legacy API data-port wiring. Initializers call these owners in +startup order and must not recreate their concrete construction. 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. diff --git a/scripts/architecture/baseline.py b/scripts/architecture/baseline.py index 2aece15b6..601fb273a 100644 --- a/scripts/architecture/baseline.py +++ b/scripts/architecture/baseline.py @@ -81,8 +81,8 @@ FOUNDATIONAL_SETTINGS_BOUNDARIES = { "app/db/session.py": "数据库会话与连接配额必须在应用组合根装配前可用", } COMPOSITION_ROOT_OPER_BOUNDARIES = { - ("app/startup/initializers/modules.py", "SystemConfigOper"): - "启动组合根负责构造唯一的系统配置数据库适配器", + ("app/startup/composition/configuration.py", "SystemConfigOper"): + "配置组合 owner 负责构造唯一的系统配置数据库适配器", } diff --git a/tests/fixtures/architecture/configuration-debt-baseline.json b/tests/fixtures/architecture/configuration-debt-baseline.json index e4c30f464..0c6afaa7e 100644 --- a/tests/fixtures/architecture/configuration-debt-baseline.json +++ b/tests/fixtures/architecture/configuration-debt-baseline.json @@ -3,9 +3,9 @@ "count": 1, "entries": [ { - "file": "app/startup/initializers/modules.py", + "file": "app/startup/composition/configuration.py", "name": "SystemConfigOper", - "reason": "启动组合根负责构造唯一的系统配置数据库适配器" + "reason": "配置组合 owner 负责构造唯一的系统配置数据库适配器" } ] }, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index c17cd73e3..586b9e905 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 8185, - "edge_sha256": "0f8fcb5fa0c3ac9bef4d185cd23d36342859ca966723dc690026e61e70625b9f", + "edge_count": 8209, + "edge_sha256": "9f7de0c9e2a95fdcaf385feae16aa582ee1d5c68c2712fbb7b3fee07862499ae", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -8628,8 +8628,18 @@ "app.sdk.utilities -> app.sdk.string", "app.startup.composition.configuration -> app.application", "app.startup.composition.configuration -> app.application.configuration", + "app.startup.composition.configuration -> app.application.database", + "app.startup.composition.configuration -> app.application.security", + "app.startup.composition.configuration -> app.application.security.userconfig", + "app.startup.composition.configuration -> app.db", + "app.startup.composition.configuration -> app.db.adapters", + "app.startup.composition.configuration -> app.db.adapters.configuration", + "app.startup.composition.configuration -> app.db.oper", + "app.startup.composition.configuration -> app.db.oper.systemconfig", + "app.startup.composition.configuration -> app.db.session", "app.startup.composition.configuration -> app.runtime", "app.startup.composition.configuration -> app.runtime.config", + "app.startup.composition.configuration -> app.runtime.settings", "app.startup.composition.configuration -> app.schemas", "app.startup.composition.configuration -> app.schemas.types", "app.startup.composition.context -> app.application", @@ -8656,15 +8666,40 @@ "app.startup.composition.database -> app.adapters.system.backup", "app.startup.composition.database -> app.adapters.system.backup.database", "app.startup.composition.database -> app.adapters.system.backup.files", + "app.startup.composition.database -> app.api", + "app.startup.composition.database -> app.api.data", "app.startup.composition.database -> app.application", "app.startup.composition.database -> app.application.backup", "app.startup.composition.database -> app.application.database", "app.startup.composition.database -> app.application.maintenance", + "app.startup.composition.database -> app.application.plugin", + "app.startup.composition.database -> app.application.plugin.transaction", + "app.startup.composition.database -> app.application.query", + "app.startup.composition.database -> app.application.workflow", "app.startup.composition.database -> app.db", + "app.startup.composition.database -> app.db.adapters", + "app.startup.composition.database -> app.db.adapters.history", + "app.startup.composition.database -> app.db.adapters.history.download", + "app.startup.composition.database -> app.db.adapters.pluginidentity", + "app.startup.composition.database -> app.db.adapters.plugininstallation", + "app.startup.composition.database -> app.db.adapters.query", + "app.startup.composition.database -> app.db.adapters.site", + "app.startup.composition.database -> app.db.adapters.subscription", + "app.startup.composition.database -> app.db.adapters.transaction", + "app.startup.composition.database -> app.db.adapters.user", + "app.startup.composition.database -> app.db.adapters.workflow", "app.startup.composition.database -> app.db.engine", "app.startup.composition.database -> app.db.health", "app.startup.composition.database -> app.db.maintenance", + "app.startup.composition.database -> app.db.oper", + "app.startup.composition.database -> app.db.oper.mediaserver", + "app.startup.composition.database -> app.db.oper.message", + "app.startup.composition.database -> app.db.oper.passkey", + "app.startup.composition.database -> app.db.oper.systemconfig", + "app.startup.composition.database -> app.db.oper.workflow", "app.startup.composition.database -> app.db.session", + "app.startup.composition.database -> app.db.uow", + "app.startup.composition.database -> app.db.worker", "app.startup.composition.database -> app.runtime", "app.startup.composition.database -> app.runtime.settings", "app.startup.composition.subscription -> app.adapters", @@ -8809,7 +8844,6 @@ "app.startup.initializers.modules -> app.application.chain.context", "app.startup.initializers.modules -> app.application.chain.events", "app.startup.initializers.modules -> app.application.configuration", - "app.startup.initializers.modules -> app.application.database", "app.startup.initializers.modules -> app.application.history", "app.startup.initializers.modules -> app.application.image", "app.startup.initializers.modules -> app.application.messaging", @@ -8820,15 +8854,11 @@ "app.startup.initializers.modules -> app.application.module", "app.startup.initializers.modules -> app.application.network", "app.startup.initializers.modules -> app.application.outbox", - "app.startup.initializers.modules -> app.application.plugin", - "app.startup.initializers.modules -> app.application.plugin.transaction", - "app.startup.initializers.modules -> app.application.query", "app.startup.initializers.modules -> app.application.security", "app.startup.initializers.modules -> app.application.security.auth", "app.startup.initializers.modules -> app.application.security.passkey", "app.startup.initializers.modules -> app.application.security.url", "app.startup.initializers.modules -> app.application.security.user", - "app.startup.initializers.modules -> app.application.security.userconfig", "app.startup.initializers.modules -> app.application.server", "app.startup.initializers.modules -> app.application.server.report", "app.startup.initializers.modules -> app.application.server.share", @@ -8856,19 +8886,14 @@ "app.startup.initializers.modules -> app.db.adapters", "app.startup.initializers.modules -> app.db.adapters.agent", "app.startup.initializers.modules -> app.db.adapters.chain", - "app.startup.initializers.modules -> app.db.adapters.configuration", "app.startup.initializers.modules -> app.db.adapters.download", "app.startup.initializers.modules -> app.db.adapters.history", "app.startup.initializers.modules -> app.db.adapters.history.download", "app.startup.initializers.modules -> app.db.adapters.history.transfer", "app.startup.initializers.modules -> app.db.adapters.mediaserver", "app.startup.initializers.modules -> app.db.adapters.outbox", - "app.startup.initializers.modules -> app.db.adapters.pluginidentity", - "app.startup.initializers.modules -> app.db.adapters.plugininstallation", - "app.startup.initializers.modules -> app.db.adapters.query", "app.startup.initializers.modules -> app.db.adapters.site", "app.startup.initializers.modules -> app.db.adapters.subscription", - "app.startup.initializers.modules -> app.db.adapters.transaction", "app.startup.initializers.modules -> app.db.adapters.transfer", "app.startup.initializers.modules -> app.db.adapters.transfer.admission", "app.startup.initializers.modules -> app.db.adapters.transfer.execution", @@ -8883,7 +8908,6 @@ "app.startup.initializers.modules -> app.db.oper.workflow", "app.startup.initializers.modules -> app.db.session", "app.startup.initializers.modules -> app.db.uow", - "app.startup.initializers.modules -> app.db.worker", "app.startup.initializers.modules -> app.runtime", "app.startup.initializers.modules -> app.runtime.cache", "app.startup.initializers.modules -> app.runtime.config", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index a61955c3b..d95ce4d45 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -3286,7 +3286,7 @@ "attr-defined": 2, "misc": 1, "no-any-return": 2, - "no-untyped-call": 28, + "no-untyped-call": 27, "no-untyped-def": 6, "return-value": 2 }, diff --git a/tests/test_agent_lifecycle.py b/tests/test_agent_lifecycle.py index bfa479d76..5acc6d82a 100644 --- a/tests/test_agent_lifecycle.py +++ b/tests/test_agent_lifecycle.py @@ -214,7 +214,7 @@ async def test_agent_initialization_failure_does_not_stop_module_startup( == [] ) finally: - await modules_initializer.stop_database_worker() + await modules_initializer.stop_database_runtime() manager.initialize.assert_awaited_once_with() start_frontend.assert_called_once_with() diff --git a/tests/test_architecture_contract_baseline.py b/tests/test_architecture_contract_baseline.py index 0cb63ecc5..888e4d77b 100644 --- a/tests/test_architecture_contract_baseline.py +++ b/tests/test_architecture_contract_baseline.py @@ -291,7 +291,7 @@ def test_configuration_debt_baseline_tracks_canonical_direct_access() -> None: } assert baseline["composition_root_oper_boundaries"]["count"] == 1 assert baseline["composition_root_oper_boundaries"]["entries"][0]["file"] == ( - "app/startup/initializers/modules.py" + "app/startup/composition/configuration.py" ) diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index d46bade9b..d0343f71f 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -734,9 +734,11 @@ def test_user_configuration_uses_typed_transactional_adapter(): } assert "Any" not in application_source - startup_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") - assert "TransactionalUserConfigurationRepository(SessionFactory)" in startup_source - assert "UserConfigOper" not in startup_source + composition_source = ( + APP_ROOT / "startup" / "composition" / "configuration.py" + ).read_text(encoding="utf-8") + assert "TransactionalUserConfigurationRepository(SessionFactory)" in composition_source + assert "UserConfigOper" not in composition_source adapter_path = APP_ROOT / "db" / "adapters" / "configuration.py" adapter_source = adapter_path.read_text(encoding="utf-8") @@ -750,6 +752,47 @@ def test_user_configuration_uses_typed_transactional_adapter(): assert "Any" not in oper_source +def test_modules_initializer_delegates_configuration_and_database_composition(): + """模块初始化器只编排配置与数据库组合 API，不再内联其构造细节。""" + initializer_source = ( + APP_ROOT / "startup" / "initializers" / "modules.py" + ).read_text(encoding="utf-8") + configuration_source = ( + APP_ROOT / "startup" / "composition" / "configuration.py" + ).read_text(encoding="utf-8") + database_source = ( + APP_ROOT / "startup" / "composition" / "database.py" + ).read_text(encoding="utf-8") + + for call in ( + "start_database_runtime()", + "stop_database_runtime", + "compose_configuration(", + "compose_database_services(", + "publish_configuration(", + "configure_database()", + ): + assert call in initializer_source + for detail in ( + "DatabaseWorker()", + "TransactionalWriteRunner(", + "PluginPersistenceService(", + "SqlAlchemyDataQueryAdapter(", + "RuntimeConfiguration(", + "RuntimeSettingsService(", + "TransactionalUserConfigurationRepository(", + ): + assert detail not in initializer_source + assert "stop_database_worker" not in initializer_source + assert "RuntimeConfiguration(" in configuration_source + assert "RuntimeSettingsService(" in configuration_source + assert "TransactionalUserConfigurationRepository(" in configuration_source + assert "DatabaseWorker()" in database_source + assert "TransactionalWriteRunner(" in database_source + assert "PluginPersistenceService(" in database_source + assert "SqlAlchemyDataQueryAdapter(" in database_source + + def test_canonical_workflow_oper_has_no_legacy_writer_or_duplicate_exports(): """工作流旧写入口只能存在于 SDK Legacy facade。""" oper_path = APP_ROOT / "db" / "oper" / "workflow.py" @@ -944,11 +987,17 @@ def test_user_chain_and_agent_ports_are_typed_and_orm_free(): def test_startup_injects_user_adapter_instead_of_raw_oper(): """启动组合根不得把无会话 UserOper 注入宿主查询调用面。""" - path = APP_ROOT / "startup" / "initializers" / "modules.py" - source = path.read_text(encoding="utf-8-sig") + initializer_source = ( + APP_ROOT / "startup" / "initializers" / "modules.py" + ).read_text(encoding="utf-8-sig") + database_source = ( + APP_ROOT / "startup" / "composition" / "database.py" + ).read_text(encoding="utf-8-sig") + source = initializer_source + database_source assert "TransactionalUserRepository" in source assert "SqlAlchemyUserRepository" in source + assert "build_transactional_user_repository" in initializer_source assert "from app.db.oper.user import UserOper" not in source assert "user=lambda: UserOper()" not in source diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index 5e1ebc6a9..1fefb3d9f 100644 --- a/tests/test_cache_system.py +++ b/tests/test_cache_system.py @@ -171,7 +171,7 @@ def test_init_modules_does_not_clear_package_tool_cache(monkeypatch): try: await modules_initializer.init_modules() finally: - await modules_initializer.stop_database_worker() + await modules_initializer.stop_database_runtime() asyncio.run(initialize_modules()) diff --git a/tests/test_configuration_initializer.py b/tests/test_configuration_initializer.py index 36b289dba..e1abfc1e4 100644 --- a/tests/test_configuration_initializer.py +++ b/tests/test_configuration_initializer.py @@ -6,6 +6,8 @@ from unittest.mock import AsyncMock import pytest from app.application.configuration import configure_runtime_settings +from app.startup.composition import configuration as configuration_composition +from app.startup.composition import database as database_composition from app.startup.initializers import modules as modules_initializer @@ -88,12 +90,39 @@ async def test_network_test_transport_enforces_safe_http_options(monkeypatch) -> } -def test_runtime_settings_service_uses_legacy_settings_from_startup_root(monkeypatch) -> None: +@pytest.mark.asyncio +async def test_runtime_settings_service_uses_legacy_settings_from_startup_root( + monkeypatch, +) -> None: """组合根装配的设置服务应直接读写唯一部署配置对象。""" legacy_settings = _MutableSettings() - monkeypatch.setattr(modules_initializer, "legacy_settings", legacy_settings) - service = modules_initializer._build_runtime_settings_service() + class _SystemConfig: + """提供无需数据库的系统配置快照桩。""" + + def load_snapshot(self): + """模拟系统配置快照加载。""" + + class _UserConfig: + """提供无需数据库的用户配置快照桩。""" + + def load_snapshot(self): + """模拟用户配置快照加载。""" + + monkeypatch.setattr(configuration_composition, "SystemConfigOper", _SystemConfig) + monkeypatch.setattr( + configuration_composition, + "TransactionalUserConfigurationRepository", + lambda _session_factory: _UserConfig(), + ) + monkeypatch.setattr(configuration_composition, "configure_system_config", lambda _service: None) + monkeypatch.setattr(configuration_composition, "configure_user_configuration", lambda _service: None) + + composition = await configuration_composition.compose_configuration( + executor=_InlineWorker(), + settings=legacy_settings, + ) + service = composition.settings configure_runtime_settings(service) assert service.snapshot(include={"VALUE"}) == {"VALUE": "before"} @@ -122,24 +151,27 @@ async def test_configuration_services_publish_after_both_snapshots_load( """登记用户快照已加载。""" events.append("load-user") - monkeypatch.setattr(modules_initializer, "SystemConfigOper", _SystemConfig) + monkeypatch.setattr(configuration_composition, "SystemConfigOper", _SystemConfig) monkeypatch.setattr( - modules_initializer, + configuration_composition, "TransactionalUserConfigurationRepository", lambda _session_factory: _UserConfig(), ) monkeypatch.setattr( - modules_initializer, + configuration_composition, "configure_system_config", lambda _service: events.append("publish-system"), ) monkeypatch.setattr( - modules_initializer, + configuration_composition, "configure_user_configuration", lambda _service: events.append("publish-user"), ) - await modules_initializer._initialize_configuration_services(_InlineWorker()) + await configuration_composition.compose_configuration( + executor=_InlineWorker(), + settings=_MutableSettings(), + ) assert events == [ "load-system", @@ -170,31 +202,77 @@ async def test_configuration_load_failure_does_not_publish_partial_service( """模拟用户配置加载失败。""" raise RuntimeError("load failed") - monkeypatch.setattr(modules_initializer, "SystemConfigOper", _SystemConfig) + monkeypatch.setattr(configuration_composition, "SystemConfigOper", _SystemConfig) monkeypatch.setattr( - modules_initializer, + configuration_composition, "TransactionalUserConfigurationRepository", lambda _session_factory: _UserConfig(), ) monkeypatch.setattr( - modules_initializer, + configuration_composition, "configure_system_config", lambda service: published.append(service), ) monkeypatch.setattr( - modules_initializer, + configuration_composition, "configure_user_configuration", lambda service: published.append(service), ) with pytest.raises(RuntimeError, match="load failed"): - await modules_initializer._initialize_configuration_services(_InlineWorker()) + await configuration_composition.compose_configuration( + executor=_InlineWorker(), + settings=_MutableSettings(), + ) assert published == [] +def test_publish_configuration_reuses_composed_runtime_and_settings(monkeypatch) -> None: + """正式运行时与兼容入口必须共享同一组配置对象。""" + events = {} + runtime = object() + settings_service = SimpleNamespace(update=lambda key, value: (key, value)) + legacy_settings = SimpleNamespace(VALUE="configured") + composition = SimpleNamespace(runtime=runtime, settings=settings_service) + + monkeypatch.setattr( + configuration_composition, + "configure_runtime_configuration", + lambda value: events.update(runtime=value), + ) + monkeypatch.setattr( + configuration_composition, + "configure_runtime_settings", + lambda value: events.update(settings=value), + ) + monkeypatch.setattr( + configuration_composition, + "configure_runtime_setting_provider", + lambda value: events.update(provider=value), + ) + monkeypatch.setattr( + configuration_composition, + "configure_runtime_setting_updater", + lambda value: events.update(updater=value), + ) + monkeypatch.setattr( + configuration_composition, + "configure_token_runtime_config", + lambda value: events.update(token=value), + ) + + configuration_composition.publish_configuration(composition, legacy_settings) + + assert events["runtime"] is runtime + assert events["settings"] is settings_service + assert events["provider"]("VALUE") == "configured" + assert events["updater"] is settings_service.update + assert callable(events["token"]) + + @pytest.mark.asyncio -async def test_modules_startup_failure_stops_database_worker(monkeypatch) -> None: +async def test_modules_startup_failure_stops_database_runtime(monkeypatch) -> None: """模块组合根启动失败时立即关闭已创建的数据库 worker。""" monkeypatch.setattr( modules_initializer, @@ -203,7 +281,7 @@ async def test_modules_startup_failure_stops_database_worker(monkeypatch) -> Non ) monkeypatch.setattr(modules_initializer, "stop_message", lambda: True) stop_worker = AsyncMock() - monkeypatch.setattr(modules_initializer, "stop_database_worker", stop_worker) + monkeypatch.setattr(modules_initializer, "stop_database_runtime", stop_worker) with pytest.raises(RuntimeError, match="startup failed"): await modules_initializer.init_modules() @@ -225,7 +303,7 @@ async def test_modules_startup_failure_preserves_original_error_when_cleanup_fai monkeypatch.setattr(modules_initializer, "stop_message", lambda: True) monkeypatch.setattr( modules_initializer, - "stop_database_worker", + "stop_database_runtime", AsyncMock(side_effect=RuntimeError("cleanup failed")), ) @@ -247,9 +325,10 @@ async def test_database_worker_owner_is_retained_when_shutdown_fails(monkeypatch raise RuntimeError("shutdown failed") worker = _FailingWorker() - monkeypatch.setattr(modules_initializer, "_database_worker", worker) + runtime = SimpleNamespace(worker=worker) + monkeypatch.setattr(database_composition, "_database_runtime", runtime) with pytest.raises(RuntimeError, match="shutdown failed"): - await modules_initializer.stop_database_worker() + await database_composition.stop_database_runtime() - assert modules_initializer._database_worker is worker + assert database_composition._database_runtime is runtime diff --git a/tests/test_host_runtime_context.py b/tests/test_host_runtime_context.py index 4186242dd..298ee8a3a 100644 --- a/tests/test_host_runtime_context.py +++ b/tests/test_host_runtime_context.py @@ -270,6 +270,7 @@ def test_string_api_data_locator_is_confined_to_compatibility_boundary() -> None assert importers == { "app/api/dependencies/data.py", + "app/startup/composition/database.py", "app/startup/initializers/modules.py", } @@ -314,11 +315,11 @@ async def test_init_modules_cleans_partial_message_owner_on_failure(monkeypatch) "_initialize_modules", failing_initialize_modules, ) - stop_database_worker = AsyncMock() + stop_database_runtime = AsyncMock() monkeypatch.setattr( modules_initializer, - "stop_database_worker", - stop_database_worker, + "stop_database_runtime", + stop_database_runtime, ) with pytest.raises(RuntimeError) as raised: @@ -327,4 +328,4 @@ async def test_init_modules_cleans_partial_message_owner_on_failure(monkeypatch) assert raised.value is startup_error close.assert_called_once_with() assert MessageHelper.get_existing_instance() is None - stop_database_worker.assert_awaited_once_with() + stop_database_runtime.assert_awaited_once_with() diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index ab5412df2..7c2ed097f 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -1240,18 +1240,23 @@ def test_stop_modules_drains_web_agent_tasks_before_persistence(monkeypatch): "get_configured_agent_chat_persistence", MagicMock(return_value=persistence), ) + database_state = {"active": True} - async def stop_database_worker() -> None: + async def stop_database_runtime() -> None: """模拟生产 worker 关闭后释放组合根句柄。""" order.append("database") - modules_initializer._database_worker = None + database_state["active"] = False monkeypatch.setattr( modules_initializer, - "stop_database_worker", - AsyncMock(side_effect=stop_database_worker), + "stop_database_runtime", + AsyncMock(side_effect=stop_database_runtime), + ) + monkeypatch.setattr( + modules_initializer, + "database_runtime_active", + lambda: database_state["active"], ) - monkeypatch.setattr(modules_initializer, "_database_worker", object()) converged = asyncio.run(modules_initializer.stop_modules()) @@ -1260,7 +1265,7 @@ def test_stop_modules_drains_web_agent_tasks_before_persistence(monkeypatch): @pytest.mark.asyncio -async def test_stop_modules_cancellation_does_not_skip_database_worker_cleanup( +async def test_stop_modules_cancellation_does_not_skip_database_runtime_cleanup( monkeypatch, ): """模块关闭收到取消请求后仍应继续收口数据库 worker。""" @@ -1290,9 +1295,19 @@ async def test_stop_modules_cancellation_does_not_skip_database_worker_cleanup( MagicMock(return_value=persistence), ) database_worker_stopped = asyncio.Event() - stop_database_worker = AsyncMock(side_effect=database_worker_stopped.set) - monkeypatch.setattr(modules_initializer, "stop_database_worker", stop_database_worker) - monkeypatch.setattr(modules_initializer, "_database_worker", object()) + database_state = {"active": True} + + def mark_database_worker_stopped() -> None: + """记录取消路径仍执行了 worker 关闭调用。""" + database_worker_stopped.set() + + stop_database_runtime = AsyncMock(side_effect=mark_database_worker_stopped) + monkeypatch.setattr(modules_initializer, "stop_database_runtime", stop_database_runtime) + monkeypatch.setattr( + modules_initializer, + "database_runtime_active", + lambda: database_state["active"], + ) shutdown = asyncio.create_task(modules_initializer.stop_modules()) await started.wait() @@ -1301,7 +1316,7 @@ async def test_stop_modules_cancellation_does_not_skip_database_worker_cleanup( assert completed is False assert database_worker_stopped.is_set() - stop_database_worker.assert_awaited_once_with() + stop_database_runtime.assert_awaited_once_with() @pytest.mark.asyncio