From 820582ab120b65d09183b6c3bead5962a012cd4c Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 23 Aug 2026 21:41:41 +0800 Subject: [PATCH] refactor: close plugin shutdown mutation races --- .../tools/impl/update_system_settings.py | 166 +++++------ app/api/endpoints/system.py | 26 +- app/application/plugin/install.py | 22 +- app/application/plugin/runtime.py | 28 ++ app/runtime/extensions/plugin/admission.py | 34 +-- app/schemas/exports.py | 1 + .../backend-architecture-next-stage.md | 4 +- .../architecture/dependency-baseline.json | 15 +- .../startup-performance-baseline.json | 234 ++++++++-------- tests/test_lifecycle_shutdown.py | 262 +++++++---------- tests/test_plugin_install_command.py | 97 +++++++ tests/test_plugin_lifecycle_status.py | 53 +++- tests/test_plugin_monitor_lifecycle.py | 17 +- tests/test_plugin_mutation_admission.py | 79 +++++- tests/test_plugin_system_setting_admission.py | 263 ++++++++++++++++++ tests/test_transfer_worker_lifecycle.py | 2 + 16 files changed, 904 insertions(+), 399 deletions(-) create mode 100644 tests/test_plugin_system_setting_admission.py diff --git a/app/agent/tools/impl/update_system_settings.py b/app/agent/tools/impl/update_system_settings.py index 2a3e02d65..f51c03af4 100644 --- a/app/agent/tools/impl/update_system_settings.py +++ b/app/agent/tools/impl/update_system_settings.py @@ -22,6 +22,7 @@ from app.application.configuration import ( get_configured_system_config as SystemConfigOper, get_runtime_settings, ) +from app.application.plugin.runtime import plugin_system_config_mutation from app.runtime.log import logger from app.schemas.event import ConfigChangeEventData from app.schemas.types import EventType @@ -259,94 +260,101 @@ class UpdateSystemSettingsTool(MoviePilotTool): ensure_ascii=False, ) - current_value = self._load_setting_value(spec) - next_value = self._prepare_next_value( - spec=spec, - current_value=current_value, - value=value, - operation=operation, - remove_keys=remove_keys, - match_field=match_field, - match_value=match_value, + mutation_key = ( + spec.systemconfig_key if spec.source == "systemconfig" else None ) + with plugin_system_config_mutation(mutation_key): + current_value = self._load_setting_value(spec) + next_value = self._prepare_next_value( + spec=spec, + current_value=current_value, + value=value, + operation=operation, + remove_keys=remove_keys, + match_field=match_field, + match_value=match_value, + ) - event_value = next_value - changed = False - message = "" - if spec.source == "settings": - success, message = get_runtime_settings().update(spec.key, next_value) - if success is False: - return json.dumps( - { - "success": False, - "message": message or f"更新设置 {spec.key} 失败", - }, - ensure_ascii=False, + event_value = next_value + changed = False + message = "" + if spec.source == "settings": + success, message = get_runtime_settings().update( + spec.key, + next_value, ) - changed = success is True - else: - normalized_value = self._normalize_systemconfig_value(next_value) - event_value = normalized_value - success = await self._get_system_config().async_set( - spec.systemconfig_key, - normalized_value, - ) - changed = success is True + if success is False: + return json.dumps( + { + "success": False, + "message": message or f"更新设置 {spec.key} 失败", + }, + ensure_ascii=False, + ) + changed = success is True + else: + normalized_value = self._normalize_systemconfig_value(next_value) + event_value = normalized_value + success = await self._get_system_config().async_set( + spec.systemconfig_key, + normalized_value, + ) + changed = success is True - if changed: - await eventmanager.async_send_event( - etype=EventType.ConfigChanged, - data=ConfigChangeEventData( - key=spec.key, - value=event_value, - change_type="update", - ), - ) + if changed: + await eventmanager.async_send_event( + etype=EventType.ConfigChanged, + data=ConfigChangeEventData( + key=spec.key, + value=event_value, + change_type="update", + ), + ) - saved_value = self._load_setting_value(spec) - redact_values = ( - should_redact_setting(spec, saved_value) - or should_redact_setting(spec, current_value) - ) - response_previous_value = ( - redact_secret_value( - current_value, - redact_scalar=is_secret_setting_key(spec.key), + saved_value = self._load_setting_value(spec) + redact_values = ( + should_redact_setting(spec, saved_value) + or should_redact_setting(spec, current_value) ) - if redact_values - else current_value - ) - response_saved_value = ( - redact_secret_value( - saved_value, - redact_scalar=is_secret_setting_key(spec.key), + response_previous_value = ( + redact_secret_value( + current_value, + redact_scalar=is_secret_setting_key(spec.key), + ) + if redact_values + else current_value ) - if redact_values - else saved_value - ) - if not changed and not message: - message = "配置值未发生变化" + response_saved_value = ( + redact_secret_value( + saved_value, + redact_scalar=is_secret_setting_key(spec.key), + ) + if redact_values + else saved_value + ) + if not changed and not message: + message = "配置值未发生变化" - return json.dumps( - { - "success": True, - "message": message or f"系统设置 {spec.key} 已更新", - "changed": changed, - "operation": operation, - "setting": { - "setting_key": spec.key, - "source": spec.source, - "group": spec.group, - "label": spec.label, + return json.dumps( + { + "success": True, + "message": message or f"系统设置 {spec.key} 已更新", + "changed": changed, + "operation": operation, + "setting": { + "setting_key": spec.key, + "source": spec.source, + "group": spec.group, + "label": spec.label, + }, + "values_redacted": redact_values, + "previous_value": response_previous_value, + "saved_value": response_saved_value, }, - "values_redacted": redact_values, - "previous_value": response_previous_value, - "saved_value": response_saved_value, - }, - ensure_ascii=False, - indent=2, - default=str, - ) + ensure_ascii=False, + indent=2, + default=str, + ) except Exception as e: logger.error(f"更新系统设置失败: {e}", exc_info=True) return json.dumps( diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index 4f07a9656..197e08ddd 100644 --- a/app/api/endpoints/system.py +++ b/app/api/endpoints/system.py @@ -45,6 +45,7 @@ from app.application.configuration import ( get_configured_system_config, get_runtime_settings, ) +from app.application.plugin.runtime import plugin_system_config_mutation from app.api.dependencies.auth import ( get_current_active_superuser, get_current_active_superuser_async, @@ -67,6 +68,7 @@ from app.runtime.state import SystemHelper from app.runtime.log import logger from app.application.scheduling import Scheduler from app.schemas.event import ConfigChangeEventData +from app.schemas.exception import PluginMutationRejectedError from app.schemas.types import SystemConfigKey, EventType from app.foundation.crypto import HashUtils from app.adapters.network.http import RequestUtils, AsyncRequestUtils @@ -1070,14 +1072,22 @@ async def set_setting( if isinstance(value, list): value = list(filter(None, value)) value = value if value else None - success = await get_configured_system_config().async_set(key, value) - if success: - # 发送配置变更事件 - await eventmanager.async_send_event( - etype=EventType.ConfigChanged, - data=ConfigChangeEventData(key=key, value=value, change_type="update"), - ) - return _SchemaResponse(success=True) + try: + with plugin_system_config_mutation(key): + success = await get_configured_system_config().async_set(key, value) + if success: + # 发送配置变更事件 + await eventmanager.async_send_event( + etype=EventType.ConfigChanged, + data=ConfigChangeEventData( + key=key, + value=value, + change_type="update", + ), + ) + return _SchemaResponse(success=True) + except PluginMutationRejectedError as error: + return _SchemaResponse(success=False, message=str(error)) else: return _SchemaResponse(success=False, message=f"配置项 '{key}' 不存在") diff --git a/app/application/plugin/install.py b/app/application/plugin/install.py index 56f7fd18e..5379f0ef3 100644 --- a/app/application/plugin/install.py +++ b/app/application/plugin/install.py @@ -30,6 +30,16 @@ PluginMutationAdmission = Callable[[str], ContextManager[None]] PluginPackageWriteGuard = Callable[[str], ContextManager[None]] +async def _await_task_to_terminal(task: asyncio.Task[Any]) -> Any: + """忽略调用方的重复取消，直到受保护子任务进入真实终态。""" + while not task.done(): + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + continue + return task.result() + + @dataclass(frozen=True, slots=True) class PluginInstallRollback: """描述失败安装中各类可补偿副作用的恢复结果。""" @@ -183,7 +193,7 @@ class PluginInstallCommand: state.checkpoint = checkpoint except asyncio.CancelledError: try: - state.checkpoint = await asyncio.shield(checkpoint_task) + state.checkpoint = await _await_task_to_terminal(checkpoint_task) except BaseException: pass raise @@ -359,14 +369,8 @@ class PluginInstallCommand: ) ) try: - result = await asyncio.shield(rollback_task) - except asyncio.CancelledError: - try: - result = await asyncio.shield(rollback_task) - except BaseException as err: - logger.error(f"插件 {plugin_id} 取消后的补偿未完成：{err}") - return - except Exception as err: + result = await _await_task_to_terminal(rollback_task) + except BaseException as err: logger.error(f"插件 {plugin_id} 取消后的补偿失败：{err}") return if result.rollback.errors: diff --git a/app/application/plugin/runtime.py b/app/application/plugin/runtime.py index eecf9f854..fe8bf9c8c 100644 --- a/app/application/plugin/runtime.py +++ b/app/application/plugin/runtime.py @@ -3,8 +3,21 @@ from __future__ import annotations from collections.abc import Callable +from contextlib import nullcontext from typing import Any, ContextManager, Protocol +from app.schemas.types import SystemConfigKey + + +PLUGIN_MUTATION_SYSTEM_CONFIG_KEYS = frozenset( + { + SystemConfigKey.UserInstalledPlugins, + SystemConfigKey.PluginInstances, + SystemConfigKey.PluginFolders, + } +) + + class PluginRuntime(Protocol): """声明入口层消费的插件宿主能力。""" @@ -36,3 +49,18 @@ def configure_plugin_runtime(provider: PluginRuntimeProvider) -> None: def get_plugin_manager() -> PluginRuntime: """返回当前组合根提供的插件运行时能力。""" return _runtime_provider() + + +def plugin_system_config_mutation( + key: str | SystemConfigKey | None, +) -> ContextManager[None]: + """仅为会改变插件运行态所有权的系统配置取得 mutation lease。""" + if key is None: + return nullcontext() + try: + normalized_key = key if isinstance(key, SystemConfigKey) else SystemConfigKey(key) + except ValueError: + return nullcontext() + if normalized_key not in PLUGIN_MUTATION_SYSTEM_CONFIG_KEYS: + return nullcontext() + return get_plugin_manager().mutation(f"更新插件系统配置 {normalized_key.value}") diff --git a/app/runtime/extensions/plugin/admission.py b/app/runtime/extensions/plugin/admission.py index 6656624f4..d0f24538f 100644 --- a/app/runtime/extensions/plugin/admission.py +++ b/app/runtime/extensions/plugin/admission.py @@ -48,34 +48,34 @@ class PluginMutationAdmission: def is_held(self) -> bool: """判断当前执行上下文是否持有仍有效的本 admission lease。""" context = self._current_context.get() - return bool( - context - and context.admission is self - and context.open - and context.holders > 0 - ) + with self._condition: + return bool( + context + and context.admission is self + and context.open + and context.holders > 0 + ) @contextmanager def hold(self, operation: str) -> Iterator[None]: """取得可变事务 lease；封口后仅允许已获准事务的嵌套调用。""" context = self._current_context.get() - nested = bool( - context - and context.admission is self - and context.open - and context.holders > 0 - ) context_token = None - if not nested: - context = _MutationContext(admission=self) - context_token = self._current_context.set(context) - assert context is not None - acquired = False try: with self._condition: + nested = bool( + context + and context.admission is self + and context.open + and context.holders > 0 + ) if not self._accepting and not nested: raise PluginMutationRejectedError(operation) + if not nested: + context = _MutationContext(admission=self) + context_token = self._current_context.set(context) + assert context is not None self._active_count += 1 context.holders += 1 acquired = True diff --git a/app/schemas/exports.py b/app/schemas/exports.py index 25e8d04f1..32d95f09f 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -270,6 +270,7 @@ SCHEMA_EXPORTS = { 'PluginMarketSyncData': ('app.schemas.system', 'PluginMarketSyncData'), 'PluginMarketSyncRequest': ('app.schemas.system', 'PluginMarketSyncRequest'), 'PluginMemoryInfo': ('app.schemas.plugin', 'PluginMemoryInfo'), + 'PluginMutationRejectedError': ('app.schemas.exception', 'PluginMutationRejectedError'), 'PluginRating': ('app.schemas.plugin', 'PluginRating'), 'PluginRatingMap': ('app.schemas.plugin', 'PluginRatingMap'), 'PluginRatingRequest': ('app.schemas.plugin', 'PluginRatingRequest'), diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 8fb3314ed..76a54974e 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -15,7 +15,7 @@ ### 长期整改阶段 0：治理门禁恢复（2026-08-23） -- 宿主依赖基线已审查本阶段归位后的语义差异：当前为 `805` 个模块、`6514` 条内部导入边，12 组重点禁止边继续全部为 `0`，唯一非平凡 SCC 仍是隔离的 TMDB 移植包。 +- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异：当前为 `805` 个模块、`6525` 条内部导入边，12 组重点禁止边继续全部为 `0`，唯一非平凡 SCC 仍是隔离的 TMDB 移植包。 - 启动性能探针会在隔离生命周期中真实创建并释放 TaskRegistry；normal/safe 组件数分别为 `23`/`11`，CI 只读检查使用稳定的宿主模块集合和生命周期组件顺序，不再把 Python/平台模块数量当作硬合同。 - 官方插件快照覆盖 `plugins.v3`、`plugins.v2` 以及 V3 实际会从 `package.json` 回退加载的 31 个默认实现；`app/plugins/**` 仍只是宿主运行副本，不进入扫描。 - SDK 快照以各模块显式 `__all__` 为公开合同，能够记录赋值别名；`typing`、`__future__` 等实现期导入不再被误冻结，既有数据库备份门面已补精确导出清单。 @@ -72,7 +72,7 @@ - 继续采用单进程控制面是正确选择，不建议现在拆成微服务；插件、调度器、工作流、事件和数据库共享进程内状态，拆分会放大部署、事务和兼容成本。 - `foundation/domain/runtime/adapters/application/chain/api/startup` 的职责方向基本成立；宿主架构基线、复杂度 ratchet、异步阻塞 ratchet 当前均通过。 -- 依赖图当前为 `805` 个 Python 模块、`6514` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。 +- 依赖图当前为 `805` 个 Python 模块、`6525` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。 - 当前主要风险已经从“目录和依赖失控”转移到运行时协议、后台副作用的可靠性和遗留兼容面。换言之，下一阶段重点应是**语义收口和可验证性**，而不是继续搬文件或机械拆大文件。 综合评价：架构方向可持续，生产可用性较高；可演进性仍处于中等水平。现阶段没有静态审计发现必须立即推倒重来的 P0 架构问题，但存在需要按 P1/P2 计划治理的真实债务。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index c009c1f58..c341aadd5 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6514, - "edge_sha256": "eda9bc500158baa80dc5013eb4e409aa57cb70446e59a52b7e782cb9c58f9c34", + "edge_count": 6525, + "edge_sha256": "72b9416b5309bb51768b95b3c9da4b167242c3bc3f27d4e561a3f097267ca6a0", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -281,6 +281,7 @@ "app.agent.middleware.skills -> app.agent.middleware", "app.agent.middleware.skills -> app.agent.middleware.utils", "app.agent.middleware.skills -> app.agent.policy", + "app.agent.middleware.skills -> app.agent.policy.sanitizer", "app.agent.middleware.skills -> app.agent.skills", "app.agent.middleware.skills -> app.agent.skills.metadata", "app.agent.middleware.skills -> app.agent.tools", @@ -309,6 +310,7 @@ "app.agent.middleware.summarization -> app.runtime.log", "app.agent.middleware.tool_selection -> app.agent", "app.agent.middleware.tool_selection -> app.agent.llm", + "app.agent.middleware.tool_selection -> app.agent.llm.helper", "app.agent.middleware.tool_selection -> app.agent.tools", "app.agent.middleware.tool_selection -> app.agent.tools.tags", "app.agent.middleware.tool_selection -> app.runtime", @@ -435,6 +437,7 @@ "app.agent.tools.base -> app.schemas.types", "app.agent.tools.catalog -> app.agent", "app.agent.tools.catalog -> app.agent.policy", + "app.agent.tools.catalog -> app.agent.policy.contracts", "app.agent.tools.factory -> app.agent", "app.agent.tools.factory -> app.agent.llm", "app.agent.tools.factory -> app.agent.llm.capability", @@ -883,6 +886,7 @@ "app.agent.tools.impl.query_doctor_report -> app.agent.tools.base", "app.agent.tools.impl.query_doctor_report -> app.agent.tools.tags", "app.agent.tools.impl.query_doctor_report -> app.doctor", + "app.agent.tools.impl.query_doctor_report -> app.doctor.runner", "app.agent.tools.impl.query_doctor_report -> app.runtime", "app.agent.tools.impl.query_doctor_report -> app.runtime.log", "app.agent.tools.impl.query_download_tasks -> app.agent", @@ -1471,6 +1475,8 @@ "app.agent.tools.impl.update_system_settings -> app.agent.tools.tags", "app.agent.tools.impl.update_system_settings -> app.application", "app.agent.tools.impl.update_system_settings -> app.application.configuration", + "app.agent.tools.impl.update_system_settings -> app.application.plugin", + "app.agent.tools.impl.update_system_settings -> app.application.plugin.runtime", "app.agent.tools.impl.update_system_settings -> app.runtime", "app.agent.tools.impl.update_system_settings -> app.runtime.events", "app.agent.tools.impl.update_system_settings -> app.runtime.log", @@ -2261,6 +2267,8 @@ "app.api.endpoints.system -> app.application.messaging.message", "app.api.endpoints.system -> app.application.module", "app.api.endpoints.system -> app.application.network", + "app.api.endpoints.system -> app.application.plugin", + "app.api.endpoints.system -> app.application.plugin.runtime", "app.api.endpoints.system -> app.application.rules", "app.api.endpoints.system -> app.application.scheduling", "app.api.endpoints.system -> app.application.security", @@ -2287,6 +2295,7 @@ "app.api.endpoints.system -> app.schemas", "app.api.endpoints.system -> app.schemas.common", "app.api.endpoints.system -> app.schemas.event", + "app.api.endpoints.system -> app.schemas.exception", "app.api.endpoints.system -> app.schemas.response", "app.api.endpoints.system -> app.schemas.system", "app.api.endpoints.system -> app.schemas.token", @@ -2688,6 +2697,8 @@ "app.application.plugin.install -> app.runtime.log", "app.application.plugin.install -> app.schemas", "app.application.plugin.install -> app.schemas.exception", + "app.application.plugin.runtime -> app.schemas", + "app.application.plugin.runtime -> app.schemas.types", "app.application.recognition -> app.application", "app.application.recognition -> app.application.configuration", "app.application.recognition -> app.schemas", diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index d0d0831b8..643acc39b 100644 --- a/tests/fixtures/architecture/startup-performance-baseline.json +++ b/tests/fixtures/architecture/startup-performance-baseline.json @@ -1,41 +1,41 @@ { "schema_version": 2, - "generated_at": "2026-08-23T11:06:42.858188+00:00", + "generated_at": "2026-08-23T13:47:08.442528+00:00", "platform": "macOS-26.5.2-arm64-arm-64bit-Mach-O", "python": "3.14.3", "repeat": 3, "targets": { "app.startup.lifecycle": { - "loaded_app_module_count": 362, - "max_ms": 1022.231, - "median_ms": 976.39, - "min_ms": 967.395, + "loaded_app_module_count": 367, + "max_ms": 1591.175, + "median_ms": 1049.37, + "min_ms": 1022.351, "samples_ms": [ - 1022.231, - 967.395, - 976.39 + 1591.175, + 1049.37, + 1022.351 ] }, "app.factory": { - "loaded_app_module_count": 374, - "max_ms": 989.21, - "median_ms": 980.605, - "min_ms": 951.564, + "loaded_app_module_count": 379, + "max_ms": 1198.166, + "median_ms": 1083.805, + "min_ms": 1009.309, "samples_ms": [ - 980.605, - 989.21, - 951.564 + 1009.309, + 1198.166, + 1083.805 ] }, "app.main": { - "loaded_app_module_count": 376, - "max_ms": 1114.04, - "median_ms": 1088.251, - "min_ms": 1080.26, + "loaded_app_module_count": 381, + "max_ms": 1229.477, + "median_ms": 1179.859, + "min_ms": 1129.055, "samples_ms": [ - 1080.26, - 1088.251, - 1114.04 + 1129.055, + 1179.859, + 1229.477 ] } }, @@ -46,26 +46,26 @@ "samples": [ { "mode": "normal", - "enabled_component_count": 21, - "startup_ms": 0.619, - "full_lifespan_ms": 0.806, + "enabled_component_count": 23, + "startup_ms": 0.64, + "full_lifespan_ms": 0.805, "stage_ms": { - "后台任务登记器": 0.075, - "数据库准备": 0.037, - "HTTP 基础能力": 0.031, + "后台任务登记器": 0.079, + "数据库准备": 0.039, + "HTTP 基础能力": 0.029, "领域依赖装配": 0.027, "数据库引擎预热": 0.024, - "数据库连接预算": 0.025, - "路由": 0.022, - "模块服务": 0.024, - "插件备份恢复": 0.021, - "插件": 0.02, - "定时器": 0.023, - "监控器": 0.021, - "待处理整理回放": 0.019, - "命令服务": 0.022, - "工作流": 0.022, - "插件同步与启动收尾": 0.034 + "数据库连接预算": 0.023, + "路由": 0.023, + "模块服务": 0.021, + "插件备份恢复": 0.024, + "插件": 0.021, + "定时器": 0.026, + "监控器": 0.022, + "待处理整理回放": 0.025, + "命令服务": 0.025, + "工作流": 0.021, + "插件同步与启动收尾": 0.037 }, "threads_before": 2, "threads_started": 2, @@ -77,26 +77,26 @@ }, { "mode": "normal", - "enabled_component_count": 21, - "startup_ms": 0.628, + "enabled_component_count": 23, + "startup_ms": 0.652, "full_lifespan_ms": 0.808, "stage_ms": { - "后台任务登记器": 0.073, - "数据库准备": 0.037, - "HTTP 基础能力": 0.039, - "领域依赖装配": 0.027, - "数据库引擎预热": 0.025, - "数据库连接预算": 0.025, + "后台任务登记器": 0.079, + "数据库准备": 0.039, + "HTTP 基础能力": 0.033, + "领域依赖装配": 0.029, + "数据库引擎预热": 0.026, + "数据库连接预算": 0.021, "路由": 0.022, - "模块服务": 0.022, + "模块服务": 0.023, "插件备份恢复": 0.021, - "插件": 0.02, - "定时器": 0.024, - "监控器": 0.022, - "待处理整理回放": 0.029, - "命令服务": 0.02, - "工作流": 0.021, - "插件同步与启动收尾": 0.032 + "插件": 0.022, + "定时器": 0.026, + "监控器": 0.023, + "待处理整理回放": 0.024, + "命令服务": 0.023, + "工作流": 0.025, + "插件同步与启动收尾": 0.035 }, "threads_before": 2, "threads_started": 2, @@ -108,26 +108,26 @@ }, { "mode": "normal", - "enabled_component_count": 21, - "startup_ms": 0.664, - "full_lifespan_ms": 0.863, + "enabled_component_count": 23, + "startup_ms": 0.739, + "full_lifespan_ms": 0.944, "stage_ms": { - "后台任务登记器": 0.083, - "数据库准备": 0.047, - "HTTP 基础能力": 0.036, + "后台任务登记器": 0.087, + "数据库准备": 0.037, + "HTTP 基础能力": 0.029, "领域依赖装配": 0.028, - "数据库引擎预热": 0.029, - "数据库连接预算": 0.025, - "路由": 0.026, - "模块服务": 0.025, - "插件备份恢复": 0.025, - "插件": 0.028, - "定时器": 0.026, - "监控器": 0.021, - "待处理整理回放": 0.021, - "命令服务": 0.024, - "工作流": 0.021, - "插件同步与启动收尾": 0.036 + "数据库引擎预热": 0.024, + "数据库连接预算": 0.021, + "路由": 0.02, + "模块服务": 0.02, + "插件备份恢复": 0.024, + "插件": 0.019, + "定时器": 0.025, + "监控器": 0.054, + "待处理整理回放": 0.041, + "命令服务": 0.034, + "工作流": 0.025, + "插件同步与启动收尾": 0.037 }, "threads_before": 2, "threads_started": 2, @@ -138,9 +138,9 @@ "database_connections_started": 0 } ], - "median_startup_ms": 0.628, + "median_startup_ms": 0.652, "median_full_lifespan_ms": 0.808, - "enabled_component_count": 21, + "enabled_component_count": 23, "enabled_components": [ "后台任务登记器", "数据库准备", @@ -157,6 +157,8 @@ "监控器", "整理后台服务", "AI智能体会话", + "插件事件入口", + "事件尾任务结算", "插件后台服务", "事件投递屏障", "待处理整理回放", @@ -170,42 +172,18 @@ { "mode": "safe", "enabled_component_count": 11, - "startup_ms": 0.464, - "full_lifespan_ms": 0.637, + "startup_ms": 0.475, + "full_lifespan_ms": 0.619, "stage_ms": { - "后台任务登记器": 0.074, + "后台任务登记器": 0.079, "数据库准备": 0.038, "HTTP 基础能力": 0.03, - "领域依赖装配": 0.029, - "数据库引擎预热": 0.026, - "数据库连接预算": 0.025, - "路由": 0.024, - "模块服务": 0.024, - "插件同步与启动收尾": 0.033 - }, - "threads_before": 2, - "threads_started": 2, - "threads_after": 2, - "tasks_before": 1, - "tasks_started": 2, - "tasks_after": 1, - "database_connections_started": 0 - }, - { - "mode": "safe", - "enabled_component_count": 11, - "startup_ms": 0.451, - "full_lifespan_ms": 0.624, - "stage_ms": { - "后台任务登记器": 0.073, - "数据库准备": 0.035, - "HTTP 基础能力": 0.028, - "领域依赖装配": 0.027, + "领域依赖装配": 0.028, "数据库引擎预热": 0.024, "数据库连接预算": 0.023, "路由": 0.023, - "模块服务": 0.02, - "插件同步与启动收尾": 0.031 + "模块服务": 0.022, + "插件同步与启动收尾": 0.036 }, "threads_before": 2, "threads_started": 2, @@ -218,18 +196,42 @@ { "mode": "safe", "enabled_component_count": 11, - "startup_ms": 0.477, - "full_lifespan_ms": 0.661, + "startup_ms": 0.493, + "full_lifespan_ms": 0.676, "stage_ms": { - "后台任务登记器": 0.081, - "数据库准备": 0.05, + "后台任务登记器": 0.089, + "数据库准备": 0.037, "HTTP 基础能力": 0.035, - "领域依赖装配": 0.03, - "数据库引擎预热": 0.026, - "数据库连接预算": 0.024, + "领域依赖装配": 0.028, + "数据库引擎预热": 0.025, + "数据库连接预算": 0.021, + "路由": 0.025, + "模块服务": 0.023, + "插件同步与启动收尾": 0.062 + }, + "threads_before": 2, + "threads_started": 2, + "threads_after": 2, + "tasks_before": 1, + "tasks_started": 2, + "tasks_after": 1, + "database_connections_started": 0 + }, + { + "mode": "safe", + "enabled_component_count": 11, + "startup_ms": 0.464, + "full_lifespan_ms": 0.638, + "stage_ms": { + "后台任务登记器": 0.075, + "数据库准备": 0.036, + "HTTP 基础能力": 0.028, + "领域依赖装配": 0.028, + "数据库引擎预热": 0.023, + "数据库连接预算": 0.022, "路由": 0.024, - "模块服务": 0.02, - "插件同步与启动收尾": 0.026 + "模块服务": 0.021, + "插件同步与启动收尾": 0.057 }, "threads_before": 2, "threads_started": 2, @@ -240,8 +242,8 @@ "database_connections_started": 0 } ], - "median_startup_ms": 0.464, - "median_full_lifespan_ms": 0.637, + "median_startup_ms": 0.475, + "median_full_lifespan_ms": 0.638, "enabled_component_count": 11, "enabled_components": [ "后台任务登记器", diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 9b74200a0..952416fe5 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -69,6 +69,8 @@ def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict: "stop_agent": AsyncMock(return_value=True), "stop_transfer": AsyncMock(return_value=True), "quiesce_plugins": AsyncMock(return_value=True), + "settle_events": AsyncMock(return_value=True), + "quiesce_plugin_services": AsyncMock(return_value=True), "drain_events": AsyncMock(return_value=True), "finalize_plugins": MagicMock(return_value=True), "stop_modules": AsyncMock(), @@ -94,6 +96,16 @@ def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict: "quiesce_plugins", shutdown_steps["quiesce_plugins"], ) + monkeypatch.setattr( + lifecycle, + "settle_events", + shutdown_steps["settle_events"], + ) + monkeypatch.setattr( + lifecycle, + "quiesce_plugin_services", + shutdown_steps["quiesce_plugin_services"], + ) monkeypatch.setattr(lifecycle, "drain_events", shutdown_steps["drain_events"]) monkeypatch.setattr(lifecycle, "stop_modules", shutdown_steps["stop_modules"]) monkeypatch.setattr( @@ -191,166 +203,83 @@ def test_lifespan_validation_failure_does_not_clear_outer_loop_owner(monkeypatch lifecycle.global_vars.clear_loop.assert_not_called() +def test_lifespan_settles_plugin_handlers_before_legacy_hooks(monkeypatch) -> None: + """整理尾事件必须在 handler 停用后结算，并先于旧插件停机 hook。""" + shutdown_steps = _patch_lifespan(monkeypatch) + order: list[str] = [] + for name in ( + "stop_transfer", + "quiesce_plugins", + "settle_events", + "quiesce_plugin_services", + "drain_events", + "finalize_plugins", + ): + shutdown_steps[name].side_effect = ( + lambda current=name: order.append(current) or True + ) + + async def run_lifespan() -> None: + """运行一个完整的隔离生命周期。""" + async with lifecycle.lifespan(FastAPI()): + pass + + asyncio.run(run_lifespan()) + + assert order == [ + "stop_transfer", + "quiesce_plugins", + "settle_events", + "quiesce_plugin_services", + "drain_events", + "finalize_plugins", + ] + + +_ORDERED_SHUTDOWN_STEPS = ( + "stop_plugin_monitor", + "backup_plugins", + "stop_workflow", + "stop_command", + "stop_monitor", + "stop_scheduler", + "stop_agent", + "stop_transfer", + "quiesce_plugins", + "settle_events", + "quiesce_plugin_services", + "drain_events", + "finalize_plugins", + "stop_modules", + "close_http", +) + + @pytest.mark.parametrize( - ("failing_step", "completed_steps", "blocked_steps"), - [ - ( - "stop_plugin_monitor", - ("stop_plugin_monitor",), - ( - "backup_plugins", - "stop_workflow", - "stop_command", - "stop_monitor", - "stop_scheduler", - "stop_agent", - "quiesce_plugins", - "stop_transfer", - "drain_events", - "finalize_plugins", - "stop_modules", - "close_http", - ), - ), - ( - "stop_monitor", - ( - "stop_plugin_monitor", - "backup_plugins", - "stop_workflow", - "stop_command", - "stop_monitor", - ), - ( - "stop_scheduler", - "stop_agent", - "quiesce_plugins", - "stop_transfer", - "drain_events", - "finalize_plugins", - "stop_modules", - "close_http", - ), - ), - ( - "stop_scheduler", - ( - "stop_plugin_monitor", - "backup_plugins", - "stop_workflow", - "stop_command", - "stop_monitor", - "stop_scheduler", - ), - ( - "stop_agent", - "quiesce_plugins", - "stop_transfer", - "drain_events", - "finalize_plugins", - "stop_modules", - "close_http", - ), - ), - ( - "stop_agent", - ( - "stop_plugin_monitor", - "backup_plugins", - "stop_workflow", - "stop_command", - "stop_monitor", - "stop_scheduler", - "stop_agent", - ), - ( - "quiesce_plugins", - "stop_transfer", - "drain_events", - "finalize_plugins", - "stop_modules", - "close_http", - ), - ), - ( - "quiesce_plugins", - ( - "stop_plugin_monitor", - "backup_plugins", - "stop_workflow", - "stop_command", - "stop_monitor", - "stop_scheduler", - "stop_agent", - "quiesce_plugins", - ), - ( - "stop_transfer", - "drain_events", - "finalize_plugins", - "stop_modules", - "close_http", - ), - ), - ( - "stop_transfer", - ( - "stop_plugin_monitor", - "backup_plugins", - "stop_workflow", - "stop_command", - "stop_monitor", - "stop_scheduler", - "stop_agent", - "quiesce_plugins", - "stop_transfer", - ), - ("drain_events", "finalize_plugins", "stop_modules", "close_http"), - ), - ( - "drain_events", - ( - "stop_plugin_monitor", - "backup_plugins", - "stop_workflow", - "stop_command", - "stop_monitor", - "stop_scheduler", - "stop_agent", - "quiesce_plugins", - "stop_transfer", - "drain_events", - ), - ("finalize_plugins", "stop_modules", "close_http"), - ), - ( - "finalize_plugins", - ( - "stop_plugin_monitor", - "backup_plugins", - "stop_workflow", - "stop_command", - "stop_monitor", - "stop_scheduler", - "stop_agent", - "quiesce_plugins", - "stop_transfer", - "drain_events", - "finalize_plugins", - ), - ("stop_modules", "close_http"), - ), - ], + "failing_step", + ( + "stop_plugin_monitor", + "stop_monitor", + "stop_scheduler", + "stop_agent", + "stop_transfer", + "quiesce_plugins", + "settle_events", + "quiesce_plugin_services", + "drain_events", + "finalize_plugins", + ), ) def test_lifespan_stops_releasing_dependencies_when_owner_does_not_converge( monkeypatch, failing_step, - completed_steps, - blocked_steps, ): """关键 owner 未收敛时不得关闭仍被活任务使用的后续依赖。""" shutdown_steps = _patch_lifespan(monkeypatch) shutdown_steps[failing_step].return_value = False + failed_index = _ORDERED_SHUTDOWN_STEPS.index(failing_step) + completed_steps = _ORDERED_SHUTDOWN_STEPS[: failed_index + 1] + blocked_steps = _ORDERED_SHUTDOWN_STEPS[failed_index + 1 :] async def run_lifespan(): """启动并关闭隔离后的应用生命周期。""" @@ -520,6 +449,8 @@ def test_lifespan_safe_mode_skips_optional_runtime(monkeypatch): "stop_scheduler", "stop_plugin_monitor", "quiesce_plugins", + "settle_events", + "quiesce_plugin_services", "finalize_plugins", ): shutdown_steps[name].assert_not_called() @@ -543,6 +474,19 @@ async def test_event_drain_does_not_materialize_manager(monkeypatch) -> None: event_manager_type.assert_not_called() +@pytest.mark.asyncio +async def test_event_settlement_keeps_tail_event_admission_open(monkeypatch) -> None: + """中间结算只等待在途 handler，不得提前封死旧 hook 的尾事件。""" + event_manager = MagicMock() + event_manager.drain_async = AsyncMock(return_value=True) + event_manager_type = MagicMock() + event_manager_type.get_existing_instance.return_value = event_manager + monkeypatch.setattr(modules_initializer, "EventManager", event_manager_type) + + assert await modules_initializer.settle_events() is True + event_manager.drain_async.assert_awaited_once_with(seal=False) + + def test_lifecycle_manifest_declares_normal_and_safe_mode_order() -> None: """组件清单应显式冻结依赖、模式、启动/关闭顺序和超时预算。""" app = FastAPI() @@ -591,8 +535,10 @@ def test_lifecycle_manifest_declares_normal_and_safe_mode_order() -> None: "监控器", "定时器", "AI智能体会话", - "插件后台服务", "整理后台服务", + "插件事件入口", + "事件尾任务结算", + "插件后台服务", "事件投递屏障", "插件", "模块服务", @@ -623,6 +569,8 @@ def test_lifecycle_manifest_declares_normal_and_safe_mode_order() -> None: "定时器", "AI智能体会话", "整理后台服务", + "插件事件入口", + "事件尾任务结算", "插件后台服务", "事件投递屏障", "插件", @@ -638,6 +586,8 @@ def test_lifecycle_manifest_declares_normal_and_safe_mode_order() -> None: "定时器", "AI智能体会话", "整理后台服务", + "插件事件入口", + "事件尾任务结算", "插件后台服务", "事件投递屏障", "插件", @@ -869,8 +819,10 @@ def test_lifespan_cleans_started_owners_after_late_startup_failure(monkeypatch): "stop_monitor", "stop_scheduler", "stop_agent", - "quiesce_plugins", "stop_transfer", + "quiesce_plugins", + "settle_events", + "quiesce_plugin_services", "drain_events", "finalize_plugins", "stop_modules", @@ -905,11 +857,13 @@ def test_startup_failure_cleanup_honors_transfer_fail_fast(monkeypatch): "stop_monitor", "stop_scheduler", "stop_agent", - "quiesce_plugins", "stop_transfer", ): _assert_completed_once(shutdown_steps[name]) for name in ( + "quiesce_plugins", + "settle_events", + "quiesce_plugin_services", "drain_events", "finalize_plugins", "stop_modules", diff --git a/tests/test_plugin_install_command.py b/tests/test_plugin_install_command.py index 632298dc6..3218ef034 100644 --- a/tests/test_plugin_install_command.py +++ b/tests/test_plugin_install_command.py @@ -477,6 +477,103 @@ async def test_cancelled_install_waits_for_rollback_before_releasing_lifecycle() rollback.assert_awaited_once() +@pytest.mark.asyncio +async def test_repeated_checkpoint_cancellation_retains_mutation_owner() -> None: + """快照等待被连续取消时，lease 必须保留到快照子任务终态。""" + admission = PluginMutationAdmission() + checkpoint_started = asyncio.Event() + checkpoint_release = asyncio.Event() + checkpoint_finished = asyncio.Event() + + async def checkpoint(_plugin_id: str) -> object: + """阻塞快照创建，直到测试确认 owner 仍被持有。""" + checkpoint_started.set() + await checkpoint_release.wait() + checkpoint_finished.set() + return object() + + task = asyncio.create_task( + _command( + checkpointer=checkpoint, + mutation=admission.hold, + ).execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + ) + await checkpoint_started.wait() + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + + assert task.done() is False + assert admission.seal() == 1 + idle_waiter = asyncio.create_task(asyncio.to_thread(admission.wait_until_idle)) + await asyncio.sleep(0.02) + assert idle_waiter.done() is False + + checkpoint_release.set() + with pytest.raises(asyncio.CancelledError): + await task + await idle_waiter + assert checkpoint_finished.is_set() + assert admission.active_count == 0 + + +@pytest.mark.asyncio +async def test_repeated_rollback_cancellation_retains_mutation_owner() -> None: + """补偿等待被再次连续取消时，lease 必须保留到补偿子任务终态。""" + admission = PluginMutationAdmission() + install_started = asyncio.Event() + rollback_started = asyncio.Event() + rollback_release = asyncio.Event() + rollback_finished = asyncio.Event() + + async def install(*_args) -> tuple[bool, str]: + """阻塞包安装，使首次取消进入补偿路径。""" + install_started.set() + await asyncio.Event().wait() + return True, "ok" + + async def rollback(_checkpoint: object) -> None: + """阻塞文件补偿，直到测试确认 owner 仍被持有。""" + rollback_started.set() + await rollback_release.wait() + rollback_finished.set() + + task = asyncio.create_task( + _command( + installer=install, + rollback=rollback, + mutation=admission.hold, + ).execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + ) + await install_started.wait() + task.cancel() + await rollback_started.wait() + assert admission.seal() == 1 + idle_waiter = asyncio.create_task(asyncio.to_thread(admission.wait_until_idle)) + + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0.02) + assert task.done() is False + assert idle_waiter.done() is False + assert admission.active_count == 1 + + rollback_release.set() + with pytest.raises(asyncio.CancelledError): + await task + await idle_waiter + assert rollback_finished.is_set() + assert admission.active_count == 0 + + @pytest.mark.asyncio async def test_cancelled_persisted_list_is_restored_conservatively() -> None: """清单写入已产生副作用但尚未返回时取消，也必须恢复原清单。""" diff --git a/tests/test_plugin_lifecycle_status.py b/tests/test_plugin_lifecycle_status.py index 47a0d2044..f4ead9d09 100644 --- a/tests/test_plugin_lifecycle_status.py +++ b/tests/test_plugin_lifecycle_status.py @@ -100,8 +100,8 @@ def test_lifecycle_records_load_failure_when_loader_returns_no_class(): assert statuses["DemoPlugin"] is PluginRuntimeStatus.LOAD_FAILED -def test_quiesce_keeps_instance_and_events_until_finalize(): - """第一阶段仅停止插件生产者，事件 handler 和实例留到屏障后释放。""" +def test_phased_quiesce_disables_events_before_hooks_and_keeps_instance(): + """宿主先停用 handler，再在事件结算后执行旧 hook 并保留实例。""" order: list[str] = [] class DemoPlugin: @@ -129,14 +129,19 @@ def test_quiesce_keeps_instance_and_events_until_finalize(): lifecycle, classes, running, _statuses = _lifecycle(plugins=[DemoPlugin]) lifecycle.start("DemoPlugin") lifecycle._disable_events.reset_mock() + lifecycle._disable_events.side_effect = ( + lambda _plugin_type: order.append("disable_events") + ) lifecycle._clear_modules.reset_mock() lifecycle._clear_tools.reset_mock() - assert lifecycle.quiesce() is True - assert order == ["close", "stop_service"] + assert lifecycle.quiesce_handlers() is True + assert order == ["disable_events"] + assert lifecycle.quiesce_services() is True + assert order == ["disable_events", "close", "stop_service"] assert classes["DemoPlugin"] is DemoPlugin assert isinstance(running["DemoPlugin"], DemoPlugin) - lifecycle._disable_events.assert_not_called() + lifecycle._disable_events.assert_called_once_with(DemoPlugin) lifecycle._clear_modules.assert_not_called() lifecycle._clear_tools.assert_not_called() @@ -194,6 +199,44 @@ def test_quiesce_runs_stop_service_after_close_failure_and_retries_missing_hook( assert running == {} +def test_quiesce_does_not_close_resources_until_all_handlers_are_disabled(): + """任一 handler 停用失败时不得调用可能破坏共享资源的旧停机 hook。""" + close = MagicMock() + + class DemoPlugin: + """提供可观察 close hook 的测试插件。""" + + plugin_name = "演示插件" + plugin_version = "1.0.0" + + def init_plugin(self, _config): + """接受宿主初始化配置。""" + + @staticmethod + def get_state(): + """保持插件事件 handler 启用。""" + return True + + def close(self): + """记录资源关闭调用。""" + close() + + lifecycle, classes, running, _statuses = _lifecycle(plugins=[DemoPlugin]) + lifecycle.start("DemoPlugin") + lifecycle._disable_events.side_effect = RuntimeError("disable failed") + + assert lifecycle.quiesce() is False + close.assert_not_called() + assert lifecycle.finalize() is False + assert "DemoPlugin" in classes + assert "DemoPlugin" in running + + lifecycle._disable_events.side_effect = None + assert lifecycle.quiesce() is True + close.assert_called_once_with() + assert lifecycle.finalize() is True + + def test_legacy_stop_entry_remains_idempotent(): """旧 stop 先解绑事件、保持 None 返回，且重复调用不重复执行 hook。""" order: list[str] = [] diff --git a/tests/test_plugin_monitor_lifecycle.py b/tests/test_plugin_monitor_lifecycle.py index 2cc37c60d..fb8b10fa8 100644 --- a/tests/test_plugin_monitor_lifecycle.py +++ b/tests/test_plugin_monitor_lifecycle.py @@ -582,11 +582,12 @@ def test_stop_plugin_monitor_returns_existing_manager_result(monkeypatch) -> Non @pytest.mark.asyncio -async def test_two_phase_plugin_shutdown_does_not_materialize_manager() -> None: - """两阶段入口在插件管理器尚未创建时都直接视为已收敛。""" +async def test_phased_plugin_shutdown_does_not_materialize_manager() -> None: + """三个停机入口在插件管理器尚未创建时都直接视为已收敛。""" _reset_plugin_manager() assert await plugins_initializer.quiesce_plugins(timeout=0) is True + assert await plugins_initializer.quiesce_plugin_services(timeout=0) is True assert plugins_initializer.finalize_plugins() is True assert PluginManager.get_existing_instance() is None @@ -616,7 +617,10 @@ async def test_quiesce_timeout_retains_future_owner_until_worker_finishes( release.wait(timeout=2) return True - manager._plugin_lifecycle.quiesce = MagicMock(side_effect=blocking_quiesce) + manager._plugin_lifecycle.quiesce_handlers = MagicMock(return_value=True) + manager._plugin_lifecycle.quiesce_services = MagicMock( + side_effect=blocking_quiesce, + ) manager._plugin_lifecycle.finalize = MagicMock(return_value=True) try: @@ -627,9 +631,10 @@ async def test_quiesce_timeout_retains_future_owner_until_worker_finishes( lambda: thread_helper, ) - assert await manager.quiesce_plugins(timeout=0.01) is False + assert await manager.quiesce_plugins(timeout=1) is True + assert await manager.quiesce_plugin_services(timeout=0.01) is False assert started.is_set() - owner = manager._plugin_quiesce_future + owner = manager._plugin_service_quiesce_future assert owner is not None assert owner.done() is False assert manager.finalize_plugins() is False @@ -659,7 +664,7 @@ async def test_quiesce_seals_runtime_until_new_lifespan_reopens(monkeypatch) -> ), ) manager = PluginManager() - manager._plugin_lifecycle.quiesce = MagicMock(return_value=True) + manager._plugin_lifecycle.quiesce_handlers = MagicMock(return_value=True) manager._plugin_lifecycle.start = MagicMock( return_value={"DemoPlugin": PluginRuntimeStatus.ACTIVE} ) diff --git a/tests/test_plugin_mutation_admission.py b/tests/test_plugin_mutation_admission.py index 93b6e5b0e..be72d87c9 100644 --- a/tests/test_plugin_mutation_admission.py +++ b/tests/test_plugin_mutation_admission.py @@ -1,6 +1,7 @@ """插件可变事务停机准入的确定性测试。""" import asyncio +import threading from collections.abc import Iterator from concurrent.futures import ThreadPoolExecutor from contextvars import copy_context @@ -20,6 +21,42 @@ from app.schemas.plugin import PluginRuntimeStatus from app.schemas.types import EventType +class _GatedCondition: + """让指定测试线程在取得真实 Condition 前停住。""" + + def __init__(self, blocked_thread_prefix: str) -> None: + """初始化内部 Condition 和可控的竞态闸门。""" + self._condition = threading.Condition() + self._blocked_thread_prefix = blocked_thread_prefix + self._blocked = False + self.attempted = threading.Event() + self.release = threading.Event() + + def __enter__(self): + """在目标线程首次进入时等待测试放行。""" + if ( + threading.current_thread().name.startswith(self._blocked_thread_prefix) + and not self._blocked + ): + self._blocked = True + self.attempted.set() + if not self.release.wait(timeout=1): + raise TimeoutError("测试未及时放行 Condition") + return self._condition.__enter__() + + def __exit__(self, exc_type, exc_value, traceback): + """把上下文退出委托给内部 Condition。""" + return self._condition.__exit__(exc_type, exc_value, traceback) + + def wait(self) -> bool: + """等待 admission 活动计数变化。""" + return self._condition.wait() + + def notify_all(self) -> None: + """唤醒全部等待 admission 空闲的线程。""" + self._condition.notify_all() + + @pytest.fixture def plugin_manager() -> Iterator[PluginManager]: """构造隔离的插件管理器，并在用例结束后清除单例状态。""" @@ -60,6 +97,46 @@ def test_seal_rejects_new_root_but_allows_propagated_nested_lease() -> None: assert admission.reopen() is True +def test_stale_propagated_context_is_rechecked_atomically_after_seal() -> None: + """外层退出后才取得 Condition 的复制上下文不得冒充嵌套事务。""" + admission = PluginMutationAdmission() + calls: list[str] = [] + outer = admission.hold("外层事务") + outer.__enter__() + propagated = copy_context() + condition = _GatedCondition("stale-admission") + admission._condition = condition + outer_closed = False + + def mutate() -> None: + """使用复制上下文尝试执行延迟到封口后的写入。""" + with admission.hold("延迟嵌套事务"): + calls.append("mutated") + + try: + with ThreadPoolExecutor( + max_workers=1, + thread_name_prefix="stale-admission", + ) as executor: + future = executor.submit(propagated.run, mutate) + assert condition.attempted.wait(timeout=1) + outer.__exit__(None, None, None) + outer_closed = True + assert admission.seal() == 0 + admission.wait_until_idle() + + condition.release.set() + with pytest.raises(PluginMutationRejectedError): + future.result(timeout=1) + finally: + condition.release.set() + if not outer_closed: + outer.__exit__(None, None, None) + + assert calls == [] + assert admission.active_count == 0 + + @pytest.mark.asyncio async def test_quiesce_timeout_retains_admitted_owner_and_nested_reload( plugin_manager: PluginManager, @@ -72,7 +149,7 @@ async def test_quiesce_timeout_retains_admitted_owner_and_nested_reload( manager._plugin_lifecycle.reload = MagicMock( return_value=PluginRuntimeStatus.ACTIVE ) - manager._plugin_lifecycle.quiesce = MagicMock(return_value=True) + manager._plugin_lifecycle.quiesce_handlers = MagicMock(return_value=True) manager._plugin_lifecycle.finalize = MagicMock(return_value=True) async def mutate() -> PluginRuntimeStatus: diff --git a/tests/test_plugin_system_setting_admission.py b/tests/test_plugin_system_setting_admission.py new file mode 100644 index 000000000..ada3d157f --- /dev/null +++ b/tests/test_plugin_system_setting_admission.py @@ -0,0 +1,263 @@ +"""通用系统设置入口的插件 mutation 准入测试。""" + +import asyncio +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.agent.tools.impl.update_system_settings import UpdateSystemSettingsTool +from app.api.endpoints import system as system_endpoint +from app.application.plugin import runtime as plugin_runtime +from app.runtime.extensions.plugin.admission import PluginMutationAdmission +from app.schemas.types import SystemConfigKey + + +PLUGIN_RUNTIME_KEYS = ( + SystemConfigKey.UserInstalledPlugins, + SystemConfigKey.PluginInstances, + SystemConfigKey.PluginFolders, +) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("config_key", PLUGIN_RUNTIME_KEYS) +async def test_system_http_rejects_plugin_keys_after_admission_seal( + config_key: SystemConfigKey, + monkeypatch, +) -> None: + """HTTP 通用设置入口在封口后不得写入插件运行态配置。""" + admission = PluginMutationAdmission() + admission.seal() + config = MagicMock() + config.async_set = AsyncMock() + monkeypatch.setattr( + plugin_runtime, + "get_plugin_manager", + lambda: SimpleNamespace(mutation=admission.hold), + ) + monkeypatch.setattr( + system_endpoint, + "get_runtime_settings", + lambda: SimpleNamespace(contains=lambda _key: False), + ) + monkeypatch.setattr( + system_endpoint, + "get_configured_system_config", + lambda: config, + ) + + response = await system_endpoint.set_setting(config_key.value, {}, None) + + assert response.success is False + assert "停机阶段" in response.message + config.async_set.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("config_key", PLUGIN_RUNTIME_KEYS) +async def test_agent_rejects_plugin_keys_after_admission_seal( + config_key: SystemConfigKey, + monkeypatch, +) -> None: + """Agent 通用设置入口在封口后不得读取或写入插件运行态配置。""" + admission = PluginMutationAdmission() + admission.seal() + config = MagicMock() + config.async_set = AsyncMock() + monkeypatch.setattr( + plugin_runtime, + "get_plugin_manager", + lambda: SimpleNamespace(mutation=admission.hold), + ) + tool = UpdateSystemSettingsTool( + session_id="session-1", + user_id="10001", + system_config=config, + ) + + payload = json.loads(await tool.run(setting_key=config_key.value, value={})) + + assert payload["success"] is False + assert "停机阶段" in payload["message"] + config.get.assert_not_called() + config.async_set.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_system_http_plugin_write_remains_owned_until_settled( + monkeypatch, +) -> None: + """HTTP 在途插件配置写入完成前，封口等待不得误判为空闲。""" + admission = PluginMutationAdmission() + write_started = asyncio.Event() + write_release = asyncio.Event() + + async def async_set(_key: str, _value: object) -> bool: + """阻塞配置写入，暴露 quiesce 等待窗口。""" + write_started.set() + await write_release.wait() + return True + + config = MagicMock() + config.async_set = AsyncMock(side_effect=async_set) + monkeypatch.setattr( + plugin_runtime, + "get_plugin_manager", + lambda: SimpleNamespace(mutation=admission.hold), + ) + monkeypatch.setattr( + system_endpoint, + "get_runtime_settings", + lambda: SimpleNamespace(contains=lambda _key: False), + ) + monkeypatch.setattr( + system_endpoint, + "get_configured_system_config", + lambda: config, + ) + monkeypatch.setattr(system_endpoint.eventmanager, "async_send_event", AsyncMock()) + + task = asyncio.create_task( + system_endpoint.set_setting( + SystemConfigKey.UserInstalledPlugins.value, + ["DemoPlugin"], + None, + ) + ) + await write_started.wait() + assert admission.seal() == 1 + idle_waiter = asyncio.create_task(asyncio.to_thread(admission.wait_until_idle)) + await asyncio.sleep(0.02) + assert idle_waiter.done() is False + + write_release.set() + response = await task + await idle_waiter + assert response.success is True + assert admission.active_count == 0 + + +@pytest.mark.asyncio +async def test_agent_plugin_write_remains_owned_until_saved_value_read( + monkeypatch, +) -> None: + """Agent 插件配置事务在写入和结果读取完成前持续持有 lease。""" + admission = PluginMutationAdmission() + write_started = asyncio.Event() + write_release = asyncio.Event() + + async def async_set(_key: SystemConfigKey, _value: object) -> bool: + """阻塞 Agent 配置写入，暴露 quiesce 等待窗口。""" + write_started.set() + await write_release.wait() + return True + + config = MagicMock() + config.get.side_effect = [[], ["DemoPlugin"]] + config.async_set = AsyncMock(side_effect=async_set) + monkeypatch.setattr( + plugin_runtime, + "get_plugin_manager", + lambda: SimpleNamespace(mutation=admission.hold), + ) + monkeypatch.setattr( + "app.agent.tools.impl.update_system_settings.eventmanager.async_send_event", + AsyncMock(), + ) + tool = UpdateSystemSettingsTool( + session_id="session-1", + user_id="10001", + system_config=config, + ) + + task = asyncio.create_task( + tool.run( + setting_key=SystemConfigKey.UserInstalledPlugins.value, + value=["DemoPlugin"], + ) + ) + await write_started.wait() + assert admission.seal() == 1 + idle_waiter = asyncio.create_task(asyncio.to_thread(admission.wait_until_idle)) + await asyncio.sleep(0.02) + assert idle_waiter.done() is False + + write_release.set() + payload = json.loads(await task) + await idle_waiter + assert payload["success"] is True + assert payload["saved_value"] == ["DemoPlugin"] + assert admission.active_count == 0 + + +@pytest.mark.asyncio +async def test_non_plugin_system_config_does_not_resolve_plugin_runtime( + monkeypatch, +) -> None: + """非插件 SystemConfig 写入保持原路径且不依赖插件组合根。""" + config = MagicMock() + config.async_set = AsyncMock(return_value=True) + + def fail_runtime_resolution(): + """若非插件配置误取插件运行时则立即暴露回归。""" + raise AssertionError("非插件配置不应解析插件运行时") + + monkeypatch.setattr(plugin_runtime, "get_plugin_manager", fail_runtime_resolution) + monkeypatch.setattr( + system_endpoint, + "get_runtime_settings", + lambda: SimpleNamespace(contains=lambda _key: False), + ) + monkeypatch.setattr( + system_endpoint, + "get_configured_system_config", + lambda: config, + ) + monkeypatch.setattr(system_endpoint.eventmanager, "async_send_event", AsyncMock()) + + response = await system_endpoint.set_setting( + SystemConfigKey.Directories.value, + [], + None, + ) + + assert response.success is True + config.async_set.assert_awaited_once_with(SystemConfigKey.Directories.value, None) + + +@pytest.mark.asyncio +async def test_agent_non_plugin_config_does_not_resolve_plugin_runtime( + monkeypatch, +) -> None: + """Agent 非插件配置写入保持既有读写和事件语义。""" + config = MagicMock() + config.get.side_effect = [{}, {"chatgpt": {"enabled": True}}] + config.async_set = AsyncMock(return_value=True) + + def fail_runtime_resolution(): + """若 Agent 非插件配置误取插件运行时则立即暴露回归。""" + raise AssertionError("非插件配置不应解析插件运行时") + + monkeypatch.setattr(plugin_runtime, "get_plugin_manager", fail_runtime_resolution) + monkeypatch.setattr( + "app.agent.tools.impl.update_system_settings.eventmanager.async_send_event", + AsyncMock(), + ) + tool = UpdateSystemSettingsTool( + session_id="session-1", + user_id="10001", + system_config=config, + ) + + payload = json.loads( + await tool.run( + setting_key=SystemConfigKey.AIAgentConfig.value, + value={"chatgpt": {"enabled": True}}, + ) + ) + + assert payload["success"] is True + assert payload["changed"] is True + config.async_set.assert_awaited_once() diff --git a/tests/test_transfer_worker_lifecycle.py b/tests/test_transfer_worker_lifecycle.py index 193d14cc1..54eb46ed4 100644 --- a/tests/test_transfer_worker_lifecycle.py +++ b/tests/test_transfer_worker_lifecycle.py @@ -297,6 +297,8 @@ def test_failed_retry_schedule_registers_future_observer(monkeypatch) -> None: chain.retry_scheduler = MagicMock(schedule_retry=schedule_retry) future = MagicMock(spec=Future) event_loop = MagicMock() + event_loop.is_running.return_value = True + event_loop.is_closed.return_value = False monkeypatch.setattr(global_vars, "CURRENT_EVENT_LOOP", event_loop) def submit(coroutine, loop):