diff --git a/app/adapters/cache/backends.py b/app/adapters/cache/backends.py index 874e13833..f1def0355 100644 --- a/app/adapters/cache/backends.py +++ b/app/adapters/cache/backends.py @@ -51,10 +51,11 @@ class FileBackend(CacheBackend): :param key: 区域内的相对缓存文件路径 :param value: 可写入二进制文件的字节数据；本层不序列化对象 + :param ttl: 新式TTL关键字，文件后端当前不管理过期；字符串第三位置参数沿用旧region合同 :param region: 根目录下的缓存区域名称，使用时须为字符串 :param _kwargs: 为缓存接口兼容保留，文件后端不处理这些选项 """ - cache_region = region or DEFAULT_CACHE_REGION + cache_region = ttl if isinstance(ttl, str) else (region or DEFAULT_CACHE_REGION) cache_path = self.base / cache_region / key cache_path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( @@ -186,10 +187,11 @@ class AsyncFileBackend(AsyncCacheBackend): :param key: 区域内的相对缓存文件路径 :param value: 可写入二进制文件的字节数据；本层不序列化对象 + :param ttl: 新式TTL关键字，文件后端当前不管理过期；字符串第三位置参数沿用旧region合同 :param region: 根目录下的缓存区域名称，使用时须为字符串 :param _kwargs: 为缓存接口兼容保留，文件后端不处理这些选项 """ - cache_region = region or DEFAULT_CACHE_REGION + cache_region = ttl if isinstance(ttl, str) else (region or DEFAULT_CACHE_REGION) cache_path = AsyncPath(self.base) / cache_region / key await cache_path.parent.mkdir(parents=True, exist_ok=True) async with aiofiles.tempfile.NamedTemporaryFile( diff --git a/app/adapters/network/browser.py b/app/adapters/network/browser.py index 8749b4382..c9661b775 100644 --- a/app/adapters/network/browser.py +++ b/app/adapters/network/browser.py @@ -486,17 +486,25 @@ class BrowserSessionHelper: :param owner: 可选宿主作用域；提供时会隔离并校验会话归属 :return: 回调函数返回值 """ - self._prune_sessions() - return self._run_in_session_thread( - session_key, - self._with_session_in_thread, - session_key, - callback, - user_agent=user_agent, - cookies=cookies, - timeout=timeout, - owner=owner, - ) + # 会话线程只能执行自己的浏览器任务；不能在其中同步等待其他会话关闭。 + with self._sessions_lock: + in_session_thread = threading.get_ident() in self._session_thread_ids.values() + if not in_session_thread: + self._prune_sessions() + try: + return self._run_in_session_thread( + session_key, + self._with_session_in_thread, + session_key, + callback, + user_agent=user_agent, + cookies=cookies, + timeout=timeout, + owner=owner, + ) + finally: + if not in_session_thread: + self._enforce_session_limit(protect_session_key=session_key) def _with_session_in_thread( self, @@ -1011,7 +1019,6 @@ class BrowserSessionHelper: self._close_session_state(session) return existing_session self._sessions[session_key] = session - self._enforce_session_limit(protect_session_key=session_key) return session @staticmethod diff --git a/app/adapters/system/plugin/dependency.py b/app/adapters/system/plugin/dependency.py index 9d05023b7..6ae7c4495 100644 --- a/app/adapters/system/plugin/dependency.py +++ b/app/adapters/system/plugin/dependency.py @@ -6,12 +6,13 @@ from __future__ import annotations import asyncio import json +import re from collections.abc import Callable from dataclasses import dataclass, field from importlib.metadata import PackageNotFoundError, distribution, distributions from pathlib import Path from typing import Any, Optional, Protocol -from urllib.parse import urlsplit +from urllib.parse import parse_qsl, unquote, urlsplit from packaging.markers import default_environment from packaging.requirements import Requirement @@ -258,8 +259,8 @@ class PluginDependencyInstaller: """ return (name or "").lower().replace("-", "_").replace(".", "_") - @staticmethod - def _direct_url_matches(installed_distribution: Any, required_url: str) -> bool: + @classmethod + def _direct_url_matches(cls, installed_distribution: Any, required_url: str) -> bool: """ 校验安装发行版记录的 PEP 610 URL 与清单来源一致。 """ @@ -267,15 +268,69 @@ class PluginDependencyInstaller: payload = installed_distribution.read_text("direct_url.json") if not payload: return False - direct_url = json.loads(payload).get("url") + metadata = json.loads(payload) + direct_url = metadata.get("url") if not isinstance(direct_url, str): return False - return PluginDependencyInstaller._canonical_direct_url( - required_url - ) == PluginDependencyInstaller._canonical_direct_url(direct_url) + required = urlsplit(required_url) + if "+" in required.scheme: + required = cls._vcs_source_url(required, metadata.get("vcs_info")) + if required is None: + return False + if cls._canonical_direct_url(required._replace(fragment="").geturl()) != cls._canonical_direct_url( + urlsplit(direct_url)._replace(fragment="").geturl() + ): + return False + return cls._direct_url_fragment_matches(required.fragment, metadata) except (AttributeError, json.JSONDecodeError, TypeError, ValueError): return False + @staticmethod + def _vcs_source_url(required: Any, vcs_info: Any) -> Any: + """ + 将VCS清单URL还原为PEP610仓库地址，并检查声明的版本或固定提交。 + + :param required: 已解析的清单URL + :param vcs_info: 安装元数据中的VCS身份 + :return: 去除VCS前缀和版本的URL；身份不匹配时为None + """ + vcs, scheme = required.scheme.split("+", 1) + if not isinstance(vcs_info, dict) or vcs_info.get("vcs") != vcs: + return None + path, separator, revision = required.path.rpartition("@") + if not separator: + return required._replace(scheme=scheme) + revision = unquote(revision) + commit = vcs_info.get("commit_id") + if re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", revision): + matched = isinstance(commit, str) and commit.casefold() == revision.casefold() + else: + matched = revision in {vcs_info.get("requested_revision"), commit} + return required._replace(scheme=scheme, path=path) if matched else None + + @staticmethod + def _direct_url_fragment_matches(fragment: str, metadata: dict[str, Any]) -> bool: + """ + 从PEP610独立字段核对子目录与归档摘要，不把URL片段静默丢弃。 + + :param fragment: 清单URL的片段字段 + :param metadata: 安装来源元数据 + :return: 全部来源约束都有对应证据时为True + """ + fields = dict(parse_qsl(fragment, keep_blank_values=True)) + if fields.pop("subdirectory", None) != metadata.get("subdirectory"): + return False + archive = metadata.get("archive_info") or {} + hashes = dict(archive.get("hashes") or {}) + if archive.get("hash"): + algorithm, _, digest = archive["hash"].partition("=") + hashes.setdefault(algorithm, digest) + return all( + isinstance(hashes.get(algorithm), str) + and hashes[algorithm].casefold() == digest.casefold() + for algorithm, digest in fields.items() + ) + @staticmethod def _canonical_direct_url(value: str) -> tuple[str, str, str, str, str]: """ diff --git a/app/adapters/web/plugin/routes.py b/app/adapters/web/plugin/routes.py index dc153f17b..8e7263f71 100644 --- a/app/adapters/web/plugin/routes.py +++ b/app/adapters/web/plugin/routes.py @@ -157,13 +157,11 @@ class FastAPIDynamicRouteRegistry: auth_mode = api.pop("auth", "apikey") dependencies = api.setdefault("dependencies", []) if not allow_anonymous: - if ( - auth_mode == "bear" - and Depends(self._verify_token) not in dependencies - ): - dependencies.append(Depends(self._verify_token)) - elif Depends(self._verify_apikey) not in dependencies: - dependencies.append(Depends(self._verify_apikey)) + authentication = Depends( + self._verify_token if auth_mode == "bear" else self._verify_apikey + ) + if authentication not in dependencies: + dependencies.append(authentication) # 插件 API 自行决定响应结构，不使用宿主统一 Response 路由。 api.setdefault("route_class_override", APIRoute) self._app.router.add_api_route(**api, tags=["plugin"]) @@ -191,11 +189,11 @@ class FastAPIDynamicRouteRegistry: """ if not plugin_id: return False - prefix = f"{self._prefix}/{plugin_id}/" + root_path = f"{self._prefix}/{plugin_id}" routes = [ route for route in self._app.routes if (path := self._route_path(route)) is not None - and path.startswith(prefix) + and (path == root_path or path.startswith(root_path + "/")) ] removed = False for route in routes: diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index cd81cfd2e..9de74b9b2 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -980,13 +980,6 @@ def uninstall_plugin(plugin_id: str, _: ApiPrincipal = Depends(get_current_activ message=f"请先卸载该插件的分身：{instance_ids}", ) config_oper = get_configured_system_config() - # 删除已安装信息 - install_plugins = config_oper.get(SystemConfigKey.UserInstalledPlugins) or [] - for plugin in install_plugins: - if plugin == plugin_id: - install_plugins.remove(plugin) - break - config_oper.set(SystemConfigKey.UserInstalledPlugins, install_plugins) # 移除插件API remove_plugin_api(plugin_id) # 移除插件服务 @@ -994,7 +987,12 @@ def uninstall_plugin(plugin_id: str, _: ApiPrincipal = Depends(get_current_activ # 判断是否为分身 plugin_class = plugin_manager.plugins.get(plugin_id) # 删除必须晚于停止：停机钩子会重建刚删的自有库；停止同时注销插件类，故删除一律按 force - plugin_manager.stop(plugin_id) + if plugin_manager.stop(plugin_id) is False: + return _SchemaResponse(success=False, message="插件后台服务尚未停止，已保留实例和数据供重试") + config_oper.update_atomically( + SystemConfigKey.UserInstalledPlugins, + lambda current: (None, [item for item in (current or []) if item != plugin_id]), + ) if virtual_instance: plugin_manager.delete_plugin_config(plugin_id, force=True) plugin_manager.delete_plugin_data(plugin_id, force=True) diff --git a/app/api/endpoints/pluginfolder.py b/app/api/endpoints/pluginfolder.py index b8edde77e..dd8a6c92c 100644 --- a/app/api/endpoints/pluginfolder.py +++ b/app/api/endpoints/pluginfolder.py @@ -44,7 +44,9 @@ async def save_plugin_folders( """ 保存插件文件夹分组配置。 """ - result = await get_plugin_folder_service().save(folders.root) + result = await get_plugin_folder_service().save( + folders.model_dump(mode="json", by_alias=True) + ) return Response(success=result.success, message=result.message) diff --git a/app/application/configuration.py b/app/application/configuration.py index b7b275357..6ace3c3bc 100644 --- a/app/application/configuration.py +++ b/app/application/configuration.py @@ -732,17 +732,18 @@ class SystemConfigService: normalized_value=normalized_value, ) - async def async_update_atomically( + def update_atomically( self, key: Any, mutation: Callable[[Any], tuple[T, Any]], ) -> T: """ - 在线程化短事务内原子读取旧值、规范化并写入新值。 - """ - if self._async_executor is None: - raise RuntimeError("系统配置异步数据库执行端口尚未配置") + 在持久化写锁和短事务内原子变更配置，不向应用回调暴露数据库会话。 + :param key: 系统配置键 + :param mutation: 根据最新值返回业务结果和新值的纯变更回调 + :return: 事务提交后返回变更回调的业务结果 + """ def apply(_session: Any, current: Any) -> tuple[T, Any]: """ 把不暴露数据库会话的应用层 mutation 适配到底层原子仓储。 @@ -750,12 +751,23 @@ class SystemConfigService: result, value = mutation(current) return result, self.normalize_value(key, value) - return cast( - T, - await self._async_executor.run( - partial(self._writer.update_atomically, key, apply) - ), - ) + return self._writer.update_atomically(key, apply) + + async def async_update_atomically( + self, + key: Any, + mutation: Callable[[Any], tuple[T, Any]], + ) -> T: + """ + 在线程化短事务内复用同步原子变更，避免事件循环中的阻塞数据库I/O。 + + :param key: 系统配置键 + :param mutation: 根据最新值返回业务结果和新值的纯变更回调 + :return: 事务提交后返回变更回调的业务结果 + """ + if self._async_executor is None: + raise RuntimeError("系统配置异步数据库执行端口尚未配置") + return cast(T, await self._async_executor.run(partial(self.update_atomically, key, mutation))) async def async_delete(self, key: Any) -> Any: """ diff --git a/app/application/plugin/gateway.py b/app/application/plugin/gateway.py index 3a5219d49..15b50918f 100644 --- a/app/application/plugin/gateway.py +++ b/app/application/plugin/gateway.py @@ -4,8 +4,8 @@ from __future__ import annotations -from collections.abc import Awaitable, Callable -from dataclasses import dataclass +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass, replace from datetime import datetime from typing import Protocol @@ -29,6 +29,7 @@ from app.application.plugin.source import ( list_effective_online_candidates, select_plugin_candidate, ) +from app.domain.plugin import PLUGIN_SYSTEM_VERSION_FIELD InventoryProvider = Callable[[bool], Awaitable[CandidateInventory]] IdentityReader = Callable[[str], Awaitable[PluginIdentity | None]] @@ -151,7 +152,7 @@ class PluginInstallGateway: now=self.__clock(), ) compatible, message = self.__candidate_compatibility( - admission.candidate + self._compatibility_candidate(admission.candidate, release_version) ) if not compatible: raise PluginSourceAdmissionError( @@ -174,6 +175,22 @@ class PluginInstallGateway: failure_stage="source_admission", ) + @staticmethod + def _compatibility_candidate(candidate: Candidate, release_version: str | None) -> Candidate: + """ + 历史远端Release的宿主范围交给实际版本选择，保留市场的来源和运行时要求。 + + :param candidate: 已通过来源与代际准入的冻结候选 + :param release_version: 请求安装的Release版本 + :return: 仅兼容校验使用的候选视图，不修改事务使用的原候选 + """ + if (not release_version or release_version == candidate.plugin_version + or isinstance(candidate, PluginLocalCandidate) or not isinstance(candidate.dto, Mapping)): + return candidate + metadata = dict(candidate.dto) + metadata.pop(PLUGIN_SYSTEM_VERSION_FIELD, None) + return replace(candidate, dto=metadata) + async def inspect_source( self, *, diff --git a/app/application/plugin/install.py b/app/application/plugin/install.py index f371d51a0..263923291 100644 --- a/app/application/plugin/install.py +++ b/app/application/plugin/install.py @@ -37,6 +37,8 @@ PluginMutationAdmission = Callable[[str], ContextManager[None]] PluginPackageWriteGuard = Callable[[str], ContextManager[None]] RestartRequiredRecorder = Callable[[str, tuple[str, ...]], None] LoadableMarker = Callable[[str], None] +HostEnabledReader = Callable[[str], bool] +HostEnabledRestorer = Callable[[str, bool], Awaitable[None]] T = TypeVar("T") @@ -240,6 +242,8 @@ class _InstallState: :param native_dependency_changes: 本次安装替换的已加载原生依赖 :param committed: 已确认数据库目标提交完成 :param commit_unknown: 数据库提交结果不确定，禁止按未提交状态盲目回滚 + :param host_enabled_before: 任何启用副作用之前冻结的本体启用位 + :param host_enabled_touched: 已尝试修改本体启用位，需补偿后才可删除journal """ transaction_id: str @@ -255,6 +259,8 @@ class _InstallState: native_dependency_changes: tuple[NativeDependencyChange, ...] = () committed: bool = False commit_unknown: bool = False + host_enabled_before: bool | None = None + host_enabled_touched: bool = False class PluginInstallCommand: @@ -268,6 +274,8 @@ class PluginInstallCommand: persistence: PluginPersistenceService, installed_plugins_reader: InstalledPluginsReader, loadable_marker: LoadableMarker, + host_enabled_reader: HostEnabledReader, + host_enabled_restorer: HostEnabledRestorer, plugin_ids_provider: PluginIdsProvider, packages: PluginPackageTransactionPort, install_reporter: InstallReporter, @@ -286,6 +294,8 @@ class PluginInstallCommand: self.__persistence = persistence self.__installed_plugins_reader = installed_plugins_reader self.__loadable_marker = loadable_marker + self.__host_enabled_reader = host_enabled_reader + self.__host_enabled_restorer = host_enabled_restorer self.__plugin_ids_provider = plugin_ids_provider self.__packages = packages self.__install_reporter = install_reporter @@ -370,6 +380,8 @@ class PluginInstallCommand: ) try: + state.host_enabled_before = self.__host_enabled_reader(plugin_id) + async def create_checkpoint() -> None: """ 在取消传播前保存已经创建完成的恢复材料引用。 @@ -393,6 +405,7 @@ class PluginInstallCommand: plugin_id=plugin_id, phase=PluginInstallationPhase.PREPARED, membership_before=membership_before, + host_enabled_before=state.host_enabled_before, membership_target=None, identity_before_revision=admission.expected_revision, identity_target_revision=None, @@ -509,6 +522,7 @@ class PluginInstallCommand: ) ) # 装载判据在实例表启用位上，不登记则插件装完当次能跑、重启即消失 + state.host_enabled_touched = True self.__loadable_marker(plugin_id) state.stage = "persistent_backup_stage" @@ -900,22 +914,36 @@ class PluginInstallCommand: runtime_restored = False registrations_restored = False - if file_restored and state.runtime_touched: + if file_restored and state.runtime_touched and state.host_enabled_before is not False: try: - await self.__rollback_reloader(plugin_id) + status = await self.__rollback_reloader(plugin_id) + if status is not PluginRuntimeStatus.ACTIVE: + raise RuntimeError(f"原插件重载状态为 {getattr(status, 'value', status)}") runtime_restored = True # 返回完整补偿诊断 except Exception as error: # noqa: BLE001 errors.append(f"插件运行态恢复失败：{error}") - if runtime_restored: - try: - await self.__registration_refresher(plugin_id) - registrations_restored = True - # 补偿失败需保留完整诊断，不覆盖原安装失败结果。 - except Exception as error: # noqa: BLE001 - errors.append(f"插件注册恢复失败：{error}") + host_restored = not state.host_enabled_touched + if file_restored and state.host_enabled_touched: + try: + if state.host_enabled_before is None: + raise RuntimeError("缺少安装前本体启用位快照") + await self.__host_enabled_restorer(plugin_id, state.host_enabled_before) + host_restored = True + except Exception as error: # noqa: BLE001 + errors.append(f"插件本体启用位恢复失败：{error}") + if file_restored and state.runtime_touched and state.host_enabled_before is False: + # 原本停用的插件没有待重载的运行实例；恢复端口已经确认候选停止。 + runtime_restored = host_restored + if runtime_restored and host_restored: + try: + await self.__registration_refresher(plugin_id) + registrations_restored = True + # 补偿失败需保留完整诊断，不覆盖原安装失败结果。 + except Exception as error: # noqa: BLE001 + errors.append(f"插件注册恢复失败：{error}") - rollback_complete = file_restored and ( + rollback_complete = file_restored and host_restored and ( not state.runtime_touched or (runtime_restored and registrations_restored) ) diff --git a/app/application/plugin/management.py b/app/application/plugin/management.py index 8e59b6f1e..2a8fbd9dd 100644 --- a/app/application/plugin/management.py +++ b/app/application/plugin/management.py @@ -335,13 +335,6 @@ async def uninstall_plugin_runtime(plugin_id: str) -> dict[str, Any]: raise ValueError(f"请先卸载该插件的分身：{instance_ids}") config_oper = get_configured_system_config() - install_plugins = config_oper.get(SystemConfigKey.UserInstalledPlugins) or [] - if plugin_id in install_plugins: - install_plugins = [plugin for plugin in install_plugins if plugin != plugin_id] - await config_oper.async_set( - SystemConfigKey.UserInstalledPlugins, - install_plugins, - ) remove_plugin_api(plugin_id) remove_plugin_job(plugin_id) @@ -352,7 +345,12 @@ async def uninstall_plugin_runtime(plugin_id: str) -> dict[str, Any]: # 删除数据必须发生在插件停止之后：插件的停机钩子只要取一次自有库句柄，就会把刚 # 删除的数据重新建出来。停止同时注销插件类，故其后的删除一律按 force 执行 - plugin_manager.stop(plugin_id) + if plugin_manager.stop(plugin_id) is False: + raise RuntimeError("插件后台服务尚未停止，已保留实例和数据供重试") + await config_oper.async_update_atomically( + SystemConfigKey.UserInstalledPlugins, + lambda current: (None, [item for item in (current or []) if item != plugin_id]), + ) if virtual_instance: plugin_manager.delete_plugin_config(plugin_id, force=True) diff --git a/app/application/plugin/recovery.py b/app/application/plugin/recovery.py index 8e4de82b1..144579962 100644 --- a/app/application/plugin/recovery.py +++ b/app/application/plugin/recovery.py @@ -8,6 +8,7 @@ from dataclasses import dataclass from typing import Protocol from app.application.plugin.install import ( + HostEnabledRestorer, PluginPackageCheckpoint, PluginPackageTransactionPort, ) @@ -87,12 +88,14 @@ class PluginInstallationRecoveryService: *, persistence: PluginPersistenceService, packages: PluginRecoveryPackagePort, + restore_host_enabled: HostEnabledRestorer | None = None, ) -> None: """ 保存数据库最终事实和文件恢复端口。 """ self.__persistence = persistence self.__packages = packages + self.__restore_host_enabled = restore_host_enabled async def replay(self) -> PluginInstallationRecoveryResult: """ @@ -153,6 +156,11 @@ class PluginInstallationRecoveryService: """ try: await self.__packages.async_restore(checkpoint) + if record.host_enabled_before is None: + raise PluginInstallationRecoveryError("旧journal缺少安装前本体启用位，需人工复核") + if self.__restore_host_enabled is None: + raise PluginInstallationRecoveryError("本体启用位恢复端口尚未配置") + await self.__restore_host_enabled(record.plugin_id, record.host_enabled_before) await self.__persistence.delete_installation( record.transaction_id, expected_phase=PluginInstallationPhase.PREPARED, diff --git a/app/application/plugin/transaction.py b/app/application/plugin/transaction.py index 324207754..2e1876a11 100644 --- a/app/application/plugin/transaction.py +++ b/app/application/plugin/transaction.py @@ -74,6 +74,7 @@ class PluginInstallationRecord: :param created_at: 审计事实首次创建时间，必须包含时区 :param updated_at: 审计事实最近更新时间，必须包含时区 :param schema_version: 安装日志结构版本，必须为当前支持版本 + :param host_enabled_before: 安装前本体启用位；旧日志未知时为None，不能推测恢复值 """ transaction_id: str @@ -88,6 +89,7 @@ class PluginInstallationRecord: created_at: datetime updated_at: datetime schema_version: int = INSTALLATION_JOURNAL_SCHEMA_VERSION + host_enabled_before: bool | None = None def __post_init__(self) -> None: """ @@ -105,6 +107,8 @@ class PluginInstallationRecord: if not isinstance(self.membership_before, bool): raise PluginInstallationRecordError("membership_before 必须是布尔值") + if self.host_enabled_before is not None and not isinstance(self.host_enabled_before, bool): + raise PluginInstallationRecordError("host_enabled_before 必须是布尔值或null") if not isinstance(self.package_existed, bool): raise PluginInstallationRecordError("package_existed 必须是布尔值") if not isinstance(self.persistent_backup_existed, bool): diff --git a/app/db/adapters/plugininstallation.py b/app/db/adapters/plugininstallation.py index 55294de8d..3842c8214 100644 --- a/app/db/adapters/plugininstallation.py +++ b/app/db/adapters/plugininstallation.py @@ -167,6 +167,7 @@ class TransactionalPluginInstallationStore(PluginInstallationStore): plugin_id=model.plugin_id, phase=PluginInstallationPhase(model.phase), membership_before=model.membership_before, + host_enabled_before=model.host_enabled_before, membership_target=model.membership_target, identity_before_revision=model.identity_before_revision, identity_target_revision=model.identity_target_revision, @@ -382,6 +383,7 @@ class TransactionalPluginInstallationStore(PluginInstallationStore): plugin_id=record.plugin_id, phase=record.phase.value, membership_before=record.membership_before, + host_enabled_before=record.host_enabled_before, membership_target=record.membership_target, identity_before_revision=record.identity_before_revision, identity_target_revision=record.identity_target_revision, diff --git a/app/db/models/plugindata.py b/app/db/models/plugindata.py index 0e60bc24e..10723f6ed 100644 --- a/app/db/models/plugindata.py +++ b/app/db/models/plugindata.py @@ -23,7 +23,7 @@ class PluginData(Base): value: Mapped[Optional[Any]] = mapped_column(JSON) __table_args__ = ( - Index('ix_plugindata_plugin_id_key', 'plugin_id', 'key'), + Index('ix_plugindata_plugin_id_key', 'plugin_id', 'key', unique=True), ) @classmethod diff --git a/app/db/models/plugininstallation.py b/app/db/models/plugininstallation.py index b3c384b6f..eb85740f4 100644 --- a/app/db/models/plugininstallation.py +++ b/app/db/models/plugininstallation.py @@ -20,6 +20,8 @@ class PluginInstallation(Base): plugin_id: Mapped[str] = mapped_column(String(128), nullable=False) phase: Mapped[str] = mapped_column(String(16), nullable=False) membership_before: Mapped[bool] = mapped_column(Boolean, nullable=False) + # 旧journal无法还原安装前启用位，NULL明确表示恢复证据不足。 + host_enabled_before: Mapped[Optional[bool]] = mapped_column(Boolean) membership_target: Mapped[Optional[bool]] = mapped_column(Boolean) identity_before_revision: Mapped[Optional[int]] = mapped_column(Integer) identity_target_revision: Mapped[Optional[int]] = mapped_column(Integer) diff --git a/app/db/oper/plugindata.py b/app/db/oper/plugindata.py index 5cec17d3b..9d450125f 100644 --- a/app/db/oper/plugindata.py +++ b/app/db/oper/plugindata.py @@ -2,9 +2,12 @@ 插件持久化数据的数据访问入口，适配调用方会话及兼容事务执行器。 """ -from typing import Any, Optional +from typing import Any, Optional, Union -from sqlalchemy import delete +from sqlalchemy import Executable, delete +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.dialects.sqlite import insert as sqlite_insert +from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session from app.db.base import DbOper @@ -24,37 +27,50 @@ class PluginDataOper(DbOper): :param key: 数据key :param value: 数据值 """ - plugin = self._execute_sync_query( - lambda session: PluginData.get_plugin_data_by_key( - session, plugin_id, key - ) - ) - if plugin: - self._stage_update(plugin, { - "value": value - }) - else: - self._stage_create(PluginData(plugin_id=plugin_id, key=key, value=value)) + def stage(session: Session) -> None: + """ + 在调用方事务内原子写入并刷新已加载的同键模型。 + """ + session.execute( + self._save_statement(session, plugin_id, key, value), + execution_options={"populate_existing": True}, + ).all() + + self._execute_sync_write(stage) async def async_save(self, plugin_id: str, key: str, value: Any) -> None: """ - 异步保存插件数据 + 在调用方异步事务中原子保存插件同键数据。 - :param plugin_id: 插件ID + :param plugin_id: 插件标识 :param key: 数据键 - :param value: 数据值 + :param value: 可JSON序列化的数据值 """ - plugin = await self._execute_async_query( - lambda session: PluginData.async_get_plugin_data_by_key( - session, plugin_id, key - ) - ) - if plugin: - await self._stage_async_update(plugin, {"value": value}) - else: - await self._stage_async_create( - PluginData(plugin_id=plugin_id, key=key, value=value) + async def stage(session: AsyncSession) -> None: + """ + 使用唯一键upsert，避免并发首次保存各自插入新行。 + """ + result = await session.execute( + self._save_statement(session, plugin_id, key, value), + execution_options={"populate_existing": True}, ) + result.all() + + await self._execute_async_write(stage) + + @staticmethod + def _save_statement( + session: Union[Session, AsyncSession], plugin_id: str, key: str, value: Any, + ) -> Executable: + """ + 为受支持的SQLite或PostgreSQL生成同一唯一键的原子写入语句。 + """ + insert = pg_insert if session.get_bind().dialect.name == "postgresql" else sqlite_insert + statement = insert(PluginData).values(plugin_id=plugin_id, key=key, value=value) + return statement.on_conflict_do_update( + index_elements=[PluginData.plugin_id, PluginData.key], + set_={"value": statement.excluded.value}, + ).returning(PluginData) def get_data(self, plugin_id: str, key: Optional[str] = None) -> Any: """ diff --git a/app/domain/plugin.py b/app/domain/plugin.py index 68fe3d0fc..9c530dd59 100644 --- a/app/domain/plugin.py +++ b/app/domain/plugin.py @@ -301,17 +301,15 @@ def build_plugin_release_install_plan( if release_version: if not is_release: return None, f"{plugin_id} 未声明 Release 安装，无法安装指定版本" - if not any( - item.get("version") == release_version for item in release_items - ): + selected_release = next((item for item in release_items if item.get("version") == release_version), None) + if selected_release is None: return None, f"{plugin_id} 未找到可安装的 Release 版本：{release_version}" - if release_version == plugin_version: - compatible, message = check_plugin_system_version( - metadata, - current_version=current_version, - ) - if not compatible: - return None, message + compatible, message = check_plugin_system_version( + metadata if release_version == plugin_version else selected_release, + current_version=current_version, + ) + if not compatible: + return None, message return PluginReleaseInstallPlan( release_tag=f"{plugin_id}_v{release_version}", fallback_to_filelist=False, diff --git a/app/locales/en-US.json b/app/locales/en-US.json index aa7a5531b..78125c293 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -592,7 +592,8 @@ "源文件路径为空": "Source file path is empty", "未识别到媒体类型": "Media type could not be identified", "站点域名未配置": "Site domain is not configured", - "站点配置不完整": "Site configuration is incomplete" + "站点配置不完整": "Site configuration is incomplete", + "插件后台服务尚未停止，已保留实例和数据供重试": "Plugin background services have not stopped. The instance and data were retained so you can retry." }, "message_patterns": [ { diff --git a/app/runtime/compat/resources.py b/app/runtime/compat/resources.py index a0ffcaed7..d41ef13d9 100644 --- a/app/runtime/compat/resources.py +++ b/app/runtime/compat/resources.py @@ -5,6 +5,7 @@ from __future__ import annotations import ast +import hashlib import threading import tokenize from dataclasses import dataclass @@ -46,7 +47,7 @@ RESOURCE_IMPORT_RULES: tuple[ResourceImportRule, ...] = ( _scan_cache_lock = threading.RLock() -_scan_cache: Dict[Path, Tuple[int, int, int, int, int, FrozenSet[str]]] = {} +_scan_cache: Dict[Path, Tuple[bytes, FrozenSet[str]]] = {} class PluginResourceImportScanError(RuntimeError): @@ -147,8 +148,7 @@ def _scan_source(plugin_id: str, path: Path) -> FrozenSet[str]: """ try: before_stat = path.stat() - # 热加载工具可能保留 mtime，等长替换也不会改变 size；ctime 与 inode/device - # 一并参与身份判断，避免把已替换源码误认为旧缓存。 + # 文件系统时间粒度不足以证明内容不变，每次读取内容摘要；只复用相同内容的AST结论。 cache_key = ( before_stat.st_mtime_ns, before_stat.st_ctime_ns, @@ -156,13 +156,9 @@ def _scan_source(plugin_id: str, path: Path) -> FrozenSet[str]: before_stat.st_dev, before_stat.st_ino, ) - with _scan_cache_lock: - cached = _scan_cache.get(path) - if cached and cached[:5] == cache_key: - return cached[5] with tokenize.open(path) as source_file: source = source_file.read() - tree = ast.parse(source, filename=str(path)) + content_key = hashlib.sha256(source.encode("utf-8")).digest() after_stat = path.stat() except (OSError, SyntaxError, UnicodeError) as error: raise PluginResourceImportScanError( @@ -180,6 +176,17 @@ def _scan_source(plugin_id: str, path: Path) -> FrozenSet[str]: f"扫描插件 {plugin_id} 时源码 {path.name} 发生变化" ) + with _scan_cache_lock: + cached = _scan_cache.get(path) + if cached and cached[0] == content_key: + return cached[1] + try: + tree = ast.parse(source, filename=str(path)) + except (SyntaxError, UnicodeError) as error: + raise PluginResourceImportScanError( + f"无法扫描插件 {plugin_id} 源码 {path.name}：{error}" + ) from error + capabilities: Set[str] = set() for module_name in _imported_modules(tree): for rule in RESOURCE_IMPORT_RULES: @@ -187,7 +194,7 @@ def _scan_source(plugin_id: str, path: Path) -> FrozenSet[str]: capabilities.add(rule.capability_id) result = frozenset(capabilities) with _scan_cache_lock: - _scan_cache[path] = (*cache_key, result) + _scan_cache[path] = (content_key, result) return result diff --git a/app/runtime/events.py b/app/runtime/events.py index edc6e6c05..bf1b2101a 100644 --- a/app/runtime/events.py +++ b/app/runtime/events.py @@ -790,7 +790,7 @@ class EventManager(metaclass=Singleton): self.__event_queue.put((event.priority, event)) record_metric( "event.queue.depth", - self.__event_queue.qsize(), + 1, delivery="broadcast", ) @@ -914,7 +914,7 @@ class EventManager(metaclass=Singleton): break record_metric( "event.queue.depth", - self.__event_queue.qsize(), + -1, delivery="broadcast", ) rate_limiter.reset() diff --git a/app/runtime/extensions/plugin/catalog.py b/app/runtime/extensions/plugin/catalog.py index af7955083..f8eb01b59 100644 --- a/app/runtime/extensions/plugin/catalog.py +++ b/app/runtime/extensions/plugin/catalog.py @@ -245,6 +245,8 @@ class PluginCatalogFacade: spec = importlib.util.find_spec(package_name) if spec is None or spec.origin is None: return False + if version is None: + return True local_version = self._plugin_attr(plugin_id, "plugin_version") if not local_version and instance: local_version = self._plugin_attr( diff --git a/app/runtime/extensions/plugin/lifecycle.py b/app/runtime/extensions/plugin/lifecycle.py index 6aba7ce82..700962c03 100644 --- a/app/runtime/extensions/plugin/lifecycle.py +++ b/app/runtime/extensions/plugin/lifecycle.py @@ -158,6 +158,17 @@ class PluginLifecycle: current_id = plugin.__name__ if plugin_id and current_id.casefold() != plugin_id.casefold(): continue + runtime_id = self._resolve_runtime_id(current_id) + if runtime_id in self._running: + # 未收敛实例仍由运行表持有，不能被重试构造的新实例覆盖。 + status = ( + PluginRuntimeStatus.LOAD_FAILED + if runtime_id in self._quiesced_hooks + else PluginRuntimeStatus.ACTIVE + ) + self._runtime_status_writer(plugin_id or current_id, status) + results[plugin_id or current_id] = status + continue try: if not self._auth_checker(plugin): self._remove_classification(current_id) @@ -175,6 +186,7 @@ class PluginLifecycle: self._gil_fallback_recorder, ): instance = plugin() + self._running[current_id] = instance instance.init_plugin(self._plugin_config(current_id)) self._ensure_database(current_id, instance) enabled = bool(instance.get_state()) @@ -200,9 +212,7 @@ class PluginLifecycle: status = PluginRuntimeStatus.LOAD_FAILED self._runtime_status_writer(plugin_id or current_id, status) results[plugin_id or current_id] = status - # 建库发生在进入运行态之前：失败的插件不会出现在 _running 里，卸载路径 - # 因此够不到它，句柄只能在这里释放 - self._release_databases((current_id,)) + self._cleanup_failed_candidate(current_id) self._logger.error( f"加载插件 {current_id} 出错：{error} - {traceback.format_exc()}" ) @@ -228,6 +238,20 @@ class PluginLifecycle: self._clear_tools() return results + def _cleanup_failed_candidate(self, plugin_id: str) -> None: + """ + 收尾已经构造的失败候选；未收敛时保留实例和数据库供后续停止重试。 + + :param plugin_id: 初始化失败的插件运行标识 + """ + instance = self._running.get(plugin_id) + if instance is not None: + if not self._quiesce_selected({plugin_id: instance}): + return + self._running.pop(plugin_id, None) + self._quiesced_hooks.pop(plugin_id, None) + self._release_databases((plugin_id,)) + @staticmethod def _declaration(instance: Any, hook_name: str) -> Any: """ @@ -306,18 +330,35 @@ class PluginLifecycle: ) @observe_plugin_lifecycle("stop") - def stop(self, plugin_id: Optional[str] = None) -> None: + def stop(self, plugin_id: Optional[str] = None) -> Optional[bool]: """ - 按旧单阶段 ABI 先解绑 handler，再停止并强制卸载插件。 + 先解绑handler并停止生产者，未收敛时返回False且保留owner；成功沿用None返回。 + + :param plugin_id: 指定插件标识；为空时停止全部实例 + :return: 停止或卸载失败时为False，成功时保留旧ABI的None """ with self._lifecycle_lock: plugins = self._select_running_plugins(plugin_id) - self._quiesce_selected(plugins) - self._finalize( + if not self._quiesce_selected(plugins): + self._record_stop_failure(plugins) + return False + if not self._finalize( plugin_id, - require_quiesced=False, - disable_events=not self._handlers_quiesced(plugins), - ) + require_quiesced=True, + disable_events=False, + ): + self._record_stop_failure(plugins) + return False + return None + + def _record_stop_failure(self, plugins: dict[str, Any]) -> None: + """ + 将停止失败的实例标记为加载失败，运行注册表继续持有其资源。 + + :param plugins: 本次停止入口冻结的实例快照 + """ + for current_id in plugins: + self._runtime_status_writer(current_id, PluginRuntimeStatus.LOAD_FAILED) @observe_plugin_lifecycle("quiesce") def quiesce(self, plugin_id: Optional[str] = None) -> bool: @@ -523,13 +564,16 @@ class PluginLifecycle: """ 重启指定插件并返回本次加载结果。 """ - self._runtime_status_writer(plugin_id, PluginRuntimeStatus.READY) - self.stop(plugin_id) - results = self.start(plugin_id) - status = next( - status - for result_id, status in results.items() - if result_id.casefold() == plugin_id.casefold() - ) - self._event_sender(reload_event, data={"plugin_id": plugin_id}) - return status + with self._lifecycle_lock: + self._runtime_status_writer(plugin_id, PluginRuntimeStatus.READY) + if self.stop(plugin_id) is False: + self._runtime_status_writer(plugin_id, PluginRuntimeStatus.LOAD_FAILED) + return PluginRuntimeStatus.LOAD_FAILED + results = self.start(plugin_id) + status = next( + status + for result_id, status in results.items() + if result_id.casefold() == plugin_id.casefold() + ) + self._event_sender(reload_event, data={"plugin_id": plugin_id}) + return status diff --git a/app/runtime/extensions/plugin/manager.py b/app/runtime/extensions/plugin/manager.py index 20d758212..c362937c5 100644 --- a/app/runtime/extensions/plugin/manager.py +++ b/app/runtime/extensions/plugin/manager.py @@ -376,7 +376,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): """ return self._plugin_tool_catalog.revision - def stop(self, pid: Optional[str] = None) -> None: + def stop(self, pid: Optional[str] = None) -> Optional[bool]: """ 停止插件服务。 @@ -385,9 +385,12 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): try: with self.mutation("停止插件"): with self._plugin_quiesce_lock: - self._plugin_lifecycle.stop(pid) + if self._plugin_lifecycle.stop(pid) is False: + return False except PluginMutationRejectedError as error: logger.warning(str(error)) + return False + return None def mutation(self, operation: str) -> ContextManager[None]: """ @@ -697,7 +700,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): self._monitor_suppressed_until.pop(normalized_id, None) return False - def remove_plugin(self, plugin_id: str) -> None: + def remove_plugin(self, plugin_id: str) -> Optional[bool]: """ 从内存中移除一个插件。 @@ -706,10 +709,13 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): try: with self.mutation("移除插件实例"): with self._plugin_quiesce_lock: - self._plugin_lifecycle.stop(plugin_id) + if self._plugin_lifecycle.stop(plugin_id) is False: + return False self._plugin_registry.remove(plugin_id) except PluginMutationRejectedError as error: logger.warning(str(error)) + return False + return None def remove_plugin_package(self, plugin_id: str) -> bool: """ @@ -1567,6 +1573,31 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): """ self._plugin_instance_store.enable_host(plugin_id) + def is_plugin_host_enabled(self, plugin_id: str) -> bool: + """ + 读取安装事务需要冻结的本体启用位，不把运行实例状态当作持久启用状态。 + + :param plugin_id: 物理插件标识 + :return: 本体行存在且处于启用状态时为True + """ + instance = self._plugin_instance_store.get_host(plugin_id) + return bool(instance and instance.is_enabled) + + def restore_plugin_host_enabled(self, plugin_id: str, enabled: bool) -> None: + """ + 恢复安装前启用位，原本停用时先确认候选资源停止，不清除用户展示与日志配置。 + + :param plugin_id: 物理插件标识 + :param enabled: 安装前冻结的本体启用位 + :raises RuntimeError: 候选运行实例尚未停止，不能宣称补偿完成 + """ + if enabled: + self._plugin_instance_store.enable_host(plugin_id) + return + if self.stop(plugin_id) is False: + raise RuntimeError(f"插件 {plugin_id} 尚未停止，启用位补偿未收敛") + self._plugin_instance_store.disable_host(plugin_id) + def disable_plugin_host(self, plugin_id: str) -> bool: """ 卸载收尾：停用源插件本体但保留其业务参数。 diff --git a/app/runtime/log.py b/app/runtime/log.py index 99ae95347..043098db6 100644 --- a/app/runtime/log.py +++ b/app/runtime/log.py @@ -751,9 +751,12 @@ class NonBlockingFileHandler: file_groups.setdefault(entry.file_path, []).append(entry) for file_path, entries in file_groups.items(): try: - handler = self._get_rotating_handler(file_path) - for entry in entries: - handler.handle(self._to_record(entry)) + with self._handlers_lock: + if file_path.parent.name == "subscriptions": + self._maintain_scoped_logs(file_path) + handler = self._get_rotating_handler(file_path) + for entry in entries: + handler.handle(self._to_record(entry)) except Exception as err: print(f"批量写入失败 {file_path}: {err}") for entry in entries: diff --git a/app/startup/initializers/plugins.py b/app/startup/initializers/plugins.py index 5546588a8..e67a2ac26 100644 --- a/app/startup/initializers/plugins.py +++ b/app/startup/initializers/plugins.py @@ -636,13 +636,27 @@ def configure_plugin_services() -> None: force=force, ) - async def reload_plugin_tree(plugin_id: str) -> PluginRuntimeStatus: + async def reload_plugin_tree( + plugin_id: str, *, restore_enabled: bool | None = None, + ) -> PluginRuntimeStatus: """ - 在线程池中重建源插件及其全部虚拟实例。 + 通过同一受管生命周期线程入口重建实例树或恢复安装前启用状态。 + + :param plugin_id: 物理插件标识 + :param restore_enabled: 不为None时恢复冻结启用位；否则重载实例树 + :return: 实例树重载状态；启用位恢复成功时为ACTIVE """ + def apply_lifecycle() -> PluginRuntimeStatus: + """ + 在生命周期owner中运行完整同步操作，恢复失败继续向应用端口传播。 + """ + if restore_enabled is not None: + plugin_manager.restore_plugin_host_enabled(plugin_id, restore_enabled) + return PluginRuntimeStatus.ACTIVE + return cast(PluginRuntimeStatus, plugin_manager.reload_plugin_tree(plugin_id)) + return await run_in_threadpool_to_completion( - plugin_manager.reload_plugin_tree, - plugin_id, + apply_lifecycle, ) async def refresh_plugin_registrations(plugin_id: str) -> None: @@ -655,12 +669,23 @@ def configure_plugin_services() -> None: target_id, ) + async def restore_host_enabled(plugin_id: str, enabled: bool) -> None: + """ + 在线程化owner中恢复本体启用位，并确认原本停用的候选资源已收敛。 + + :param plugin_id: 物理插件标识 + :param enabled: journal冻结的安装前启用位 + """ + await reload_plugin_tree(plugin_id, restore_enabled=enabled) + command = PluginInstallCommand( persistence=persistence, installed_plugins_reader=lambda: get_configured_system_config().get( SystemConfigKey.UserInstalledPlugins ) or [], loadable_marker=plugin_manager.mark_plugin_loadable, + host_enabled_reader=plugin_manager.is_plugin_host_enabled, + host_enabled_restorer=restore_host_enabled, plugin_ids_provider=plugin_manager.get_plugin_ids, packages=cast(PluginPackageTransactionPort, package_manager), install_reporter=lambda plugin_id, repo_url: ( @@ -694,6 +719,7 @@ def configure_plugin_services() -> None: PluginInstallationRecoveryService( persistence=persistence, packages=cast(PluginRecoveryPackagePort, package_manager), + restore_host_enabled=restore_host_enabled, ) ) configure_plugin_identity_migration( diff --git a/database/versions/b8d2f5a01962_plugin_data_identity.py b/database/versions/b8d2f5a01962_plugin_data_identity.py new file mode 100644 index 000000000..86c792d7e --- /dev/null +++ b/database/versions/b8d2f5a01962_plugin_data_identity.py @@ -0,0 +1,48 @@ +""" +合并插件同键旧重复数据，并为安装补偿保留本体启用位快照。 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "b8d2f5a01962" +down_revision = "e6a4c9f20813" +branch_labels = None +depends_on = None + + +_INDEX = "ix_plugindata_plugin_id_key" + + +def upgrade() -> None: + """ + 保留原同步读取最先可见的最低ID记录，去除其余同插件同键重复行后创建唯一索引。 + """ + connection = op.get_bind() + table = sa.table("plugindata", sa.column("id", sa.Integer), sa.column("plugin_id", sa.String), + sa.column("key", sa.String)) + keep_ids = sa.select(sa.func.min(table.c.id)).group_by(table.c.plugin_id, table.c.key) + connection.execute(table.delete().where(table.c.id.not_in(keep_ids))) + indexes = {index["name"]: index for index in sa.inspect(connection).get_indexes("plugindata")} + if _INDEX in indexes and not indexes[_INDEX].get("unique"): + op.drop_index(_INDEX, table_name="plugindata") + if _INDEX not in indexes or not indexes[_INDEX].get("unique"): + op.create_index(_INDEX, "plugindata", ["plugin_id", "key"], unique=True) + if "plugininstallation" in sa.inspect(connection).get_table_names(): + columns = {column["name"] for column in sa.inspect(connection).get_columns("plugininstallation")} + if "host_enabled_before" not in columns: + # 旧journal无法推断被提前覆盖的启用位，保留NULL表示需人工复核。 + op.add_column("plugininstallation", sa.Column("host_enabled_before", sa.Boolean, nullable=True)) + + +def downgrade() -> None: + """ + 恢复非唯一索引；升级已合并的旧重复值不重新制造。 + """ + connection = op.get_bind() + if "plugininstallation" in sa.inspect(connection).get_table_names(): + columns = {column["name"] for column in sa.inspect(connection).get_columns("plugininstallation")} + if "host_enabled_before" in columns: + op.drop_column("plugininstallation", "host_enabled_before") + op.drop_index(_INDEX, table_name="plugindata") + op.create_index(_INDEX, "plugindata", ["plugin_id", "key"], unique=False) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index d5d93561c..7d3cd151d 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -817,7 +817,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1129 | -| 内部导入边 | 9,707 | +| 内部导入边 | 9,709 | | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） | diff --git a/scripts/generate_agent_skill_docs.py b/scripts/generate_agent_skill_docs.py index 580050940..f3ba0ddad 100644 --- a/scripts/generate_agent_skill_docs.py +++ b/scripts/generate_agent_skill_docs.py @@ -18,7 +18,7 @@ if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) MOVIEPILOT_API_SKILL_VERSION = "60" -DATABASE_OPERATION_SKILL_VERSION = "12" +DATABASE_OPERATION_SKILL_VERSION = "13" DATABASE_TABLE_GUIDES: dict[str, tuple[str, str, str]] = { diff --git a/skills/database-operation/SKILL.md b/skills/database-operation/SKILL.md index 3007493c2..616279ac8 100644 --- a/skills/database-operation/SKILL.md +++ b/skills/database-operation/SKILL.md @@ -1,6 +1,6 @@ --- name: database-operation -version: 12 +version: 13 description: >- Use this skill when you need to inspect, query, maintain, or carefully modify the MoviePilot database. This skill uses the bundled scripts/mp-db.py helper, @@ -223,7 +223,7 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123" - Purpose: Stores plugin installation phase, membership target, identity revisions, and backup state. - Useful queries: Diagnosing interrupted installations, rollback conditions, and package or backup presence. - Write boundary: Owned by the plugin installation state machine; never advance phase or overwrite evidence manually. -- Columns: `id`, `transaction_id`, `plugin_id`, `phase`, `membership_before`, `membership_target`, `identity_before_revision`, `identity_target_revision`, `package_existed`, `persistent_backup_existed`, `created_at`, `updated_at`, `schema_version` +- Columns: `id`, `transaction_id`, `plugin_id`, `phase`, `membership_before`, `host_enabled_before`, `membership_target`, `identity_before_revision`, `identity_target_revision`, `package_existed`, `persistent_backup_existed`, `created_at`, `updated_at`, `schema_version` ### `plugininstance` - Purpose: Stores one row per shared-source plugin runtime instance, covering both clones and the host plugin itself (instance_id equals source_plugin_id, so equality identifies the host and inequality a clone), together with that instance's display overrides, its own log-level override and the moment that override expires, and the plugin's own configuration payload. A clone exists exactly while its row exists, so deleting the row uninstalls the clone and discards its configuration. diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 806e75bb8..42cdd3090 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -961,8 +961,8 @@ "runtime_only": true } }, - "edge_count": 9707, - "edge_sha256": "eb91f16e795a22fafba1b5742c8f4c591c9e254e6995799b2ce1b4d838275bc5", + "edge_count": 9709, + "edge_sha256": "d50236fde9c2615a7212c95348b872b2bae5a8513424c833bb4968a62d7d15c1", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3970,6 +3970,8 @@ "app.application.plugin.gateway -> app.application.plugin.inventory", "app.application.plugin.gateway -> app.application.plugin.lifecycle", "app.application.plugin.gateway -> app.application.plugin.source", + "app.application.plugin.gateway -> app.domain", + "app.application.plugin.gateway -> app.domain.plugin", "app.application.plugin.identity -> app.application", "app.application.plugin.identity -> app.application.plugin", "app.application.plugin.identity -> app.application.plugin.declaration", diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index f6c1b6b9d..b36c7fbfb 100644 --- a/tests/test_builtin_skill_boundaries.py +++ b/tests/test_builtin_skill_boundaries.py @@ -127,7 +127,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None: expected_versions = { "browser-use": "4", "command-dispatch": "2", - "database-operation": "12", + "database-operation": "13", "feedback-issue": "12", "anysearch": "3", "moviepilot-api": "60", diff --git a/tests/test_plugin_database_lifecycle.py b/tests/test_plugin_database_lifecycle.py index f8c079d97..856d7d011 100644 --- a/tests/test_plugin_database_lifecycle.py +++ b/tests/test_plugin_database_lifecycle.py @@ -580,7 +580,7 @@ def test_runtime_uninstall_stops_the_plugin_before_deleting_its_data(monkeypatch plugin_manager = _uninstall_manager() config = MagicMock() config.get.return_value = ["DemoPluginwork"] - config.async_set = AsyncMock() + config.async_update_atomically = AsyncMock() monkeypatch.setattr(plugin_management, "get_plugin_manager", lambda: plugin_manager) monkeypatch.setattr( plugin_management, @@ -619,8 +619,8 @@ def test_uninstall_virtual_instance_also_stops_before_deleting(monkeypatch): result = plugin_endpoint.uninstall_plugin("DemoPluginwork", None) assert result.success is True - config.set.assert_called_once_with( - SystemConfigKey.UserInstalledPlugins, - ["DemoPlugin"], - ) + config.update_atomically.assert_called_once() + key, mutation = config.update_atomically.call_args.args + assert key is SystemConfigKey.UserInstalledPlugins + assert mutation(["DemoPlugin"]) == (None, ["DemoPlugin"]) _assert_stop_precedes_deletion(plugin_manager) diff --git a/tests/test_plugin_endpoint.py b/tests/test_plugin_endpoint.py index a6bf1526f..d08ca402e 100644 --- a/tests/test_plugin_endpoint.py +++ b/tests/test_plugin_endpoint.py @@ -1397,10 +1397,10 @@ def test_uninstall_virtual_instance_never_removes_source_package(monkeypatch): result = uninstall_plugin("DemoPluginwork", None) assert result.success is True - config.set.assert_called_once_with( - SystemConfigKey.UserInstalledPlugins, - ["DemoPlugin"], - ) + config.update_atomically.assert_called_once() + key, mutation = config.update_atomically.call_args.args + assert key is SystemConfigKey.UserInstalledPlugins + assert mutation(["DemoPlugin"]) == (None, ["DemoPlugin"]) plugin_manager.delete_plugin_config.assert_called_once_with( "DemoPluginwork", force=True, diff --git a/tests/test_plugin_external_install_boundary.py b/tests/test_plugin_external_install_boundary.py index 45ade45cb..c34bb48a3 100644 --- a/tests/test_plugin_external_install_boundary.py +++ b/tests/test_plugin_external_install_boundary.py @@ -18,6 +18,7 @@ from app.application.plugin import management as plugin_management from app.runtime.config import global_vars from app.schemas.plugin import ( PluginInstance, + PluginRuntimeStatus, PluginSourceChangeRequest, PluginSourceIdentity, PluginSourceInstallRequest, @@ -779,6 +780,8 @@ def test_startup_composition_configures_external_helper_gateway(monkeypatch) -> gateway = Mock() sync_runner = Mock(return_value=(True, "installed")) async_runner = AsyncMock(return_value=(True, "installed")) + command_factory = Mock(return_value=Mock()) + recovery_factory = Mock(return_value=Mock()) monkeypatch.setattr( plugins_initializer, @@ -793,7 +796,7 @@ def test_startup_composition_configures_external_helper_gateway(monkeypatch) -> monkeypatch.setattr( plugins_initializer, "PluginInstallCommand", - lambda **_kwargs: Mock(), + command_factory, ) monkeypatch.setattr( plugins_initializer, @@ -803,7 +806,7 @@ def test_startup_composition_configures_external_helper_gateway(monkeypatch) -> monkeypatch.setattr( plugins_initializer, "PluginInstallationRecoveryService", - lambda **_kwargs: Mock(), + recovery_factory, ) monkeypatch.setattr( plugins_initializer, @@ -866,6 +869,21 @@ def test_startup_composition_configures_external_helper_gateway(monkeypatch) -> plugins_initializer.configure_plugin_services() + manager = plugins_initializer.get_plugin_manager() + manager.reload_plugin_tree.return_value = PluginRuntimeStatus.ACTIVE + reloader = command_factory.call_args.kwargs["target_reloader"] + restorer = recovery_factory.call_args.kwargs["restore_host_enabled"] + assert asyncio.run(reloader("DemoPlugin")) is PluginRuntimeStatus.ACTIVE + manager.reload_plugin_tree.assert_called_once_with("DemoPlugin") + for enabled in (True, False): + asyncio.run(restorer("DemoPlugin", enabled)) + assert manager.restore_plugin_host_enabled.call_args_list == [ + call("DemoPlugin", True), call("DemoPlugin", False), + ] + manager.restore_plugin_host_enabled.side_effect = RuntimeError("候选资源未停止") + with pytest.raises(RuntimeError, match="候选资源未停止"): + asyncio.run(restorer("DemoPlugin", False)) + monkeypatch.setattr( plugins_initializer, "get_plugin_instance_directory", diff --git a/tests/test_plugin_install_command.py b/tests/test_plugin_install_command.py index 22c665e4c..9114090f7 100644 --- a/tests/test_plugin_install_command.py +++ b/tests/test_plugin_install_command.py @@ -335,12 +335,14 @@ def _command( persistence=persistence, installed_plugins_reader=lambda: installed or [], loadable_marker=lambda _plugin_id: calls.append("mark_loadable"), + host_enabled_reader=lambda _plugin_id: True, + host_enabled_restorer=AsyncMock(), plugin_ids_provider=lambda: plugin_ids or [], packages=packages, install_reporter=reporter or default_reporter, target_reloader=target_reloader or AsyncMock(return_value=PluginRuntimeStatus.ACTIVE), - rollback_reloader=rollback_reloader or AsyncMock(), + rollback_reloader=rollback_reloader or AsyncMock(return_value=PluginRuntimeStatus.ACTIVE), registration_refresher=registration_refresher or AsyncMock(), mutation=mutation or (lambda _operation: nullcontext()), package_write_guard=package_write_guard @@ -366,6 +368,7 @@ def _journal_record( plugin_id="DemoPlugin", phase=phase, membership_before=True, + host_enabled_before=True, membership_target=True if committed else None, identity_before_revision=1, identity_target_revision=2 if committed else None, @@ -518,7 +521,7 @@ async def test_native_dependency_change_commits_payload_when_reload_waits_for_re ) package_restore = AsyncMock() package_cleanup = AsyncMock() - rollback_reloader = AsyncMock() + rollback_reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) registration_refresher = AsyncMock() restart_required_recorder = Mock() command, persistence, calls = _command( @@ -831,7 +834,7 @@ async def test_precommit_failure_restores_files_and_runtime( 每个提交前阶段失败都恢复文件，运行态失败还要恢复旧运行态。 """ package_restore = AsyncMock() - rollback_reloader = AsyncMock() + rollback_reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) registration_refresher = AsyncMock() async def failing_receipt(_plugin_id): @@ -894,7 +897,7 @@ async def test_precommit_failure_keeps_previous_declared_metadata_snapshot(): """ previous = _identity(version="1.0.0", revision=3) persistence = _PersistenceSpy([], identity=previous) - rollback_reloader = AsyncMock() + rollback_reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) target_reloader = AsyncMock(side_effect=RuntimeError("reload failed")) command, _, _ = _command( persistence=persistence, @@ -922,7 +925,7 @@ async def test_database_commit_failure_restores_runtime_before_deleting_journal( 数据库最终提交失败时，文件、运行态和 PREPARED journal 必须一起补偿。 """ package_restore = AsyncMock() - rollback_reloader = AsyncMock() + rollback_reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) persistence = _PersistenceSpy( [], commit_error=RuntimeError("commit failed"), @@ -950,7 +953,7 @@ async def test_cleanup_failure_keeps_committed_journal_for_replay(): 数据库已提交但清理失败时不能回滚载荷，journal 必须保留供启动重放。 """ package_finalize = AsyncMock(side_effect=RuntimeError("cleanup unavailable")) - rollback_reloader = AsyncMock() + rollback_reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) command, persistence, _ = _command( package_finalize_backup=package_finalize, rollback_reloader=rollback_reloader, @@ -1194,7 +1197,7 @@ async def test_cancelled_prepared_commit_rolls_back_after_outcome_check(): started = asyncio.Event() release = asyncio.Event() package_restore = AsyncMock() - rollback_reloader = AsyncMock() + rollback_reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) persistence = _PersistenceSpy([]) async def commit( @@ -1318,7 +1321,7 @@ async def test_unknown_commit_result_preserves_current_payload_and_journal(): 数据库最终状态无法读取时不得猜测回滚，必须留待启动恢复。 """ package_restore = AsyncMock() - rollback_reloader = AsyncMock() + rollback_reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) persistence = _PersistenceSpy([], commit_error=RuntimeError("commit failed")) persistence.get_installation = AsyncMock( side_effect=RuntimeError("database unavailable") diff --git a/tests/test_plugin_installation_migration.py b/tests/test_plugin_installation_migration.py index dfbee6e9a..3d394d409 100644 --- a/tests/test_plugin_installation_migration.py +++ b/tests/test_plugin_installation_migration.py @@ -57,7 +57,7 @@ def test_plugin_installation_migration_upgrade_downgrade_reupgrade( assert "plugininstallation" in inspector.get_table_names() assert { column["name"] for column in inspector.get_columns("plugininstallation") - } == {column.name for column in PluginInstallation.__table__.columns} + } == {column.name for column in PluginInstallation.__table__.columns} - {"host_enabled_before"} assert { index["name"] for index in inspector.get_indexes("plugininstallation") } == { diff --git a/tests/test_plugin_installation_recovery.py b/tests/test_plugin_installation_recovery.py index 7cb1411b5..1b95ff42e 100644 --- a/tests/test_plugin_installation_recovery.py +++ b/tests/test_plugin_installation_recovery.py @@ -68,6 +68,7 @@ def _record( plugin_id="DemoPlugin", phase=phase, membership_before=True, + host_enabled_before=True, membership_target=True if committed else None, identity_before_revision=1, identity_target_revision=2 if committed else None, @@ -159,6 +160,7 @@ async def test_prepared_replay_restores_before_releasing_journal() -> None: service = PluginInstallationRecoveryService( persistence=persistence, packages=packages, + restore_host_enabled=AsyncMock(), ) result = await service.replay() @@ -185,6 +187,7 @@ async def test_prepared_delete_failure_keeps_replayable_journal() -> None: service = PluginInstallationRecoveryService( persistence=persistence, packages=packages, + restore_host_enabled=AsyncMock(), ) first = await service.replay() @@ -219,6 +222,7 @@ async def test_prepared_restore_failure_does_not_block_other_plugins() -> None: service = PluginInstallationRecoveryService( persistence=persistence, packages=packages, + restore_host_enabled=AsyncMock(), ) result = await service.replay() diff --git a/tests/test_plugin_lifecycle_status.py b/tests/test_plugin_lifecycle_status.py index 8880d16f5..bbce1ba03 100644 --- a/tests/test_plugin_lifecycle_status.py +++ b/tests/test_plugin_lifecycle_status.py @@ -483,15 +483,15 @@ def test_legacy_stop_entry_remains_idempotent(): assert running == {} -def test_legacy_stop_force_finalizes_after_hook_failure(): +def test_legacy_stop_preserves_owner_after_hook_failure(): """ - 旧 stop 即使 hook 报错也释放实例，并保持历史 None 返回 ABI。 + 旧stop入口在hook报错时返回False并保留实例，成功场景另行保持None返回ABI。 """ order: list[str] = [] class DemoPlugin: """ - close 失败但仍应完成兼容卸载的测试插件。 + close失败后仍需保留所有权供后续重试的测试插件。 """ plugin_name = "演示插件" @@ -527,17 +527,18 @@ def test_legacy_stop_force_finalizes_after_hook_failure(): lifecycle._disable_events.reset_mock() lifecycle._clear_modules.reset_mock() - assert lifecycle.stop("DemoPlugin") is None + previous = running["DemoPlugin"] + assert lifecycle.stop("DemoPlugin") is False assert order == ["close", "stop_service"] lifecycle._disable_events.assert_called_once_with(DemoPlugin) - lifecycle._clear_modules.assert_called_once_with("DemoPlugin") - assert classes == {} - assert running == {} + lifecycle._clear_modules.assert_not_called() + assert classes["DemoPlugin"] is DemoPlugin + assert running["DemoPlugin"] is previous -def test_legacy_reload_continues_after_old_instance_hook_failure(): +def test_legacy_reload_stops_after_old_instance_hook_failure(): """ - 旧实例 hook 失败只影响诊断，不得阻止 reload 创建新实例。 + 旧实例hook失败必须阻止reload创建新实例，保留可追踪的资源owner。 """ instances: list[object] = [] stop_service = MagicMock() @@ -587,12 +588,9 @@ def test_legacy_reload_continues_after_old_instance_hook_failure(): result = lifecycle.reload("DemoPlugin", "plugin-reload") - assert result is PluginRuntimeStatus.ACTIVE - assert statuses["DemoPlugin"] is PluginRuntimeStatus.ACTIVE - assert running["DemoPlugin"] is not previous - assert len(instances) == 2 + assert result is PluginRuntimeStatus.LOAD_FAILED + assert statuses["DemoPlugin"] is PluginRuntimeStatus.LOAD_FAILED + assert running["DemoPlugin"] is previous + assert len(instances) == 1 stop_service.assert_called_once_with() - lifecycle._event_sender.assert_called_once_with( - "plugin-reload", - data={"plugin_id": "DemoPlugin"}, - ) + lifecycle._event_sender.assert_not_called() diff --git a/tests/test_plugin_monitor_lifecycle.py b/tests/test_plugin_monitor_lifecycle.py index b8fc17fbf..f0bed5ed6 100644 --- a/tests/test_plugin_monitor_lifecycle.py +++ b/tests/test_plugin_monitor_lifecycle.py @@ -980,8 +980,8 @@ async def test_quiesce_seals_runtime_until_new_lifespan_reopens(monkeypatch) -> assert manager.start("DemoPlugin") == { "DemoPlugin": PluginRuntimeStatus.LOAD_FAILED } - assert manager.stop("DemoPlugin") is None - assert manager.remove_plugin("DemoPlugin") is None + assert manager.stop("DemoPlugin") is False + assert manager.remove_plugin("DemoPlugin") is False assert ( manager.reload_plugin("DemoPlugin") is PluginRuntimeStatus.LOAD_FAILED diff --git a/tests/test_plugin_platform_browser_regressions.py b/tests/test_plugin_platform_browser_regressions.py new file mode 100644 index 000000000..ac21bd899 --- /dev/null +++ b/tests/test_plugin_platform_browser_regressions.py @@ -0,0 +1,75 @@ +""" +用隔离子进程验证浏览器并发容量淘汰不会互相等待。 +""" + +import json +import os +import subprocess +import sys +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from unittest.mock import patch + +from app.adapters.network.browser import BrowserSessionHelper +from tests.test_browser_helper import _FakeContext + + +def _concurrent_capacity_case(): + """ + 两个会话同时达到淘汰点，保留真实会话线程和关闭线程归属。 + """ + BrowserSessionHelper.MAX_SESSIONS = 1 + helper = BrowserSessionHelper() + original = BrowserSessionHelper._enforce_session_limit.__func__ + capacity_barrier = threading.Barrier(2) + contexts = [] + + def launch(**_kwargs): + """ + 仅替换真实浏览器资源边界，并记录创建线程。 + """ + context = _FakeContext() + contexts.append((context, threading.get_ident())) + return context + + def enforce(cls, protect_session_key=None): + """ + 固定两个新增会话互相进入容量淘汰的竞争窗口。 + """ + capacity_barrier.wait(5) + return original(cls, protect_session_key) + + try: + with patch.object(helper, "_launch_context", side_effect=launch), patch.object( + BrowserSessionHelper, "_enforce_session_limit", classmethod(enforce), + ), ThreadPoolExecutor(max_workers=2) as callers: + futures = [callers.submit(helper.with_session, key, lambda session: session.session_key) + for key in ("one", "two")] + assert {future.result(6) for future in futures} == {"one", "two"} + assert len(BrowserSessionHelper._sessions) <= 1 + finally: + BrowserSessionHelper.close_all_sessions() + assert not BrowserSessionHelper._session_executors + assert len(contexts) == 2 + assert all(context.closed and context.close_thread_id == thread_id + for context, thread_id in contexts) + print(json.dumps({"completed": True, "contexts_closed_on_owner_thread": 2})) + + +def test_parallel_session_capacity_eviction_converges_in_subprocess(tmp_path): + """ + 真正死锁由父进程超时终止，红测不会把不可回收线程留在pytest进程中。 + """ + env = dict(os.environ, CONFIG_DIR=str(tmp_path / "config"), DEBUG="false") + script = ( + "from app.testing.bootstrap import prepare_backend; prepare_backend(); " + "from tests.test_plugin_platform_browser_regressions import _concurrent_capacity_case; " + "_concurrent_capacity_case()" + ) + result = subprocess.run( + [sys.executable, "-c", script], cwd=Path(__file__).resolve().parents[1], + env=env, capture_output=True, text=True, timeout=12, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert '"completed": true' in result.stdout diff --git a/tests/test_plugin_platform_catalog_dependency_regressions.py b/tests/test_plugin_platform_catalog_dependency_regressions.py new file mode 100644 index 000000000..3539cba2a --- /dev/null +++ b/tests/test_plugin_platform_catalog_dependency_regressions.py @@ -0,0 +1,76 @@ +""" +验证停用插件源码存在判定与PEP610安装来源身份。 +""" + +import json +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from app.adapters.system.plugin.dependency import PluginDependencyInstaller +from app.runtime.extensions.plugin.catalog import PluginCatalogFacade + + +def test_disabled_physical_plugin_exists_without_running_instance(monkeypatch): + """ + 不带版本条件的存在性查询只需物理源码，不要求停用插件已有运行实例。 + """ + catalog = object.__new__(PluginCatalogFacade) + catalog._plugin_instance = lambda _: None + catalog._plugin_attr = Mock(return_value=None) + catalog._logger = Mock() + monkeypatch.setattr("app.runtime.extensions.plugin.catalog.importlib.util.find_spec", + lambda _: SimpleNamespace(origin="/isolated/demo/__init__.py")) + assert catalog.exists("Demo") + catalog._plugin_attr.assert_not_called() + assert not catalog.exists("Demo", "2") + catalog._plugin_attr.return_value = "1" + assert not catalog.exists("Demo", "2") + catalog._plugin_attr.return_value = "2" + assert catalog.exists("Demo", "2") + + +@pytest.mark.parametrize("revision,installed_revision,commit,expected", [ + ("v1", "v1", "a" * 40, True), + ("v1", "v2", "a" * 40, False), + ("a" * 40, "v1", "a" * 40, True), + ("a" * 40, "a" * 40, "b" * 40, False), + ("feature%2Ffix", "feature/fix", "a" * 40, True), +]) +def test_pep610_vcs_revision_and_commit_identity(revision, installed_revision, commit, expected): + """ + VCS清单URL按实际仓库与revision/commit匹配，不能把错误提交当作已满足。 + """ + payload = {"url": "https://example.com/acme/pkg.git", "vcs_info": { + "vcs": "git", "requested_revision": installed_revision, "commit_id": commit, + }} + distribution = SimpleNamespace(read_text=lambda _: json.dumps(payload)) + required = f"git+https://example.com/acme/pkg.git@{revision}" + assert PluginDependencyInstaller._direct_url_matches(distribution, required) is expected + + +@pytest.mark.parametrize("subdirectory,expected", [("src", True), ("other", False), (None, False)]) +def test_pep610_vcs_subdirectory_is_part_of_identity(subdirectory, expected): + """ + 同一仓库不同子目录的发行版不能混用。 + """ + payload = {"url": "https://example.com/pkg.git", "subdirectory": subdirectory, + "vcs_info": {"vcs": "git", "requested_revision": "v1", "commit_id": "a" * 40}} + distribution = SimpleNamespace(read_text=lambda _: json.dumps(payload)) + assert PluginDependencyInstaller._direct_url_matches( + distribution, "git+https://example.com/pkg.git@v1#subdirectory=src", + ) is expected + + +@pytest.mark.parametrize("hash_value,expected", [("abc123", True), ("bad", False), (None, False)]) +def test_pep610_archive_hash_is_verified_in_archive_info(hash_value, expected): + """ + 归档哈希从PEP610 archive_info验证，不因去除URL片段而放宽来源约束。 + """ + payload = {"url": "https://example.com/pkg.whl", + "archive_info": {"hashes": {"sha256": hash_value}}} + distribution = SimpleNamespace(read_text=lambda _: json.dumps(payload)) + assert PluginDependencyInstaller._direct_url_matches( + distribution, "https://example.com/pkg.whl#sha256=abc123", + ) is expected diff --git a/tests/test_plugin_platform_coverage_regressions.py b/tests/test_plugin_platform_coverage_regressions.py new file mode 100644 index 000000000..bde703e64 --- /dev/null +++ b/tests/test_plugin_platform_coverage_regressions.py @@ -0,0 +1,110 @@ +""" +补充插件停止拒绝和本体启用位恢复的真实入口边界。 +""" + +from contextlib import nullcontext +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from app.application import scheduling +from app.application.plugin import folders, management, routes +from app.runtime.extensions.plugin.manager import PluginManager +from tests.test_plugin_instance_enablement import _host, _store + + +@pytest.mark.asyncio +async def test_async_uninstall_preserves_membership_and_data_when_stop_is_refused(monkeypatch): + """ + 异步卸载遇到真实停止失败回执时保留成员、配置、实例和数据，允许稍后重试。 + """ + manager = Mock() + manager.mutation.side_effect = lambda _operation: nullcontext() + manager.get_plugin_instance.return_value = SimpleNamespace(instance_id="ReviewClone") + manager.get_plugin_source_instances.return_value = [] + manager.plugins = {} + manager.stop.return_value = False + configuration = SimpleNamespace(async_update_atomically=AsyncMock()) + monkeypatch.setattr(management, "get_plugin_manager", lambda: manager) + monkeypatch.setattr(management, "get_configured_system_config", lambda: configuration) + monkeypatch.setattr(routes, "remove_plugin_api", Mock()) + monkeypatch.setattr(scheduling, "remove_plugin_job", Mock()) + remove_folder = Mock() + monkeypatch.setattr(folders, "remove_plugin_from_folders", remove_folder) + with pytest.raises(RuntimeError, match="尚未停止"): + await management.uninstall_plugin_runtime("ReviewClone") + manager.stop.assert_called_once_with("ReviewClone") + configuration.async_update_atomically.assert_not_awaited() + manager.delete_plugin_config.assert_not_called() + manager.delete_plugin_data.assert_not_called() + manager.delete_plugin_instance.assert_not_called() + manager.remove_plugin.assert_not_called() + remove_folder.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("package_result", [True, False, RuntimeError("清理旧分身文件失败")]) +async def test_legacy_clone_uninstall_reports_actual_package_cleanup(monkeypatch, package_result): + """ + 旧版磁盘分身清理按真实结果回执，拒绝或异常不能误报文件已删除。 + """ + manager = Mock() + manager.mutation.side_effect = lambda _operation: nullcontext() + manager.get_plugin_instance.return_value = None + manager.get_plugin_source_instances.return_value = [] + manager.plugins = {"LegacyClone": SimpleNamespace(is_clone=True)} + manager.stop.return_value = None + if isinstance(package_result, Exception): + manager.remove_plugin_package.side_effect = package_result + else: + manager.remove_plugin_package.return_value = package_result + configuration = SimpleNamespace(async_update_atomically=AsyncMock()) + monkeypatch.setattr(management, "get_plugin_manager", lambda: manager) + monkeypatch.setattr(management, "get_configured_system_config", lambda: configuration) + monkeypatch.setattr(routes, "remove_plugin_api", Mock()) + monkeypatch.setattr(scheduling, "remove_plugin_job", Mock()) + monkeypatch.setattr(folders, "remove_plugin_from_folders", Mock()) + result = await management.uninstall_plugin_runtime("LegacyClone") + expected = package_result is True + assert result == {"was_clone": True, "clone_files_removed": expected} + configuration.async_update_atomically.assert_awaited_once() + manager.remove_plugin_package.assert_called_once_with("LegacyClone") + manager.delete_plugin_config.assert_called_once_with("LegacyClone", force=True) + manager.delete_plugin_data.assert_called_once_with("LegacyClone", force=True) + manager.remove_plugin.assert_called_once_with("LegacyClone") + + +@pytest.mark.parametrize("enabled", [True, False]) +def test_manager_restores_host_enabled_without_discarding_its_descriptor(enabled): + """ + 通过真实实例存储恢复冻结启用位，保留已有展示字段且不会为启用操作停止实例。 + """ + host = _host("ReviewHost", enabled=not enabled).model_copy(update={"plugin_name": "保留名称"}) + store, directory = _store({"ReviewHost": host}) + manager = object.__new__(PluginManager) + manager._plugin_instance_store = store + manager.stop = Mock(return_value=None) + assert manager.is_plugin_host_enabled("MissingHost") is False + assert manager.is_plugin_host_enabled("ReviewHost") is not enabled + manager.restore_plugin_host_enabled("ReviewHost", enabled) + assert manager.is_plugin_host_enabled("ReviewHost") is enabled + assert directory.records["ReviewHost"].plugin_name == "保留名称" + if enabled: + manager.stop.assert_not_called() + else: + manager.stop.assert_called_once_with("ReviewHost") + + +def test_manager_does_not_persist_disabled_flag_until_running_owner_stops(): + """ + 停止失败时不能先写停用位，原实例及其描述留给下一轮补偿。 + """ + store, directory = _store({"ReviewHost": _host("ReviewHost")}) + manager = object.__new__(PluginManager) + manager._plugin_instance_store = store + manager.stop = Mock(return_value=False) + with pytest.raises(RuntimeError, match="尚未停止"): + manager.restore_plugin_host_enabled("ReviewHost", False) + assert manager.is_plugin_host_enabled("ReviewHost") is True + assert directory.records["ReviewHost"].is_enabled is True diff --git a/tests/test_plugin_platform_data_regressions.py b/tests/test_plugin_platform_data_regressions.py new file mode 100644 index 000000000..dd16e8659 --- /dev/null +++ b/tests/test_plugin_platform_data_regressions.py @@ -0,0 +1,210 @@ +""" +验证插件同键持久化、源码资源热更新与公开文件缓存兼容回归。 +""" + +import asyncio +import importlib +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Barrier + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.orm import Session + +from app.db.models.plugindata import PluginData +from app.db.oper.plugindata import PluginDataOper +from app.runtime.compat.resources import scan_plugin_resource_imports +from app.sdk.cache import AsyncFileBackend, FileBackend + + +def test_concurrent_plugin_key_save_has_one_sync_and_async_value(tmp_path): + """ + 同步首次保存同键的竞争不能产生重复行，异步读取必须返回同一有效值。 + """ + database = tmp_path / "plugin-data.db" + engine = sa.create_engine(f"sqlite:///{database}", connect_args={"timeout": 10}) + PluginData.__table__.create(engine) + first_reads = Barrier(2) + writers = Barrier(2) + + def synchronize_legacy_reads(_connection, _cursor, statement, _parameters, _context, _executemany): + """ + 将旧版读后插入的两个首次查询固定在同一竞争窗口。 + """ + if statement.lstrip().upper().startswith("SELECT") and "plugindata" in statement: + first_reads.wait(5) + + sa.event.listen(engine, "after_cursor_execute", synchronize_legacy_reads) + + def save(value): + """ + 每个写者持有独立事务，提交由调用方负责。 + """ + writers.wait(5) + with Session(engine) as session, session.begin(): + PluginDataOper(session).save("Demo", "key", value) + + try: + with ThreadPoolExecutor(max_workers=2) as pool: + list(pool.map(save, ["A", "B"])) + finally: + sa.event.remove(engine, "after_cursor_execute", synchronize_legacy_reads) + with Session(engine) as session: + rows = session.execute(sa.select(PluginData)).scalars().all() + assert len(rows) == 1 + expected = PluginDataOper(session).get_data("Demo", "key") + assert expected in {"A", "B"} + + async def read_async(): + """ + 用独立异步会话核对同一物理数据文件。 + """ + async_engine = create_async_engine(f"sqlite+aiosqlite:///{database}") + try: + async with async_sessionmaker(async_engine)() as session: + return await PluginDataOper(session).async_get_data("Demo", "key") + finally: + await async_engine.dispose() + + assert asyncio.run(read_async()) == expected + engine.dispose() + + +def test_plugin_key_unique_migration_preserves_previously_visible_value(monkeypatch): + """ + 升级去除旧重复行并保留同步读取原先可见的最低ID值，降级只移除唯一约束。 + """ + migration = importlib.import_module("database.versions.b8d2f5a01962_plugin_data_identity") + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + table = sa.Table("plugindata", metadata, + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("plugin_id", sa.String, nullable=False), + sa.Column("key", sa.String, nullable=False), sa.Column("value", sa.JSON)) + sa.Index("ix_plugindata_plugin_id_key", table.c.plugin_id, table.c.key) + with engine.begin() as connection: + metadata.create_all(connection) + connection.execute(table.insert(), [ + {"id": 1, "plugin_id": "Demo", "key": "key", "value": "visible"}, + {"id": 2, "plugin_id": "Demo", "key": "key", "value": "duplicate"}, + {"id": 3, "plugin_id": "Other", "key": "key", "value": "other"}, + ]) + monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection))) + migration.upgrade() + migration.upgrade() + rows = connection.execute(sa.select(table).order_by(table.c.id)).mappings().all() + assert [(row["plugin_id"], row["value"]) for row in rows] == [("Demo", "visible"), ("Other", "other")] + index = next(index for index in sa.inspect(connection).get_indexes("plugindata") + if index["name"] == "ix_plugindata_plugin_id_key") + assert index["unique"] + migration.downgrade() + index = next(index for index in sa.inspect(connection).get_indexes("plugindata") + if index["name"] == "ix_plugindata_plugin_id_key") + assert not index["unique"] + engine.dispose() + + +@pytest.mark.asyncio +async def test_async_concurrent_upsert_and_caller_rollback_are_visible_consistently(tmp_path): + """ + 异步首次写入也不能重复，已加载模型刷新但调用方回滚仍恢复原值。 + """ + database = tmp_path / "async-plugin-data.db" + engine = create_async_engine(f"sqlite+aiosqlite:///{database}", connect_args={"timeout": 10}) + async with engine.begin() as connection: + await connection.run_sync(PluginData.__table__.create) + sessions = async_sessionmaker(engine, expire_on_commit=False) + ready = asyncio.Event() + + async def save(value): + """ + 每个异步写者使用独立事务，并在共同事件后开始。 + """ + async with sessions() as session, session.begin(): + await ready.wait() + await PluginDataOper(session).async_save("Demo", "key", value) + + try: + writers = [asyncio.create_task(save(value)) for value in ("first", "second")] + ready.set() + await asyncio.gather(*writers) + async with sessions() as session: + rows = (await session.execute(sa.select(PluginData))).scalars().all() + assert len(rows) == 1 + original = rows[0].value + await PluginDataOper(session).async_save("Demo", "key", "temporary") + assert rows[0].value == "temporary" + await session.rollback() + async with sessions() as session: + assert await PluginDataOper(session).async_get_data("Demo", "key") == original + finally: + await engine.dispose() + + +def test_sync_upsert_refreshes_loaded_model_without_committing(tmp_path): + """ + 同步upsert更新调用方已加载对象，同时保留调用方对提交和回滚的完整控制。 + """ + engine = sa.create_engine(f"sqlite:///{tmp_path / 'rollback.db'}") + PluginData.__table__.create(engine) + try: + with Session(engine) as session, session.begin(): + PluginDataOper(session).save("Demo", "key", "original") + with Session(engine) as session: + loaded = session.execute(sa.select(PluginData)).scalar_one() + PluginDataOper(session).save("Demo", "key", "temporary") + assert loaded.value == "temporary" + session.rollback() + with Session(engine) as session: + assert PluginDataOper(session).get_data("Demo", "key") == "original" + finally: + engine.dispose() + + +def test_resource_scan_observes_equal_size_change_with_identical_stat(tmp_path, monkeypatch): + """ + 热更新在相同时间粒度内等长改写时，缓存必须按实际内容失效，无需sleep。 + """ + plugin = tmp_path / "Demo" + plugin.mkdir() + source = plugin / "__init__.py" + source.write_text("import cloakbrowsex\n", encoding="utf-8") + initial = source.stat() + real_stat = Path.stat + monkeypatch.setattr(Path, "stat", lambda path, *args, **kwargs: initial if path == source else real_stat(path, *args, **kwargs)) + assert scan_plugin_resource_imports("Demo", plugin) == () + source.write_text("import cloakbrowser\n", encoding="utf-8") + assert scan_plugin_resource_imports("Demo", plugin) == ("host.display",) + source.write_text("import cloakbrowsex\n", encoding="utf-8") + assert scan_plugin_resource_imports("Demo", plugin) == () + + +def test_sdk_file_cache_preserves_third_positional_region(tmp_path): + """ + 公开同步文件后端沿用旧第三位置region，同时接纳新ttl关键字。 + """ + cache = FileBackend(tmp_path) + cache.set("same", b"images", "images") + cache.set("same", b"lyrics", "lyrics") + cache.set("new", b"ttl", region="images", ttl=30) + assert cache.get("same", "images") == b"images" + assert cache.get("same", "lyrics") == b"lyrics" + assert cache.get("new", "images") == b"ttl" + + +@pytest.mark.asyncio +async def test_sdk_async_file_cache_preserves_third_positional_region(tmp_path): + """ + 公开异步文件后端的旧第三位置region不再混写默认区域。 + """ + cache = AsyncFileBackend(tmp_path) + await cache.set("same", b"images", "images") + await cache.set("same", b"lyrics", "lyrics") + await cache.set("new", b"ttl", region="images", ttl=30) + assert await cache.get("same", "images") == b"images" + assert await cache.get("same", "lyrics") == b"lyrics" + assert await cache.get("new", "images") == b"ttl" diff --git a/tests/test_plugin_platform_endpoint_regressions.py b/tests/test_plugin_platform_endpoint_regressions.py new file mode 100644 index 000000000..17e8bb504 --- /dev/null +++ b/tests/test_plugin_platform_endpoint_regressions.py @@ -0,0 +1,122 @@ +""" +验证动态插件认证、路由卸载与文件夹JSON持久化边界。 +""" + +from contextlib import nullcontext +from unittest.mock import Mock + +import pytest +import sqlalchemy as sa +from fastapi import Depends, FastAPI, Header, HTTPException +from fastapi.testclient import TestClient + +from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry +from app.api.endpoints import pluginfolder +from app.application.plugin.folders import PluginFolderService +from app.schemas.plugin import PluginFoldersData + + +def test_existing_bearer_dependency_does_not_also_require_apikey(): + """ + 插件已声明同一bearer依赖时，认证去重不能意外增加API密钥要求。 + """ + app = FastAPI() + + def bearer(authorization: str = Header("")): + """ + 只接受本测试的合成Bearer凭据。 + """ + if authorization != "Bearer synthetic": + raise HTTPException(401) + + def apikey(x_api_key: str = Header("")): + """ + API密钥入口保持独立，缺失时拒绝。 + """ + if x_api_key != "synthetic-key": + raise HTTPException(403) + + def endpoint(): + """ + 返回可观察的插件业务响应。 + """ + return {"ok": True} + + api = {"path": "/Demo", "endpoint": endpoint, "methods": ["GET"], + "auth": "bear", "dependencies": [Depends(bearer)]} + registry = FastAPIDynamicRouteRegistry( + app=app, plugin_ids=lambda: ["Demo"], plugin_apis=lambda _: [api], + verify_token=bearer, verify_apikey=apikey, prefix="/plugin", log=Mock(), + ) + registry.update("Demo", "add") + with TestClient(app) as client: + assert client.get("/plugin/Demo", headers={"Authorization": "Bearer synthetic"}).status_code == 200 + assert client.get("/plugin/Demo").status_code == 401 + assert len(api["dependencies"]) == 1 + + +def test_unloading_root_and_children_preserves_other_plugin_prefix(): + """ + 卸载同时删除插件根与子路由，不误删共享文本前缀的其他插件。 + """ + app = FastAPI() + + def endpoint(): + """ + 提供内存路由的成功响应。 + """ + return {"ok": True} + + registry = FastAPIDynamicRouteRegistry( + app=app, plugin_ids=lambda: ["Demo"], + plugin_apis=lambda _: [ + {"path": path, "endpoint": endpoint, "methods": ["GET"], "allow_anonymous": True} + for path in ("/Demo", "/Demo/child", "/DemoOther") + ], + verify_token=lambda: None, verify_apikey=lambda: None, + prefix="/plugin", log=Mock(), + ) + registry.update("Demo", "add") + registry.update("Demo", "remove") + with TestClient(app) as client: + assert client.get("/plugin/Demo").status_code == 404 + assert client.get("/plugin/Demo/child").status_code == 404 + assert client.get("/plugin/DemoOther").status_code == 200 + + +@pytest.mark.asyncio +async def test_folder_endpoint_persists_objects_and_legacy_lists_as_json(tmp_path, monkeypatch): + """ + 实际JSON列必须接收可序列化对象，保存后保留showIcon别名与旧列表格式。 + """ + engine = sa.create_engine(f"sqlite:///{tmp_path / 'folders.db'}") + table = sa.Table("folder_config", sa.MetaData(), sa.Column("value", sa.JSON)) + table.metadata.create_all(engine) + + async def write(folders): + """ + 在真实临时SQLite事务中触发JSON序列化和持久化。 + """ + with engine.begin() as connection: + connection.execute(table.insert().values(value=folders)) + + service = PluginFolderService( + read=dict, write=write, write_sync=lambda _: None, + mutation=lambda _: nullcontext(), + ) + monkeypatch.setattr(pluginfolder, "get_plugin_folder_service", lambda: service) + payload = PluginFoldersData.model_validate({ + "object": {"plugins": ["Demo"], "showIcon": True, "color": "red"}, + "legacy": ["Other"], + }) + try: + result = await pluginfolder.save_plugin_folders(payload, None) + assert result.success + with engine.connect() as connection: + saved = connection.execute(sa.select(table.c.value)).scalar_one() + assert saved["object"]["showIcon"] is True + assert "show_icon" not in saved["object"] + assert saved["object"]["plugins"] == ["Demo"] + assert saved["legacy"] == ["Other"] + finally: + engine.dispose() diff --git a/tests/test_plugin_platform_install_regressions.py b/tests/test_plugin_platform_install_regressions.py new file mode 100644 index 000000000..b07cfb1f8 --- /dev/null +++ b/tests/test_plugin_platform_install_regressions.py @@ -0,0 +1,141 @@ +""" +验证安装回滚运行状态与历史Release的版本准入。 +""" + +from dataclasses import replace +from unittest.mock import AsyncMock + +import pytest + +from app.application.plugin.gateway import PluginInstallGateway +from app.application.plugin.install import PluginInstallResult +from app.domain.plugin import ( + build_plugin_release_install_plan, + check_plugin_runtime_compatibility, + check_plugin_system_version, +) +from app.schemas.plugin import PluginRuntimeStatus +from tests.test_plugin_install_command import _admission, _command +from tests.test_plugin_install_gateway import NOW, REPO_URL, _inventory + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [PluginRuntimeStatus.LOAD_FAILED, PluginRuntimeStatus.BLOCKED_BY_POLICY, + PluginRuntimeStatus.DEPENDENCY_PENDING]) +async def test_rollback_requires_successful_old_runtime_and_keeps_journal(status): + """ + 回滚重载返回非active时不宣称恢复，不删journal或执行注册刷新。 + """ + refresh = AsyncMock() + command, persistence, _ = _command( + target_reloader=AsyncMock(return_value=PluginRuntimeStatus.LOAD_FAILED), + rollback_reloader=AsyncMock(return_value=status), + registration_refresher=refresh, + ) + result = await command.execute(admission=_admission(), release_version=None, force=True) + assert not result.success + assert result.rollback.runtime_attempted + assert not result.rollback.runtime_restored + assert not result.rollback.journal_deleted + assert persistence.records + assert any(status.value in error for error in result.rollback.errors) + refresh.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_failed_install_restores_previously_disabled_host(monkeypatch): + """ + 启用位已持久修改后后续阶段失败，补偿仍恢复安装前停用状态。 + """ + enabled = [False] + command, persistence, _ = _command( + package_activate_backup=AsyncMock(side_effect=RuntimeError("activate failed")), + ) + + async def restore(_plugin_id, previous): + """ + 模拟实际启用位恢复端口。 + """ + enabled[0] = previous + + monkeypatch.setattr(command, "_PluginInstallCommand__host_enabled_reader", lambda _: enabled[0], raising=False) + monkeypatch.setattr(command, "_PluginInstallCommand__host_enabled_restorer", restore, raising=False) + monkeypatch.setattr(command, "_PluginInstallCommand__loadable_marker", lambda _: enabled.__setitem__(0, True)) + result = await command.execute(admission=_admission(), release_version=None, force=True) + assert not result.success + assert enabled[0] is False + assert result.rollback.journal_deleted + assert not persistence.records + + +@pytest.mark.asyncio +async def test_disabled_host_rollback_does_not_require_activating_old_plugin(monkeypatch): + """ + 安装前本体停用时，补偿应停止候选并恢复停用位，不强行重载旧插件才算成功。 + """ + old_reload = AsyncMock(return_value=PluginRuntimeStatus.LOAD_FAILED) + restore = AsyncMock() + command, persistence, _ = _command( + target_reloader=AsyncMock(return_value=PluginRuntimeStatus.LOAD_FAILED), + rollback_reloader=old_reload, + ) + monkeypatch.setattr(command, "_PluginInstallCommand__host_enabled_reader", lambda _: False) + monkeypatch.setattr(command, "_PluginInstallCommand__host_enabled_restorer", restore) + result = await command.execute(admission=_admission(), release_version=None, force=True) + assert not result.success + assert result.rollback.runtime_restored + assert result.rollback.journal_deleted + assert not persistence.records + old_reload.assert_not_awaited() + restore.assert_awaited_once_with("DemoPlugin", False) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("release,free_threaded,expected", [("1.0", False, True), (None, False, False), + ("2.0", False, False), ("1.0", True, False)]) +async def test_gateway_historical_release_keeps_source_runtime_checks(release, free_threaded, expected): + """ + 指定旧Release不受最新包宿主版本限制，同时保留当前Release与运行时限制。 + """ + inventory = _inventory() + candidate = replace(inventory.market_reads[0].candidates[0], plugin_version="2.0", dto={ + "version": "2.0", "release": True, "v3": True, "v3t": False, "system_version": ">=99", + }) + inventory = replace(inventory, market_reads=(replace(inventory.market_reads[0], candidates=(candidate,)),)) + + def compatibility(selected): + """ + 使用真实领域规则分别检查运行时和宿主版本。 + """ + runtime = check_plugin_runtime_compatibility(selected.dto, free_threaded=free_threaded) + return runtime if not runtime[0] else check_plugin_system_version(selected.dto, current_version="3.1.0") + + executor = AsyncMock() + executor.execute.return_value = PluginInstallResult(success=True) + gateway = PluginInstallGateway( + inventory=AsyncMock(return_value=inventory), identity=AsyncMock(return_value=None), + candidate_compatibility=compatibility, executor=executor, clock=lambda: NOW, + source_plugin_id=lambda value: value, + ) + result = await gateway.install(plugin_id="DemoPlugin", repo_url=REPO_URL, + explicit_source=True, release_version=release) + assert result.success is expected + if expected: + assert executor.execute.await_args.kwargs["release_version"] == release + assert executor.execute.await_args.kwargs["admission"].candidate is candidate + else: + executor.execute.assert_not_awaited() + assert candidate.dto["system_version"] == ">=99" + + +@pytest.mark.parametrize("requirement,allowed", [(">=3,<4", True), (">=99", False)]) +def test_selected_release_uses_its_own_host_constraint(requirement, allowed): + """ + 历史Release若提供宿主约束，应验证该版本自身的约束而非最新版本的约束。 + """ + plan, _ = build_plugin_release_install_plan( + plugin_id="DemoPlugin", metadata={"release": True, "version": "2.0", "system_version": ">=99"}, + release_version="1.0", release_items=[{"version": "1.0", "system_version": requirement}], + current_version="3.1.0", + ) + assert (plan is not None) is allowed diff --git a/tests/test_plugin_platform_journal_regressions.py b/tests/test_plugin_platform_journal_regressions.py new file mode 100644 index 000000000..9bac96083 --- /dev/null +++ b/tests/test_plugin_platform_journal_regressions.py @@ -0,0 +1,156 @@ +""" +验证本体启用快照持久化、崩溃恢复及旧journal未知证据处理。 +""" + +import importlib +from dataclasses import replace +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.orm import sessionmaker + +from app.application.plugin.recovery import PluginInstallationRecoveryService +from app.application.plugin.transaction import PluginInstallationPhase, PluginPersistenceService +from app.db.adapters.plugininstallation import TransactionalPluginInstallationStore +from app.db.models.plugininstallation import PluginInstallation +from app.db.models.plugininstance import PluginInstance +from app.db.models.systemconfig import SystemConfig +from tests.test_plugin_install_transaction import _AtomicSystemConfig, _record +from tests.test_plugin_installation_recovery import _packages, _Persistence + + +@pytest.mark.parametrize("enabled", [True, False]) +def test_host_enabled_snapshot_survives_new_repository_instance(tmp_path, enabled): + """ + 新建仓储重新读取后仍能取得安装前的真实启用位，不依赖进程内finally状态。 + """ + engine = sa.create_engine(f"sqlite:///{tmp_path / 'journal.db'}") + for model in (SystemConfig, PluginInstallation): + model.__table__.create(engine) + factory = sessionmaker(engine, expire_on_commit=False) + config = _AtomicSystemConfig(factory) + try: + first = TransactionalPluginInstallationStore(factory, config.update_atomically) + first.create(replace(_record(), host_enabled_before=enabled)) + restored = TransactionalPluginInstallationStore(factory, config.update_atomically).get("txn-demo-1") + assert restored.host_enabled_before is enabled + finally: + engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("enabled", [True, False]) +async def test_startup_restores_enabled_snapshot_before_deleting_journal(enabled): + """ + 模拟安装后崩溃，启动恢复先还原文件和启用位，再删journal与恢复材料。 + """ + record = replace(_record(), host_enabled_before=enabled) + persistence = _Persistence([record]) + order = [] + packages = _packages(async_restore=AsyncMock(side_effect=lambda _: order.append("files"))) + + async def restore_host(plugin_id, previous): + """ + journal仍在时恢复其持久启用位。 + """ + assert persistence.records + assert plugin_id == record.plugin_id and previous is enabled + order.append("enabled") + + service = PluginInstallationRecoveryService( + persistence=persistence, packages=packages, restore_host_enabled=restore_host, + ) + result = await service.replay() + assert result.restored == 1 and result.failed == 0 + assert order == ["files", "enabled"] + assert not persistence.records + + +@pytest.mark.asyncio +async def test_legacy_unknown_host_enabled_journal_is_retained(): + """ + 旧journal缺少被提前覆盖的启用位，不能猜测并删除唯一恢复证据。 + """ + record = _record() + persistence = _Persistence([record]) + packages = _packages() + restorer = AsyncMock() + service = PluginInstallationRecoveryService( + persistence=persistence, packages=packages, restore_host_enabled=restorer, + ) + result = await service.replay() + assert result.failed == 1 and result.restored == 0 + assert persistence.records[record.transaction_id].phase is PluginInstallationPhase.PREPARED + packages.async_restore.assert_awaited_once() + packages.async_cleanup.assert_not_awaited() + restorer.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("enabled", [True, False]) +async def test_known_enabled_snapshot_without_restorer_retains_real_journal(tmp_path, enabled): + """ + 真实journal已知旧启用位但当前状态相反时，缺少恢复端口必须保留唯一恢复证据。 + """ + engine = sa.create_engine(f"sqlite:///{tmp_path / 'missing-restorer.db'}") + for model in (SystemConfig, PluginInstallation, PluginInstance): + model.__table__.create(engine) + factory = sessionmaker(engine, expire_on_commit=False) + config = _AtomicSystemConfig(factory) + store = TransactionalPluginInstallationStore(factory, config.update_atomically) + record = replace(_record(), host_enabled_before=enabled) + store.create(record) + with factory.begin() as session: + session.add(PluginInstance(instance_id=record.plugin_id, source_plugin_id=record.plugin_id, + is_enabled=not enabled, created_at=record.created_at.isoformat(), updated_at=record.updated_at.isoformat())) + + async def run(operation): + """ + 在测试线程执行独占临时数据库的持久化操作。 + """ + return operation() + + packages = _packages() + persistence = PluginPersistenceService(executor=SimpleNamespace(run=run), identities=Mock(), installations=store) + service = PluginInstallationRecoveryService(persistence=persistence, packages=packages) + try: + result = await service.replay() + assert result.failed == 1 and result.restored == 0 + retained = TransactionalPluginInstallationStore(factory, config.update_atomically).get(record.transaction_id) + assert retained is not None and retained.host_enabled_before is enabled + with factory() as session: + current = session.execute(sa.select(PluginInstance)).scalar_one() + assert current.is_enabled is not enabled + packages.async_restore.assert_awaited_once() + packages.async_cleanup.assert_not_awaited() + finally: + engine.dispose() + + +def test_journal_enabled_migration_preserves_unknown_and_round_trips(monkeypatch): + """ + 独立升级将旧事务启用位保留NULL，两次升级幂等，降级不改变原journal字段。 + """ + migration = importlib.import_module("database.versions.b8d2f5a01962_plugin_data_identity") + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + data = sa.Table("plugindata", metadata, sa.Column("id", sa.Integer, primary_key=True), + sa.Column("plugin_id", sa.String), sa.Column("key", sa.String)) + sa.Index("ix_plugindata_plugin_id_key", data.c.plugin_id, data.c.key) + journal = sa.Table("plugininstallation", metadata, sa.Column("id", sa.Integer, primary_key=True), + sa.Column("transaction_id", sa.String)) + with engine.begin() as connection: + metadata.create_all(connection) + connection.execute(journal.insert().values(id=1, transaction_id="old")) + monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection))) + migration.upgrade() + migration.upgrade() + assert connection.execute(sa.text("SELECT host_enabled_before FROM plugininstallation")).scalar_one() is None + migration.downgrade() + assert "host_enabled_before" not in {column["name"] for column in sa.inspect(connection).get_columns("plugininstallation")} + assert connection.execute(sa.select(journal.c.transaction_id)).scalar_one() == "old" + engine.dispose() diff --git a/tests/test_plugin_platform_lifecycle_regressions.py b/tests/test_plugin_platform_lifecycle_regressions.py new file mode 100644 index 000000000..bbb4019d6 --- /dev/null +++ b/tests/test_plugin_platform_lifecycle_regressions.py @@ -0,0 +1,184 @@ +""" +验证插件候选初始化失败及停止失败时的资源所有权。 +""" + +import threading +from unittest.mock import Mock + +import pytest + +from app.runtime.extensions.plugin.database import PluginDatabase +from app.runtime.extensions.plugin.manager import PluginManager +from app.schemas.plugin import PluginRuntimeStatus +from tests.test_plugin_database_lifecycle import _build_lifecycle + + +@pytest.mark.parametrize("stage", ["init", "database", "state"]) +def test_failed_candidate_closes_started_thread_before_releasing_database(stage): + """ + 初始化、建库或状态查询失败时仍收尾已启动线程，数据库只在收敛后释放。 + """ + stopped = threading.Event() + created = [] + released = [] + + class Candidate: + """ + 初始化启动真实线程并按参数触发后续故障的隔离插件。 + """ + plugin_name = "candidate" + plugin_version = "1" + + def __init__(self): + """ + 记录唯一实例并创建尚未启动的线程。 + """ + created.append(self) + self.thread = threading.Thread(target=stopped.wait) + + def init_plugin(self, _config): + """ + 模拟初始化已产生资源后发生异常。 + """ + self.thread.start() + if stage == "init": + raise RuntimeError("init failed") + + def get_state(self): + """ + 模拟初始化之后的状态投影失败。 + """ + if stage == "state": + raise RuntimeError("state failed") + return True + + def close(self): + """ + 关闭并等待本实例持有的生产者线程。 + """ + stopped.set() + self.thread.join(5) + + def ensure(_plugin_id, _models, _migrations): + """ + 在需要时模拟插件自有数据库初始化失败。 + """ + if stage == "database": + raise RuntimeError("database failed") + + def release(plugin_id): + """ + 数据库关闭不得先于依赖数据库的生产者停止。 + """ + released.append((plugin_id, created[0].thread.is_alive())) + + running = {} + lifecycle = _build_lifecycle( + running=running, load_plugins=lambda *_: [Candidate], + database=lambda: PluginDatabase(ensure=ensure, release=release), + ) + try: + assert lifecycle.start("Candidate") == {"Candidate": PluginRuntimeStatus.LOAD_FAILED} + assert not created[0].thread.is_alive() + assert released == [("Candidate", False)] + assert not running + finally: + stopped.set() + created[0].thread.join(5) + + +def test_failed_candidate_retains_owner_and_blocks_duplicate_initialization(): + """ + 候选关闭失败时保留所有权，重试启动不能覆盖旧实例，随后可再次停止。 + """ + created = [] + allow_close = [False] + release = Mock() + + class Candidate: + """ + 初始化失败且第一次关闭不收敛的插件。 + """ + plugin_name = "candidate" + plugin_version = "1" + + def __init__(self): + """ + 记录每次构造以检测资源owner被重复创建。 + """ + created.append(self) + + def init_plugin(self, _config): + """ + 在生命周期已经持有实例时触发初始化失败。 + """ + raise RuntimeError("init failed") + + def close(self): + """ + 允许后续显式停止重试收敛。 + """ + return allow_close[0] + + running = {} + lifecycle = _build_lifecycle( + running=running, load_plugins=lambda *_: [Candidate], + database=lambda: PluginDatabase(release=release), + ) + assert lifecycle.start("Candidate")["Candidate"] is PluginRuntimeStatus.LOAD_FAILED + assert running["Candidate"] is created[0] + release.assert_not_called() + assert lifecycle.start("Candidate")["Candidate"] is PluginRuntimeStatus.LOAD_FAILED + assert len(created) == 1 + allow_close[0] = True + assert lifecycle.stop("Candidate") is not False + assert not running + release.assert_called_once_with("Candidate") + + +@pytest.mark.parametrize("hook", ["close", "stop_service"]) +def test_stop_and_reload_preserve_nonconverged_owner(hook): + """ + 停止hook未收敛时不丢弃owner，不释放数据库，也不允许reload创建新实例。 + """ + plugin = Mock() + plugin.close.return_value = None + plugin.stop_service.return_value = None + getattr(plugin, hook).return_value = False + running = {"Demo": plugin} + classes = {"Demo": type(plugin)} + loader, clear_modules, release, events = Mock(return_value=[]), Mock(), Mock(), Mock() + statuses = {} + lifecycle = _build_lifecycle( + running=running, classes=classes, load_plugins=loader, + clear_modules=clear_modules, event_sender=events, + runtime_status_writer=statuses.__setitem__, + database=lambda: PluginDatabase(release=release), + ) + assert lifecycle.stop("Demo") is False + assert running["Demo"] is plugin + assert statuses["Demo"] is PluginRuntimeStatus.LOAD_FAILED + assert lifecycle.reload("Demo", "reloaded") is PluginRuntimeStatus.LOAD_FAILED + loader.assert_not_called() + clear_modules.assert_not_called() + release.assert_not_called() + events.assert_not_called() + getattr(plugin, hook).return_value = None + assert lifecycle.stop("Demo") is not False + assert not running + release.assert_called_once_with("Demo") + + +def test_manager_remove_keeps_registry_when_lifecycle_stop_fails(): + """ + 管理器兼容移除入口必须传播停止失败，不能再次从共享注册表丢弃活owner。 + """ + manager = object.__new__(PluginManager) + from contextlib import nullcontext + manager.mutation = lambda _: nullcontext() + manager._plugin_quiesce_lock = threading.RLock() + manager._plugin_lifecycle = Mock() + manager._plugin_lifecycle.stop.return_value = False + manager._plugin_registry = Mock() + assert manager.remove_plugin("Demo") is False + manager._plugin_registry.remove.assert_not_called() diff --git a/tests/test_plugin_platform_membership_regressions.py b/tests/test_plugin_platform_membership_regressions.py new file mode 100644 index 000000000..7c9ba25f7 --- /dev/null +++ b/tests/test_plugin_platform_membership_regressions.py @@ -0,0 +1,126 @@ +""" +验证同步和异步卸载在真实SQLite清单上的原子成员更新。 +""" + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor +from contextlib import nullcontext +from copy import deepcopy +from unittest.mock import Mock + +import pytest +import sqlalchemy as sa +from sqlalchemy.orm import sessionmaker + +from app.api.endpoints import plugin as endpoint +from app.application import scheduling +from app.application.configuration import SystemConfigService +from app.application.plugin import folders, management, routes +from app.db.models.systemconfig import SystemConfig +from app.schemas.types import SystemConfigKey +from tests.test_plugin_install_transaction import _AtomicSystemConfig + + +class _MembershipConfig(_AtomicSystemConfig): + """ + 在真实SQLite中提供清单读写，旧分离读写在屏障处稳定产生丢更新。 + + :ivar readers: 将旧两个清单快照固定在同一竞争窗口的屏障 + """ + + def __init__(self, factory): + """ + 保存独立会话工厂和竞争窗口。 + """ + super().__init__(factory) + self.readers = threading.Barrier(2) + + def get(self, key): + """ + 读取已提交清单，在旧实现的首次快照之间同步。 + """ + with self._factory() as session: + result = deepcopy(session.execute(sa.select(SystemConfig.value).where( + SystemConfig.key == key.value, + )).scalar_one()) + self.readers.wait(5) + return result + + def set(self, key, value): + """ + 兼容旧接口的独立全量覆盖写入。 + """ + with self._factory.begin() as session: + session.execute(sa.update(SystemConfig).where(SystemConfig.key == key.value).values(value=value)) + return True + + def update_atomically(self, key, mutation): + """ + 与安装事务使用相同配置写锁和短事务更新成员。 + """ + return super().update_atomically(key.value, mutation) + + +class _ThreadExecutor: + """ + 提供与生产异步数据库执行器相同的独立线程调用边界。 + """ + + async def run(self, operation): + """ + 等待单个短事务结束。 + """ + return await asyncio.to_thread(operation) + + +@pytest.mark.parametrize("paths", [("sync", "sync"), ("sync", "async"), ("async", "async")]) +def test_concurrent_uninstalls_preserve_other_members_and_install_update(paths, tmp_path, monkeypatch): + """ + 两条卸载入口删除不同插件时均保留另一请求及后续原子安装的变更。 + """ + engine = sa.create_engine(f"sqlite:///{tmp_path / 'membership.db'}", connect_args={"timeout": 5}) + SystemConfig.__table__.create(engine) + factory = sessionmaker(engine, expire_on_commit=False) + with factory.begin() as session: + session.add(SystemConfig(key=SystemConfigKey.UserInstalledPlugins.value, + value=["A", "B", "Unrelated"])) + store = _MembershipConfig(factory) + config = SystemConfigService(reader=store, writer=store, async_executor=_ThreadExecutor()) + manager = Mock() + manager.plugins = {} + manager.mutation.side_effect = lambda _: nullcontext() + manager.get_plugin_instance.return_value = None + manager.get_plugin_source_instances.return_value = [] + manager.stop.return_value = None + monkeypatch.setattr(endpoint, "get_plugin_manager", lambda: manager) + monkeypatch.setattr(management, "get_plugin_manager", lambda: manager) + monkeypatch.setattr(endpoint, "get_configured_system_config", lambda: config) + monkeypatch.setattr(management, "get_configured_system_config", lambda: config) + for module, names in [(endpoint, ["remove_plugin_api", "remove_plugin_job", "remove_plugin_from_folders"]), + (routes, ["remove_plugin_api"]), (scheduling, ["remove_plugin_job"]), + (folders, ["remove_plugin_from_folders"])]: + for name in names: + monkeypatch.setattr(module, name, lambda _: None) + + def uninstall(arguments): + """ + 每个请求持有自己的调用线程及异步事件循环。 + """ + path, plugin_id = arguments + if path == "sync": + assert endpoint.uninstall_plugin(plugin_id, None).success + else: + asyncio.run(management.uninstall_plugin_runtime(plugin_id)) + + try: + with ThreadPoolExecutor(max_workers=2) as pool: + list(pool.map(uninstall, zip(paths, ["A", "B"]))) + with factory() as session: + assert session.execute(sa.select(SystemConfig.value)).scalar_one() == ["Unrelated"] + config.update_atomically(SystemConfigKey.UserInstalledPlugins, + lambda current: (None, [*current, "InstalledLater"])) + with factory() as session: + assert session.execute(sa.select(SystemConfig.value)).scalar_one() == ["Unrelated", "InstalledLater"] + finally: + engine.dispose() diff --git a/tests/test_plugin_platform_observability_regressions.py b/tests/test_plugin_platform_observability_regressions.py new file mode 100644 index 000000000..185499a37 --- /dev/null +++ b/tests/test_plugin_platform_observability_regressions.py @@ -0,0 +1,80 @@ +""" +验证异步日志容量与事件队列深度观测的实际运行路径。 +""" + +import threading + +from app.runtime import events as event_module +from app.runtime import log as log_module +from app.runtime.events import Event, EventManager +from tests.test_subscription_logging import fixture_file_writer # noqa: F401 + + +def test_subscription_batch_bounds_handlers_and_reclaims_idle(file_writer, tmp_path, monkeypatch): + """ + 批量写入44条订阅时句柄不超过32，时钟推进后释放旧句柄而保留新日志。 + """ + now = [100.0] + monkeypatch.setattr(log_module.time, "monotonic", lambda: now[0]) + entries = [log_module.LogEntry("INFO", "payload", tmp_path / "subscriptions" / f"{i}.log") + for i in range(1, 45)] + file_writer._write_batch(entries) + assert len(file_writer._rotating_handlers) == 32 + assert len(file_writer._scoped_handler_times) == 32 + now[0] += 61 + path = tmp_path / "subscriptions" / "45.log" + file_writer._write_batch([log_module.LogEntry("INFO", "latest", path)]) + assert set(file_writer._rotating_handlers) == {path} + assert "latest" in path.read_text() + + +def test_subscription_batch_trims_directory_quota(file_writer, tmp_path, monkeypatch): + """ + 异步批量路径执行目录配额清理，仅删除旧订阅日志而保留非日志文件。 + """ + directory = tmp_path / "subscriptions" + directory.mkdir() + for name in ("1.log", "1.log.1", "notes.txt"): + (directory / name).write_text("x" * 100) + monkeypatch.setattr(log_module, "_SCOPED_LOG_TOTAL_BYTES", 50) + active = directory / "2.log" + file_writer._write_batch([log_module.LogEntry("INFO", "latest", active)]) + assert not (directory / "1.log").exists() + assert (directory / "notes.txt").exists() + assert "latest" in active.read_text() + + +def test_broadcast_depth_is_balanced_after_real_consumer_and_sentinel(monkeypatch): + """ + 累加型队列指标只记录入队加一和出队减一，真实消费者处理后回零且不计哨兵。 + """ + manager = object.__new__(EventManager) + EventManager.__init__(manager) + deltas = [] + processed = [] + + def metric(name, value, **_labels): + """ + 模拟UpDownCounter的累加语义。 + """ + if name == "event.queue.depth": + deltas.append(value) + + monkeypatch.setattr(event_module, "record_metric", metric) + monkeypatch.setattr(manager, "_EventManager__dispatch_broadcast_event", processed.append) + events = [Event("review-depth", {"value": index}, priority=index) for index in range(3)] + for event in events: + manager._EventManager__trigger_broadcast_event(event) + manager._EventManager__event_queue.put((999, event_module._EVENT_STOP_SENTINEL)) + manager._EventManager__event.set() + thread = threading.Thread(target=manager._EventManager__broadcast_consumer_loop) + try: + thread.start() + thread.join(5) + assert not thread.is_alive() + assert processed == events + assert deltas == [1, 1, 1, -1, -1, -1] + assert sum(deltas) == 0 + finally: + manager._EventManager__event.clear() + thread.join(5)