refactor(startup): isolate configuration and database composition

This commit is contained in:
jxxghp
2026-08-29 16:20:04 +08:00
parent a1e351aa9e
commit b4547cf043
18 changed files with 531 additions and 243 deletions

View File

@@ -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:

View File

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

View File

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

View File

@@ -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 重复与测试存量

View File

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

View File

@@ -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`

View File

@@ -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.

View File

@@ -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 负责构造唯一的系统配置数据库适配器",
}

View File

@@ -3,9 +3,9 @@
"count": 1,
"entries": [
{
"file": "app/startup/initializers/modules.py",
"file": "app/startup/composition/configuration.py",
"name": "SystemConfigOper",
"reason": "启动组合根负责构造唯一的系统配置数据库适配器"
"reason": "配置组合 owner 负责构造唯一的系统配置数据库适配器"
}
]
},

View File

@@ -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",

View File

@@ -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
},

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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