diff --git a/app/adapters/external/market.py b/app/adapters/external/market.py index 0488dab94..c3d935180 100644 --- a/app/adapters/external/market.py +++ b/app/adapters/external/market.py @@ -12,6 +12,7 @@ import tempfile import threading import time import traceback +import uuid import zipfile from pathlib import Path, PurePosixPath, PureWindowsPath from typing import Dict, List, Optional, Tuple, Set, Callable, Awaitable, Sequence @@ -1129,21 +1130,38 @@ class PluginHelper(metaclass=WeakSingleton): backup_root = settings.CONFIG_PATH / "plugins_backup" backup_dir = backup_root / pid.lower() + staging_dir = backup_root / f".{pid.lower()}.tmp-{uuid.uuid4().hex}" + previous_dir = backup_root / f".{pid.lower()}.old-{uuid.uuid4().hex}" try: backup_root.mkdir(parents=True, exist_ok=True) - if backup_dir.exists(): - shutil.rmtree(backup_dir, ignore_errors=True) shutil.copytree( plugin_dir, - backup_dir, - dirs_exist_ok=True, + staging_dir, ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store") ) + if backup_dir.exists(): + backup_dir.replace(previous_dir) + staging_dir.replace(backup_dir) + if previous_dir.exists(): + shutil.rmtree(previous_dir, ignore_errors=True) logger.info(f"已刷新插件备份: {pid}") return True except Exception as e: + if not backup_dir.exists() and previous_dir.exists(): + try: + previous_dir.replace(backup_dir) + except Exception as rollback_error: + logger.error( + f"恢复插件旧备份失败，已保留恢复材料 {previous_dir}: " + f"{rollback_error}" + ) logger.error(f"刷新插件备份失败: {pid} - {e}") return False + finally: + if staging_dir.exists(): + shutil.rmtree(staging_dir, ignore_errors=True) + if backup_dir.exists() and previous_dir.exists(): + shutil.rmtree(previous_dir, ignore_errors=True) def __collect_plugin_wheels_dirs(self) -> List[Path]: """ diff --git a/app/adapters/system/plugin/dependency.py b/app/adapters/system/plugin/dependency.py index e6e7c179d..5b7d269ec 100644 --- a/app/adapters/system/plugin/dependency.py +++ b/app/adapters/system/plugin/dependency.py @@ -297,6 +297,39 @@ class PluginDependencyInstaller: logger.error(f"收集所有需要安装或更新的依赖项时发生错误：{err}") return [] + def classify_plugins(self) -> tuple[list[str], list[str], list[str]]: + """按源码和依赖状态划分已安装插件。""" + ready: list[str] = [] + missing_dependencies: list[str] = [] + missing_source: list[str] = [] + installed_packages = self._installed_packages() + + for plugin_id in self._installed_plugins_provider() or []: + plugin_dir = self._plugin_dir / plugin_id.lower() + if not plugin_dir.is_dir(): + missing_source.append(plugin_id) + continue + try: + manifest = load_dependency_manifest(plugin_dir) + requirements = [] if manifest is None else [ + requirement + for requirement in manifest.dependencies + if not requirement.marker or requirement.marker.evaluate() + ] + except PluginDependencyManifestError as error: + logger.error(f"插件 {plugin_id} 依赖清单无效：{error}") + missing_dependencies.append(plugin_id) + continue + if all( + self._requirement_satisfied(requirement, installed_packages) + for requirement in requirements + ): + ready.append(plugin_id) + else: + missing_dependencies.append(plugin_id) + + return ready, missing_dependencies, missing_source + def _wheels_dirs(self) -> list[Path]: """收集已安装插件附带的本地 wheels 目录。""" result = [] diff --git a/app/agent/tools/impl/_plugin_tool_utils.py b/app/agent/tools/impl/_plugin_tool_utils.py index fc56dd7e1..13e331c5c 100644 --- a/app/agent/tools/impl/_plugin_tool_utils.py +++ b/app/agent/tools/impl/_plugin_tool_utils.py @@ -11,6 +11,7 @@ from app.application.configuration import get_configured_system_config as System from app.adapters.external.server import MoviePilotServerHelper from app.adapters.external.market import PluginHelper from app.adapters.system.plugin.package import PluginPackageManager +from app.schemas.plugin import PluginRuntimeStatus from app.schemas.types import SystemConfigKey # 默认只向智能体返回一个可读预览，避免超大插件数据挤爆上下文窗口。 @@ -79,10 +80,11 @@ def refresh_plugin_registrations(plugin_id: str) -> None: register_plugin_api(plugin_id) -def reload_plugin_runtime(plugin_id: str) -> None: +def reload_plugin_runtime(plugin_id: str) -> PluginRuntimeStatus: """重载插件实例并重新注册其命令、定时任务和 API。""" - get_plugin_manager().reload_plugin(plugin_id) + runtime_status = get_plugin_manager().reload_plugin(plugin_id) refresh_plugin_registrations(plugin_id) + return runtime_status def summarize_plugin(plugin: Any) -> dict[str, Any]: @@ -349,30 +351,31 @@ async def install_plugin_runtime( target_id, ) - result = await PluginInstallCommand( - installed_plugins_reader=lambda: SystemConfigOper().get( - SystemConfigKey.UserInstalledPlugins - ) or [], - installed_plugins_writer=save_installed_plugins, - plugin_ids_provider=plugin_manager.get_plugin_ids, - compatibility_checker=skip_compatibility_check, - package_installer=install_package, - package_checkpointer=package_manager.async_checkpoint, - package_committer=package_manager.async_commit, - package_rollback=package_manager.async_rollback, - install_reporter=lambda target_id, target_repo: ( - MoviePilotServerHelper.async_install_plugin_reg( - plugin_id=target_id, - repo_url=target_repo, - ) - ), - plugin_reloader=reload_runtime, - registration_refresher=refresh_registrations, - ).execute( - plugin_id=plugin_id, - repo_url=repo_url, - force=force, - ) + with plugin_manager.suppress_plugin_monitor(plugin_id): + result = await PluginInstallCommand( + installed_plugins_reader=lambda: SystemConfigOper().get( + SystemConfigKey.UserInstalledPlugins + ) or [], + installed_plugins_writer=save_installed_plugins, + plugin_ids_provider=plugin_manager.get_plugin_ids, + compatibility_checker=skip_compatibility_check, + package_installer=install_package, + package_checkpointer=package_manager.async_checkpoint, + package_committer=package_manager.async_commit, + package_rollback=package_manager.async_rollback, + install_reporter=lambda target_id, target_repo: ( + MoviePilotServerHelper.async_install_plugin_reg( + plugin_id=target_id, + repo_url=target_repo, + ) + ), + plugin_reloader=reload_runtime, + registration_refresher=refresh_registrations, + ).execute( + plugin_id=plugin_id, + repo_url=repo_url, + force=force, + ) return result.success, result.message, result.refreshed_only diff --git a/app/agent/tools/impl/reload_plugin.py b/app/agent/tools/impl/reload_plugin.py index 70ec228b5..7582f6056 100644 --- a/app/agent/tools/impl/reload_plugin.py +++ b/app/agent/tools/impl/reload_plugin.py @@ -12,6 +12,7 @@ from app.agent.tools.impl._plugin_tool_utils import ( reload_plugin_runtime, ) from app.runtime.log import logger +from app.schemas.plugin import PluginRuntimeStatus class ReloadPluginInput(BaseModel): @@ -57,9 +58,26 @@ class ReloadPluginTool(MoviePilotTool): ensure_ascii=False, ) - reload_plugin_runtime(plugin_id) + runtime_status = reload_plugin_runtime(plugin_id) refreshed_plugin = get_plugin_snapshot(plugin_id) or plugin_info + if runtime_status is not PluginRuntimeStatus.ACTIVE: + return json.dumps( + { + "success": False, + **refreshed_plugin, + "runtime_status": runtime_status, + "message": ( + "未通过用户认证，请查看日志" + if runtime_status is PluginRuntimeStatus.BLOCKED_BY_POLICY + else "插件加载失败，请查看插件日志" + ), + }, + ensure_ascii=False, + indent=2, + default=str, + ) + return json.dumps( { "success": True, diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index ff0fc50d8..ab37c7037 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -20,6 +20,8 @@ from app.schemas.plugin import PluginRatingMap as _SchemaPluginRatingMap from app.schemas.plugin import PluginRatingRequest as _SchemaPluginRatingRequest from app.schemas.plugin import PluginReleaseData as _SchemaPluginReleaseData from app.schemas.plugin import PluginRemoteInfo as _SchemaPluginRemoteInfo +from app.schemas.plugin import PluginRuntimeStatus as _SchemaPluginRuntimeStatus +from app.schemas.plugin import PluginRuntimeSummary as _SchemaPluginRuntimeSummary from app.schemas.plugin import PluginSidebarNavItem as _SchemaPluginSidebarNavItem from app.schemas.response import Response as _SchemaResponse from app.schemas.token import TokenPayload as _SchemaTokenPayload @@ -254,7 +256,7 @@ async def all_plugins( # 已安装插件 installed_plugins = [plugin for plugin in local_plugins if plugin.installed] if state == "installed": - return installed_plugins + return plugin_manager.get_installed_plugins() # 未安装的本地插件 not_installed_plugins = [plugin for plugin in local_plugins if not plugin.installed] @@ -306,6 +308,34 @@ async def installed(_: ApiPrincipal = Depends(get_current_active_superuser_async return get_configured_system_config().get(SystemConfigKey.UserInstalledPlugins) or [] +@router.get( + "/runtime", + summary="插件运行时收敛状态", + response_model=_SchemaPluginRuntimeSummary, +) +async def runtime_status( + _: ApiPrincipal = Depends(get_current_active_superuser_async), +) -> _SchemaPluginRuntimeSummary: + """返回插件页轮询所需的轻量状态摘要。""" + plugin_manager = PluginManager() + statuses = plugin_manager.get_plugin_runtime_statuses() + pending = { + _SchemaPluginRuntimeStatus.SOURCE_MISSING, + _SchemaPluginRuntimeStatus.DEPENDENCY_PENDING, + _SchemaPluginRuntimeStatus.READY, + } + failed = { + _SchemaPluginRuntimeStatus.BLOCKED_BY_POLICY, + _SchemaPluginRuntimeStatus.LOAD_FAILED, + } + return _SchemaPluginRuntimeSummary( + ready=not plugin_manager.is_plugin_settling(), + generation=plugin_manager.get_plugin_runtime_generation(), + pending_count=sum(status in pending for status in statuses.values()), + failed_count=sum(status in failed for status in statuses.values()), + ) + + @router.get("/history/{plugin_id}", summary="获取插件更新说明", response_model=_SchemaPlugin) async def plugin_history( plugin_id: str, @@ -474,10 +504,19 @@ def reload_plugin( 重新加载插件 """ # 重新加载插件 - PluginManager().reload_plugin(plugin_id) + runtime_status = PluginManager().reload_plugin(plugin_id) # 注册插件服务 register_plugin(plugin_id) - return _SchemaResponse(success=True) + if runtime_status is _SchemaPluginRuntimeStatus.ACTIVE: + return _SchemaResponse(success=True) + return _SchemaResponse( + success=False, + message=( + "未通过用户认证，请查看日志" + if runtime_status is _SchemaPluginRuntimeStatus.BLOCKED_BY_POLICY + else "插件加载失败，请查看插件日志" + ), + ) @router.get("/install/{plugin_id}", summary="安装插件", response_model=_SchemaResponse[None]) @@ -493,6 +532,7 @@ async def install( """ plugin_helper = PluginHelper() package_manager = PluginPackageManager(plugin_helper) + plugin_manager = PluginManager() async def save_installed_plugins(plugin_ids: List[str]) -> object: """保存安装用例确认后的插件列表。""" @@ -543,12 +583,13 @@ async def install( plugin_reloader=reload_runtime, registration_refresher=refresh_registrations, ) - result = await command.execute( - plugin_id=plugin_id, - repo_url=repo_url, - release_version=release_version, - force=bool(force), - ) + with plugin_manager.suppress_plugin_monitor(plugin_id): + result = await command.execute( + plugin_id=plugin_id, + repo_url=repo_url, + release_version=release_version, + force=bool(force), + ) if not result.success: return _SchemaResponse(success=False, message=result.message) return _SchemaResponse(success=True) diff --git a/app/application/plugin/install.py b/app/application/plugin/install.py index 9966e8a2c..22419a70b 100644 --- a/app/application/plugin/install.py +++ b/app/application/plugin/install.py @@ -356,27 +356,9 @@ class PluginInstallCommand: dependency_supported=False, errors=tuple(errors), ) - rollback_message = [] - rollback_message.append("插件文件已恢复" if file_restored else "插件文件恢复失败") - if installed_list_persisted: - rollback_message.append( - "已安装列表已恢复" - if installed_list_restored - else "已安装列表恢复失败" - ) - if runtime_touched: - rollback_message.append( - "旧运行态已恢复" if runtime_restored else "旧运行态恢复失败" - ) - rollback_message.append( - "旧路由和服务注册已恢复" - if registrations_restored - else "旧路由和服务注册恢复失败" - ) - rollback_message.append("Python依赖变更不支持自动回滚") return PluginInstallResult( success=False, - message=f"{message}；{'；'.join(rollback_message)}", + message=message, package_installed=package_installed, installed_list_persisted=installed_list_persisted, failure_stage=stage, diff --git a/app/chain/system.py b/app/chain/system.py index 8d99e1da9..7c072af6a 100644 --- a/app/chain/system.py +++ b/app/chain/system.py @@ -1,12 +1,13 @@ +import errno import json import re import shutil +import uuid from pathlib import Path from typing import Union, Optional from app.chain import ChainBase from app.runtime.config import settings -from app.application.plugin.runtime import get_plugin_manager from app.runtime.state import SystemHelper from app.runtime.log import logger from app.schemas.message import Message @@ -22,6 +23,7 @@ class SystemChain(ChainBase): """ _restart_file = "__system_restart__" + _plugin_restore_pending_file = "__plugin_restore_pending__" def remote_clear_cache(self, channel: NotificationChannel, userid: Union[int, str], source: Optional[str] = None): """ @@ -77,31 +79,46 @@ class SystemChain(ChainBase): # 确保备份目录存在 backup_dir.mkdir(parents=True, exist_ok=True) - + pending_file = backup_dir / SystemChain._plugin_restore_pending_file + pending_items = ( + SystemChain.__read_plugin_restore_pending(pending_file) + if pending_file.exists() + else None + ) # 需要排除的文件和目录 exclude_items = {"__init__.py", "__pycache__", ".DS_Store"} + backup_failed = False + # 遍历插件目录，备份除排除项外的所有内容 for item in plugins_dir.iterdir(): if item.name in exclude_items: continue - + # 失败项目的原快照是下一次恢复的唯一材料，关停备份不能覆盖它。 + if pending_file.exists() and ( + pending_items is None or item.name in pending_items + ): + logger.debug(f"插件 {item.name} 有待重试恢复标记，保留原快照") + continue target_path = backup_dir / item.name - # 如果是目录 - if item.is_dir(): - if target_path.exists(): - continue - shutil.copytree(item, target_path) - logger.debug(f"已备份插件目录: {item.name}") - # 如果是文件 - elif item.is_file(): - if target_path.exists(): - continue - shutil.copy2(item, target_path) - logger.info(f"已备份插件文件: {item.name}") + try: + SystemChain.__replace_snapshot( + item, + target_path, + ignore=shutil.ignore_patterns( + "__pycache__", "*.pyc", ".DS_Store" + ) if item.is_dir() else None, + ) + logger.debug(f"已备份插件项目: {item.name}") + except Exception as e: + backup_failed = True + logger.error(f"备份插件 {item.name} 失败: {e}") - logger.info(f"插件备份完成，备份位置: {backup_dir}") + if backup_failed: + logger.warning(f"插件备份部分失败，保留可用旧快照: {backup_dir}") + else: + logger.info(f"插件备份完成，备份位置: {backup_dir}") except Exception as e: logger.error(f"插件备份失败: {str(e)}") @@ -124,44 +141,164 @@ class SystemChain(ChainBase): logger.info("插件备份目录不存在，跳过恢复") return - # 系统被重置才恢复插件 - if SystemHelper().is_system_reset(): + pending_file = backup_dir / SystemChain._plugin_restore_pending_file - # 确保插件目录存在 - plugins_dir.mkdir(parents=True, exist_ok=True) + # 系统重置或上次恢复未完成时才消费备份。 + system_reset = SystemHelper().is_system_reset() + should_restore = system_reset or pending_file.exists() + if not should_restore: + logger.info("当前不是系统重置，保留插件备份供后续重置使用") + return - # 遍历备份目录，恢复所有内容 - restored_count = 0 - for item in backup_dir.iterdir(): - target_path = plugins_dir / item.name - try: - # 如果是目录，且目录内有内容 - if item.is_dir() and any(item.iterdir()): - if target_path.exists(): - shutil.rmtree(target_path) - shutil.copytree(item, target_path) - logger.debug(f"已恢复插件目录: {item.name}") - restored_count += 1 - # 如果是文件 - elif item.is_file(): - shutil.copy2(item, target_path) - logger.debug(f"已恢复插件文件: {item.name}") - restored_count += 1 - except Exception as e: - logger.error(f"恢复插件 {item.name} 时发生错误: {str(e)}") + # 确保插件目录存在 + plugins_dir.mkdir(parents=True, exist_ok=True) + + # 遍历备份目录，恢复所有内容 + restored_count = 0 + restore_failed = False + failed_items: dict[str, bool] = {} + pending_items = ( + SystemChain.__read_plugin_restore_pending(pending_file) + if pending_file.exists() and not system_reset + else None + ) + for item in backup_dir.iterdir(): + if ( + item.name == SystemChain._plugin_restore_pending_file + or item.name.startswith(".") + ): + continue + target_path = plugins_dir / item.name + if pending_items is not None: + if item.name not in pending_items: continue + if not pending_items[item.name] and target_path.exists(): + logger.info(f"插件 {item.name} 已在恢复失败后重新安装，跳过备份覆盖") + continue + target_existed = target_path.exists() + try: + if item.is_dir() or item.is_file(): + SystemChain.__replace_snapshot(item, target_path) + logger.debug(f"已恢复插件文件: {item.name}") + restored_count += 1 + except Exception as e: + restore_failed = True + failed_items[item.name] = target_existed + logger.error(f"恢复插件 {item.name} 时发生错误: {str(e)}") + continue - logger.info(f"插件恢复完成，共恢复 {restored_count} 个项目") + logger.info(f"插件恢复完成，共恢复 {restored_count} 个项目") - # 安装缺少的依赖 - get_plugin_manager().install_plugin_missing_dependencies() + if restore_failed: + if SystemChain.__write_plugin_restore_pending(pending_file, failed_items): + logger.warning("插件恢复未完成，保留备份并标记为下次启动重试") + else: + logger.warning("插件恢复未完成，已保留备份，但无法写入下次启动重试标记") + return - # 删除备份目录 + # 源码恢复完成后即可消费备份；依赖由启动后的统一后台任务处理。 try: shutil.rmtree(backup_dir) logger.info(f"已删除插件备份目录: {backup_dir}") except Exception as e: logger.warning(f"删除备份目录失败: {str(e)}") + if backup_dir.exists(): + SystemChain.__write_plugin_restore_pending(pending_file, {}) + + @staticmethod + def __read_plugin_restore_pending(pending_file: Path) -> Optional[dict[str, bool]]: + """读取仍需恢复的插件项目；无效内容按全部项目重试。""" + try: + payload = json.loads(pending_file.read_text(encoding="utf-8")) + failed_items = payload.get("failed_items") + if not isinstance(failed_items, dict): + return None + return { + str(name): target_existed + for name, target_existed in failed_items.items() + if isinstance(name, str) and isinstance(target_existed, bool) + } + except (OSError, ValueError, TypeError): + return None + + @staticmethod + def __write_plugin_restore_pending( + pending_file: Path, + failed_items: dict[str, bool], + ) -> bool: + """记录失败项目及其原目标状态，供普通重启继续未完成恢复。""" + try: + pending_file.write_text( + json.dumps({"failed_items": failed_items}, ensure_ascii=False), + encoding="utf-8", + ) + return True + except Exception as e: + logger.error(f"写入插件恢复重试标记失败: {e}") + return False + + @staticmethod + def __replace_snapshot(source: Path, target: Path, *, ignore=None) -> None: + """复制到同级临时路径后替换目标，避免失败时丢失旧快照。""" + target.parent.mkdir(parents=True, exist_ok=True) + suffix = uuid.uuid4().hex + staging = target.with_name(f".{target.name}.tmp-{suffix}") + previous = target.with_name(f".{target.name}.old-{suffix}") + previous_available = False + published = False + try: + if source.is_dir(): + shutil.copytree(source, staging, ignore=ignore) + else: + shutil.copy2(source, staging) + if target.exists(): + try: + target.replace(previous) + except OSError as error: + if error.errno != errno.EXDEV: + raise + # overlayfs 可能拒绝把镜像层目录直接 rename 到可写层， + # 先复制旧目标保留恢复材料，再删除旧目录继续发布快照。 + if target.is_dir(): + shutil.copytree(target, previous, symlinks=True) + else: + shutil.copy2(target, previous, follow_symlinks=False) + previous_available = True + SystemChain.__remove_snapshot_path(target) + else: + previous_available = True + staging.replace(target) + published = True + except Exception: + if previous_available and not published: + try: + SystemChain.__remove_snapshot_path(target) + previous.replace(target) + previous_available = False + except Exception as rollback_error: + logger.error( + f"恢复旧快照失败，已保留恢复材料 {previous}: " + f"{rollback_error}" + ) + raise + finally: + if staging.is_dir(): + shutil.rmtree(staging, ignore_errors=True) + elif staging.exists(): + staging.unlink(missing_ok=True) + if published and previous.exists(): + if previous.is_dir(): + shutil.rmtree(previous, ignore_errors=True) + else: + previous.unlink(missing_ok=True) + + @staticmethod + def __remove_snapshot_path(path: Path) -> None: + """删除待替换目标，保留失败回滚所需的旧快照副本。""" + if path.is_dir() and not path.is_symlink(): + shutil.rmtree(path) + elif path.exists() or path.is_symlink(): + path.unlink() def __get_version_message(self) -> str: """ diff --git a/app/command.py b/app/command.py index 03df6e2e5..fc2782346 100644 --- a/app/command.py +++ b/app/command.py @@ -1,6 +1,7 @@ import copy import threading import traceback +from concurrent.futures import Future from typing import Any, Union, Dict, Optional from app.chain import ChainBase @@ -154,12 +155,11 @@ class Command(metaclass=Singleton): # 初始化命令 self.init_commands() - def init_commands(self, pid: Optional[str] = None) -> None: + def init_commands(self, pid: Optional[str] = None) -> Future: """ - 初始化菜单命令 + 提交菜单命令重建任务，并返回可等待的完成信号。 """ - # 使用线程池提交后台任务，避免引起阻塞 - ThreadHelper().submit(self.__init_commands_background, pid) + return ThreadHelper().submit(self.__init_commands_background, pid) def __init_commands_background(self, pid: Optional[str] = None) -> None: """ diff --git a/app/runtime/extensions/plugin/catalog.py b/app/runtime/extensions/plugin/catalog.py index 3ca77632e..af42fd6b4 100644 --- a/app/runtime/extensions/plugin/catalog.py +++ b/app/runtime/extensions/plugin/catalog.py @@ -11,7 +11,7 @@ from app.runtime.config import settings from app.runtime.extensions.plugin.contracts import supports_plugin_hook from app.runtime.extensions.plugin.storage import PluginStorage from app.runtime.extensions.plugin.system import PluginSystemServices -from app.schemas.plugin import Plugin +from app.schemas.plugin import Plugin, PluginRuntimeStatus from app.schemas.types import SystemConfigKey @@ -31,6 +31,7 @@ class PluginCatalogFacade: map_plugin: Callable[..., Optional[Plugin]], auth_checker: Callable[..., bool], plugin_attr: Callable[[str, str], Any], + runtime_status: Callable[[str], Optional[PluginRuntimeStatus]], log: Any, ) -> None: """保存注册表、目录服务和插件外部系统端口。""" @@ -44,6 +45,7 @@ class PluginCatalogFacade: self._map_plugin = map_plugin self._auth_checker = auth_checker self._plugin_attr = plugin_attr + self._runtime_status = runtime_status self._logger = log def online(self, force: bool = False) -> list[Plugin]: @@ -70,6 +72,7 @@ class PluginCatalogFacade: id=plugin_id, installed=plugin_id in installed, state=self._safe_state(plugin_id, plugin_instance), + runtime_status=self._runtime_status(plugin_id), has_page=supports_plugin_hook(plugin_class, "get_page"), plugin_public_key=getattr(plugin_class, "plugin_public_key", None), plugin_name=getattr(plugin_class, "plugin_name", None), @@ -88,6 +91,32 @@ class PluginCatalogFacade: plugins.sort(key=lambda item: getattr(item, "plugin_order", 0)) return plugins + def installed(self) -> list[Plugin]: + """按安装清单投影插件，未加载项目仍返回可观察占位卡片。""" + installed_ids = self._storage().read(SystemConfigKey.UserInstalledPlugins) or [] + local_by_id = { + plugin.id: plugin + for plugin in self.local() + if plugin.installed and plugin.id + } + result = [] + for plugin_id in installed_ids: + plugin = local_by_id.get(plugin_id) + if plugin: + result.append(plugin) + continue + result.append(Plugin( + id=plugin_id, + plugin_name=plugin_id, + installed=True, + state=False, + runtime_status=self._runtime_status(plugin_id), + is_local=True, + )) + # 展示顺序由持久化安装清单保留，避免后台恢复或占位卡片出现后改变用户看到的位置。 + # 前端可用用户级 PluginOrder 覆盖，plugin_order 只用于运行期插件发现顺序。 + return result + def local_version(self, plugin_id: str) -> Optional[str]: """读取指定已安装插件版本，不触发全量目录投影。""" installed = self._storage().read(SystemConfigKey.UserInstalledPlugins) or [] diff --git a/app/runtime/extensions/plugin/dependency.py b/app/runtime/extensions/plugin/dependency.py index 1bef06c6f..691c9c6b1 100644 --- a/app/runtime/extensions/plugin/dependency.py +++ b/app/runtime/extensions/plugin/dependency.py @@ -2,11 +2,29 @@ import time from collections.abc import Callable +from dataclasses import dataclass from typing import Any from app.runtime.extensions.plugin.system import PluginSystemServices +@dataclass(frozen=True) +class PluginDependencyInstallResult: + """记录插件依赖检查结果，区分无缺失、安装成功和安装失败。""" + + missing: list[str] + success: bool + + +@dataclass(frozen=True) +class PluginDependencyClassification: + """按当前源码和 Python 环境划分已安装插件。""" + + ready: tuple[str, ...] + missing_dependencies: tuple[str, ...] + missing_source: tuple[str, ...] + + class PluginDependencyService: """执行缺失插件依赖的发现和安装，不参与插件生命周期。""" @@ -20,12 +38,12 @@ class PluginDependencyService: self._system = system self._logger = log - def install_missing(self) -> list[str]: - """安装当前环境缺失的插件依赖并返回检查到的依赖名。""" + def install_missing_with_status(self) -> PluginDependencyInstallResult: + """安装缺失依赖并返回安装器的明确结果。""" installer = self._system().dependency missing = installer.find_missing() if not missing: - return missing + return PluginDependencyInstallResult(missing=[], success=True) self._logger.debug(f"检测到缺失的依赖项: {missing}") self._logger.info(f"开始安装缺失的依赖项，共 {len(missing)} 个...") started = time.time() @@ -39,4 +57,19 @@ class PluginDependencyService: self._logger.warning( f"存在缺失依赖项安装失败，请尝试手动安装，总耗时：{elapsed:.2f} 秒" ) - return missing + return PluginDependencyInstallResult(missing=missing, success=success) + + def install_missing(self) -> list[str]: + """安装当前环境缺失的插件依赖并保持历史列表返回合同。""" + return self.install_missing_with_status().missing + + def classify_plugins(self) -> PluginDependencyClassification: + """返回启动编排使用的轻量插件分类。""" + ready, missing_dependencies, missing_source = ( + self._system().dependency.classify_plugins() + ) + return PluginDependencyClassification( + ready=tuple(ready), + missing_dependencies=tuple(missing_dependencies), + missing_source=tuple(missing_source), + ) diff --git a/app/runtime/extensions/plugin/lifecycle.py b/app/runtime/extensions/plugin/lifecycle.py index c83b474e9..99bdee618 100644 --- a/app/runtime/extensions/plugin/lifecycle.py +++ b/app/runtime/extensions/plugin/lifecycle.py @@ -6,6 +6,8 @@ import traceback from collections.abc import Callable from typing import Any, Optional +from app.schemas.plugin import PluginRuntimeStatus + class PluginLifecycle: """管理插件发现、初始化、启停和热重载，不持有市场或 HTTP 路由职责。""" @@ -23,6 +25,7 @@ class PluginLifecycle: clear_tools: Callable[[], None], enable_events: Callable[[Any], None], disable_events: Callable[[Any], None], + runtime_status_writer: Callable[[str, PluginRuntimeStatus], None], log: Any, event_sender: Callable[..., Any], ) -> None: @@ -37,12 +40,19 @@ class PluginLifecycle: self._clear_tools = clear_tools self._enable_events = enable_events self._disable_events = disable_events + self._runtime_status_writer = runtime_status_writer self._logger = log self._event_sender = event_sender - def start(self, plugin_id: Optional[str] = None) -> None: - """加载并初始化指定插件或全部已安装插件。""" + def start( + self, + plugin_id: Optional[str] = None, + ) -> dict[str, PluginRuntimeStatus]: + """加载并初始化插件，返回每个目标的明确运行结果。""" installed_plugins = self._installed_plugins() + results: dict[str, PluginRuntimeStatus] = {} + if plugin_id: + self._runtime_status_writer(plugin_id, PluginRuntimeStatus.READY) def check_module(module: Any) -> bool: """判断模块是否具备宿主插件最小生命周期钩子。""" @@ -58,6 +68,9 @@ class PluginLifecycle: if not self._auth_checker(plugin): if current_id in self._classes: self._classes[current_id] = plugin + status = PluginRuntimeStatus.BLOCKED_BY_POLICY + self._runtime_status_writer(current_id, status) + results[current_id] = status continue self._classes[current_id] = plugin instance = plugin() @@ -70,11 +83,22 @@ class PluginLifecycle: self._enable_events(plugin) else: self._disable_events(plugin) + status = PluginRuntimeStatus.ACTIVE + self._runtime_status_writer(current_id, status) + results[current_id] = status except Exception as error: # noqa: BLE001 + status = PluginRuntimeStatus.LOAD_FAILED + self._runtime_status_writer(current_id, status) + results[current_id] = status self._logger.error( f"加载插件 {current_id} 出错：{error} - {traceback.format_exc()}" ) + if plugin_id and plugin_id not in results: + status = PluginRuntimeStatus.LOAD_FAILED + self._runtime_status_writer(plugin_id, status) + results[plugin_id] = status self._clear_tools() + return results def initialize(self, plugin_id: str, config: dict) -> None: """重新应用指定插件配置并刷新事件注册状态。""" @@ -115,11 +139,17 @@ class PluginLifecycle: self._clear_tools() self._logger.info("插件停止完成") - def reload(self, plugin_id: str, reload_event: Any) -> None: - """重启指定插件并广播插件重载事件。""" + def reload( + self, + plugin_id: str, + reload_event: Any, + ) -> PluginRuntimeStatus: + """重启指定插件并返回本次加载结果。""" + self._runtime_status_writer(plugin_id, PluginRuntimeStatus.READY) self.stop(plugin_id) - self.start(plugin_id) + status = self.start(plugin_id)[plugin_id] self._event_sender(reload_event, data={"plugin_id": plugin_id}) + return status def _stop_plugin(self, plugin: Any) -> None: """按插件旧 ABI 顺序关闭资源和服务。""" diff --git a/app/runtime/extensions/plugin/monitor.py b/app/runtime/extensions/plugin/monitor.py index fa2fd48c2..342972c2b 100644 --- a/app/runtime/extensions/plugin/monitor.py +++ b/app/runtime/extensions/plugin/monitor.py @@ -10,6 +10,7 @@ from typing import Any, Optional FederatedChangeResolver = Callable[[Path], Optional[tuple[str, Optional[dict], bool]]] RuntimePluginResolver = Callable[[Path], Optional[str]] +MonitorSuppression = Callable[[str], bool] LocalCandidateResolver = Callable[[Path], Optional[dict]] LocalPluginSync = Callable[[str, Optional[dict]], bool] PluginReloader = Callable[[str], Any] @@ -80,6 +81,7 @@ class PluginChangeMonitor: dependency_manifest_status: DependencyManifestStatus, watch: WatchFunction, log: Any, + monitor_suppressed: Optional[MonitorSuppression] = None, ) -> None: """保存监控路径、变化解析器和副作用回调。""" self._runtime_root = runtime_root @@ -88,6 +90,7 @@ class PluginChangeMonitor: self._recent_sync = recent_sync self._federated_change = federated_change self._runtime_plugin = runtime_plugin + self._monitor_suppressed = monitor_suppressed or (lambda _plugin_id: False) self._local_candidate = local_candidate self._sync_local = sync_local self._reload_plugin = reload_plugin @@ -150,6 +153,11 @@ class PluginChangeMonitor: if event_path.suffix != ".py": continue runtime_plugin_id = self._runtime_plugin(event_path) + if runtime_plugin_id and self._monitor_suppressed(runtime_plugin_id): + self._logger.debug( + f"插件 {runtime_plugin_id} 正在写入，跳过本批文件监控重载" + ) + continue candidate = ( self._local_candidate(event_path) if not runtime_plugin_id diff --git a/app/runtime/extensions/plugin/registry.py b/app/runtime/extensions/plugin/registry.py index 61a0fea1b..a6a1db398 100644 --- a/app/runtime/extensions/plugin/registry.py +++ b/app/runtime/extensions/plugin/registry.py @@ -2,6 +2,8 @@ from typing import Any, Dict, Optional +from app.schemas.plugin import PluginRuntimeStatus + class PluginRegistry: """集中持有插件类和运行实例，并为读取方提供稳定快照。""" @@ -10,6 +12,9 @@ class PluginRegistry: """创建彼此独立但生命周期一致的类表和实例表。""" self._classes: Dict[str, Any] = {} self._running: Dict[str, Any] = {} + self._runtime_statuses: Dict[str, PluginRuntimeStatus] = {} + self._settling = False + self._generation = 0 @property def classes(self) -> Dict[str, Any]: @@ -45,12 +50,53 @@ class PluginRegistry: """复制运行实例表，避免插件重载期间迭代失效。""" return dict(self._running) + def set_runtime_status( + self, + plugin_id: str, + status: PluginRuntimeStatus, + ) -> None: + """记录插件当前状态，并在实际变化时推进前端刷新代次。""" + if self._runtime_statuses.get(plugin_id) == status: + return + self._runtime_statuses[plugin_id] = status + self._generation += 1 + + def runtime_status(self, plugin_id: str) -> Optional[PluginRuntimeStatus]: + """读取指定插件状态。""" + return self._runtime_statuses.get(plugin_id) + + def runtime_status_snapshot(self) -> Dict[str, PluginRuntimeStatus]: + """复制插件状态表，避免后台加载期间迭代失效。""" + return dict(self._runtime_statuses) + + def set_settling(self, settling: bool) -> None: + """标记启动后的插件源码与依赖收敛任务是否仍在执行。""" + if self._settling == settling: + return + self._settling = settling + self._generation += 1 + + @property + def settling(self) -> bool: + """返回插件后台收敛任务是否仍在执行。""" + return self._settling + + @property + def generation(self) -> int: + """返回状态变化代次，供读取方识别刷新边界。""" + return self._generation + def remove(self, plugin_id: str) -> None: - """同时移除指定插件类和运行实例。""" + """同时移除指定插件类、运行实例和状态。""" self._classes.pop(plugin_id, None) self._running.pop(plugin_id, None) + if self._runtime_statuses.pop(plugin_id, None) is not None: + self._generation += 1 def clear(self) -> None: """原地清空注册表，保持外部持有的兼容字典引用有效。""" self._classes.clear() self._running.clear() + if self._runtime_statuses: + self._runtime_statuses.clear() + self._generation += 1 diff --git a/app/runtime/extensions/plugin_manager.py b/app/runtime/extensions/plugin_manager.py index 737a91286..aba00117c 100644 --- a/app/runtime/extensions/plugin_manager.py +++ b/app/runtime/extensions/plugin_manager.py @@ -1,5 +1,7 @@ import asyncio import posixpath +import threading +from contextlib import contextmanager from pathlib import Path from typing import Any, Dict, List, Optional, Type, Union, Callable, Tuple @@ -7,6 +9,7 @@ from watchfiles import watch from app.schemas.plugin import Plugin as _SchemaPlugin from app.schemas.plugin import PluginDashboard as _SchemaPluginDashboard +from app.schemas.plugin import PluginRuntimeStatus from app.foundation.crypto import RSAUtils from app.foundation.singleton import Singleton from app.foundation.version import compare_version @@ -34,7 +37,11 @@ from app.runtime.extensions.plugin.clone import PluginCloneService from app.runtime.extensions.plugin.access import PluginAccessPolicy from app.runtime.extensions.plugin.catalog import PluginCatalogFacade from app.runtime.extensions.plugin.paths import PluginPathResolver -from app.runtime.extensions.plugin.dependency import PluginDependencyService +from app.runtime.extensions.plugin.dependency import ( + PluginDependencyClassification, + PluginDependencyInstallResult, + PluginDependencyService, +) from app.runtime.extensions.plugin.storage import PluginConfigStore from app.schemas.types import EventType, SystemConfigKey @@ -151,10 +158,13 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): map_plugin=lambda **kwargs: self._process_plugin_info(**kwargs), auth_checker=lambda **kwargs: self.__set_and_check_auth_level(**kwargs), plugin_attr=lambda pid, attr: self.get_plugin_attr(pid, attr), + runtime_status=self._plugin_registry.runtime_status, log=logger, ) # 本地插件同步写入运行目录后的短时忽略窗口 self._recent_local_sync: Dict[str, float] = {} + self._monitor_suppression_lock = threading.Lock() + self._suppressed_monitor_plugins: Dict[str, int] = {} self._plugin_paths = PluginPathResolver( runtime_root=settings.ROOT_PATH / "app" / "plugins", running=lambda: self._running_plugins, @@ -205,6 +215,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): clear_tools=self.clear_plugin_agent_tools_cache, enable_events=eventmanager.enable_event_handler, disable_events=eventmanager.disable_event_handler, + runtime_status_writer=self._plugin_registry.set_runtime_status, log=logger, event_sender=eventmanager.send_event, ) @@ -298,17 +309,19 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): """按最新系统配置完整重启插件。""" # 停止已有插件 self.stop() - # 启动插件 - self.start() + classification = self.classify_plugins() + self.apply_plugin_dependency_classification(classification) + for plugin_id in classification.ready: + self.start(plugin_id) - def start(self, pid: Optional[str] = None): + def start(self, pid: Optional[str] = None) -> Dict[str, PluginRuntimeStatus]: """ 启动加载插件 :param pid: 插件ID，为空加载所有插件 """ _legacy_diagnostics_configurator(enabled=settings.DEBUG, emitter=logger.warning) - self._plugin_lifecycle.start(pid) + return self._plugin_lifecycle.start(pid) def init_plugin(self, plugin_id: str, conf: dict): """ @@ -385,7 +398,10 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): def start_monitor(self): """按当前配置启动插件文件修改监测。""" - if settings.DEV or settings.PLUGIN_AUTO_RELOAD: + if ( + not self.is_plugin_settling() + and (settings.DEV or settings.PLUGIN_AUTO_RELOAD) + ): self._plugin_monitor.start() def reload_monitor(self): @@ -393,7 +409,10 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): 重新加载插件文件修改监测 """ self._plugin_monitor.reload( - enabled=settings.DEV or settings.PLUGIN_AUTO_RELOAD + enabled=( + not self.is_plugin_settling() + and (settings.DEV or settings.PLUGIN_AUTO_RELOAD) + ) ) def stop_monitor(self): @@ -413,6 +432,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): recent_sync=self._recent_local_sync, federated_change=self._get_federated_plugin_change, runtime_plugin=self._get_plugin_id_from_path, + monitor_suppressed=self.is_plugin_monitor_suppressed, local_candidate=self._get_local_plugin_candidate_from_path, sync_local=self._sync_local_plugin_if_installed, reload_plugin=self.reload_plugin, @@ -454,19 +474,43 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): """ return self._local_plugin_sync.sync(pid, candidate) + @contextmanager + def suppress_plugin_monitor(self, plugin_id: str): + """在插件目录原子更新期间阻止文件监控抢先重载半成品。""" + normalized_id = plugin_id.lower() + with self._monitor_suppression_lock: + self._suppressed_monitor_plugins[normalized_id] = ( + self._suppressed_monitor_plugins.get(normalized_id, 0) + 1 + ) + try: + yield + finally: + with self._monitor_suppression_lock: + count = self._suppressed_monitor_plugins.get(normalized_id, 0) + if count <= 1: + self._suppressed_monitor_plugins.pop(normalized_id, None) + else: + self._suppressed_monitor_plugins[normalized_id] = count - 1 + + def is_plugin_monitor_suppressed(self, plugin_id: str) -> bool: + """判断指定插件是否处于安装或替换写入阶段。""" + with self._monitor_suppression_lock: + return self._suppressed_monitor_plugins.get(plugin_id.lower(), 0) > 0 + def remove_plugin(self, plugin_id: str): """ 从内存中移除一个插件 :param plugin_id: 插件ID """ self._plugin_lifecycle.stop(plugin_id) + self._plugin_registry.remove(plugin_id) - def reload_plugin(self, plugin_id: str): + def reload_plugin(self, plugin_id: str) -> PluginRuntimeStatus: """ 将一个插件重新加载到内存 :param plugin_id: 插件ID """ - self._plugin_lifecycle.reload(plugin_id, EventType.PluginReload) + return self._plugin_lifecycle.reload(plugin_id, EventType.PluginReload) @staticmethod def _clear_plugin_modules(plugin_id: Optional[str] = None): @@ -499,6 +543,66 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): log=logger, ).install_missing() + @staticmethod + def install_plugin_missing_dependencies_with_status() -> PluginDependencyInstallResult: + """安装插件缺失依赖并返回缺失项及安装成功状态。""" + return PluginDependencyService( + system=get_plugin_system, + log=logger, + ).install_missing_with_status() + + @staticmethod + def classify_plugins() -> PluginDependencyClassification: + """按源码和依赖是否就绪划分已安装插件。""" + return PluginDependencyService( + system=get_plugin_system, + log=logger, + ).classify_plugins() + + def apply_plugin_dependency_classification( + self, + classification: PluginDependencyClassification, + ) -> None: + """把源码和依赖分类写入运行状态，已激活插件保持当前结果。""" + running_ids = set(self._plugin_registry.running_ids()) + for plugin_id in classification.missing_source: + self._plugin_registry.set_runtime_status( + plugin_id, + PluginRuntimeStatus.SOURCE_MISSING, + ) + for plugin_id in classification.missing_dependencies: + self._plugin_registry.set_runtime_status( + plugin_id, + PluginRuntimeStatus.DEPENDENCY_PENDING, + ) + for plugin_id in classification.ready: + current_status = self._plugin_registry.runtime_status(plugin_id) + if ( + plugin_id in running_ids + and current_status is not PluginRuntimeStatus.DEPENDENCY_PENDING + ): + continue + self._plugin_registry.set_runtime_status( + plugin_id, + PluginRuntimeStatus.READY, + ) + + def set_plugin_settling(self, settling: bool) -> None: + """更新启动后的插件恢复任务状态。""" + self._plugin_registry.set_settling(settling) + + def get_plugin_runtime_statuses(self) -> Dict[str, PluginRuntimeStatus]: + """返回插件运行状态快照。""" + return self._plugin_registry.runtime_status_snapshot() + + def get_plugin_runtime_generation(self) -> int: + """返回插件状态变化代次。""" + return self._plugin_registry.generation + + def is_plugin_settling(self) -> bool: + """返回插件源码和依赖是否仍在后台恢复。""" + return self._plugin_registry.settling + def get_plugin_config(self, pid: str) -> dict: """ 获取插件配置 @@ -777,6 +881,10 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): """ return self._plugin_catalog_view.local() + def get_installed_plugins(self) -> List[_SchemaPlugin]: + """按安装清单返回插件，即使运行时尚未加载也保留卡片。""" + return self._plugin_catalog_view.installed() + def get_local_plugin_version(self, pid: str) -> Optional[str]: """ 获取指定已安装插件的本地版本，不触发全部插件的状态、页面和权限计算。 diff --git a/app/schemas/exports.py b/app/schemas/exports.py index 149e80b88..06eaad182 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -259,6 +259,8 @@ SCHEMA_EXPORTS = { 'PluginReleaseData': ('app.schemas.plugin', 'PluginReleaseData'), 'PluginReleaseItem': ('app.schemas.plugin', 'PluginReleaseItem'), 'PluginRemoteInfo': ('app.schemas.plugin', 'PluginRemoteInfo'), + 'PluginRuntimeStatus': ('app.schemas.plugin', 'PluginRuntimeStatus'), + 'PluginRuntimeSummary': ('app.schemas.plugin', 'PluginRuntimeSummary'), 'PluginSidebarNavItem': ('app.schemas.plugin', 'PluginSidebarNavItem'), 'PluginWorkflowActionGroup': ('app.schemas.workflow', 'PluginWorkflowActionGroup'), 'ProcessInfo': ('app.schemas.dashboard', 'ProcessInfo'), diff --git a/app/schemas/plugin.py b/app/schemas/plugin.py index f6035ee33..72b67ae6a 100644 --- a/app/schemas/plugin.py +++ b/app/schemas/plugin.py @@ -1,3 +1,4 @@ +from enum import Enum as _Enum from typing import Optional, List, Dict, Union from pydantic import BaseModel, Field, RootModel @@ -5,6 +6,17 @@ from pydantic import BaseModel, Field, RootModel from app.schemas.common import JsonData +class PluginRuntimeStatus(str, _Enum): + """插件从源码准备到运行激活的六类状态。""" + + SOURCE_MISSING = "source_missing" + DEPENDENCY_PENDING = "dependency_pending" + READY = "ready" + ACTIVE = "active" + BLOCKED_BY_POLICY = "blocked_by_policy" + LOAD_FAILED = "load_failed" + + class Plugin(BaseModel): """ 插件信息 @@ -34,6 +46,8 @@ class Plugin(BaseModel): installed: Optional[bool] = False # 运行状态 state: Optional[bool] = False + # 插件源码、依赖和运行时加载状态 + runtime_status: Optional[PluginRuntimeStatus] = None # 是否有详情页面 has_page: Optional[bool] = False # 是否有新版本 @@ -60,6 +74,15 @@ class Plugin(BaseModel): plugin_public_key: Optional[str] = None +class PluginRuntimeSummary(BaseModel): + """插件后台收敛状态和前端刷新代次。""" + + ready: bool = Field(description="本轮插件源码、依赖和加载是否已收敛") + generation: int = Field(description="插件运行状态变化代次") + pending_count: int = Field(description="仍处于准备阶段的插件数量") + failed_count: int = Field(description="加载失败或被策略阻止的插件数量") + + class PluginDashboard(Plugin): """ 插件仪表盘 diff --git a/app/startup/command_initializer.py b/app/startup/command_initializer.py index 51102d3ab..aa88b6f14 100644 --- a/app/startup/command_initializer.py +++ b/app/startup/command_initializer.py @@ -1,3 +1,5 @@ +from concurrent.futures import Future + from app.application.commands import register_command_class from app.command import Command @@ -19,8 +21,8 @@ def stop_command(): pass -def restart_command(): +def restart_command() -> Future: """ - 重启命令 + 重建命令并返回完成信号。 """ - Command().init_commands() + return Command().init_commands() diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index 351d4b8e3..82ca03522 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -25,6 +25,7 @@ except Exception: pass from app.chain.system import SystemChain +from app.application.plugin.runtime import get_plugin_manager from app.runtime.config import global_vars, settings from app.adapters.external.server import MoviePilotServerHelper from app.runtime.state import SystemHelper @@ -35,6 +36,7 @@ from app.startup.modules_initializer import init_modules, stop_modules from app.startup.monitor_initializer import stop_monitor, init_monitor from app.startup.plugins_initializer import ( configure_plugin_services, + execute_task, init_plugins, stop_plugins, sync_plugins, @@ -67,11 +69,18 @@ async def init_extra(): SystemHelper().set_system_modified() SystemChain().restart_finish() return - if await sync_plugins(): - # 重新注册插件定时服务 - init_plugin_scheduler() - # 重新注册命令 - restart_command() + plugin_manager = get_plugin_manager() + try: + if await sync_plugins(): + await execute_task( + global_vars.loop, + init_plugin_scheduler, + "插件定时服务刷新", + ) + await asyncio.wrap_future(restart_command()) + finally: + plugin_manager.set_plugin_settling(False) + plugin_manager.start_monitor() # 设置系统已修改标志 SystemHelper().set_system_modified() # 重启完成 @@ -320,12 +329,9 @@ async def lifespan(app: FastAPI): finally: print("Shutting down...") global_vars.stop_system() - # 取消同步插件任务 + # 插件恢复会在线程池中修改源码与依赖，必须完成后再进入资源关闭阶段。 try: - sync_plugins_task.cancel() await sync_plugins_task - except asyncio.CancelledError: - pass except Exception as e: print(str(e)) try: diff --git a/app/startup/plugins_initializer.py b/app/startup/plugins_initializer.py index 439c2e14e..6f2ce2043 100644 --- a/app/startup/plugins_initializer.py +++ b/app/startup/plugins_initializer.py @@ -16,6 +16,7 @@ from app.runtime.extensions.plugin_manager import ( configure_plugin_resource_import_preparer, configure_site_auth_level_provider, ) +from app.runtime.extensions.plugin.dependency import PluginDependencyInstallResult from app.application.plugin.catalog import PluginCatalogService from app.adapters.external.plugin.client import PluginMarketClient from app.runtime.extensions.plugin.storage import ( @@ -42,6 +43,7 @@ from app.db.oper.plugindata import PluginDataOper from app.db.oper.systemconfig import SystemConfigOper from app.runtime.log import logger from app.foundation.version import compare_version +from app.schemas.plugin import PluginRuntimeStatus from app.schemas.types import SystemConfigKey @@ -121,46 +123,105 @@ async def sync_plugins() -> bool: """ 初始化安装插件，并动态注册后台任务及API """ + plugin_manager = None try: configure_plugin_services() loop = global_vars.loop plugin_manager = PluginManager() + plugin_manager.set_plugin_settling(True) sync_result = await execute_task(loop, plugin_manager.sync, "插件同步到本地") - resolved_dependencies = await execute_task(loop, plugin_manager.install_plugin_missing_dependencies, - "缺失依赖项安装") - # 判断是否需要进行插件初始化 - if not sync_result and not resolved_dependencies: - logger.debug("没有新的插件同步到本地或缺失依赖项需要安装") + dependency_result = await execute_task( + loop, + plugin_manager.install_plugin_missing_dependencies_with_status, + "缺失依赖项安装", + ) + if dependency_result is None: + return False + if not isinstance(dependency_result, PluginDependencyInstallResult): + logger.error("缺失依赖项安装返回了无效结果，跳过插件重新初始化") + return False + previous_statuses = plugin_manager.get_plugin_runtime_statuses() + classification = plugin_manager.classify_plugins() + plugin_manager.apply_plugin_dependency_classification(classification) + if not dependency_result.success: + logger.error("缺失依赖项安装未完成，将继续激活当前已就绪插件") + changed_ids = await execute_task( + loop, + lambda: _activate_ready_plugins( + plugin_manager, + classification.ready, + sync_result or [], + previous_statuses, + ), + "插件运行态激活", + ) + if changed_ids is None: return False - # 继续执行后续的插件初始化步骤 - logger.info("正在重新初始化插件") - # 重新初始化插件 - plugin_manager.init_config() - # 重新注册插件API - register_plugin_api() - logger.info("所有插件初始化完成") + if not changed_ids: + logger.debug("没有新的插件进入可运行状态") + return False + + for plugin_id in changed_ids: + register_plugin_api(plugin_id) + if dependency_result.success: + logger.info(f"后台插件加载完成，共处理 {len(changed_ids)} 个插件") + else: + logger.warning( + f"缺失依赖项仍未全部恢复，已激活 {len(changed_ids)} 个就绪插件" + ) return True except Exception as e: logger.error(f"插件初始化过程中出现异常: {e}") return False +def _activate_ready_plugins( + plugin_manager: PluginManager, + ready_ids: tuple[str, ...], + synced_ids: list[str], + previous_statuses: dict[str, PluginRuntimeStatus], +) -> list[str]: + """在线程池中完成插件导入和初始化，避免阻塞 Web 事件循环。""" + running_ids = set(plugin_manager.running_plugins) + synced = set(synced_ids) + changed_ids: list[str] = [] + for plugin_id in ready_ids: + dependency_recovered = ( + previous_statuses.get(plugin_id) + is PluginRuntimeStatus.DEPENDENCY_PENDING + ) + if plugin_id in running_ids and (plugin_id in synced or dependency_recovered): + plugin_manager.reload_plugin(plugin_id) + changed_ids.append(plugin_id) + continue + if plugin_id not in running_ids: + plugin_manager.start(plugin_id) + changed_ids.append(plugin_id) + return changed_ids + + async def execute_task(loop, task_func, task_name): """ 执行后台任务 """ try: result = await loop.run_in_executor(None, task_func) - if isinstance(result, list) and result: - logger.debug(f"{task_name} 已完成，共处理 {len(result)} 个项目") + if isinstance(result, PluginDependencyInstallResult): + processed_count = len(result.missing) + elif isinstance(result, list): + processed_count = len(result) + else: + processed_count = 0 + if processed_count: + logger.debug(f"{task_name} 已完成，共处理 {processed_count} 个项目") else: logger.debug(f"没有新的 {task_name} 需要处理") return result except Exception as e: logger.error(f"{task_name} 时发生错误：{e}", exc_info=True) - return [] + return None def init_plugins(): @@ -169,9 +230,19 @@ def init_plugins(): """ configure_plugin_services() plugin_manager = PluginManager() - plugin_manager.start() + classification = plugin_manager.classify_plugins() + plugin_manager.apply_plugin_dependency_classification(classification) + plugin_manager.set_plugin_settling(True) + for plugin_id in classification.ready: + plugin_manager.start(plugin_id) register_plugin_api() plugin_manager.start_monitor() + logger.info( + "插件启动分类：立即加载=%s，等待依赖=%s，等待源码=%s", + len(classification.ready), + len(classification.missing_dependencies), + len(classification.missing_source), + ) def stop_plugins(): diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 9de3eecd1..52c4ac76f 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": 6067, - "edge_sha256": "a10a5353df10ba2817b49f6994eefd99266c853eda5119f36b2a68bafb9221ed", + "edge_count": 6075, + "edge_sha256": "3cea396b7570abbb70a8a30a9cd18006b52e21aa9d8d8106d2596baa52573d83", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -540,6 +540,7 @@ "app.agent.tools.impl._plugin_tool_utils -> app.runtime", "app.agent.tools.impl._plugin_tool_utils -> app.runtime.config", "app.agent.tools.impl._plugin_tool_utils -> app.schemas", + "app.agent.tools.impl._plugin_tool_utils -> app.schemas.plugin", "app.agent.tools.impl._plugin_tool_utils -> app.schemas.types", "app.agent.tools.impl._system_setting_utils -> app.agent", "app.agent.tools.impl._system_setting_utils -> app.agent.policy", @@ -1136,6 +1137,8 @@ "app.agent.tools.impl.reload_plugin -> app.agent.tools.tags", "app.agent.tools.impl.reload_plugin -> app.runtime", "app.agent.tools.impl.reload_plugin -> app.runtime.log", + "app.agent.tools.impl.reload_plugin -> app.schemas", + "app.agent.tools.impl.reload_plugin -> app.schemas.plugin", "app.agent.tools.impl.run_agent_task -> app.agent", "app.agent.tools.impl.run_agent_task -> app.agent.tools", "app.agent.tools.impl.run_agent_task -> app.agent.tools.base", @@ -3118,9 +3121,6 @@ "app.chain.system -> app.adapters.network.http", "app.chain.system -> app.adapters.system", "app.chain.system -> app.adapters.system.host", - "app.chain.system -> app.application", - "app.chain.system -> app.application.plugin", - "app.chain.system -> app.application.plugin.runtime", "app.chain.system -> app.chain", "app.chain.system -> app.runtime", "app.chain.system -> app.runtime.config", @@ -5327,6 +5327,8 @@ "app.runtime.extensions.plugin.dependency -> app.runtime.extensions", "app.runtime.extensions.plugin.dependency -> app.runtime.extensions.plugin", "app.runtime.extensions.plugin.dependency -> app.runtime.extensions.plugin.system", + "app.runtime.extensions.plugin.lifecycle -> app.schemas", + "app.runtime.extensions.plugin.lifecycle -> app.schemas.plugin", "app.runtime.extensions.plugin.metadata -> app.runtime", "app.runtime.extensions.plugin.metadata -> app.runtime.extensions", "app.runtime.extensions.plugin.metadata -> app.runtime.extensions.plugin", @@ -5344,6 +5346,8 @@ "app.runtime.extensions.plugin.projection -> app.runtime.log", "app.runtime.extensions.plugin.projection -> app.schemas", "app.runtime.extensions.plugin.projection -> app.schemas.plugin", + "app.runtime.extensions.plugin.registry -> app.schemas", + "app.runtime.extensions.plugin.registry -> app.schemas.plugin", "app.runtime.extensions.plugin.sync -> app.runtime", "app.runtime.extensions.plugin.sync -> app.runtime.extensions", "app.runtime.extensions.plugin.sync -> app.runtime.extensions.plugin", @@ -5735,6 +5739,9 @@ "app.startup.lifecycle -> app.adapters.external.server", "app.startup.lifecycle -> app.adapters.network", "app.startup.lifecycle -> app.adapters.network.http", + "app.startup.lifecycle -> app.application", + "app.startup.lifecycle -> app.application.plugin", + "app.startup.lifecycle -> app.application.plugin.runtime", "app.startup.lifecycle -> app.chain", "app.startup.lifecycle -> app.chain.system", "app.startup.lifecycle -> app.db", @@ -5889,6 +5896,7 @@ "app.startup.plugins_initializer -> app.runtime.config", "app.startup.plugins_initializer -> app.runtime.extensions", "app.startup.plugins_initializer -> app.runtime.extensions.plugin", + "app.startup.plugins_initializer -> app.runtime.extensions.plugin.dependency", "app.startup.plugins_initializer -> app.runtime.extensions.plugin.storage", "app.startup.plugins_initializer -> app.runtime.extensions.plugin.system", "app.startup.plugins_initializer -> app.runtime.extensions.plugin_manager", diff --git a/tests/test_agent_plugin_tools.py b/tests/test_agent_plugin_tools.py index acc9fae2b..595c66c53 100644 --- a/tests/test_agent_plugin_tools.py +++ b/tests/test_agent_plugin_tools.py @@ -11,6 +11,7 @@ from app.agent.tools.impl.query_market_plugins import QueryMarketPluginsTool from app.agent.tools.impl.query_plugin_config import QueryPluginConfigTool from app.agent.tools.impl.query_plugin_data import QueryPluginDataTool from app.agent.tools.impl.reload_plugin import ReloadPluginTool +from app.schemas.plugin import PluginRuntimeStatus from app.agent.tools.impl.uninstall_plugin import UninstallPluginTool from app.agent.tools.impl.update_plugin_config import UpdatePluginConfigTool @@ -215,6 +216,7 @@ def test_reload_plugin_triggers_runtime_refresh() -> None: "app.agent.tools.impl.reload_plugin.reload_plugin_runtime" ) as reload_plugin_runtime, ): + reload_plugin_runtime.return_value = PluginRuntimeStatus.ACTIVE result = asyncio.run(tool.run(plugin_id="DemoPlugin")) payload = json.loads(result) @@ -223,6 +225,27 @@ def test_reload_plugin_triggers_runtime_refresh() -> None: reload_plugin_runtime.assert_called_once_with("DemoPlugin") +def test_reload_plugin_reports_runtime_failure() -> None: + """重载未进入 active 时不得继续向智能体报告成功。""" + tool = ReloadPluginTool(session_id="session-1", user_id="10001") + + with ( + patch( + "app.agent.tools.impl.reload_plugin.get_plugin_snapshot", + side_effect=[_plugin_snapshot(), _plugin_snapshot(state=False)], + ), + patch( + "app.agent.tools.impl.reload_plugin.reload_plugin_runtime", + return_value=PluginRuntimeStatus.LOAD_FAILED, + ), + ): + result = asyncio.run(tool.run(plugin_id="DemoPlugin")) + + payload = json.loads(result) + assert payload["success"] is False + assert payload["runtime_status"] == "load_failed" + + def test_install_plugin_installs_market_candidate() -> None: """ 安装插件工具会使用市场候选携带的仓库地址。 diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 4f215ffba..9f8de7c6a 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -139,6 +139,31 @@ def test_lifespan_normal_mode_starts_full_runtime(monkeypatch): _assert_completed_once(step) +def test_lifespan_waits_for_plugin_settlement_before_shutdown(monkeypatch): + """关停必须等待插件恢复线程结束，避免与备份和资源释放并发。""" + shutdown_steps = _patch_lifespan(monkeypatch) + order = [] + shutdown_steps["backup_plugins"].side_effect = lambda: order.append("backup") + + async def run_lifespan(): + started = asyncio.Event() + release = asyncio.Event() + + async def settle_plugins(): + started.set() + await release.wait() + order.append("settled") + + lifecycle.init_extra.side_effect = settle_plugins + async with lifecycle.lifespan(FastAPI()): + await started.wait() + asyncio.get_running_loop().call_later(0.02, release.set) + + asyncio.run(run_lifespan()) + + assert order[:2] == ["settled", "backup"] + + def test_lifespan_configures_plugin_services_before_restore(monkeypatch): """插件恢复依赖的外部系统服务必须先于恢复阶段完成装配。""" shutdown_steps = _patch_lifespan(monkeypatch) diff --git a/tests/test_plugin_backup_restore.py b/tests/test_plugin_backup_restore.py new file mode 100644 index 000000000..ecb79444e --- /dev/null +++ b/tests/test_plugin_backup_restore.py @@ -0,0 +1,334 @@ +"""插件持久化备份与 Docker 重置恢复合同测试。""" + +import errno +from pathlib import Path +from types import SimpleNamespace + +from app.adapters.external import market as market_module +from app.chain import system as system_module +from app.chain.system import SystemChain + + +def _patch_docker_paths(monkeypatch, tmp_path: Path, *, reset: bool) -> Path: + """把插件恢复路径和 Docker 重置条件隔离到临时目录。""" + config_dir = tmp_path / "config" + runtime_dir = tmp_path / "app" / "plugins" + config_dir.mkdir(parents=True) + runtime_dir.mkdir(parents=True) + monkeypatch.setattr( + system_module, + "settings", + SimpleNamespace(ROOT_PATH=tmp_path, CONFIG_PATH=config_dir), + ) + monkeypatch.setattr( + system_module.SystemUtils, + "is_docker", + staticmethod(lambda: True), + ) + monkeypatch.setattr( + system_module.SystemHelper, + "is_system_reset", + lambda _self: reset, + ) + return runtime_dir + + +def _patch_market_paths(monkeypatch, tmp_path: Path) -> tuple[Path, Path]: + """把插件更新后的持久化备份路径隔离到临时目录。""" + plugin_root = tmp_path / "app" / "plugins" + config_dir = tmp_path / "config" + plugin_root.mkdir(parents=True) + config_dir.mkdir(parents=True) + monkeypatch.setattr(market_module, "PLUGIN_DIR", plugin_root) + monkeypatch.setattr( + market_module, + "settings", + SimpleNamespace(CONFIG_PATH=config_dir), + ) + monkeypatch.setattr( + market_module.SystemUtils, + "is_docker", + staticmethod(lambda: True), + ) + return plugin_root, config_dir / "plugins_backup" + + +def _write_plugin(root: Path, plugin_id: str, filename: str, content: str) -> Path: + plugin_dir = root / plugin_id + plugin_dir.mkdir(parents=True, exist_ok=True) + target = plugin_dir / filename + target.write_text(content, encoding="utf-8") + return target + + +def test_backup_plugins_refreshes_existing_snapshot(monkeypatch, tmp_path): + """关停备份应刷新同名插件并移除旧快照中的遗留文件。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=False) + backup_root = tmp_path / "config" / "plugins_backup" + backup_dir = backup_root / "demo" + _write_plugin(runtime_dir, "demo", "plugin.py", "new") + _write_plugin(backup_root, "demo", "plugin.py", "old") + _write_plugin(backup_root, "demo", "stale.py", "stale") + + SystemChain.backup_plugins() + + assert (backup_dir / "plugin.py").read_text(encoding="utf-8") == "new" + assert not (backup_dir / "stale.py").exists() + + +def test_backup_plugins_failure_preserves_previous_snapshot(monkeypatch, tmp_path): + """复制新快照失败时应保留上一份可恢复内容。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=False) + backup_root = tmp_path / "config" / "plugins_backup" + backup_dir = backup_root / "demo" + _write_plugin(runtime_dir, "demo", "plugin.py", "new") + _write_plugin(backup_root, "demo", "plugin.py", "old") + + def fail_copy(*_args, **_kwargs): + raise OSError("copy failed") + + monkeypatch.setattr(system_module.shutil, "copytree", fail_copy) + + SystemChain.backup_plugins() + + assert (backup_dir / "plugin.py").read_text(encoding="utf-8") == "old" + + +def test_backup_plugins_keeps_snapshot_missing_from_runtime(monkeypatch, tmp_path): + """运行目录缺失时不得删除唯一的持久化备份。""" + _patch_docker_paths(monkeypatch, tmp_path, reset=False) + backup_root = tmp_path / "config" / "plugins_backup" + backup_file = _write_plugin(backup_root, "demo", "plugin.py", "recoverable") + + SystemChain.backup_plugins() + + assert backup_file.read_text(encoding="utf-8") == "recoverable" + + +def test_restore_plugins_keeps_backup_on_regular_start(monkeypatch, tmp_path): + """普通重启保留备份，等待真正的容器重置场景消费。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=False) + backup_dir = tmp_path / "config" / "plugins_backup" + _write_plugin(backup_dir, "demo", "plugin.py", "stable") + + SystemChain.restore_plugins() + + assert not (runtime_dir / "demo").exists() + assert (backup_dir / "demo" / "plugin.py").exists() + + +def test_restore_plugins_consumes_backup_after_source_restore(monkeypatch, tmp_path): + """源码恢复完成即可消费备份，依赖恢复由启动后台任务统一处理。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=True) + backup_dir = tmp_path / "config" / "plugins_backup" + _write_plugin(backup_dir, "DemoPlugin", "plugin.py", "stable") + + SystemChain.restore_plugins() + + assert (runtime_dir / "DemoPlugin" / "plugin.py").read_text( + encoding="utf-8" + ) == "stable" + assert not backup_dir.exists() + + +def test_restore_plugins_retries_only_missing_sources(monkeypatch, tmp_path): + """恢复失败后只补仍缺失的目录，不覆盖用户随后重新安装的插件。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=True) + backup_dir = tmp_path / "config" / "plugins_backup" + _write_plugin(backup_dir, "DemoPlugin", "plugin.py", "backup") + reset_state = {"value": True} + monkeypatch.setattr( + system_module.SystemHelper, + "is_system_reset", + lambda _self: reset_state["value"], + ) + original_copytree = system_module.shutil.copytree + + def fail_copy(source, target, *args, **kwargs): + if Path(source).name == "DemoPlugin": + raise OSError("copy failed") + return original_copytree(source, target, *args, **kwargs) + + monkeypatch.setattr(system_module.shutil, "copytree", fail_copy) + SystemChain.restore_plugins() + pending = backup_dir / SystemChain._plugin_restore_pending_file + assert pending.exists() + + _write_plugin(runtime_dir, "DemoPlugin", "plugin.py", "reinstalled") + reset_state["value"] = False + monkeypatch.setattr(system_module.shutil, "copytree", original_copytree) + SystemChain.restore_plugins() + + assert (runtime_dir / "DemoPlugin" / "plugin.py").read_text( + encoding="utf-8" + ) == "reinstalled" + assert not backup_dir.exists() + + +def test_restore_plugins_retries_existing_target_after_copy_failure( + monkeypatch, + tmp_path, +): + """原目标已存在时，失败重试仍应完成备份版本的原子替换。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=True) + backup_dir = tmp_path / "config" / "plugins_backup" + _write_plugin(runtime_dir, "DemoPlugin", "plugin.py", "runtime-old") + _write_plugin(backup_dir, "DemoPlugin", "plugin.py", "backup-new") + reset_state = {"value": True} + monkeypatch.setattr( + system_module.SystemHelper, + "is_system_reset", + lambda _self: reset_state["value"], + ) + original_copytree = system_module.shutil.copytree + + def fail_copy(source, target, *args, **kwargs): + if Path(source).name == "DemoPlugin": + raise OSError("copy failed") + return original_copytree(source, target, *args, **kwargs) + + monkeypatch.setattr(system_module.shutil, "copytree", fail_copy) + SystemChain.restore_plugins() + assert (runtime_dir / "DemoPlugin" / "plugin.py").read_text( + encoding="utf-8" + ) == "runtime-old" + + reset_state["value"] = False + monkeypatch.setattr(system_module.shutil, "copytree", original_copytree) + SystemChain.restore_plugins() + + assert (runtime_dir / "DemoPlugin" / "plugin.py").read_text( + encoding="utf-8" + ) == "backup-new" + assert not backup_dir.exists() + + +def test_restore_plugins_falls_back_when_overlay_rename_returns_exdev( + monkeypatch, + tmp_path, +): + """镜像层目录拒绝 rename 时仍能完成可恢复的快照替换。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=True) + _write_plugin(runtime_dir, "DemoPlugin", "plugin.py", "runtime-old") + backup_dir = tmp_path / "config" / "plugins_backup" + _write_plugin(backup_dir, "DemoPlugin", "plugin.py", "backup-new") + + original_replace = Path.replace + + def exdev_for_existing_target(self, target): + if self == runtime_dir / "DemoPlugin": + raise OSError(errno.EXDEV, "cross-device link") + return original_replace(self, target) + + monkeypatch.setattr(Path, "replace", exdev_for_existing_target) + + SystemChain.restore_plugins() + + assert (runtime_dir / "DemoPlugin" / "plugin.py").read_text( + encoding="utf-8" + ) == "backup-new" + assert not backup_dir.exists() + + +def test_restore_plugins_restores_previous_after_partial_overlay_removal( + monkeypatch, + tmp_path, +): + """overlayfs 删除旧目录部分失败时仍恢复完整旧快照。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=True) + _write_plugin(runtime_dir, "DemoPlugin", "plugin.py", "runtime-old") + _write_plugin(runtime_dir, "DemoPlugin", "settings.json", "settings-old") + backup_dir = tmp_path / "config" / "plugins_backup" + _write_plugin(backup_dir, "DemoPlugin", "plugin.py", "backup-new") + + original_replace = Path.replace + original_rmtree = system_module.shutil.rmtree + removal_attempts = 0 + + def exdev_for_existing_target(self, target): + if self == runtime_dir / "DemoPlugin": + raise OSError(errno.EXDEV, "cross-device link") + return original_replace(self, target) + + def fail_after_partial_removal(path, *args, **kwargs): + nonlocal removal_attempts + if Path(path) == runtime_dir / "DemoPlugin" and removal_attempts == 0: + removal_attempts += 1 + (runtime_dir / "DemoPlugin" / "plugin.py").unlink() + raise OSError("directory removal interrupted") + return original_rmtree(path, *args, **kwargs) + + monkeypatch.setattr(Path, "replace", exdev_for_existing_target) + monkeypatch.setattr(system_module.shutil, "rmtree", fail_after_partial_removal) + + SystemChain.restore_plugins() + + assert (runtime_dir / "DemoPlugin" / "plugin.py").read_text( + encoding="utf-8" + ) == "runtime-old" + assert (runtime_dir / "DemoPlugin" / "settings.json").read_text( + encoding="utf-8" + ) == "settings-old" + assert backup_dir.exists() + assert (backup_dir / SystemChain._plugin_restore_pending_file).exists() + + +def test_backup_keeps_restore_retry_marker(monkeypatch, tmp_path): + """关停备份不得清除尚未完成的恢复标记。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=False) + backup_dir = tmp_path / "config" / "plugins_backup" + pending = backup_dir / SystemChain._plugin_restore_pending_file + pending.parent.mkdir(parents=True) + pending.touch() + _write_plugin(runtime_dir, "demo", "plugin.py", "current") + + SystemChain.backup_plugins() + + assert pending.exists() + + +def test_backup_does_not_overwrite_failed_restore_snapshot(monkeypatch, tmp_path): + """待重试项目的原快照必须跨关停保留，避免恢复材料被当前目录覆盖。""" + runtime_dir = _patch_docker_paths(monkeypatch, tmp_path, reset=False) + backup_dir = tmp_path / "config" / "plugins_backup" + pending = backup_dir / SystemChain._plugin_restore_pending_file + pending.parent.mkdir(parents=True) + pending.write_text( + '{"failed_items": {"demo": false}}', encoding="utf-8" + ) + _write_plugin(runtime_dir, "demo", "plugin.py", "reinstalled") + _write_plugin(backup_dir, "demo", "plugin.py", "recoverable") + + SystemChain.backup_plugins() + + assert (backup_dir / "demo" / "plugin.py").read_text(encoding="utf-8") == "recoverable" + + +def test_market_refresh_replaces_snapshot_and_removes_stale_files(monkeypatch, tmp_path): + """插件更新成功后应刷新对应持久化快照。""" + plugin_root, backup_root = _patch_market_paths(monkeypatch, tmp_path) + backup_dir = backup_root / "demo" + _write_plugin(plugin_root, "demo", "plugin.py", "new") + _write_plugin(backup_root, "demo", "plugin.py", "old") + _write_plugin(backup_root, "demo", "stale.py", "stale") + + assert market_module.PluginHelper.refresh_persistent_plugin_backup("demo") is True + + assert (backup_dir / "plugin.py").read_text(encoding="utf-8") == "new" + assert not (backup_dir / "stale.py").exists() + + +def test_market_refresh_failure_preserves_previous_snapshot(monkeypatch, tmp_path): + """插件更新备份失败时应继续保留旧快照。""" + plugin_root, backup_root = _patch_market_paths(monkeypatch, tmp_path) + backup_dir = backup_root / "demo" + _write_plugin(plugin_root, "demo", "plugin.py", "new") + _write_plugin(backup_root, "demo", "plugin.py", "old") + + def fail_copy(*_args, **_kwargs): + raise OSError("copy failed") + + monkeypatch.setattr(market_module.shutil, "copytree", fail_copy) + + assert market_module.PluginHelper.refresh_persistent_plugin_backup("demo") is False + assert (backup_dir / "plugin.py").read_text(encoding="utf-8") == "old" diff --git a/tests/test_plugin_catalog_runtime.py b/tests/test_plugin_catalog_runtime.py new file mode 100644 index 000000000..99fa7bbad --- /dev/null +++ b/tests/test_plugin_catalog_runtime.py @@ -0,0 +1,57 @@ +"""插件安装事实与运行状态投影测试。""" + +from types import SimpleNamespace + +from app.runtime.extensions.plugin.catalog import PluginCatalogFacade +from app.schemas.plugin import PluginRuntimeStatus +from app.schemas.types import SystemConfigKey + + +def test_installed_catalog_keeps_plugins_that_are_not_loaded(): + """已安装清单中的插件即使缺依赖或源码也必须保留可观察卡片。""" + class ActivePlugin: + plugin_name = "已运行插件" + plugin_version = "1.0.0" + plugin_order = 0 + + active_instance = SimpleNamespace(get_state=lambda: True) + statuses = { + "ActivePlugin": PluginRuntimeStatus.ACTIVE, + "DependencyPending": PluginRuntimeStatus.DEPENDENCY_PENDING, + "SourceMissing": PluginRuntimeStatus.SOURCE_MISSING, + } + facade = PluginCatalogFacade( + classes=lambda: {"ActivePlugin": ActivePlugin}, + running=lambda: {"ActivePlugin": active_instance}, + storage=lambda: SimpleNamespace( + read=lambda key: [ + "ActivePlugin", + "DependencyPending", + "SourceMissing", + ] if key is SystemConfigKey.UserInstalledPlugins else None, + ), + system=lambda: SimpleNamespace(), + market_catalog=lambda: None, + market_loader=lambda *_args, **_kwargs: [], + async_market_loader=lambda *_args, **_kwargs: [], + map_plugin=lambda **_kwargs: None, + auth_checker=lambda **_kwargs: True, + plugin_attr=lambda _plugin_id, _attr: None, + runtime_status=statuses.get, + log=SimpleNamespace(error=lambda *_args: None, info=lambda *_args: None), + ) + + plugins = facade.installed() + + assert [plugin.id for plugin in plugins] == [ + "ActivePlugin", + "DependencyPending", + "SourceMissing", + ] + assert [plugin.runtime_status for plugin in plugins] == [ + PluginRuntimeStatus.ACTIVE, + PluginRuntimeStatus.DEPENDENCY_PENDING, + PluginRuntimeStatus.SOURCE_MISSING, + ] + assert plugins[1].plugin_name == "DependencyPending" + assert plugins[2].installed is True diff --git a/tests/test_plugin_dependency_installer.py b/tests/test_plugin_dependency_installer.py index 09342bf63..1c9b2b76b 100644 --- a/tests/test_plugin_dependency_installer.py +++ b/tests/test_plugin_dependency_installer.py @@ -27,6 +27,36 @@ def _write_pyproject(root: Path, plugin_id: str, content: str) -> Path: return plugin_dir +def test_classify_plugins_preserves_ids_and_separates_startup_paths( + tmp_path, + monkeypatch, +): + """启动分类保留规范插件 ID，并区分可加载、缺依赖和缺源码。""" + plugin_root = tmp_path / "plugins" + (plugin_root / "readyplugin").mkdir(parents=True) + _write_requirements(plugin_root, "DependencyPending", "demo>=2\n") + installer = PluginDependencyInstaller( + Mock(), + installed_plugins_provider=lambda: [ + "ReadyPlugin", + "DependencyPending", + "SourcePending", + ], + plugin_dir=plugin_root, + ) + monkeypatch.setattr( + installer, + "_installed_packages", + lambda: {"demo": Version("1.0")}, + ) + + ready, missing_dependencies, missing_source = installer.classify_plugins() + + assert ready == ["ReadyPlugin"] + assert missing_dependencies == ["DependencyPending"] + assert missing_source == ["SourcePending"] + + def test_find_missing_merges_only_installed_plugin_constraints(tmp_path, monkeypatch): """依赖扫描只覆盖安装清单，并合并同名包的多插件约束。""" plugin_root = tmp_path / "plugins" diff --git a/tests/test_plugin_dependency_service.py b/tests/test_plugin_dependency_service.py new file mode 100644 index 000000000..b18981756 --- /dev/null +++ b/tests/test_plugin_dependency_service.py @@ -0,0 +1,37 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock + +from app.runtime.extensions.plugin.dependency import PluginDependencyService + + +def test_install_missing_skips_installer_when_environment_is_satisfied() -> None: + """依赖均满足时只执行轻量检查，不进入包安装链。""" + installer = SimpleNamespace( + find_missing=MagicMock(return_value=[]), + install=MagicMock(), + ) + service = PluginDependencyService( + system=lambda: SimpleNamespace(dependency=installer), + log=MagicMock(), + ) + + result = service.install_missing_with_status() + + assert result.success is True + assert result.missing == [] + installer.install.assert_not_called() + + +def test_install_missing_preserves_list_return_contract() -> None: + """旧入口继续返回缺失项列表，供现有调用方按真值判断。""" + installer = SimpleNamespace( + find_missing=MagicMock(return_value=["demo>=1"]), + install=MagicMock(return_value=(True, "")), + ) + service = PluginDependencyService( + system=lambda: SimpleNamespace(dependency=installer), + log=MagicMock(), + ) + + assert service.install_missing() == ["demo>=1"] + installer.install.assert_called_once_with(["demo>=1"]) diff --git a/tests/test_plugin_endpoint.py b/tests/test_plugin_endpoint.py index a79ce6bff..fdb879aef 100644 --- a/tests/test_plugin_endpoint.py +++ b/tests/test_plugin_endpoint.py @@ -6,11 +6,14 @@ from app import schemas from app.api.endpoints.plugin import plugin_history from app.api.endpoints.plugin import plugin_releases from app.api.endpoints.plugin import reset_plugin +from app.api.endpoints.plugin import reload_plugin +from app.api.endpoints.plugin import runtime_status from app.api.endpoints.system import sync_plugin_market_from_wiki from app.application.plugin.config import PluginConfigCommand from app.runtime.config import settings from app.runtime.extensions.plugin_manager import PluginManager from app.schemas.event import PluginDataResetEventData +from app.schemas.plugin import PluginRuntimeStatus from app.schemas.types import ChainEventType from app.foundation.singleton import Singleton @@ -48,6 +51,42 @@ def test_plugin_history_merges_remote_metadata(): assert result.has_update +def test_runtime_status_reports_pending_and_terminal_counts(): + """插件页摘要区分后台收敛、准备态和终态失败。""" + plugin_manager = MagicMock() + plugin_manager.get_plugin_runtime_statuses.return_value = { + "SourcePending": PluginRuntimeStatus.SOURCE_MISSING, + "DependencyPending": PluginRuntimeStatus.DEPENDENCY_PENDING, + "ActivePlugin": PluginRuntimeStatus.ACTIVE, + "FailedPlugin": PluginRuntimeStatus.LOAD_FAILED, + } + plugin_manager.is_plugin_settling.return_value = True + plugin_manager.get_plugin_runtime_generation.return_value = 7 + + with patch("app.api.endpoints.plugin.PluginManager", return_value=plugin_manager): + result = asyncio.run(runtime_status(None)) + + assert result.ready is False + assert result.generation == 7 + assert result.pending_count == 2 + assert result.failed_count == 1 + + +def test_reload_endpoint_reports_load_failure(monkeypatch): + """插件重载失败时接口返回失败，同时仍刷新旧注册投影。""" + plugin_manager = MagicMock() + plugin_manager.reload_plugin.return_value = PluginRuntimeStatus.LOAD_FAILED + register = MagicMock() + monkeypatch.setattr(plugin_endpoint, "PluginManager", lambda: plugin_manager) + monkeypatch.setattr(plugin_endpoint, "register_plugin", register) + + result = reload_plugin("DemoPlugin", None) + + assert result.success is False + assert result.message == "插件加载失败，请查看插件日志" + register.assert_called_once_with("DemoPlugin") + + def test_plugin_history_returns_installed_plugin_when_remote_missing(): """ 远端仓库不可用时，接口仍返回本地已安装插件信息，前端可继续展示兜底状态。 diff --git a/tests/test_plugin_install_command.py b/tests/test_plugin_install_command.py index a56ed306f..c0c559d00 100644 --- a/tests/test_plugin_install_command.py +++ b/tests/test_plugin_install_command.py @@ -60,6 +60,8 @@ async def test_install_failure_stops_before_report_persistence_and_reload(): assert result.failure_stage == "package_install" assert result.rollback.file_restored is True assert result.rollback.dependency_supported is False + assert "插件文件已恢复" not in result.message + assert "Python依赖变更不支持自动回滚" not in result.message rollback.assert_awaited_once() reporter.assert_not_awaited() writer.assert_not_awaited() diff --git a/tests/test_plugin_lifecycle_status.py b/tests/test_plugin_lifecycle_status.py new file mode 100644 index 000000000..3da4e131a --- /dev/null +++ b/tests/test_plugin_lifecycle_status.py @@ -0,0 +1,100 @@ +"""插件生命周期六类状态中的运行结果测试。""" + +from types import SimpleNamespace +from unittest.mock import MagicMock + +from app.runtime.extensions.plugin.lifecycle import PluginLifecycle +from app.schemas.plugin import PluginRuntimeStatus + + +def _plugin_class(*, init_error: Exception | None = None): + """构造满足插件最小生命周期合同的测试类。""" + class DemoPlugin: + plugin_name = "演示插件" + plugin_version = "1.0.0" + + def init_plugin(self, _config): + if init_error: + raise init_error + + @staticmethod + def get_state(): + return True + + return DemoPlugin + + +def _lifecycle(*, plugins, auth=True): + """构造隔离外部事件和模块清理的生命周期实例。""" + classes = {} + running = {} + statuses = {} + lifecycle = PluginLifecycle( + classes=classes, + running=running, + load_plugins=lambda _plugin_id, _installed, _check: list(plugins), + installed_plugins=lambda: ["DemoPlugin"], + plugin_config=lambda _plugin_id: {}, + auth_checker=lambda _plugin: auth, + clear_modules=MagicMock(), + clear_tools=MagicMock(), + enable_events=MagicMock(), + disable_events=MagicMock(), + runtime_status_writer=statuses.__setitem__, + log=MagicMock(), + event_sender=MagicMock(), + ) + return lifecycle, classes, running, statuses + + +def test_lifecycle_records_active_result(): + """插件完成构造和初始化后进入 active。""" + lifecycle, classes, running, statuses = _lifecycle( + plugins=[_plugin_class()], + ) + + result = lifecycle.start("DemoPlugin") + + assert result == {"DemoPlugin": PluginRuntimeStatus.ACTIVE} + assert "DemoPlugin" in classes + assert "DemoPlugin" in running + assert statuses["DemoPlugin"] is PluginRuntimeStatus.ACTIVE + + +def test_lifecycle_records_policy_block_without_runtime_instance(): + """类已发现但权限策略拒绝时进入 blocked_by_policy。""" + lifecycle, _classes, running, statuses = _lifecycle( + plugins=[_plugin_class()], + auth=False, + ) + + result = lifecycle.start("DemoPlugin") + + assert result == {"DemoPlugin": PluginRuntimeStatus.BLOCKED_BY_POLICY} + assert running == {} + assert statuses["DemoPlugin"] is PluginRuntimeStatus.BLOCKED_BY_POLICY + + +def test_lifecycle_records_load_failure_for_init_exception(): + """插件初始化异常时保留类信息并进入 load_failed。""" + lifecycle, classes, running, statuses = _lifecycle( + plugins=[_plugin_class(init_error=RuntimeError("init failed"))], + ) + + result = lifecycle.start("DemoPlugin") + + assert result == {"DemoPlugin": PluginRuntimeStatus.LOAD_FAILED} + assert "DemoPlugin" in classes + assert running == {} + assert statuses["DemoPlugin"] is PluginRuntimeStatus.LOAD_FAILED + + +def test_lifecycle_records_load_failure_when_loader_returns_no_class(): + """目标源码无法产生合法插件类时进入 load_failed。""" + lifecycle, _classes, running, statuses = _lifecycle(plugins=[]) + + result = lifecycle.start("DemoPlugin") + + assert result == {"DemoPlugin": PluginRuntimeStatus.LOAD_FAILED} + assert running == {} + assert statuses["DemoPlugin"] is PluginRuntimeStatus.LOAD_FAILED diff --git a/tests/test_plugin_monitor_lifecycle.py b/tests/test_plugin_monitor_lifecycle.py index eab3b3a17..c76952a14 100644 --- a/tests/test_plugin_monitor_lifecycle.py +++ b/tests/test_plugin_monitor_lifecycle.py @@ -1,12 +1,23 @@ +import asyncio +import threading +import time from types import SimpleNamespace from unittest.mock import MagicMock import pytest from app.foundation.singleton import Singleton -from app.runtime.extensions.plugin.monitor import PluginMonitorController +from app.runtime.extensions.plugin.dependency import ( + PluginDependencyClassification, + PluginDependencyInstallResult, +) +from app.runtime.extensions.plugin.monitor import ( + PluginChangeMonitor, + PluginMonitorController, +) from app.runtime.extensions.plugin.system import reset_plugin_system from app.runtime.extensions.plugin_manager import PluginManager +from app.schemas.plugin import PluginRuntimeStatus from app.startup import plugins_initializer @@ -45,10 +56,15 @@ def test_plugin_manager_constructor_does_not_start_monitor_before_runtime( def test_init_plugins_starts_monitor_after_runtime_and_routes(monkeypatch) -> None: - """插件运行时和动态路由就绪后，启动层才允许文件监控接收变化。""" + """启动阶段只加载依赖已就绪的插件，再开放路由和文件监控。""" order: list[str] = [] manager = MagicMock() - manager.start.side_effect = lambda: order.append("plugins") + manager.classify_plugins.return_value = PluginDependencyClassification( + ready=("ReadyPlugin",), + missing_dependencies=("DependencyPending",), + missing_source=("SourcePending",), + ) + manager.start.side_effect = lambda plugin_id: order.append(f"plugin:{plugin_id}") manager.start_monitor.side_effect = lambda: order.append("monitor") monkeypatch.setattr( plugins_initializer, @@ -64,7 +80,238 @@ def test_init_plugins_starts_monitor_after_runtime_and_routes(monkeypatch) -> No plugins_initializer.init_plugins() - assert order == ["services", "plugins", "routes", "monitor"] + assert order == ["services", "plugin:ReadyPlugin", "routes", "monitor"] + manager.set_plugin_settling.assert_called_once_with(True) + + +def test_plugin_manager_projects_dependency_classification_to_runtime_status() -> None: + """真实管理器按分类字段写入三类启动状态，避免测试替身掩盖字段漂移。""" + _reset_plugin_manager() + manager = PluginManager() + + manager.apply_plugin_dependency_classification( + PluginDependencyClassification( + ready=("ReadyPlugin",), + missing_dependencies=("DependencyPending",), + missing_source=("SourcePending",), + ) + ) + + assert manager.get_plugin_runtime_statuses() == { + "ReadyPlugin": PluginRuntimeStatus.READY, + "DependencyPending": PluginRuntimeStatus.DEPENDENCY_PENDING, + "SourcePending": PluginRuntimeStatus.SOURCE_MISSING, + } + _reset_plugin_manager() + + +def test_plugin_manager_promotes_running_dependency_after_recovery() -> None: + """依赖恢复后，运行中的插件状态必须允许后台流程触发重载。""" + _reset_plugin_manager() + manager = PluginManager() + manager._plugin_registry.running["DependencyRecovered"] = object() + manager._plugin_registry.set_runtime_status( + "DependencyRecovered", + PluginRuntimeStatus.DEPENDENCY_PENDING, + ) + + manager.apply_plugin_dependency_classification( + PluginDependencyClassification( + ready=("DependencyRecovered",), + missing_dependencies=(), + missing_source=(), + ) + ) + + assert manager.get_plugin_runtime_statuses()["DependencyRecovered"] is ( + PluginRuntimeStatus.READY + ) + _reset_plugin_manager() + + +def _patch_sync_plugins(monkeypatch, manager: MagicMock) -> MagicMock: + """隔离后台执行器并返回动态路由注册替身。""" + async def execute(_loop, task_func, _task_name): + return task_func() + + register = MagicMock() + monkeypatch.setattr(plugins_initializer, "configure_plugin_services", lambda: None) + monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager) + monkeypatch.setattr(plugins_initializer, "execute_task", execute) + monkeypatch.setattr(plugins_initializer, "register_plugin_api", register) + manager.get_plugin_runtime_statuses.return_value = {} + return register + + +@pytest.mark.asyncio +async def test_sync_plugins_activates_ready_plugins_when_dependencies_fail( + monkeypatch, +) -> None: + """依赖恢复失败时仍激活无关的已就绪插件。""" + manager = MagicMock() + manager.sync.return_value = ["demo"] + manager.install_plugin_missing_dependencies_with_status.return_value = ( + PluginDependencyInstallResult(missing=["demo>=1"], success=False) + ) + manager.classify_plugins.return_value = PluginDependencyClassification( + ready=("ReadyPlugin",), + missing_dependencies=("DependencyPending",), + missing_source=(), + ) + manager.running_plugins = {} + register = _patch_sync_plugins(monkeypatch, manager) + + assert await plugins_initializer.sync_plugins() is True + + manager.start.assert_called_once_with("ReadyPlugin") + manager.reload_plugin.assert_not_called() + register.assert_called_once_with("ReadyPlugin") + + +@pytest.mark.asyncio +async def test_sync_plugins_loads_only_plugins_that_become_ready( + monkeypatch, +) -> None: + """后台依赖恢复后只启动尚未运行且当前已就绪的插件。""" + manager = MagicMock() + manager.sync.return_value = [] + manager.install_plugin_missing_dependencies_with_status.return_value = ( + PluginDependencyInstallResult(missing=["demo>=1"], success=True) + ) + manager.classify_plugins.return_value = PluginDependencyClassification( + ready=("ReadyPlugin", "DependencyRecovered"), + missing_dependencies=(), + missing_source=("SourcePending",), + ) + running = {"ReadyPlugin": object()} + manager.running_plugins = running + + def start(plugin_id: str) -> None: + running[plugin_id] = object() + + manager.start.side_effect = start + register = _patch_sync_plugins(monkeypatch, manager) + + assert await plugins_initializer.sync_plugins() is True + + manager.start.assert_called_once_with("DependencyRecovered") + manager.reload_plugin.assert_not_called() + register.assert_called_once_with("DependencyRecovered") + + +@pytest.mark.asyncio +async def test_sync_plugins_reloads_only_updated_running_plugins(monkeypatch) -> None: + """源码同步只重载对应运行实例，不重启其他插件。""" + manager = MagicMock() + manager.sync.return_value = ["UpdatedPlugin"] + manager.install_plugin_missing_dependencies_with_status.return_value = ( + PluginDependencyInstallResult(missing=[], success=True) + ) + manager.classify_plugins.return_value = PluginDependencyClassification( + ready=("StablePlugin", "UpdatedPlugin"), + missing_dependencies=(), + missing_source=(), + ) + manager.running_plugins = { + "StablePlugin": object(), + "UpdatedPlugin": object(), + } + register = _patch_sync_plugins(monkeypatch, manager) + + assert await plugins_initializer.sync_plugins() is True + + manager.reload_plugin.assert_called_once_with("UpdatedPlugin") + manager.start.assert_not_called() + register.assert_called_once_with("UpdatedPlugin") + + +@pytest.mark.asyncio +async def test_sync_plugins_reloads_running_plugin_after_dependency_recovery( + monkeypatch, +) -> None: + """依赖恢复后，已运行的旧实例必须切换到新源码。""" + manager = MagicMock() + manager.sync.return_value = [] + manager.install_plugin_missing_dependencies_with_status.return_value = ( + PluginDependencyInstallResult(missing=["demo>=1"], success=True) + ) + manager.classify_plugins.return_value = PluginDependencyClassification( + ready=("DependencyRecovered",), + missing_dependencies=(), + missing_source=(), + ) + manager.running_plugins = {"DependencyRecovered": object()} + register = _patch_sync_plugins(monkeypatch, manager) + manager.get_plugin_runtime_statuses.return_value = { + "DependencyRecovered": PluginRuntimeStatus.DEPENDENCY_PENDING, + } + + assert await plugins_initializer.sync_plugins() is True + + manager.reload_plugin.assert_called_once_with("DependencyRecovered") + manager.start.assert_not_called() + register.assert_called_once_with("DependencyRecovered") + + +@pytest.mark.asyncio +async def test_sync_plugins_keeps_runtime_when_nothing_changed(monkeypatch) -> None: + """源码和依赖均无变化时保留首次初始化结果。""" + manager = MagicMock() + manager.sync.return_value = [] + manager.install_plugin_missing_dependencies_with_status.return_value = ( + PluginDependencyInstallResult(missing=[], success=True) + ) + manager.classify_plugins.return_value = PluginDependencyClassification( + ready=("ReadyPlugin",), + missing_dependencies=(), + missing_source=(), + ) + manager.running_plugins = {"ReadyPlugin": object()} + register = _patch_sync_plugins(monkeypatch, manager) + + assert await plugins_initializer.sync_plugins() is False + + manager.start.assert_not_called() + manager.reload_plugin.assert_not_called() + register.assert_not_called() + + +@pytest.mark.asyncio +async def test_sync_plugins_keeps_event_loop_responsive_during_activation( + monkeypatch, +) -> None: + """插件初始化运行在线程池时，Web 事件循环仍可继续调度。""" + manager = MagicMock() + manager.sync.return_value = [] + manager.install_plugin_missing_dependencies_with_status.return_value = ( + PluginDependencyInstallResult(missing=[], success=True) + ) + manager.classify_plugins.return_value = PluginDependencyClassification( + ready=("SlowPlugin",), + missing_dependencies=(), + missing_source=(), + ) + manager.running_plugins = {} + activation_started = threading.Event() + + def slow_start(_plugin_id: str) -> None: + activation_started.set() + time.sleep(0.1) + + manager.start.side_effect = slow_start + monkeypatch.setattr(plugins_initializer, "configure_plugin_services", lambda: None) + monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager) + monkeypatch.setattr(plugins_initializer, "register_plugin_api", MagicMock()) + monkeypatch.setattr( + plugins_initializer.global_vars, + "CURRENT_EVENT_LOOP", + asyncio.get_running_loop(), + ) + + sync_task = asyncio.create_task(plugins_initializer.sync_plugins()) + assert await asyncio.to_thread(activation_started.wait, 1) + assert sync_task.done() is False + assert await sync_task is True @pytest.mark.parametrize( @@ -98,6 +345,86 @@ def test_start_monitor_respects_runtime_configuration( _reset_plugin_manager() +def test_plugin_monitor_waits_until_dependency_settlement(monkeypatch) -> None: + """后台依赖收敛期间不启动文件监控，避免源码写入触发重复重载。""" + _reset_plugin_manager() + reset_plugin_system() + monkeypatch.setattr( + "app.runtime.extensions.plugin_manager.settings", + SimpleNamespace( + DEV=True, + PLUGIN_AUTO_RELOAD=False, + ROOT_PATH=MagicMock(), + ), + ) + manager = PluginManager() + start = MagicMock() + reload_monitor = MagicMock() + manager._plugin_monitor.start = start + manager._plugin_monitor.reload = reload_monitor + + manager.set_plugin_settling(True) + manager.start_monitor() + manager.reload_monitor() + + start.assert_not_called() + reload_monitor.assert_called_once_with(enabled=False) + + manager.set_plugin_settling(False) + manager.start_monitor() + + start.assert_called_once_with() + _reset_plugin_manager() + + +def test_plugin_monitor_skips_installing_plugin_until_package_write_finishes(tmp_path) -> None: + """安装替换目录期间，文件事件不得抢先导入未完成的插件包。""" + reload_plugin = MagicMock() + monitor = PluginChangeMonitor( + runtime_root=tmp_path, + local_roots=lambda: [], + stop_event=threading.Event(), + recent_sync={}, + federated_change=lambda _path: None, + runtime_plugin=lambda _path: "DemoPlugin", + local_candidate=lambda _path: None, + sync_local=MagicMock(), + reload_plugin=reload_plugin, + dependency_manifest_status=lambda _path: None, + watch=lambda *_args, **_kwargs: (), + log=MagicMock(), + monitor_suppressed=lambda plugin_id: plugin_id.lower() == "demoplugin", + ) + + monitor._process_changes({("modified", str(tmp_path / "demo" / "plugin.py"))}) + + reload_plugin.assert_not_called() + + +def test_plugin_monitor_suppression_is_reference_counted(monkeypatch) -> None: + """同一插件的重叠写入必须等最后一个事务退出后才解除监控抑制。""" + _reset_plugin_manager() + reset_plugin_system() + monkeypatch.setattr( + "app.runtime.extensions.plugin_manager.settings", + SimpleNamespace( + DEV=False, + PLUGIN_AUTO_RELOAD=False, + ROOT_PATH=MagicMock(), + ), + ) + manager = PluginManager() + + with manager.suppress_plugin_monitor("DemoPlugin"): + assert manager.is_plugin_monitor_suppressed("demoplugin") is True + with manager.suppress_plugin_monitor("demoplugin"): + assert manager.is_plugin_monitor_suppressed("DemoPlugin") is True + assert manager.is_plugin_monitor_suppressed("DemoPlugin") is True + + assert manager.is_plugin_monitor_suppressed("DemoPlugin") is False + _reset_plugin_manager() + + def test_config_change_reloads_monitor(monkeypatch) -> None: """配置热更新继续使用重建语义，不复用首次启动入口。""" _reset_plugin_manager() diff --git a/tests/test_plugin_registry.py b/tests/test_plugin_registry.py index f0f0781b1..23149dc72 100644 --- a/tests/test_plugin_registry.py +++ b/tests/test_plugin_registry.py @@ -1,6 +1,7 @@ from types import SimpleNamespace from app.runtime.extensions.plugin.registry import PluginRegistry +from app.schemas.plugin import PluginRuntimeStatus def test_registry_owns_classes_instances_and_stable_snapshots(): @@ -36,3 +37,24 @@ def test_registry_clear_preserves_compatibility_mapping_identity(): assert registry.running is running assert classes == {} assert running == {} + + +def test_registry_tracks_runtime_status_generation_and_settling(): + """状态与后台收敛变化只在真实改变时推进刷新代次。""" + registry = PluginRegistry() + + registry.set_runtime_status("Demo", PluginRuntimeStatus.READY) + first_generation = registry.generation + registry.set_runtime_status("Demo", PluginRuntimeStatus.READY) + registry.set_settling(True) + + assert registry.runtime_status("Demo") is PluginRuntimeStatus.READY + assert registry.runtime_status_snapshot() == { + "Demo": PluginRuntimeStatus.READY, + } + assert registry.generation == first_generation + 1 + assert registry.settling is True + + registry.remove("Demo") + + assert registry.runtime_status("Demo") is None diff --git a/tests/test_plugin_settlement_lifecycle.py b/tests/test_plugin_settlement_lifecycle.py new file mode 100644 index 000000000..b84836789 --- /dev/null +++ b/tests/test_plugin_settlement_lifecycle.py @@ -0,0 +1,66 @@ +import asyncio +from concurrent.futures import Future +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.startup import lifecycle + + +@pytest.mark.asyncio +async def test_runtime_ready_waits_for_scheduler_and_command_refresh(monkeypatch) -> None: + """插件 ready 只在调度任务和命令注册完成后对外可见。""" + order: list[str] = [] + manager = MagicMock() + command_future = Future() + + async def sync_plugins() -> bool: + order.append("plugins") + return True + + async def execute_task(_loop, task_func, _task_name): + task_func() + return [] + + monkeypatch.setattr(lifecycle.settings, "MOVIEPILOT_SAFE_MODE", False) + monkeypatch.setattr(lifecycle, "get_plugin_manager", lambda: manager) + monkeypatch.setattr(lifecycle, "sync_plugins", sync_plugins) + monkeypatch.setattr(lifecycle, "execute_task", execute_task) + monkeypatch.setattr( + lifecycle, + "init_plugin_scheduler", + lambda: order.append("scheduler"), + ) + monkeypatch.setattr( + lifecycle, + "restart_command", + lambda: (order.append("commands"), command_future)[1], + ) + monkeypatch.setattr(lifecycle, "SystemHelper", MagicMock()) + monkeypatch.setattr(lifecycle, "SystemChain", MagicMock()) + monkeypatch.setattr( + lifecycle.MoviePilotServerHelper, + "async_report_usage", + AsyncMock(), + ) + manager.set_plugin_settling.side_effect = lambda value: order.append( + f"settling:{value}" + ) + manager.start_monitor.side_effect = lambda: order.append("monitor") + + settle_task = asyncio.create_task(lifecycle.init_extra()) + await asyncio.sleep(0) + + assert order == ["plugins", "scheduler", "commands"] + manager.set_plugin_settling.assert_not_called() + + command_future.set_result(None) + await settle_task + + assert order == [ + "plugins", + "scheduler", + "commands", + "settling:False", + "monitor", + ]