From b8b59ae20a5abf875adab2851066a513d2d7eac9 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Sun, 16 Aug 2026 16:44:45 +0800 Subject: [PATCH] refactor(runtime): activate managed resources on demand (#6334) --- AGENTS.md | 14 +- app/adapters/network/browser.py | 62 ++- app/adapters/system/display.py | 29 -- app/adapters/system/display/__init__.py | 53 +++ app/adapters/system/display/capability.toml | 12 + app/adapters/system/display/resource.py | 45 ++ app/runtime/compat/resource_imports.py | 196 ++++++++ .../extensions/managed_resource_adapter.py | 194 ++++++++ app/runtime/extensions/plugin_manager.py | 23 + app/runtime/managed_resources.py | 182 +++++++ app/sdk/browser.py | 34 ++ app/startup/managed_resources_initializer.py | 47 ++ app/startup/modules_initializer.py | 19 +- app/startup/plugins_initializer.py | 15 + docs/rules/05-architecture.md | 21 +- scripts/perf/README.md | 31 ++ scripts/perf/instrument/sitecustomize.py | 344 +++++++++++++- scripts/perf/moviepilot_docker_ab.py | 449 +++++++++++++++--- scripts/perf/test_scenarios.py | 430 +++++++++++++++++ tests/test_agent_lifecycle.py | 2 +- tests/test_browser_helper.py | 65 ++- tests/test_cache_system.py | 2 +- tests/test_display_resource.py | 99 ++++ tests/test_legacy_plugin_resource_imports.py | 382 +++++++++++++++ tests/test_lifecycle_shutdown.py | 33 +- tests/test_managed_resources.py | 339 +++++++++++++ tests/test_plugin_sdk.py | 62 +++ 27 files changed, 3068 insertions(+), 116 deletions(-) delete mode 100644 app/adapters/system/display.py create mode 100644 app/adapters/system/display/__init__.py create mode 100644 app/adapters/system/display/capability.toml create mode 100644 app/adapters/system/display/resource.py create mode 100644 app/runtime/compat/resource_imports.py create mode 100644 app/runtime/extensions/managed_resource_adapter.py create mode 100644 app/runtime/managed_resources.py create mode 100644 app/sdk/browser.py create mode 100644 app/startup/managed_resources_initializer.py create mode 100644 scripts/perf/test_scenarios.py create mode 100644 tests/test_display_resource.py create mode 100644 tests/test_legacy_plugin_resource_imports.py create mode 100644 tests/test_managed_resources.py diff --git a/AGENTS.md b/AGENTS.md index 23fd86dc7..4e9b9ea53 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -60,19 +60,19 @@ The legacy roots have no physical directories in the source tree. Current images |---|---|---|---| | `app/foundation/` | 无状态、无配置和无 I/O 的底层机制：反射/动态导入、加密、DOM、身份、集合、单例、文本、URL 和版本比较 | `settings`、DB/SystemConfig、网络请求、运行日志、MoviePilot 业务规则、旧导入路径 | `reflection.py`, `crypto.py`, `collections.py`, `text.py`, `url.py` | | `app/domain/` | Pure MoviePilot business semantics and models for media, recognition, sites, and torrents | Persistence, global settings reads, network/filesystem clients, Rust imports, service discovery, process lifecycle | `context.py`, `media.py`, `metainfo.py`, `scraper.py`, `meta/` | -| `app/runtime/` | 进程级运行机制和策略：配置、事件、完整日志、缓存契约/内存行为、并发、调度、限流、本地化、GC 和重启状态 | 具体外部产品、业务流程、Redis/文件缓存实现 | `config.py`, `events.py`, `log.py`, `cache.py`, `thread.py`, `state.py` | -| `app/runtime/extensions/` | 模块、插件和配置化服务实现的发现、注册与生命周期 | 通用反射机制、插件公开 API、无关业务流程 | `module_manager.py`, `plugin_manager.py`, `service_registry.py` | +| `app/runtime/` | 进程级运行机制和策略：配置、事件、完整日志、缓存契约/内存行为、托管资源门面、并发、调度、限流、本地化、GC 和重启状态 | 具体外部产品、业务流程、Redis/文件缓存实现 | `config.py`, `events.py`, `log.py`, `cache.py`, `managed_resources.py`, `thread.py`, `state.py` | +| `app/runtime/extensions/` | 模块、插件、配置化服务和托管资源实现的发现、注册与生命周期适配 | 通用反射机制、插件公开 API、无关业务流程 | `module_manager.py`, `plugin_manager.py`, `managed_resource_adapter.py`, `service_registry.py` | | `app/adapters/network/` | HTTP、浏览器、DNS、Cloudflare 和 IP 等通用网络技术适配 | RSS/站点业务编排、身份认证策略、命名外部产品流程 | `http.py`, `browser.py`, `doh.py`, `ip.py` | | `app/adapters/cache/` | Redis 与文件缓存等具体持久化实现 | 缓存协议、装饰器和进程内缓存策略 | `backends.py`, `redis.py` | -| `app/adapters/system/` | 操作系统、文件、进程、标准流、包/资源安装、显示和 Rust 加速适配 | 业务规则、进程重启决策 | `host.py`, `stdio.py`, `package.py`, `resource.py`, `rust.py`, `fsproxy.py` | +| `app/adapters/system/` | 操作系统、文件、进程、标准流、包/资源安装、显示和 Rust 加速适配 | 业务规则、进程重启决策 | `host.py`, `display/`, `stdio.py`, `package.py`, `resource.py`, `rust.py`, `fsproxy.py` | | `app/adapters/external/` | CookieCloud、插件市场、OCR、IP 归属和 MoviePilot Server 等命名外部生态 | 通用 HTTP/DNS/文件机制或可复用领域语义 | `market.py`, `server.py`, `cookiecloud.py`, `ocr.py`, `location.py`, `wechat_crypt.py` | | `app/application/` | 读取配置/持久化状态的聚焦应用服务和服务族规则 | 多领域 Chain 编排、底层通用机制、通用传输协议 | `recognition.py`, `filter.py`, `filter_rules.py`, `notification.py`, `mediaserver.py`, `rss.py`, `site/sites.*` | | `app/application/messaging/` | 消息渲染/路由、交互和 Agent 到消息桥接：`interaction.py` 通用交互契约和视图工具；`router.py` 统一交互优先级和回调分发；`site.py`/`subscribe.py`/`skill.py` 对应命令的会话、输入解析和视图；`media.py` 媒体交互状态（业务工作流仍由 `MediaInteractionChain` 执行）；`plugin.py` 插件输入接管和插件按钮回调；`agent.py` Agent 选择状态、回调协议和 WebAgent 消息桥接；`message.py` 通知渲染、模板和队列。不作为推荐给插件直接使用的公开 SDK | 认证策略、通用 HTTP、服务发现、仅端点使用的 Web Push 行为 | `message.py`, `interaction.py`, `router.py`, `agent.py` | | `app/application/security/` | 认证、授权、Cookie、Passkey、OTP/二次认证、路径/URL 安全、SSRF 和签名策略 | 通用 URL 解析、进程运行策略、普通业务校验 | `access.py`, `auth.py`, `cookie.py`, `passkey.py`, `otp.py`, `twofactor.py`, `url.py` | | `app/chain/` | Reusable use-case orchestration across modules, services, Oper classes, events, and caches; chains reach modules only through `run_module` dispatch on method-name contracts | Transport schemas, backend-specific protocol details, generic primitives, direct imports of module internals (classes, exceptions, constants) | `media.py`, `download.py`, `subscribe.py`, `transfer.py` | -| `app/startup/` | Composition root: inject providers/adapters, order initialization and shutdown, decide restart/lifecycle policy | Reusable business rules or adapter implementation details | `lifecycle.py`, `domain_initializer.py`, `cache_initializer.py`, `modules_initializer.py` | -| `app/sdk/` | Deliberately curated stable imports for new plugins | Canonical implementation logic or host-internal dependencies | `cache.py`, `logging.py`, `media.py`, `network.py`, `services.py` | -| `app/runtime/compat/` | 仅依赖标准库的精确旧导入路由和 DEBUG 诊断 | 业务实现、通配猜测、目标模块的提前导入 | `manifest.py`, `imports.py`, `diagnostics.py` | +| `app/startup/` | Composition root: inject providers/adapters, order initialization and shutdown, decide restart/lifecycle policy | Reusable business rules or adapter implementation details | `lifecycle.py`, `domain_initializer.py`, `cache_initializer.py`, `managed_resources_initializer.py`, `modules_initializer.py` | +| `app/sdk/` | Deliberately curated stable imports for new plugins | Canonical implementation logic or host-internal dependencies | `browser.py`, `cache.py`, `logging.py`, `media.py`, `network.py`, `services.py` | +| `app/runtime/compat/` | 仅依赖标准库的精确旧导入路由、资源前置扫描和 DEBUG 诊断 | 业务实现、通配猜测、目标模块的提前导入 | `manifest.py`, `imports.py`, `resource_imports.py`, `diagnostics.py` | 容易误分的三个边界必须按实际职责判断：`application/rss.py` 同时承担 Feed/种子语义、站点规则和浏览器回退，不是单纯 HTTP 传输；`application/site/sites.*` 及 `user.sites.v3.bin` 共同构成站点目录、认证和索引应用能力，只有下载安装机制留在 `adapters/system/resource.py`；`foundation/crypto.py` 只提供无状态 RSA/摘要/AES 算法，认证、签名、令牌和二次验证策略仍属于 `application/security/`。 @@ -156,4 +156,4 @@ For the full documentation map and cross-references, refer to: **[Documentation Hub Index](./docs/rules/README.md)** -*Last Updated: 2026-08-15* +*Last Updated: 2026-08-16* diff --git a/app/adapters/network/browser.py b/app/adapters/network/browser.py index 8a3202f09..06149c556 100644 --- a/app/adapters/network/browser.py +++ b/app/adapters/network/browser.py @@ -9,6 +9,10 @@ from urllib.parse import urlparse from app.runtime.config import settings from app.runtime.log import logger +from app.runtime.managed_resources import ( + acquire_managed_resource, + acquire_managed_resource_async, +) from app.adapters.network.http import RequestUtils, cookie_parse @@ -117,6 +121,47 @@ class BrowserPage(Protocol): ... +def launch_browser_context(headless: bool = True, **kwargs: Any) -> BrowserContext: + """ + 启动同步浏览器上下文；有界面模式先显式获取宿主显示资源。 + + :param headless: 是否使用无头模式 + :param kwargs: 浏览器实现接受的其余启动参数 + :return: 浏览器上下文 + """ + if not headless: + acquire_managed_resource( + "host.display", + reason="headed_browser_launch", + retry=True, + ) + from cloakbrowser import launch_context + + return launch_context(headless=headless, **kwargs) + + +async def launch_browser_context_async( + headless: bool = True, + **kwargs: Any, +) -> Any: + """ + 启动异步浏览器上下文；有界面模式等待宿主显示资源就绪。 + + :param headless: 是否使用无头模式 + :param kwargs: 浏览器实现接受的其余启动参数 + :return: 浏览器上下文 + """ + if not headless: + await acquire_managed_resource_async( + "host.display", + reason="headed_browser_launch", + retry=True, + ) + from cloakbrowser import launch_context_async + + return await launch_context_async(headless=headless, **kwargs) + + @dataclass class _BrowserSessionState: """保存一个可复用浏览器上下文及其页面游标。""" @@ -662,10 +707,7 @@ class BrowserSessionHelper: viewport: Optional[dict[str, int]] = None, ) -> BrowserContext: """按宿主反检测配置创建 CloakBrowser 上下文。""" - from cloakbrowser import launch_context - context_kwargs = { - "headless": headless, "humanize": settings.CLOAKBROWSER_HUMANIZE, "human_preset": settings.CLOAKBROWSER_HUMAN_PRESET, } @@ -673,7 +715,7 @@ class BrowserSessionHelper: context_kwargs["user_agent"] = user_agent if viewport: context_kwargs["viewport"] = viewport - return launch_context(**context_kwargs) + return launch_browser_context(headless=headless, **context_kwargs) def _get_or_create_session( self, @@ -883,13 +925,11 @@ class PlaywrightHelper: """ 启动 CloakBrowser 上下文。 """ - from cloakbrowser import launch_context - - return launch_context(headless=headless, - proxy=proxies, - user_agent=user_agent, - humanize=settings.CLOAKBROWSER_HUMANIZE, - human_preset=settings.CLOAKBROWSER_HUMAN_PRESET) + return launch_browser_context(headless=headless, + proxy=proxies, + user_agent=user_agent, + humanize=settings.CLOAKBROWSER_HUMANIZE, + human_preset=settings.CLOAKBROWSER_HUMAN_PRESET) @staticmethod def __fs_cookie_str(cookies: list) -> str: diff --git a/app/adapters/system/display.py b/app/adapters/system/display.py deleted file mode 100644 index 614e58dca..000000000 --- a/app/adapters/system/display.py +++ /dev/null @@ -1,29 +0,0 @@ -from pyvirtualdisplay import Display - -from app.runtime.log import logger -from app.foundation.singleton import Singleton -from app.adapters.system.host import SystemUtils - -import os - - -class DisplayHelper(metaclass=Singleton): - """在容器环境中管理浏览器所需的虚拟显示。""" - - def __init__(self): - """仅在 Docker 内启动虚拟显示服务。""" - self._display = None - if not SystemUtils.is_docker(): - return - try: - self._display = Display(visible=False, size=(1024, 768), extra_args=[os.environ['DISPLAY']]) - self._display.start() - except Exception as err: - logger.error(f"DisplayHelper init error: {str(err)}") - - def stop(self): - """停止已经启动的虚拟显示服务。""" - if self._display: - logger.info("正在停止虚拟显示...") - self._display.stop() - logger.info("虚拟显示已停止") diff --git a/app/adapters/system/display/__init__.py b/app/adapters/system/display/__init__.py new file mode 100644 index 000000000..f64aa0797 --- /dev/null +++ b/app/adapters/system/display/__init__.py @@ -0,0 +1,53 @@ +"""虚拟显示适配器及旧 DisplayHelper 兼容入口。""" + +from __future__ import annotations + +from importlib import import_module +from typing import Any + +from app.foundation.singleton import Singleton +from app.runtime.log import logger +from app.runtime.managed_resources import ( + acquire_managed_resource, + stop_managed_resource, +) + + +DISPLAY_CAPABILITY_ID = "host.display" + + +class DisplayHelper(metaclass=Singleton): + """保留旧构造 API，并把资源所有权委托给 host.display 能力。""" + + def __init__(self) -> None: + """显式构造旧门面时激活虚拟显示，失败保持旧 API 的日志语义。""" + try: + acquire_managed_resource( + DISPLAY_CAPABILITY_ID, + reason="legacy_display_helper", + retry=True, + ) + except Exception as error: + logger.error("DisplayHelper init error: %s", error) + + def stop(self) -> None: + """停止已激活的虚拟显示；未配置 Runtime 时保持幂等。""" + stop_managed_resource( + DISPLAY_CAPABILITY_ID, + reason="legacy_display_helper_stop", + ) + + +__all__ = ["DISPLAY_CAPABILITY_ID", "DisplayHelper", "VirtualDisplayResource"] + + +def __getattr__(name: str) -> Any: + """按需公开资源实现，普通兼容导入不加载显示后端。""" + if name != "VirtualDisplayResource": + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + value = getattr( + import_module("app.adapters.system.display.resource"), + "VirtualDisplayResource", + ) + globals()[name] = value + return value diff --git a/app/adapters/system/display/capability.toml b/app/adapters/system/display/capability.toml new file mode 100644 index 000000000..ee17d37c4 --- /dev/null +++ b/app/adapters/system/display/capability.toml @@ -0,0 +1,12 @@ +schema_version = 1 +id = "host.display" +kind = "managed_resource.sync" +entrypoint = "app.adapters.system.display.resource:VirtualDisplayResource" +depends_on = [] + +[metadata] +name = "Virtual Display" + +[activation] +policy = "on_first_use" +watch = [] diff --git a/app/adapters/system/display/resource.py b/app/adapters/system/display/resource.py new file mode 100644 index 000000000..c91e730fb --- /dev/null +++ b/app/adapters/system/display/resource.py @@ -0,0 +1,45 @@ +"""虚拟显示进程的托管资源实现。""" + +from __future__ import annotations + +import os +from typing import Any, Optional + +from app.adapters.system.host import SystemUtils +from app.runtime.log import logger + + +class VirtualDisplayResource: + """按需拥有一个容器内虚拟显示进程。""" + + def __init__(self) -> None: + self._display: Optional[Any] = None + + @property + def display(self) -> Optional[Any]: + """返回当前拥有的显示对象；未启动或已停止时为 None。""" + return self._display + + def start(self) -> None: + """仅在容器环境启动虚拟显示，重复启动保持幂等。""" + if self._display is not None or not SystemUtils.is_docker(): + return + from pyvirtualdisplay import Display + + display = Display( + visible=False, + size=(1024, 768), + extra_args=[os.environ["DISPLAY"]], + ) + self._display = display + display.start() + + def stop(self) -> None: + """停止当前资源拥有的显示进程，失败时保留句柄供 Runtime 重试。""" + display = self._display + if display is None: + return + logger.info("正在停止虚拟显示...") + display.stop() + self._display = None + logger.info("虚拟显示已停止") diff --git a/app/runtime/compat/resource_imports.py b/app/runtime/compat/resource_imports.py new file mode 100644 index 000000000..9abf6f36a --- /dev/null +++ b/app/runtime/compat/resource_imports.py @@ -0,0 +1,196 @@ +"""从旧插件源码导入中识别必须提前就绪的宿主资源。""" + +from __future__ import annotations + +import ast +import threading +import tokenize +from dataclasses import dataclass +from pathlib import Path +from typing import Dict, FrozenSet, Iterable, Set, Tuple + + +@dataclass(frozen=True, slots=True) +class ResourceImportRule: + """描述第三方模块导入与宿主资源能力之间的静态映射。""" + + capability_id: str # 导入前必须准备的宿主能力标识 + module_prefixes: tuple[str, ...] # 按完整包边界匹配的第三方模块前缀 + headed_entrypoints: tuple[str, ...] # 已确认允许 headed 模式的公开入口 + + +# 旧插件可能绕过宿主浏览器门面直接调用 CloakBrowser。其六个 launch +# 入口均允许 headed 模式，因此导入该包或任意子模块时保守准备虚拟显示。 +RESOURCE_IMPORT_RULES: tuple[ResourceImportRule, ...] = ( + ResourceImportRule( + capability_id="host.display", + module_prefixes=("cloakbrowser",), + headed_entrypoints=( + "launch", + "launch_async", + "launch_context", + "launch_context_async", + "launch_persistent_context", + "launch_persistent_context_async", + ), + ), +) + + +_scan_cache_lock = threading.RLock() +_scan_cache: Dict[Path, Tuple[int, int, int, int, int, FrozenSet[str]]] = {} + + +class PluginResourceImportScanError(RuntimeError): + """表示单个插件源码无法生成可靠的精确资源集合。""" + + +def _all_resource_capabilities() -> FrozenSet[str]: + """扫描不完整时返回全部已登记资源，避免漏失导入前置条件。""" + return frozenset(rule.capability_id for rule in RESOURCE_IMPORT_RULES) + + +def _matches_module(module_name: str, module_prefixes: Iterable[str]) -> bool: + """按完整包边界匹配模块，避免相似名称产生误报。""" + return any( + module_name == prefix or module_name.startswith(f"{prefix}.") + for prefix in module_prefixes + ) + + +def _dynamic_import_aliases(tree: ast.AST) -> tuple[Set[str], Set[str]]: + """收集 importlib 模块及 import_module 函数的本地别名。""" + module_aliases = {"importlib"} + function_aliases: Set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for imported in node.names: + if imported.name == "importlib": + module_aliases.add(imported.asname or imported.name) + elif isinstance(node, ast.ImportFrom) and node.module == "importlib": + for imported in node.names: + if imported.name == "import_module": + function_aliases.add(imported.asname or imported.name) + return module_aliases, function_aliases + + +def _constant_dynamic_import( + node: ast.Call, + *, + importlib_aliases: Set[str], + import_module_aliases: Set[str], +) -> str | None: + """提取受支持动态导入调用中的常量模块名。""" + if not node.args: + return None + is_import_call = isinstance(node.func, ast.Name) and ( + node.func.id == "__import__" or node.func.id in import_module_aliases + ) + if ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id in importlib_aliases + and node.func.attr == "import_module" + ): + is_import_call = True + if not is_import_call: + return None + argument = node.args[0] + if isinstance(argument, ast.Constant) and isinstance(argument.value, str): + return argument.value + return None + + +def _imported_modules(tree: ast.AST) -> FrozenSet[str]: + """提取静态导入以及可确定目标的动态导入模块名。""" + modules: Set[str] = set() + importlib_aliases, import_module_aliases = _dynamic_import_aliases(tree) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + modules.update(imported.name for imported in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + modules.add(node.module) + elif isinstance(node, ast.Call): + module_name = _constant_dynamic_import( + node, + importlib_aliases=importlib_aliases, + import_module_aliases=import_module_aliases, + ) + if module_name: + modules.add(module_name) + return frozenset(modules) + + +def _scan_source(plugin_id: str, path: Path) -> FrozenSet[str]: + """读取并解析单个源码文件；不完整结果不能进入插件导入阶段。""" + try: + before_stat = path.stat() + # 热加载工具可能保留 mtime，等长替换也不会改变 size；ctime 与 inode/device + # 一并参与身份判断，避免把已替换源码误认为旧缓存。 + cache_key = ( + before_stat.st_mtime_ns, + before_stat.st_ctime_ns, + before_stat.st_size, + 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)) + after_stat = path.stat() + except (OSError, SyntaxError, UnicodeError) as error: + raise PluginResourceImportScanError( + f"无法扫描插件 {plugin_id} 源码 {path.name}：{error}" + ) from error + after_key = ( + after_stat.st_mtime_ns, + after_stat.st_ctime_ns, + after_stat.st_size, + after_stat.st_dev, + after_stat.st_ino, + ) + if cache_key != after_key: + raise PluginResourceImportScanError( + f"扫描插件 {plugin_id} 时源码 {path.name} 发生变化" + ) + + capabilities: Set[str] = set() + for module_name in _imported_modules(tree): + for rule in RESOURCE_IMPORT_RULES: + if _matches_module(module_name, rule.module_prefixes): + capabilities.add(rule.capability_id) + result = frozenset(capabilities) + with _scan_cache_lock: + _scan_cache[path] = (*cache_key, result) + return result + + +def scan_plugin_resource_imports( + plugin_id: str, + plugin_dir: Path, +) -> tuple[str, ...]: + """递归扫描插件源码并返回导入前必须准备的 capability ID。""" + if not plugin_dir.is_dir(): + raise PluginResourceImportScanError( + f"插件 {plugin_id} 源码目录不存在：{plugin_dir}" + ) + + capabilities: Set[str] = set() + try: + source_files = sorted(plugin_dir.rglob("*.py")) + except OSError: + return tuple(sorted(_all_resource_capabilities())) + for path in source_files: + if "__pycache__" in path.parts: + continue + try: + capabilities.update(_scan_source(plugin_id, path)) + except PluginResourceImportScanError: + # Python 最终只会导入真实依赖链；无法解析的残留或平台专用文件不应 + # 阻断整个插件，但必须按最保守资源集合准备后再交给 loader 判断。 + capabilities.update(_all_resource_capabilities()) + return tuple(sorted(capabilities)) diff --git a/app/runtime/extensions/managed_resource_adapter.py b/app/runtime/extensions/managed_resource_adapter.py new file mode 100644 index 000000000..86c9e5d47 --- /dev/null +++ b/app/runtime/extensions/managed_resource_adapter.py @@ -0,0 +1,194 @@ +"""Managed Resource 的声明发现与 Capability Runtime 适配器。""" + +from __future__ import annotations + +import asyncio +import importlib +import inspect +from pathlib import Path +from typing import Any, Iterable + +from app.runtime.capabilities.errors import CapabilityAdapterContractError +from app.runtime.capabilities.model import ( + ActivationPolicy, + AdapterExecutionMode, + CapabilitySpec, +) +from app.runtime.capabilities.registry import CapabilityRegistry +from app.runtime.managed_resources import ( + MANAGED_RESOURCE_ASYNC_KIND, + MANAGED_RESOURCE_SYNC_KIND, +) + + +_DEFAULT_RESOURCE_ROOT = Path(__file__).resolve().parents[2] / "adapters" +_RESOURCE_KINDS = {MANAGED_RESOURCE_SYNC_KIND, MANAGED_RESOURCE_ASYNC_KIND} + + +def _load_entrypoint(spec: CapabilitySpec) -> Any: + """解析声明中的 canonical 实现对象，不创建资源实例。""" + module_name, symbol_name = spec.entrypoint.split(":", maxsplit=1) + module = importlib.import_module(module_name) + try: + return getattr(module, symbol_name) + except AttributeError as error: + raise CapabilityAdapterContractError( + f"{spec.entrypoint} 未公开 Managed Resource 实现" + ) from error + + +def _create_candidate(spec: CapabilitySpec, implementation: Any) -> Any: + """通过零参数工厂创建资源候选，实例在 start 成功前不可见。""" + if not callable(implementation): + raise CapabilityAdapterContractError( + f"{spec.entrypoint} 不是可调用的 Managed Resource 工厂" + ) + candidate = implementation() + if candidate is None or inspect.isawaitable(candidate): + close = getattr(candidate, "close", None) + if callable(close): + close() + raise CapabilityAdapterContractError( + f"{spec.entrypoint} 必须同步返回资源候选" + ) + return candidate + + +def _resource_method(spec: CapabilitySpec, candidate: Any, name: str) -> Any: + """读取必需生命周期方法并生成稳定合同错误。""" + callback = getattr(candidate, name, None) + if not callable(callback): + raise CapabilityAdapterContractError( + f"{spec.entrypoint} 的资源候选缺少 {name}()" + ) + return callback + + +class SyncManagedResourceAdapter: + """把同步 start/stop 资源接入 Capability Runtime。""" + + execution_mode = AdapterExecutionMode.SYNC + + @staticmethod + def materialize(spec: CapabilitySpec) -> Any: + """解析资源工厂。""" + return _load_entrypoint(spec) + + @staticmethod + def create( + spec: CapabilitySpec, + implementation: Any, + _generation: int, + _previous: Any = None, + ) -> Any: + """创建尚未发布的同步资源候选。""" + return _create_candidate(spec, implementation) + + @staticmethod + def start(spec: CapabilitySpec, candidate: Any, _generation: int) -> None: + """启动同步候选；同步 kind 不接受 awaitable 返回值。""" + result = _resource_method(spec, candidate, "start")() + if inspect.isawaitable(result): + close = getattr(result, "close", None) + if callable(close): + close() + raise CapabilityAdapterContractError( + f"{spec.entrypoint}.start() 返回 awaitable，与同步 kind 不匹配" + ) + + @staticmethod + def stop(spec: CapabilitySpec, instance: Any, _generation: int) -> None: + """停止同步资源，异常交由 Runtime 保留资源所有权并支持重试。""" + result = _resource_method(spec, instance, "stop")() + if inspect.isawaitable(result): + close = getattr(result, "close", None) + if callable(close): + close() + raise CapabilityAdapterContractError( + f"{spec.entrypoint}.stop() 返回 awaitable，与同步 kind 不匹配" + ) + + @staticmethod + def cleanup( + spec: CapabilitySpec, + candidate: Any, + generation: int, + _error: BaseException, + ) -> None: + """启动失败时按同一 stop 合同清理尚未发布的候选。""" + SyncManagedResourceAdapter.stop(spec, candidate, generation) + + +class AsyncManagedResourceAdapter: + """把异步 start/stop 资源接入 Capability Runtime。""" + + execution_mode = AdapterExecutionMode.ASYNC + + @staticmethod + async def materialize(spec: CapabilitySpec) -> Any: + """在线程中解析资源工厂，避免第三方导入阻塞事件循环。""" + return await asyncio.to_thread(_load_entrypoint, spec) + + @staticmethod + async def create( + spec: CapabilitySpec, + implementation: Any, + _generation: int, + _previous: Any = None, + ) -> Any: + """创建尚未发布的异步资源候选。""" + return _create_candidate(spec, implementation) + + @staticmethod + async def start(spec: CapabilitySpec, candidate: Any, _generation: int) -> None: + """等待异步候选完成启动。""" + result = _resource_method(spec, candidate, "start")() + if not inspect.isawaitable(result): + raise CapabilityAdapterContractError( + f"{spec.entrypoint}.start() 必须返回 awaitable" + ) + await result + + @staticmethod + async def stop(spec: CapabilitySpec, instance: Any, _generation: int) -> None: + """等待异步资源完成停止。""" + result = _resource_method(spec, instance, "stop")() + if not inspect.isawaitable(result): + raise CapabilityAdapterContractError( + f"{spec.entrypoint}.stop() 必须返回 awaitable" + ) + await result + + @staticmethod + async def cleanup( + spec: CapabilitySpec, + candidate: Any, + generation: int, + _error: BaseException, + ) -> None: + """启动失败时等待同一 stop 合同清理候选。""" + await AsyncManagedResourceAdapter.stop(spec, candidate, generation) + + +def _validate_registry(registry: CapabilityRegistry) -> None: + """固定类别级声明合同，资源只能由显式首用触发。""" + for spec in registry.list_specs(): + if set(spec.metadata) != {"name"}: + raise ValueError(f"{spec.source}: Managed Resource metadata 只能包含 name") + if spec.activation is not ActivationPolicy.ON_FIRST_USE: + raise ValueError(f"{spec.source}: Managed Resource 必须使用 on_first_use") + if spec.selector is not None or spec.watch: + raise ValueError(f"{spec.source}: Managed Resource 不接受配置 selector 或 watch") + + +def build_managed_resource_registry( + roots: Iterable[Path | str] | None = None, +) -> CapabilityRegistry: + """从 data-only manifest 构建不导入资源实现的注册表。""" + registry = CapabilityRegistry.discover( + tuple(roots) if roots is not None else (_DEFAULT_RESOURCE_ROOT,), + kinds=_RESOURCE_KINDS, + selector_schemas={}, + ) + _validate_registry(registry) + return registry diff --git a/app/runtime/extensions/plugin_manager.py b/app/runtime/extensions/plugin_manager.py index 4c73b1395..7ae04c49a 100644 --- a/app/runtime/extensions/plugin_manager.py +++ b/app/runtime/extensions/plugin_manager.py @@ -37,6 +37,7 @@ from app.schemas.types import EventType, SystemConfigKey LegacyDiagnosticsConfigurator = Callable[..., None] LegacyImportScanner = Callable[..., None] +LegacyPluginImportPreparer = Callable[..., None] PluginInstallReporter = Callable[..., None] SiteAuthLevelProvider = Callable[[], int] @@ -45,6 +46,10 @@ def _ignore_legacy_diagnostics(**_kwargs) -> None: """在启动组合根尚未注入兼容服务时保持插件加载可用。""" +def _ignore_plugin_resource_imports(**_kwargs) -> None: + """未进入应用启动组合时不主动创建进程级宿主资源。""" + + def _unavailable_site_auth_level() -> int: """站点能力尚未装配时返回未认证等级。""" return 0 @@ -54,6 +59,9 @@ _legacy_diagnostics_configurator: LegacyDiagnosticsConfigurator = ( _ignore_legacy_diagnostics ) _legacy_import_scanner: LegacyImportScanner = _ignore_legacy_diagnostics +_legacy_plugin_import_preparer: LegacyPluginImportPreparer = ( + _ignore_plugin_resource_imports +) _plugin_install_reporter: PluginInstallReporter = _ignore_legacy_diagnostics _site_auth_level_provider: SiteAuthLevelProvider = _unavailable_site_auth_level @@ -69,6 +77,14 @@ def configure_plugin_legacy_import_services( _legacy_import_scanner = import_scanner +def configure_plugin_resource_import_preparer( + preparer: LegacyPluginImportPreparer, +) -> None: + """注入旧插件导入前的宿主资源准备器。""" + global _legacy_plugin_import_preparer + _legacy_plugin_import_preparer = preparer + + def configure_plugin_install_reporter(reporter: PluginInstallReporter) -> None: """由启动组合根注入插件安装上报器，避免扩展层依赖远程服务。""" global _plugin_install_reporter @@ -318,6 +334,13 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): module_name = f"app.plugins.{plugin_dir.name}" logger.debug(f"正在导入插件模块：{module_name}") + # 旧插件可能直接导入带宿主资源前置条件的第三方包。资源必须在 + # Python 执行插件模块顶层代码前就绪，否则导入副作用无法安全回滚。 + _legacy_plugin_import_preparer( + plugin_id=plugin_dir.name, + plugin_dir=plugin_dir, + ) + _legacy_import_scanner( plugin_id=plugin_dir.name, plugin_dir=plugin_dir, diff --git a/app/runtime/managed_resources.py b/app/runtime/managed_resources.py new file mode 100644 index 000000000..d6a21122b --- /dev/null +++ b/app/runtime/managed_resources.py @@ -0,0 +1,182 @@ +"""进程级托管资源的轻量调用门面。""" + +from __future__ import annotations + +import asyncio +import threading +from typing import Any, Optional, Protocol + + +MANAGED_RESOURCE_SYNC_KIND = "managed_resource.sync" +MANAGED_RESOURCE_ASYNC_KIND = "managed_resource.async" + + +class ManagedResourceRuntime(Protocol): + """Managed Resource 门面依赖的最小 Capability Runtime 合同。""" + + @property + def is_shutdown(self) -> bool: + """返回 Runtime 是否已进入不可逆关闭态。""" + + def get_spec(self, capability_id: str) -> Any: + """返回资源声明。""" + + def get_running(self, capability_id: str) -> Any: + """只查询已发布实例。""" + + def snapshot(self, capability_id: str) -> Any: + """返回资源状态快照。""" + + def observations(self, capability_id: Optional[str] = None) -> tuple[Any, ...]: + """返回资源转换观测。""" + + def activate(self, capability_id: str, *, reason: str, retry: bool = False) -> Any: + """通过同步 adapter 激活资源。""" + + async def activate_async( + self, + capability_id: str, + *, + reason: str, + retry: bool = False, + ) -> Any: + """通过异步 adapter 激活资源。""" + + def stop(self, capability_id: str, *, reason: str) -> None: + """通过同步 adapter 停止资源。""" + + async def stop_async(self, capability_id: str, *, reason: str) -> None: + """通过异步 adapter 停止资源。""" + + async def shutdown_async(self, *, reason: str) -> None: + """关闭混合同步和异步 adapter 的 Runtime。""" + + +_runtime_lock = threading.RLock() +_managed_resource_runtime: Optional[ManagedResourceRuntime] = None + + +def configure_managed_resource_runtime(runtime: ManagedResourceRuntime) -> None: + """由启动组合层注入唯一的 Managed Resource Runtime。""" + if runtime is None: + raise ValueError("Managed Resource Runtime 不能为空") + global _managed_resource_runtime + with _runtime_lock: + _managed_resource_runtime = runtime + + +def _runtime(*, required: bool) -> Optional[ManagedResourceRuntime]: + """读取当前 Runtime；资源使用路径要求启动组合已经完成装配。""" + with _runtime_lock: + runtime = _managed_resource_runtime + if runtime is None and required: + raise RuntimeError("Managed Resource Runtime 尚未初始化") + return runtime + + +def _resource_kind(runtime: ManagedResourceRuntime, capability_id: str) -> str: + """返回声明的执行模式；未知资源继续沿用 Runtime 的领域错误。""" + spec = runtime.get_spec(capability_id) + if spec is None: + runtime.get_running(capability_id) + raise RuntimeError(f"未知 Managed Resource：{capability_id}") + return str(spec.kind) + + +def acquire_managed_resource( + capability_id: str, + *, + reason: str, + retry: bool = True, +) -> Any: + """同步激活一个声明为同步模式的托管资源。""" + runtime = _runtime(required=True) + kind = _resource_kind(runtime, capability_id) + if kind != MANAGED_RESOURCE_SYNC_KIND: + raise RuntimeError(f"异步 Managed Resource 不能通过同步入口激活：{capability_id}") + return runtime.activate(capability_id, reason=reason, retry=retry) + + +async def acquire_managed_resource_async( + capability_id: str, + *, + reason: str, + retry: bool = True, +) -> Any: + """异步激活资源；同步资源移交工作线程，避免阻塞事件循环。""" + runtime = _runtime(required=True) + kind = _resource_kind(runtime, capability_id) + if kind == MANAGED_RESOURCE_ASYNC_KIND: + return await runtime.activate_async( + capability_id, + reason=reason, + retry=retry, + ) + if kind == MANAGED_RESOURCE_SYNC_KIND: + return await asyncio.to_thread( + runtime.activate, + capability_id, + reason=reason, + retry=retry, + ) + raise RuntimeError(f"未知 Managed Resource kind：{kind}") + + +def get_running_managed_resource(capability_id: str) -> Any: + """只查询已发布资源；Runtime 未配置时返回 None，不触发初始化。""" + runtime = _runtime(required=False) + if runtime is None: + return None + return runtime.get_running(capability_id) + + +def managed_resource_snapshot(capability_id: str) -> Any: + """返回资源状态快照；Runtime 未配置时返回 None。""" + runtime = _runtime(required=False) + if runtime is None: + return None + return runtime.snapshot(capability_id) + + +def managed_resource_observations( + capability_id: Optional[str] = None, +) -> tuple[Any, ...]: + """返回资源转换观测；Runtime 未配置时返回空快照。""" + runtime = _runtime(required=False) + if runtime is None: + return () + return runtime.observations(capability_id) + + +def stop_managed_resource(capability_id: str, *, reason: str) -> None: + """同步停止资源；Runtime 未配置时保持幂等且不反向初始化。""" + runtime = _runtime(required=False) + if runtime is None: + return + kind = _resource_kind(runtime, capability_id) + if kind != MANAGED_RESOURCE_SYNC_KIND: + raise RuntimeError(f"异步 Managed Resource 不能通过同步入口停止：{capability_id}") + runtime.stop(capability_id, reason=reason) + + +async def stop_managed_resource_async(capability_id: str, *, reason: str) -> None: + """异步停止资源；同步资源移交工作线程。""" + runtime = _runtime(required=False) + if runtime is None: + return + kind = _resource_kind(runtime, capability_id) + if kind == MANAGED_RESOURCE_ASYNC_KIND: + await runtime.stop_async(capability_id, reason=reason) + return + if kind == MANAGED_RESOURCE_SYNC_KIND: + await asyncio.to_thread(runtime.stop, capability_id, reason=reason) + return + raise RuntimeError(f"未知 Managed Resource kind：{kind}") + + +async def shutdown_managed_resource_runtime(*, reason: str) -> None: + """关闭已配置 Runtime；未配置时直接返回，绝不因关闭而创建资源。""" + runtime = _runtime(required=False) + if runtime is None: + return + await runtime.shutdown_async(reason=reason) diff --git a/app/sdk/browser.py b/app/sdk/browser.py new file mode 100644 index 000000000..c98e00413 --- /dev/null +++ b/app/sdk/browser.py @@ -0,0 +1,34 @@ +"""插件可依赖的轻量浏览器启动接口。""" + +from __future__ import annotations + +from typing import Any + + +def launch_browser_context(headless: bool = True, **kwargs: Any) -> Any: + """ + 启动同步浏览器上下文，并由宿主协调所需进程资源。 + + :param headless: 是否使用无头模式 + :param kwargs: 浏览器实现接受的其余启动参数 + :return: 浏览器上下文 + """ + from app.adapters.network.browser import launch_browser_context as launch + + return launch(headless=headless, **kwargs) + + +async def launch_browser_context_async(headless: bool = True, **kwargs: Any) -> Any: + """ + 启动异步浏览器上下文，并由宿主协调所需进程资源。 + + :param headless: 是否使用无头模式 + :param kwargs: 浏览器实现接受的其余启动参数 + :return: 浏览器上下文 + """ + from app.adapters.network.browser import launch_browser_context_async as launch + + return await launch(headless=headless, **kwargs) + + +__all__ = ["launch_browser_context", "launch_browser_context_async"] diff --git a/app/startup/managed_resources_initializer.py b/app/startup/managed_resources_initializer.py new file mode 100644 index 000000000..4a394768a --- /dev/null +++ b/app/startup/managed_resources_initializer.py @@ -0,0 +1,47 @@ +"""Managed Resource 的启动组合与进程关闭入口。""" + +from __future__ import annotations + +import threading +from typing import Optional + +from app.runtime.capabilities.runtime import CapabilityRuntime +from app.runtime.extensions.managed_resource_adapter import ( + AsyncManagedResourceAdapter, + SyncManagedResourceAdapter, + build_managed_resource_registry, +) +from app.runtime.managed_resources import ( + MANAGED_RESOURCE_ASYNC_KIND, + MANAGED_RESOURCE_SYNC_KIND, + configure_managed_resource_runtime, +) + + +_runtime_lock = threading.RLock() +_managed_resource_runtime: Optional[CapabilityRuntime] = None + + +def init_managed_resources() -> CapabilityRuntime: + """构建并注入资源 Runtime；只发现声明，不物化或启动任何资源。""" + global _managed_resource_runtime + with _runtime_lock: + if _managed_resource_runtime is None: + _managed_resource_runtime = CapabilityRuntime( + build_managed_resource_registry(), + adapters={ + MANAGED_RESOURCE_SYNC_KIND: SyncManagedResourceAdapter(), + MANAGED_RESOURCE_ASYNC_KIND: AsyncManagedResourceAdapter(), + }, + ) + configure_managed_resource_runtime(_managed_resource_runtime) + return _managed_resource_runtime + + +async def stop_managed_resources() -> None: + """关闭已经初始化的资源 Runtime；未初始化时不执行发现或激活。""" + with _runtime_lock: + runtime = _managed_resource_runtime + if runtime is None: + return + await runtime.shutdown_async(reason="application_shutdown") diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index a54b29ef9..f59a41b7b 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -22,7 +22,6 @@ from app.runtime.extensions.module_manager import ModuleManager from app.runtime.events import EventManager from app.runtime.state import SystemHelper from app.runtime.thread import ThreadHelper -from app.adapters.system.display import DisplayHelper from app.adapters.network.doh import DohHelper from app.adapters.system.resource import ( ResourceHelper, @@ -36,6 +35,10 @@ from app.command import CommandChain from app.schemas import Notification, NotificationType from app.schemas.types import SystemConfigKey from app.startup.agent_initializer import init_agent, stop_agent +from app.startup.managed_resources_initializer import ( + init_managed_resources, + stop_managed_resources, +) from app.application.security.access import set_superuser_token_payload_provider from app.application.security.auth import build_superuser_token_payload from app.application.image import configure_wallpaper_providers @@ -170,6 +173,13 @@ def update_resources() -> None: logger.error(f"资源更新完成但自动重启失败：{message}") +def close_browser_sessions() -> None: + """在托管资源关闭前释放所有浏览器上下文及其工作线程。""" + from app.adapters.network.browser import BrowserSessionHelper + + BrowserSessionHelper.close_all_sessions() + + async def stop_modules(): """ 服务关闭 @@ -186,7 +196,8 @@ async def stop_modules(): await run_step("AI智能体", stop_agent) await run_step("模块", lambda: ModuleManager().shutdown()) await run_step("事件消费", lambda: EventManager().stop()) - await run_step("虚拟显示", lambda: DisplayHelper().stop()) + await run_step("浏览器会话", close_browser_sessions) + await run_step("托管资源", stop_managed_resources) await run_step("DoH服务", lambda: DohHelper().shutdown()) await run_step("线程池", lambda: ThreadHelper().shutdown()) await run_step("消息服务", stop_message) @@ -201,12 +212,12 @@ async def init_modules(): """ 启动模块 """ + # 托管资源只在这里装配声明与 adapter，具体资源仍由首个消费者显式激活。 + init_managed_resources() # 应用服务不反向依赖 Chain，由启动组合层注入壁纸来源。 configure_wallpaper_services() # 认证访问层不反向依赖数据库实现，由启动组合层注入载荷提供器。 set_superuser_token_payload_provider(build_superuser_token_payload) - # 虚拟显示 - DisplayHelper() # DoH DohHelper() # 站点管理 diff --git a/app/startup/plugins_initializer.py b/app/startup/plugins_initializer.py index b311e0e0c..2fac784b0 100644 --- a/app/startup/plugins_initializer.py +++ b/app/startup/plugins_initializer.py @@ -1,25 +1,40 @@ +from pathlib import Path + from app.runtime.compat.diagnostics import ( configure_legacy_import_diagnostics, scan_plugin_legacy_imports, ) +from app.runtime.compat.resource_imports import scan_plugin_resource_imports from app.runtime.config import global_vars from app.runtime.extensions.plugin_manager import ( PluginManager, configure_plugin_install_reporter, configure_plugin_legacy_import_services, + configure_plugin_resource_import_preparer, configure_site_auth_level_provider, ) +from app.runtime.managed_resources import acquire_managed_resource from app.application.site.sites import SitesHelper # pylint: disable=no-name-in-module from app.adapters.external.server import MoviePilotServerHelper from app.runtime.log import logger +def _prepare_legacy_plugin_import(*, plugin_id: str, plugin_dir: Path) -> None: + """在执行旧插件顶层代码前准备其静态导入所需的宿主资源。""" + for capability_id in scan_plugin_resource_imports(plugin_id, plugin_dir): + acquire_managed_resource( + capability_id, + reason="legacy_plugin_import", + ) + + def _configure_plugin_services() -> None: """把兼容诊断、远程上报和站点认证等级装配到插件管理器。""" configure_plugin_legacy_import_services( diagnostics_configurator=configure_legacy_import_diagnostics, import_scanner=scan_plugin_legacy_imports, ) + configure_plugin_resource_import_preparer(_prepare_legacy_plugin_import) configure_plugin_install_reporter(MoviePilotServerHelper.install_plugin_reg) configure_site_auth_level_provider(lambda: SitesHelper().auth_level) diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 9cf6b55f3..80fc531d3 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -78,9 +78,10 @@ create additional top-level directory categories. | `app/runtime/events.py` | Event contracts, dispatch and resolver registration | | `app/runtime/log.py` | Complete console/plugin/file logging runtime and shutdown | | `app/runtime/cache.py` | Cache protocols, memory implementations, decorators and proxies | +| `app/runtime/managed_resources.py` | Provider-neutral acquisition, observation and shutdown facade for process-owned optional resources | | `app/runtime/state.py` | Process restart and update state | -| `app/runtime/extensions/` | Module, plugin and configured-service discovery/registration/lifecycle | -| `app/runtime/compat/` | Standard-library-only exact legacy import routing and DEBUG diagnostics | +| `app/runtime/extensions/` | Module, plugin, configured-service and managed-resource discovery/registration/lifecycle adapters | +| `app/runtime/compat/` | Standard-library-only exact legacy import routing, resource preflight scanning and DEBUG diagnostics | `app/startup/` remains the established composition root and is not nested under runtime. It injects providers and callbacks, orders initialization/shutdown and @@ -109,6 +110,16 @@ site extension owns the configured catalog/authentication/index capability and lives in `app/application/site/`; only its download and file installation mechanism remains in `app/adapters/system/resource.py`. +可选的进程级技术资源使用 Managed Resource 合同：实现及其 data-only +`capability.toml` 与适配器同目录，`runtime/extensions` 只解释通用的同步/异步 +`start`、`stop` 生命周期，`startup` 负责构建 Capability Runtime。声明必须使用 +`on_first_use`，普通启动只发现声明；消费者通过 `app/runtime/managed_resources.py` +显式获取资源。关闭路径先释放消费者，再关闭已初始化 Runtime，未使用的资源不得因关闭而物化。 +Runtime 关闭后不可逆；完整应用生命周期的再次启动必须由新进程承载，不能在同一解释器中重建局部资源域。 +插件需要浏览器时使用 `app.sdk.browser`，由宿主浏览器适配器协调资源，不直接依赖资源实现。 +旧插件若直接导入有资源前置条件的第三方包，compat 在插件 import 前递归扫描源码并保守准备资源； +无法精确解析的文件按全部已登记资源降级，最终可导入性仍由 Python loader 判断。 + `app/foundation/crypto.py` stays in foundation because it contains only generic RSA, digest and CryptoJS-compatible AES primitives and has no settings, policy, I/O or logging. Authentication, token, passkey, signing and two-factor policy @@ -355,8 +366,12 @@ policy. `app/db` therefore has no dependency on `app/domain`. | `app/runtime/events.py` | `EventManager`, `Event` and event resolver registration | | `app/runtime/extensions/module_manager.py` | Module discovery and lifecycle | | `app/runtime/extensions/plugin_manager.py` | Plugin discovery and lifecycle | +| `app/runtime/extensions/managed_resource_adapter.py` | Data-only managed-resource registry and sync/async lifecycle adapters | +| `app/runtime/managed_resources.py` | Lightweight acquisition, state observation and shutdown facade | | `app/foundation/reflection.py` | Generic reflection and Python module discovery | | `app/adapters/network/http.py` | Shared synchronous and asynchronous HTTP clients | +| `app/adapters/network/browser.py` | Browser launch facade and browser session implementation | +| `app/adapters/system/display/` | On-first-use virtual display resource and legacy `DisplayHelper` facade | | `app/application/rss.py` | Configured RSS retrieval and parsing | | `app/application/site/sites.*` | Generated site catalog, authentication and index capability plus its colocated data bundle | | `app/runtime/cache.py` | Cache contracts, memory backend, decorators and proxies | @@ -369,7 +384,7 @@ policy. `app/db` therefore has no dependency on `app/domain`. | `app/application/security/url.py` | URL/path validation, SSRF protection and signed image policy | | `app/application/mediaserver.py` | Configured media-server discovery and identity matching | | `app/runtime/compat/manifest.py` | Exact legacy-to-canonical import manifest | -| `app/sdk/` | Stable plugin imports | +| `app/sdk/` | Stable plugin imports, including provider-neutral browser launch functions | Run `tests/test_architecture_dependencies.py` after every ownership or import change. It rejects physical legacy or retired canonical sources, forbidden diff --git a/scripts/perf/README.md b/scripts/perf/README.md index 42597b6ac..a2b074069 100644 --- a/scripts/perf/README.md +++ b/scripts/perf/README.md @@ -50,6 +50,37 @@ ${PYTHON} scripts/perf/moviepilot_docker_ab.py \ 开发 harness 时可以用小数分钟做短冒烟，例如 `--points 0,0.02`。正式数据必须保持 `1,5,10,30`。 +未指定 `--scenario` 时仍使用 `idle-default`，样本目录和 Docker 资源名称与既有命令保持一致。 + +## 浏览器激活场景 + +浏览器场景使用 campaign browser seed 的独立克隆卷，不直接挂载或写入固定来源卷，也不会在样本 +阶段下载浏览器。容器保持 internal network；探针只发送信号，`app.sdk.browser` 的导入、浏览器上下文 +创建和本地 `data:` 页面校验都发生在主 MoviePilot Python 进程中。 + +非默认浏览器场景用于候选实现的 After 激活门禁；Before 不具备新 SDK，且旧实现启动时已经常驻 +Xvfb，因此不能用同一个 `0 → 0` / `0 → 1` 不变量衡量。三轮 Before/After 空载收益仍由默认 +`run` 的 `idle-default` 场景完成，浏览器场景用三个隔离的 After sample 记录冷激活成本。 + +```bash +../.venv/bin/python scripts/perf/moviepilot_docker_ab.py \ + --campaign v3-perf-002-headless \ + sample --variant after --index 1 --scenario browser-headless --points 1,5,10,30 + +../.venv/bin/python scripts/perf/moviepilot_docker_ab.py \ + --campaign v3-perf-002-headed \ + sample --variant after --index 1 --scenario browser-headed --points 1,5,10,30 +``` + +- `browser-headless`：一次真实 headless context 激活，要求 Xvfb `0 → 0`； +- `browser-headed`：主进程内两个线程通过屏障并发调用 + `launch_browser_context(headless=False)`；要求两个真实 SDK 冷启动调用成功、额外上下文关闭后只保留一个、 + Capability observation 只有一个 `headed_browser_launch` generation/start，且 Xvfb `0 → 1`； +- 激活完成后再开始 `1/5/10/30m` 计时，JSON 保留激活前后 Engine 网络、working set、进程 + PSS/USS/RSS/线程、Xvfb 数量/PSS、`sys.modules` 和进程内 marker； +- 非默认场景结果保存在 `samples//-/`，可与同 campaign 的 idle 样本并存， + Markdown 中位数会按场景分组，不会混算。 + ## 完整三组 A/B ```bash diff --git a/scripts/perf/instrument/sitecustomize.py b/scripts/perf/instrument/sitecustomize.py index 086e242dc..2fabc4f58 100644 --- a/scripts/perf/instrument/sitecustomize.py +++ b/scripts/perf/instrument/sitecustomize.py @@ -1,12 +1,41 @@ -"""MoviePilot Docker A/B 测量时使用的最小 ``sys.modules`` 快照探针。""" +"""MoviePilot Docker 测量进程内使用的最小诊断探针。""" + +from __future__ import annotations import os import signal import sys +# 场景激活依赖只在收到信号后加载，避免改变 idle-default 的 import 基线。 +# pylint: disable=import-outside-toplevel _OUTPUT_DIR = os.environ.get("MP_PERF_OUTPUT_DIR") +_SCENARIO = os.environ.get("MP_PERF_SCENARIO", "idle-default") +_ACTIVATION_TIMEOUT = float(os.environ.get("MP_PERF_ACTIVATION_TIMEOUT", "120")) _snapshot_index = 0 +_activation_started = False +_browser_resources: list[object] = [] + + +def _utc_now() -> str: + """返回稳定、可机器解析的 UTC 时间。""" + from datetime import datetime, timezone + + return datetime.now(timezone.utc).isoformat() + + +def _atomic_write_json(path, payload: dict[str, object]) -> None: + """原子发布结果，避免采集端读取到半写 marker。""" + import json + + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path = path.with_suffix(f"{path.suffix}.tmp") + with temporary_path.open("w", encoding="utf-8") as output: + json.dump(payload, output, ensure_ascii=False, indent=2, sort_keys=True) + output.write("\n") + output.flush() + os.fsync(output.fileno()) + os.replace(temporary_path, path) def _dump_modules(_signum, _frame) -> None: @@ -31,5 +60,318 @@ def _dump_modules(_signum, _frame) -> None: os.replace(temporary_path, final_path) +def _launch_one_browser( + *, + index: int, + headless: bool, + launcher, + start_gate, +) -> tuple[dict[str, object], list[object]]: + """启动一个本地 data URL 浏览器上下文并返回可序列化结果。""" + import time + + if start_gate is not None: + start_gate.wait(timeout=min(_ACTIVATION_TIMEOUT, 10)) + started_at = time.perf_counter() + retained: list[object] = [] + result: dict[str, object] = { + "index": index, + "headless": headless, + "started_at_monotonic": started_at, + } + try: + context = launcher(headless=headless) + retained.append(context) + page = context.new_page() + retained.append(page) + page.goto("data:text/html,MoviePilot Browser Probe") + title = page.title() + result.update( + { + "success": title == "MoviePilot Browser Probe", + "page_title": title, + "context_type": type(context).__name__, + } + ) + if not result["success"]: + result["error"] = "本地 data URL 标题校验失败" + except Exception as error: # pragma: no cover - 真实浏览器错误由 marker 保存 + result.update( + { + "success": False, + "error_type": type(error).__name__, + "error": str(error), + } + ) + result["elapsed_seconds"] = time.perf_counter() - started_at + return result, retained + + +def _enum_value(value): + """把 runtime 枚举降为 JSON 标量。""" + return getattr(value, "value", value) + + +def _read_display_runtime() -> dict[str, object]: + """读取 host.display 的只读状态和观测，不触发资源激活。""" + try: + from app.runtime.managed_resources import ( + managed_resource_observations, + managed_resource_snapshot, + ) + + snapshot = managed_resource_snapshot("host.display") + observations = managed_resource_observations("host.display") + return { + "available": True, + "snapshot": { + "capability_id": snapshot.capability_id, + "materialization": _enum_value(snapshot.materialization), + "lifecycle": _enum_value(snapshot.lifecycle), + "generation": snapshot.generation, + "visible": snapshot.visible, + "error": snapshot.error, + }, + "observations": [ + { + "capability_id": item.capability_id, + "generation": item.generation, + "operation": item.operation, + "outcome": item.outcome, + "reason": item.reason, + "materialization": _enum_value(item.materialization), + "lifecycle": _enum_value(item.lifecycle), + "duration_ms": item.duration_ms, + "error": item.error, + } + for item in observations + ], + } + except Exception as error: # pragma: no cover - 核心未就绪或真实 runtime 错误 + return { + "available": False, + "error_type": type(error).__name__, + "error": str(error), + } + + +def _close_browser_resources(resources: list[object]) -> list[dict[str, str]]: + """逆序关闭一次探针创建的页面与上下文，并返回可序列化错误。""" + errors: list[dict[str, str]] = [] + for resource in reversed(resources): + close = getattr(resource, "close", None) + if not callable(close): + continue + try: + close() + except Exception as error: # pragma: no cover - 真实浏览器错误由 marker 保存 + errors.append( + { + "resource_type": type(resource).__name__, + "error_type": type(error).__name__, + "error": str(error), + } + ) + return errors + + +def _activate_browser_scenario( + scenario: str, + launcher=None, +) -> dict[str, object]: + """通过公开 SDK 执行真实浏览器激活，headed 使用并发冷启动探针。""" + import threading + import time + + if scenario not in {"browser-headless", "browser-headed"}: + raise ValueError(f"场景不支持浏览器激活：{scenario}") + if launcher is None: + from app.sdk.browser import launch_browser_context + + launcher = launch_browser_context + + display_before = _read_display_runtime() + headless = scenario == "browser-headless" + concurrency = 1 if headless else 2 + launch_results: list[dict[str, object] | None] = [None] * concurrency + cleanup_error_slots: list[list[dict[str, str]]] = [ + [] for _index in range(concurrency) + ] + retained_slots = [False] * concurrency + start_gate = None if headless else threading.Barrier(concurrency) + completion_gate = None if headless else threading.Barrier(concurrency) + + def launch(index: int) -> None: + result, resources = _launch_one_browser( + index=index, + headless=headless, + launcher=launcher, + start_gate=start_gate, + ) + launch_results[index] = result + if headless: + if result.get("success"): + _browser_resources.extend(resources) + result["retained"] = True + retained_slots[index] = True + else: + cleanup_error_slots[index] = _close_browser_resources(resources) + return + + try: + completion_gate.wait(timeout=min(_ACTIVATION_TIMEOUT, 30)) + except threading.BrokenBarrierError: + cleanup_error_slots[index] = _close_browser_resources(resources) + result.update( + { + "success": False, + "error_type": "BrokenBarrierError", + "error": "并发浏览器启动未能完成同线程清理协调", + } + ) + return + + successful_indices = [ + candidate_index + for candidate_index, item in enumerate(launch_results) + if item is not None and bool(item.get("success")) + ] + retained_index = min(successful_indices) if successful_indices else None + if index == retained_index: + # 保留对象不再跨线程使用；容器退出会回收浏览器及其 worker 进程。 + _browser_resources.extend(resources) + result["retained"] = True + retained_slots[index] = True + else: + # Playwright sync/greenlet 对象必须在创建它的线程内关闭。 + cleanup_error_slots[index] = _close_browser_resources(resources) + + if headless: + launch(0) + else: + threads = [ + threading.Thread( + target=launch, + args=(index,), + name=f"mp-perf-browser-launch-{index}", + daemon=True, + ) + for index in range(concurrency) + ] + for thread in threads: + thread.start() + deadline = time.monotonic() + _ACTIVATION_TIMEOUT + for thread in threads: + thread.join(timeout=max(deadline - time.monotonic(), 0)) + + serialized_launches = [ + item + if item is not None + else { + "index": index, + "success": False, + "error_type": "TimeoutError", + "error": "浏览器启动未在进程内超时前完成", + } + for index, item in enumerate(launch_results) + ] + launch_starts = [ + float(item["started_at_monotonic"]) + for item in serialized_launches + if "started_at_monotonic" in item + ] + successful_indices = [ + index + for index, item in enumerate(serialized_launches) + if bool(item.get("success")) + ] + cleanup_errors = [ + error for slot_errors in cleanup_error_slots for error in slot_errors + ] + retained_count = sum(retained_slots) + + expected_successes = concurrency + browser_success = ( + len(successful_indices) == expected_successes and not cleanup_errors + ) + return { + "requested": True, + "headless": headless, + "concurrency": concurrency, + "successes": len(successful_indices), + "retained_contexts": retained_count, + "launches": serialized_launches, + "cleanup_errors": cleanup_errors, + "success": browser_success, + "managed_resource": { + "before": display_before, + "after": _read_display_runtime(), + }, + "single_flight_probe": { + "requested": not headless, + "concurrent_callers": concurrency if not headless else 0, + "successful_callers": len(successful_indices) if not headless else 0, + "barrier_used": not headless, + "launch_start_spread_ms": ( + (max(launch_starts) - min(launch_starts)) * 1000 + if launch_starts + else None + ), + "all_callers_succeeded": len(successful_indices) == expected_successes, + "calls": serialized_launches if not headless else [], + }, + } + + +def _run_activation() -> None: + """在目标解释器的工作线程中运行激活并发布完成 marker。""" + if not _OUTPUT_DIR: + return + import time + from pathlib import Path + + started_at = time.perf_counter() + result: dict[str, object] = { + "schema_version": 1, + "scenario": _SCENARIO, + "pid": os.getpid(), + "started_at": _utc_now(), + } + try: + result["browser"] = _activate_browser_scenario(_SCENARIO) + result["success"] = bool(result["browser"]["success"]) + except Exception as error: # pragma: no cover - 真实集成错误由 marker 保存 + result.update( + { + "success": False, + "error_type": type(error).__name__, + "error": str(error), + } + ) + result["elapsed_seconds"] = time.perf_counter() - started_at + result["completed_at"] = _utc_now() + _atomic_write_json( + Path(_OUTPUT_DIR) / f"activation-{os.getpid()}.json", + result, + ) + + +def _request_activation(_signum, _frame) -> None: + """SIGUSR2 只调度一次工作线程，真实 import 与启动仍在目标进程内完成。""" + global _activation_started + if not _OUTPUT_DIR or _activation_started: + return + import threading + + _activation_started = True + threading.Thread( + target=_run_activation, + name="mp-perf-scenario-activation", + daemon=True, + ).start() + + if _OUTPUT_DIR and hasattr(signal, "SIGUSR1"): signal.signal(signal.SIGUSR1, _dump_modules) +if _OUTPUT_DIR and hasattr(signal, "SIGUSR2"): + signal.signal(signal.SIGUSR2, _request_activation) diff --git a/scripts/perf/moviepilot_docker_ab.py b/scripts/perf/moviepilot_docker_ab.py index 797085cda..5a2a97829 100644 --- a/scripts/perf/moviepilot_docker_ab.py +++ b/scripts/perf/moviepilot_docker_ab.py @@ -31,6 +31,8 @@ DEFAULT_SUBSTRATE = ( "sha256:925de1fdf1bb0312144bc818bc8ebaa999a9a159c6d14f1b48b0ff05edb7f720" ) DEFAULT_BROWSER_SOURCE_VOLUME = "mp-perf-v3-browser-seed" +DEFAULT_SCENARIO = "idle-default" +SCENARIOS = (DEFAULT_SCENARIO, "browser-headless", "browser-headed") CAMPAIGN_LABEL = "org.moviepilot.perf.campaign" ROLE_LABEL = "org.moviepilot.perf.role" SOURCE_LABEL = "org.moviepilot.perf.source-commit" @@ -587,6 +589,10 @@ def fixed_environment(args: argparse.Namespace, instrument: bool) -> dict[str, s { "PYTHONPATH": "/opt/moviepilot-perf/instrument", "MP_PERF_OUTPUT_DIR": "/opt/moviepilot-perf/out/modules", + "MP_PERF_SCENARIO": getattr(args, "scenario", DEFAULT_SCENARIO), + "MP_PERF_ACTIVATION_TIMEOUT": str( + getattr(args, "activation_timeout", 180) + ), } ) return environment @@ -1103,7 +1109,9 @@ def sample_volume_names( index: int, ) -> tuple[str, str]: """返回单个样本的隔离配置和浏览器卷名称。""" - prefix = f"{resource_prefix(args)}-{variant}-{index}" + scenario = getattr(args, "scenario", DEFAULT_SCENARIO) + scenario_segment = "" if scenario == DEFAULT_SCENARIO else f"-{scenario}" + prefix = f"{resource_prefix(args)}{scenario_segment}-{variant}-{index}" return f"{prefix}-config", f"{prefix}-browser" @@ -1113,11 +1121,190 @@ def sample_result_directory( index: int, ) -> Path: """返回单个样本的原始结果目录。""" - return campaign_directory(args) / "samples" / f"{variant}-{index}" + scenario = getattr(args, "scenario", DEFAULT_SCENARIO) + sample_root = campaign_directory(args) / "samples" + if scenario == DEFAULT_SCENARIO: + return sample_root / f"{variant}-{index}" + return sample_root / scenario / f"{variant}-{index}" + + +def capture_activation_snapshot( + container, + output_dir: Path, + phase: str, +) -> dict[str, Any]: + """采集浏览器激活边界的 Engine、进程和进程内 import 状态。""" + engine = capture_engine_stats(container) + processes = capture_processes(container) + modules = capture_modules(container, output_dir, processes["main_python"]) + return { + "phase": phase, + "captured_at": utc_now(), + "engine": engine, + "processes": processes, + "modules": modules, + } + + +def evaluate_browser_activation( + scenario: str, + pre: dict[str, Any], + post: dict[str, Any], + marker: dict[str, Any], + expected_pid: Optional[int] = None, +) -> dict[str, Any]: + """按场景不变量判断浏览器与 display 的真实激活是否有效。""" + pre_xvfb = pre["processes"]["xvfb"] + post_xvfb = post["processes"]["xvfb"] + browser = marker.get("browser") or {} + managed_resource = browser.get("managed_resource") or {} + managed_before = managed_resource.get("before") or {} + managed_after = managed_resource.get("after") or {} + before_observations = managed_before.get("observations") or [] + after_observations = managed_after.get("observations") or [] + observation_prefix_matches = ( + after_observations[: len(before_observations)] == before_observations + ) + new_observations = ( + after_observations[len(before_observations) :] + if observation_prefix_matches + else after_observations + ) + display_starts = [ + item + for item in new_observations + if item.get("operation") == "activate" and item.get("outcome") == "started" + ] + display_successes = [ + item + for item in new_observations + if item.get("operation") == "activate" and item.get("outcome") == "succeeded" + ] + display_start_reasons = [item.get("reason") for item in display_starts] + before_generation = (managed_before.get("snapshot") or {}).get("generation") + after_generation = (managed_after.get("snapshot") or {}).get("generation") + errors: list[str] = [] + if marker.get("scenario") != scenario: + errors.append("进程内 marker 的场景与采集请求不一致") + if expected_pid is not None and marker.get("pid") != expected_pid: + errors.append("进程内 marker 不是目标 MoviePilot Python 进程写出") + if not marker.get("success") or not browser.get("success"): + errors.append("主 MoviePilot Python 进程未完成浏览器激活") + if browser.get("retained_contexts") != 1: + errors.append("激活后必须保留一个浏览器上下文供 post activation 采样") + if not managed_before.get("available") or not managed_after.get("available"): + errors.append("主进程未提供 host.display managed resource 观测") + process_single_flight = browser.get("single_flight_probe") or {} + + single_flight = { + "requested": scenario == "browser-headed", + "concurrent_callers": int(process_single_flight.get("concurrent_callers") or 0), + "successful_callers": int(process_single_flight.get("successful_callers") or 0), + "xvfb_process_delta": int(post_xvfb["count"]) - int(pre_xvfb["count"]), + "generation_before": before_generation, + "generation_after": after_generation, + "activation_start_count": len(display_starts), + "activation_success_count": len(display_successes), + "activation_start_reasons": display_start_reasons, + "observation_prefix_matches": observation_prefix_matches, + "passed": None, + } + if scenario == "browser-headless": + if pre_xvfb["count"] != 0 or post_xvfb["count"] != 0: + errors.append("headless 激活前后都不得存在 Xvfb") + if display_starts or before_generation != after_generation: + errors.append("headless 激活不得申请 host.display") + elif scenario == "browser-headed": + if pre_xvfb["count"] != 0: + errors.append("headed 冷激活前必须没有 Xvfb") + if post_xvfb["count"] != 1: + errors.append("headed 并发激活后必须恰好存在一个 Xvfb") + single_flight["passed"] = ( + single_flight["concurrent_callers"] == 2 + and single_flight["successful_callers"] == 2 + and single_flight["xvfb_process_delta"] == 1 + and single_flight["activation_start_count"] == 1 + and single_flight["activation_success_count"] == 1 + and single_flight["activation_start_reasons"] == ["headed_browser_launch"] + and before_generation is not None + and after_generation == before_generation + 1 + ) + if not single_flight["passed"]: + errors.append("headed 并发请求未证明 display single-flight") + else: + errors.append(f"未知浏览器场景：{scenario}") + + return { + "passed": not errors, + "errors": errors, + "expected": ("Xvfb 0→0" if scenario == "browser-headless" else "Xvfb 0→1"), + "observed": { + "pre_xvfb_count": pre_xvfb["count"], + "pre_xvfb_pss_kib": pre_xvfb["pss_kib"], + "post_xvfb_count": post_xvfb["count"], + "post_xvfb_pss_kib": post_xvfb["pss_kib"], + }, + "single_flight": single_flight, + } + + +def activate_browser_scenario( + container, + output_dir: Path, + scenario: str, + timeout: float, +) -> dict[str, Any]: + """通过 SIGUSR2 让目标 MoviePilot 解释器执行场景激活并回收 marker。""" + pre = capture_activation_snapshot(container, output_dir, "pre-activation") + main_python = pre["processes"]["main_python"] + if not main_python: + raise HarnessError("未找到主 Python 进程，无法触发浏览器场景") + marker_path = output_dir / "modules" / f"activation-{main_python['pid']}.json" + marker_path.unlink(missing_ok=True) + + requested_at = time.monotonic() + result = container.exec_run(["kill", "-USR2", str(main_python["pid"])]) + if result.exit_code != 0: + raise HarnessError("向主 Python 进程发送场景激活信号失败") + deadline = requested_at + timeout + while time.monotonic() < deadline: + if marker_path.exists(): + break + if not container_running(container): + raise HarnessError("等待场景激活 marker 时容器提前退出") + time.sleep(0.05) + if not marker_path.exists(): + raise HarnessError(f"浏览器场景激活在 {timeout:.0f}s 内未完成") + + marker_received_at = time.monotonic() + marker = json.loads(marker_path.read_text(encoding="utf-8")) + post = capture_activation_snapshot(container, output_dir, "post-activation") + validation = evaluate_browser_activation( + scenario, + pre, + post, + marker, + expected_pid=main_python["pid"], + ) + return { + "scenario": scenario, + "trigger": "SIGUSR2-to-main-python", + "main_python_pid": main_python["pid"], + "orchestrator_elapsed_seconds": marker_received_at - requested_at, + "post_capture_elapsed_seconds": time.monotonic() - marker_received_at, + "worker_elapsed_seconds": marker.get("elapsed_seconds"), + "pre": pre, + "post": post, + "marker": marker, + "validation": validation, + } def command_sample(args: argparse.Namespace) -> dict[str, Any]: """执行一个隔离样本并在约定时间点采集完整指标。""" + scenario = getattr(args, "scenario", DEFAULT_SCENARIO) + if scenario != DEFAULT_SCENARIO and args.variant != "after": + raise HarnessError("浏览器激活场景只用于验证包含 app.sdk.browser 的 After 候选") client = require_docker_client() build = load_build_manifest(args) config_seed, browser_seed = require_seed_volumes(client, args) @@ -1144,13 +1331,21 @@ def command_sample(args: argparse.Namespace) -> dict[str, Any]: clone_volume(client, image, browser_seed, browser_volume.name) browser_before = volume_fingerprint(client, image, browser_volume.name) network = ensure_internal_network(client, args) - container_name = f"{resource_prefix(args)}-{args.variant}-{args.index}" + scenario_segment = "" if scenario == DEFAULT_SCENARIO else f"-{scenario}" + container_name = ( + f"{resource_prefix(args)}{scenario_segment}-{args.variant}-{args.index}" + ) + role = ( + f"sample-{args.variant}-{args.index}" + if scenario == DEFAULT_SCENARIO + else f"sample-{scenario}-{args.variant}-{args.index}" + ) container = create_app_container( client, args, image=image, name=container_name, - role=f"sample-{args.variant}-{args.index}", + role=role, config_volume=config_volume.name, browser_volume=browser_volume.name, network_name=network.name, @@ -1161,6 +1356,7 @@ def command_sample(args: argparse.Namespace) -> dict[str, Any]: "campaign": args.campaign, "variant": args.variant, "sample_index": args.index, + "scenario": scenario, "source_commit": build[f"{args.variant}_commit"], "image": image, "started_at": utc_now(), @@ -1171,6 +1367,7 @@ def command_sample(args: argparse.Namespace) -> dict[str, Any]: "network": "internal", "database": "sqlite-seed-clone", "browser": "prewarmed-seed-clone", + "scenario": scenario, }, "browser_before": browser_before, "measurements": [], @@ -1191,9 +1388,27 @@ def command_sample(args: argparse.Namespace) -> dict[str, Any]: result["http_ready_seconds"] = ready_seconds result["settled_seconds"] = settled_at - started_at result["settled_wait_seconds_after_ready"] = settled_at - ready_at + measurement_origin_at = settled_at + + if scenario != DEFAULT_SCENARIO: + activation = activate_browser_scenario( + container, + output_dir, + scenario, + args.activation_timeout, + ) + result["activation"] = activation + atomic_write_json(output_dir / "result.partial.json", result) + if not activation["validation"]["passed"]: + details = "; ".join(activation["validation"]["errors"]) + raise HarnessError(f"{scenario} 场景激活不满足验收条件：{details}") + measurement_origin_at = time.monotonic() + result["measurement_origin"] = "post-activation" + else: + result["measurement_origin"] = "settled" for point in args.points: - deadline = settled_at + point * 60 + deadline = measurement_origin_at + point * 60 remaining = deadline - time.monotonic() if remaining > 0: time.sleep(remaining) @@ -1201,7 +1416,12 @@ def command_sample(args: argparse.Namespace) -> dict[str, Any]: raise HarnessError(f"容器在 {point:g}m 采样前退出") print(f"[{args.variant}-{args.index}] sampling {point:g}m") result["measurements"].append( - capture_measurement(container, output_dir, point, settled_at) + capture_measurement( + container, + output_dir, + point, + measurement_origin_at, + ) ) atomic_write_json(output_dir / "result.partial.json", result) assert_no_app_env(container) @@ -1236,7 +1456,7 @@ def load_sample_results(args: argparse.Namespace) -> list[dict[str, Any]]: results = [] if not sample_root.exists(): return results - for path in sorted(sample_root.glob("*/result.json")): + for path in sorted(sample_root.rglob("result.json")): results.append(json.loads(path.read_text(encoding="utf-8"))) return results @@ -1282,6 +1502,10 @@ def build_markdown_report( samples: list[dict[str, Any]], ) -> str: """生成不含本机路径和凭据的 Markdown 汇总。""" + scenarios = sorted( + {sample.get("scenario", DEFAULT_SCENARIO) for sample in samples} + ) or [DEFAULT_SCENARIO] + show_scenario = any(scenario != DEFAULT_SCENARIO for scenario in scenarios) points = sorted( { float(measurement["target_minute"]) @@ -1314,7 +1538,8 @@ def build_markdown_report( ) headers = ( - ["版本", "样本", "HTTP ready(s)"] + (["场景"] if show_scenario else []) + + ["版本", "样本", "HTTP ready(s)"] + [f"{point:g}m WS(MiB)" for point in points] + [ "末次 Python PSS(MiB)", @@ -1333,11 +1558,12 @@ def build_markdown_report( for sample in sorted( samples, key=lambda item: ( + item.get("scenario", DEFAULT_SCENARIO), variant_order.get(item["variant"], 99), item["sample_index"], ), ): - row = [ + row = ([sample.get("scenario", DEFAULT_SCENARIO)] if show_scenario else []) + [ sample["variant"], str(sample["sample_index"]), f"{sample.get('http_ready_seconds', 0):.2f}" @@ -1388,53 +1614,149 @@ def build_markdown_report( ) lines.append("| " + " | ".join(row) + " |") + activated_samples = [sample for sample in samples if sample.get("activation")] + if activated_samples: + activation_headers = [ + "场景", + "版本", + "样本", + "激活(s)", + "Pre WS(MiB)", + "Post WS(MiB)", + "Pre Python PSS(MiB)", + "Post Python PSS(MiB)", + "Pre Xvfb", + "Post Xvfb", + "Post Xvfb PSS(MiB)", + "Activation RX Δ(KiB)", + "Activation TX Δ(KiB)", + "Browser", + "Single-flight generation/start", + "验收", + ] + lines.extend( + [ + "", + "## 场景激活", + "", + "| " + " | ".join(activation_headers) + " |", + "| " + " | ".join(["---"] * len(activation_headers)) + " |", + ] + ) + for sample in sorted( + activated_samples, + key=lambda item: ( + item.get("scenario", DEFAULT_SCENARIO), + variant_order.get(item["variant"], 99), + item["sample_index"], + ), + ): + activation = sample["activation"] + pre = activation["pre"] + post = activation["post"] + marker = activation["marker"] + validation = activation["validation"] + pre_python = pre["processes"].get("main_python") or {} + post_python = post["processes"].get("main_python") or {} + single_flight = validation["single_flight"] + activation_row = [ + sample.get("scenario", DEFAULT_SCENARIO), + sample["variant"], + str(sample["sample_index"]), + f"{float(activation.get('worker_elapsed_seconds') or 0):.2f}", + format_mib(pre["engine"]["working_set_bytes"]), + format_mib(post["engine"]["working_set_bytes"]), + format_kib_as_mib(pre_python.get("pss_kib")), + format_kib_as_mib(post_python.get("pss_kib")), + str(pre["processes"]["xvfb"]["count"]), + str(post["processes"]["xvfb"]["count"]), + format_kib_as_mib(post["processes"]["xvfb"]["pss_kib"]), + format_bytes_as_kib( + post["engine"]["network_rx_bytes"] + - pre["engine"]["network_rx_bytes"] + ), + format_bytes_as_kib( + post["engine"]["network_tx_bytes"] + - pre["engine"]["network_tx_bytes"] + ), + "成功" if marker.get("success") else "失败", + ( + f"{single_flight.get('generation_after')}/" + f"{single_flight.get('activation_start_count')}" + if single_flight.get("passed") is True + else "不适用" + if single_flight.get("passed") is None + else "失败" + ), + "通过" if validation["passed"] else "失败", + ] + lines.append("| " + " | ".join(activation_row) + " |") + lines.extend(["", "## 中位数对照", ""]) - if points: - lines.append("| 时间点 | Before(MiB) | After(MiB) | 净差(MiB) | 变化 |") - lines.append("| --- | ---: | ---: | ---: | ---: |") - for point in points: - before_values = [ - measurement_at(sample, point)["engine"]["working_set_bytes"] - for sample in samples - if sample["variant"] == "before" and measurement_at(sample, point) - ] - after_values = [ - measurement_at(sample, point)["engine"]["working_set_bytes"] - for sample in samples - if sample["variant"] == "after" and measurement_at(sample, point) - ] - before_median = median(before_values) - after_median = median(after_values) - if before_median is None or after_median is None: - lines.append(f"| {point:g}m | — | — | — | — |") - continue - delta = after_median - before_median - percent = delta / before_median * 100 if before_median else 0 - lines.append( - f"| {point:g}m | {format_mib(before_median)} | " - f"{format_mib(after_median)} | {delta / 1024 / 1024:.1f} | {percent:.1f}% |" + for scenario in scenarios: + scenario_samples = [ + sample + for sample in samples + if sample.get("scenario", DEFAULT_SCENARIO) == scenario + ] + if show_scenario: + lines.extend([f"### `{scenario}`", ""]) + if points: + lines.append("| 时间点 | Before(MiB) | After(MiB) | 净差(MiB) | 变化 |") + lines.append("| --- | ---: | ---: | ---: | ---: |") + for point in points: + before_values = [ + measurement_at(sample, point)["engine"]["working_set_bytes"] + for sample in scenario_samples + if sample["variant"] == "before" and measurement_at(sample, point) + ] + after_values = [ + measurement_at(sample, point)["engine"]["working_set_bytes"] + for sample in scenario_samples + if sample["variant"] == "after" and measurement_at(sample, point) + ] + before_median = median(before_values) + after_median = median(after_values) + if before_median is None or after_median is None: + lines.append(f"| {point:g}m | — | — | — | — |") + continue + delta = after_median - before_median + percent = delta / before_median * 100 if before_median else 0 + lines.append( + f"| {point:g}m | {format_mib(before_median)} | " + f"{format_mib(after_median)} | {delta / 1024 / 1024:.1f} | " + f"{percent:.1f}% |" + ) + lines.append("") + + lines.extend(["## 启动时间", ""]) + for scenario in scenarios: + scenario_samples = [ + sample + for sample in samples + if sample.get("scenario", DEFAULT_SCENARIO) == scenario + ] + ready_before = median( + sample["http_ready_seconds"] + for sample in scenario_samples + if sample["variant"] == "before" and "http_ready_seconds" in sample + ) + ready_after = median( + sample["http_ready_seconds"] + for sample in scenario_samples + if sample["variant"] == "after" and "http_ready_seconds" in sample + ) + scenario_prefix = f"`{scenario}`：" if show_scenario else "" + if ready_before is not None and ready_after is not None: + startup_change = ( + (ready_after - ready_before) / ready_before * 100 if ready_before else 0 ) - ready_before = median( - sample["http_ready_seconds"] - for sample in samples - if sample["variant"] == "before" and "http_ready_seconds" in sample - ) - ready_after = median( - sample["http_ready_seconds"] - for sample in samples - if sample["variant"] == "after" and "http_ready_seconds" in sample - ) - lines.extend(["", "## 启动时间", ""]) - if ready_before is not None and ready_after is not None: - startup_change = ( - (ready_after - ready_before) / ready_before * 100 if ready_before else 0 - ) - lines.append( - f"Before 中位数 {ready_before:.2f}s，After 中位数 {ready_after:.2f}s，" - f"变化 {startup_change:.1f}%。" - ) - else: - lines.append("样本尚不完整。") + lines.append( + f"{scenario_prefix}Before 中位数 {ready_before:.2f}s，" + f"After 中位数 {ready_after:.2f}s，变化 {startup_change:.1f}%。" + ) + else: + lines.append(f"{scenario_prefix}样本尚不完整。") lines.extend( [ "", @@ -1605,6 +1927,12 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--memory", default="2g") parser.add_argument("--ready-timeout", type=int, default=300) parser.add_argument("--settle-timeout", type=int, default=300) + parser.add_argument( + "--activation-timeout", + type=int, + default=180, + help="进程内场景激活完成 marker 的等待秒数", + ) parser.add_argument("--stop-timeout", type=int, default=120) subparsers = parser.add_subparsers(dest="command", required=True) @@ -1624,6 +1952,12 @@ def build_parser() -> argparse.ArgumentParser: sample.add_argument( "--points", type=parse_points, default=parse_points("1,5,10,30") ) + sample.add_argument( + "--scenario", + choices=SCENARIOS, + default=DEFAULT_SCENARIO, + help="样本场景；默认保持 PERF-001 idle-default 行为", + ) sample.add_argument("--replace", action="store_true") run = subparsers.add_parser("run", help="完整执行 build、seed 和三组平衡 A/B") @@ -1652,7 +1986,12 @@ def main(argv: Optional[list[str]] = None) -> int: args.browser_source_volume = DEFAULT_BROWSER_SOURCE_VOLUME if args.cpus <= 0: parser.error("--cpus 必须大于 0") - if args.ready_timeout <= 0 or args.settle_timeout <= 0 or args.stop_timeout <= 0: + if ( + args.ready_timeout <= 0 + or args.settle_timeout <= 0 + or args.activation_timeout <= 0 + or args.stop_timeout <= 0 + ): parser.error("timeout 必须大于 0") try: if args.command == "build": diff --git a/scripts/perf/test_scenarios.py b/scripts/perf/test_scenarios.py new file mode 100644 index 000000000..013952e2f --- /dev/null +++ b/scripts/perf/test_scenarios.py @@ -0,0 +1,430 @@ +"""PERF Docker harness 场景协议的无 Docker fake 测试。""" + +from __future__ import annotations + +import argparse +import importlib.util +import json +import os +import sys +import threading +import time +from pathlib import Path +from types import ModuleType, SimpleNamespace + +import pytest + + +PERF_DIR = Path(__file__).resolve().parent + + +def load_module(name: str, path: Path): + """从脚本路径加载模块，避免要求 scripts 变成运行时 Python package。""" + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def snapshot(xvfb_count: int, xvfb_pss_kib: int = 0) -> dict: + """构造只包含验收字段的进程快照。""" + return { + "processes": { + "xvfb": {"count": xvfb_count, "pss_kib": xvfb_pss_kib}, + } + } + + +def managed_resource(before_generation: int, after_generation: int) -> dict: + """构造 host.display single-flight 观测。""" + observations = [] + if after_generation > before_generation: + observations = [ + { + "operation": "activate", + "outcome": "started", + "generation": after_generation, + "reason": "headed_browser_launch", + }, + { + "operation": "activate", + "outcome": "succeeded", + "generation": after_generation, + }, + ] + return { + "before": { + "available": True, + "snapshot": {"generation": before_generation}, + "observations": [], + }, + "after": { + "available": True, + "snapshot": {"generation": after_generation}, + "observations": observations, + }, + } + + +class FakePage: + """验证本地 data URL 的同步页面替身。""" + + def __init__(self) -> None: + self.url = "" + + def goto(self, url: str) -> None: + self.url = url + + def title(self) -> str: + assert self.url.startswith("data:text/html,") + return "MoviePilot Browser Probe" + + def close(self) -> None: + """模拟 Playwright page 对称关闭。""" + + +class FakeContext: + """浏览器上下文替身。""" + + def new_page(self) -> FakePage: + return FakePage() + + def close(self) -> None: + """模拟 CloakBrowser context 对称关闭。""" + + +def test_default_cli_and_paths_keep_idle_contract(tmp_path: Path) -> None: + """未指定 scenario 时保持既有 idle 命令和资源路径。""" + harness = load_module("moviepilot_perf_cli", PERF_DIR / "moviepilot_docker_ab.py") + args = harness.build_parser().parse_args( + [ + "--campaign", + "fake", + "--output-dir", + str(tmp_path), + "sample", + "--variant", + "after", + "--index", + "1", + ] + ) + + assert args.scenario == "idle-default" + assert harness.sample_volume_names(args, "after", 1) == ( + "mpperf-fake-after-1-config", + "mpperf-fake-after-1-browser", + ) + assert harness.sample_result_directory(args, "after", 1) == ( + tmp_path / "fake" / "samples" / "after-1" + ) + + +def test_browser_scenario_uses_isolated_resource_and_result_names( + tmp_path: Path, +) -> None: + """不同激活场景不会覆盖 idle 样本或彼此复用可写卷。""" + harness = load_module("moviepilot_perf_paths", PERF_DIR / "moviepilot_docker_ab.py") + args = argparse.Namespace( + campaign="fake", + output_dir=tmp_path, + scenario="browser-headed", + ) + + assert harness.sample_volume_names(args, "after", 2) == ( + "mpperf-fake-browser-headed-after-2-config", + "mpperf-fake-browser-headed-after-2-browser", + ) + assert harness.sample_result_directory(args, "after", 2) == ( + tmp_path / "fake" / "samples" / "browser-headed" / "after-2" + ) + + +def test_browser_scenario_rejects_before_without_touching_docker() -> None: + """旧基线不具备 SDK/display 冷启动不变量，非默认场景只接受 After。""" + harness = load_module( + "moviepilot_perf_after_only", PERF_DIR / "moviepilot_docker_ab.py" + ) + args = argparse.Namespace(scenario="browser-headless", variant="before") + + with pytest.raises(harness.HarnessError, match="After"): + harness.command_sample(args) + + +def test_activation_validation_enforces_headless_and_headed_invariants() -> None: + """headless 保持无 Xvfb，headed 两调用只能产生一个 Xvfb。""" + harness = load_module( + "moviepilot_perf_validation", + PERF_DIR / "moviepilot_docker_ab.py", + ) + headless_marker = { + "scenario": "browser-headless", + "pid": 42, + "success": True, + "browser": { + "success": True, + "concurrency": 1, + "successes": 1, + "retained_contexts": 1, + "managed_resource": managed_resource(0, 0), + "single_flight_probe": { + "concurrent_callers": 0, + "successful_callers": 0, + }, + }, + } + headed_marker = { + "scenario": "browser-headed", + "pid": 42, + "success": True, + "browser": { + "success": True, + "concurrency": 2, + "successes": 2, + "retained_contexts": 1, + "managed_resource": managed_resource(0, 1), + "single_flight_probe": { + "concurrent_callers": 2, + "successful_callers": 2, + }, + }, + } + + headless = harness.evaluate_browser_activation( + "browser-headless", + snapshot(0), + snapshot(0), + headless_marker, + expected_pid=42, + ) + headed = harness.evaluate_browser_activation( + "browser-headed", + snapshot(0), + snapshot(1, 72 * 1024), + headed_marker, + expected_pid=42, + ) + invalid = harness.evaluate_browser_activation( + "browser-headed", + snapshot(0), + snapshot(2, 144 * 1024), + headed_marker, + expected_pid=42, + ) + + assert headless["passed"] is True + assert headed["passed"] is True + assert headed["single_flight"]["passed"] is True + assert headed["single_flight"]["generation_after"] == 1 + assert headed["single_flight"]["activation_start_count"] == 1 + assert invalid["passed"] is False + assert invalid["single_flight"]["passed"] is False + + +def test_sitecustomize_acquires_headed_display_concurrently_in_same_process() -> None: + """headed probe 并发走公开 SDK 冷启动，并只保留一个上下文。""" + probe = load_module( + "moviepilot_perf_sitecustomize", + PERF_DIR / "instrument" / "sitecustomize.py", + ) + browser_calls: list[tuple[int, bool]] = [] + closed_contexts: list[tuple[int, int]] = [] + lock = threading.Lock() + + class TrackedContext(FakeContext): + """记录并发探针关闭的额外浏览器上下文。""" + + def __init__(self, index: int) -> None: + self.index = index + + def close(self) -> None: + closed_contexts.append((self.index, threading.get_ident())) + + def launcher(*, headless: bool) -> FakeContext: + with lock: + browser_calls.append((threading.get_ident(), headless)) + index = len(browser_calls) - 1 + return TrackedContext(index) + + result = probe._activate_browser_scenario( + "browser-headed", + launcher=launcher, + ) + + assert result["success"] is True + assert result["successes"] == 2 + assert result["retained_contexts"] == 1 + assert len(browser_calls) == 2 + assert len({thread_id for thread_id, _headless in browser_calls}) == 2 + assert all(headless is False for _thread_id, headless in browser_calls) + assert len(closed_contexts) == 1 + closed_index, closed_thread_id = closed_contexts[0] + assert closed_thread_id == browser_calls[closed_index][0] + assert result["single_flight_probe"]["barrier_used"] is True + + +def test_sitecustomize_headless_uses_one_headless_context() -> None: + """headless probe 只启动一个无显示上下文。""" + probe = load_module( + "moviepilot_perf_sitecustomize_headless", + PERF_DIR / "instrument" / "sitecustomize.py", + ) + calls: list[bool] = [] + + def launcher(*, headless: bool) -> FakeContext: + calls.append(headless) + return FakeContext() + + result = probe._activate_browser_scenario("browser-headless", launcher=launcher) + + assert result["success"] is True + assert calls == [True] + assert result["single_flight_probe"]["requested"] is False + + +def test_sitecustomize_serializes_managed_resource_facade(monkeypatch) -> None: + """进程探针按公开只读 facade 记录 generation 与 activate observation。""" + observation = SimpleNamespace( + capability_id="host.display", + generation=1, + operation="activate", + outcome="started", + reason="fake", + materialization="materialized", + lifecycle="starting", + duration_ms=0.5, + error=None, + ) + runtime_snapshot = SimpleNamespace( + capability_id="host.display", + materialization="materialized", + lifecycle="running", + generation=1, + visible=True, + error=None, + ) + + facade = ModuleType("app.runtime.managed_resources") + + def managed_resource_snapshot(capability_id: str): + assert capability_id == "host.display" + return runtime_snapshot + + def managed_resource_observations(capability_id=None): + assert capability_id == "host.display" + return (observation,) + + facade.managed_resource_snapshot = managed_resource_snapshot + facade.managed_resource_observations = managed_resource_observations + monkeypatch.setitem(sys.modules, "app.runtime.managed_resources", facade) + probe = load_module( + "moviepilot_perf_sitecustomize_observation", + PERF_DIR / "instrument" / "sitecustomize.py", + ) + + result = probe._read_display_runtime() + + assert result["available"] is True + assert result["snapshot"]["generation"] == 1 + assert result["observations"][0]["operation"] == "activate" + assert result["observations"][0]["outcome"] == "started" + + +def test_sitecustomize_signal_worker_publishes_atomic_marker(tmp_path: Path) -> None: + """信号回调只调度目标进程工作线程，并发布带 PID 的完成 marker。""" + probe = load_module( + "moviepilot_perf_sitecustomize_marker", + PERF_DIR / "instrument" / "sitecustomize.py", + ) + probe._OUTPUT_DIR = str(tmp_path) + probe._SCENARIO = "browser-headless" + probe._activation_started = False + probe._activate_browser_scenario = lambda scenario: { + "success": scenario == "browser-headless" + } + + probe._request_activation(None, None) + marker_path = tmp_path / f"activation-{os.getpid()}.json" + deadline = time.monotonic() + 2 + while time.monotonic() < deadline and not marker_path.exists(): + time.sleep(0.01) + + payload = json.loads(marker_path.read_text(encoding="utf-8")) + assert payload["pid"] == os.getpid() + assert payload["scenario"] == "browser-headless" + assert payload["success"] is True + assert not list(tmp_path.glob("*.tmp")) + + +def test_markdown_reports_activation_and_keeps_scenario_medians_separate() -> None: + """非默认场景报告包含激活证据，并按场景隔离中位数。""" + harness = load_module( + "moviepilot_perf_report", PERF_DIR / "moviepilot_docker_ab.py" + ) + process_data = { + "main_python": {"pss_kib": 400 * 1024, "uss_kib": 390 * 1024, "threads": 8}, + "xvfb": {"count": 0, "pss_kib": 0}, + } + post_process_data = { + "main_python": {"pss_kib": 410 * 1024, "uss_kib": 400 * 1024, "threads": 10}, + "xvfb": {"count": 1, "pss_kib": 72 * 1024}, + } + activation = { + "worker_elapsed_seconds": 1.25, + "pre": { + "engine": { + "working_set_bytes": 500 * 1024 * 1024, + "network_rx_bytes": 1024, + "network_tx_bytes": 512, + }, + "processes": process_data, + }, + "post": { + "engine": { + "working_set_bytes": 600 * 1024 * 1024, + "network_rx_bytes": 3072, + "network_tx_bytes": 1536, + }, + "processes": post_process_data, + }, + "marker": {"success": True}, + "validation": { + "passed": True, + "single_flight": {"passed": True}, + }, + } + sample = { + "scenario": "browser-headed", + "variant": "after", + "sample_index": 1, + "http_ready_seconds": 7.0, + "activation": activation, + "measurements": [ + { + "target_minute": 1.0, + "engine": { + "working_set_bytes": 610 * 1024 * 1024, + "network_rx_bytes": 1024, + "network_tx_bytes": 512, + }, + "processes": post_process_data, + "modules": {"count": 3000}, + } + ], + } + build = { + "campaign": "fake", + "platform": "linux/arm64", + "before_commit": "before", + "after_commit": "after", + "substrate": {"reference": "frozen"}, + } + + report = harness.build_markdown_report(build, None, [sample]) + + assert "## 场景激活" in report + assert "browser-headed" in report + assert "Single-flight" in report + assert "### `browser-headed`" in report + assert "1.25" in report diff --git a/tests/test_agent_lifecycle.py b/tests/test_agent_lifecycle.py index 90f56b064..a14dce458 100644 --- a/tests/test_agent_lifecycle.py +++ b/tests/test_agent_lifecycle.py @@ -108,13 +108,13 @@ async def test_agent_initialization_failure_does_not_stop_module_startup( monkeypatch.setattr(modules_initializer, "init_agent", agent_initializer.init_agent) for name in ( - "DisplayHelper", "DohHelper", "SitesHelper", "ResourceHelper", "ModuleManager", ): monkeypatch.setattr(modules_initializer, name, MagicMock()) + monkeypatch.setattr(modules_initializer, "init_managed_resources", MagicMock()) monkeypatch.setattr(modules_initializer, "user_auth", MagicMock()) monkeypatch.setattr(modules_initializer.EventManager, "start", MagicMock()) for name in ( diff --git a/tests/test_browser_helper.py b/tests/test_browser_helper.py index a012637e7..7e66aa6b9 100644 --- a/tests/test_browser_helper.py +++ b/tests/test_browser_helper.py @@ -1,15 +1,23 @@ from __future__ import annotations import json +import asyncio +import sys import threading from concurrent.futures import ThreadPoolExecutor +from types import ModuleType from typing import Optional -from unittest.mock import patch +from unittest.mock import AsyncMock, MagicMock, call, patch import pytest from app.agent.tools.impl.browse_webpage import BrowserAction, BrowseWebpageTool -from app.adapters.network.browser import BrowserSessionHelper, PlaywrightHelper +from app.adapters.network.browser import ( + BrowserSessionHelper, + PlaywrightHelper, + launch_browser_context, + launch_browser_context_async, +) class _FakeResponse: @@ -224,6 +232,59 @@ def test_legacy_browser_type_constructor_is_accepted(): assert source == "ok" +def test_sync_browser_facade_activates_display_only_for_headed_mode(monkeypatch): + """同步启动仅在明确有界面模式获取 host.display，参数原样交给浏览器。""" + provider = ModuleType("cloakbrowser") + launch_context = MagicMock(return_value=object()) + provider.launch_context = launch_context + monkeypatch.setitem(sys.modules, "cloakbrowser", provider) + activate = MagicMock() + monkeypatch.setattr( + "app.adapters.network.browser.acquire_managed_resource", + activate, + ) + + headless_context = launch_browser_context(headless=True, locale="zh-CN") + headed_context = launch_browser_context(headless=False, locale="zh-CN") + + assert headless_context is launch_context.return_value + assert headed_context is launch_context.return_value + activate.assert_called_once_with( + "host.display", + reason="headed_browser_launch", + retry=True, + ) + assert launch_context.call_args_list == [ + call(headless=True, locale="zh-CN"), + call(headless=False, locale="zh-CN"), + ] + + +def test_async_browser_facade_waits_for_display_before_provider(monkeypatch): + """异步有界面启动必须等待显示资源完成激活后再创建浏览器上下文。""" + events: list[str] = [] + provider = ModuleType("cloakbrowser") + + async def provider_launch(**_kwargs): + events.append("provider") + return object() + + provider.launch_context_async = provider_launch + monkeypatch.setitem(sys.modules, "cloakbrowser", provider) + + async def activate(*_args, **_kwargs): + events.append("display") + + monkeypatch.setattr( + "app.adapters.network.browser.acquire_managed_resource_async", + AsyncMock(side_effect=activate), + ) + + asyncio.run(launch_browser_context_async(headless=False, timezone="Asia/Shanghai")) + + assert events == ["display", "provider"] + + def test_browser_session_helper_blocks_private_network_by_default(): """默认应阻止 Agent 浏览器访问本机或私网地址。""" with pytest.raises(ValueError, match="默认不允许访问本机或私网地址"): diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index 5bb256d45..02fc6f74c 100644 --- a/tests/test_cache_system.py +++ b/tests/test_cache_system.py @@ -142,7 +142,7 @@ def test_init_modules_does_not_clear_package_tool_cache(monkeypatch): raise AssertionError("init_modules must not clear package tool cache directly") monkeypatch.setattr(modules_initializer, "clear_package_tool_cache", fail_if_called) - monkeypatch.setattr(modules_initializer, "DisplayHelper", lambda: None) + monkeypatch.setattr(modules_initializer, "init_managed_resources", lambda: None) monkeypatch.setattr(modules_initializer, "DohHelper", lambda: None) monkeypatch.setattr(modules_initializer, "SitesHelper", lambda: None) monkeypatch.setattr( diff --git a/tests/test_display_resource.py b/tests/test_display_resource.py new file mode 100644 index 000000000..23455543a --- /dev/null +++ b/tests/test_display_resource.py @@ -0,0 +1,99 @@ +"""虚拟显示托管资源与旧 API 的兼容测试。""" + +from __future__ import annotations + +import sys +from types import ModuleType + +from app.adapters.system.display import DisplayHelper +from app.adapters.system.display.resource import VirtualDisplayResource +from app.foundation.singleton import Singleton + + +def test_virtual_display_skips_host_process_outside_docker(monkeypatch) -> None: + """非容器环境启动资源时不得创建虚拟显示进程。""" + monkeypatch.setattr( + "app.adapters.system.display.resource.SystemUtils.is_docker", + lambda: False, + ) + resource = VirtualDisplayResource() + + resource.start() + resource.stop() + + assert resource.display is None + + +def test_virtual_display_starts_and_stops_owned_process(monkeypatch) -> None: + """容器环境只停止当前资源实际拥有的显示进程。""" + events: list[object] = [] + + class FakeDisplay: + """记录 pyvirtualdisplay 的构造与生命周期。""" + + def __init__(self, **kwargs) -> None: + events.append(("create", kwargs)) + + def start(self) -> None: + events.append("start") + + def stop(self) -> None: + events.append("stop") + + pyvirtualdisplay = ModuleType("pyvirtualdisplay") + pyvirtualdisplay.Display = FakeDisplay + monkeypatch.setitem(sys.modules, "pyvirtualdisplay", pyvirtualdisplay) + monkeypatch.setattr( + "app.adapters.system.display.resource.SystemUtils.is_docker", + lambda: True, + ) + monkeypatch.setenv("DISPLAY", ":99") + resource = VirtualDisplayResource() + + resource.start() + resource.stop() + resource.stop() + + assert events == [ + ( + "create", + { + "visible": False, + "size": (1024, 768), + "extra_args": [":99"], + }, + ), + "start", + "stop", + ] + assert resource.display is None + + +def test_display_helper_keeps_legacy_constructor_and_stop_contract(monkeypatch) -> None: + """旧构造入口显式激活 host.display，stop 只停止已配置 Runtime。""" + events: list[tuple[str, str]] = [] + singleton_key = (DisplayHelper, (), frozenset()) + previous = Singleton._instances.pop(singleton_key, None) + monkeypatch.setattr( + "app.adapters.system.display.acquire_managed_resource", + lambda capability_id, *, reason, retry: events.append( + ("activate", capability_id) + ), + ) + monkeypatch.setattr( + "app.adapters.system.display.stop_managed_resource", + lambda capability_id, *, reason: events.append(("stop", capability_id)), + ) + try: + helper = DisplayHelper() + assert DisplayHelper() is helper + helper.stop() + finally: + Singleton._instances.pop(singleton_key, None) + if previous is not None: + Singleton._instances[singleton_key] = previous + + assert events == [ + ("activate", "host.display"), + ("stop", "host.display"), + ] diff --git a/tests/test_legacy_plugin_resource_imports.py b/tests/test_legacy_plugin_resource_imports.py new file mode 100644 index 000000000..5c874f7ab --- /dev/null +++ b/tests/test_legacy_plugin_resource_imports.py @@ -0,0 +1,382 @@ +"""旧插件资源导入扫描与加载前准备合同测试。""" + +from __future__ import annotations + +import importlib +import os +from pathlib import Path +from types import ModuleType, SimpleNamespace + +import pytest + +from app.runtime.compat import resource_imports +from app.runtime.compat.resource_imports import ( + PluginResourceImportScanError, + RESOURCE_IMPORT_RULES, + scan_plugin_resource_imports, +) +from app.runtime.extensions import plugin_manager as plugin_manager_module +from app.runtime.extensions.plugin_manager import PluginManager +from app.startup import plugins_initializer + + +_HEADED_CLOAKBROWSER_ENTRYPOINTS = ( + "launch", + "launch_async", + "launch_context", + "launch_context_async", + "launch_persistent_context", + "launch_persistent_context_async", +) + + +def _write_plugin(root: Path, plugin_id: str, source: str) -> Path: + """写入一个仅用于 AST 扫描的最小插件源码目录。""" + plugin_dir = root / plugin_id.lower() + plugin_dir.mkdir(parents=True) + (plugin_dir / "__init__.py").write_text(source, encoding="utf-8") + return plugin_dir + + +@pytest.mark.parametrize( + "source", + ( + "import cloakbrowser\n", + "from cloakbrowser import *\n", + "from cloakbrowser.browser import launch_context\n", + "__import__('cloakbrowser')\n", + "import importlib\nimportlib.import_module('cloakbrowser.browser')\n", + "import importlib as loader\nloader.import_module('cloakbrowser')\n", + "from importlib import import_module as load\nload('cloakbrowser')\n", + ), +) +def test_cloakbrowser_import_shapes_require_display( + tmp_path: Path, + source: str, +) -> None: + """静态、星号、子模块及常量动态导入均准备虚拟显示。""" + plugin_dir = _write_plugin(tmp_path, "SamplePlugin", source) + + assert scan_plugin_resource_imports("SamplePlugin", plugin_dir) == ("host.display",) + + +@pytest.mark.parametrize( + "entrypoint", + _HEADED_CLOAKBROWSER_ENTRYPOINTS, +) +def test_all_headed_capable_cloakbrowser_entrypoints_require_display( + tmp_path: Path, + entrypoint: str, +) -> None: + """CloakBrowser 六类允许 headed 模式的入口共用同一资源规则。""" + plugin_dir = _write_plugin( + tmp_path, + "HeadedPlugin", + f"from cloakbrowser import {entrypoint}\n", + ) + + assert scan_plugin_resource_imports("HeadedPlugin", plugin_dir) == ("host.display",) + assert RESOURCE_IMPORT_RULES[0].headed_entrypoints == ( + _HEADED_CLOAKBROWSER_ENTRYPOINTS + ) + + +@pytest.mark.parametrize( + ("plugin_id", "source"), + ( + ("DynamicWechat", "from cloakbrowser import launch_context_async\n"), + ("ContractCheck", "from cloakbrowser import launch_context\n"), + ("InvitesSignin", "from cloakbrowser import launch_context\n"), + ( + "WeatherWidget", + "__import__('cloakbrowser')\nfrom cloakbrowser import launch_context\n", + ), + ( + "P115StrmHelper", + "from cloakbrowser import launch_context as _cloak_launch_context\n", + ), + ), +) +def test_current_direct_cloakbrowser_plugin_shapes_require_display( + tmp_path: Path, + plugin_id: str, + source: str, +) -> None: + """当前五种直接 CloakBrowser 插件导入形态均命中 host.display。""" + plugin_dir = _write_plugin(tmp_path, plugin_id, source) + + assert scan_plugin_resource_imports(plugin_id, plugin_dir) == ("host.display",) + + +def test_sdk_browser_import_does_not_require_legacy_resource(tmp_path: Path) -> None: + """宿主 SDK 浏览器门面自行按 headless 参数协调资源，不应被保守扫描。""" + plugin_dir = _write_plugin( + tmp_path, + "SdkPlugin", + "from app.sdk.browser import launch_browser_context_async\n", + ) + + assert scan_plugin_resource_imports("SdkPlugin", plugin_dir) == () + + +def test_scanner_reuses_successful_file_result( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """未变化源码在热加载扫描时复用按文件状态缓存的结果。""" + plugin_dir = _write_plugin(tmp_path, "CachedPlugin", "import cloakbrowser\n") + parse_calls = 0 + original_parse = resource_imports.ast.parse + + def count_parse(*args, **kwargs): + nonlocal parse_calls + parse_calls += 1 + return original_parse(*args, **kwargs) + + monkeypatch.setattr(resource_imports.ast, "parse", count_parse) + + assert scan_plugin_resource_imports("CachedPlugin", plugin_dir) == ("host.display",) + assert scan_plugin_resource_imports("CachedPlugin", plugin_dir) == ("host.display",) + assert parse_calls == 1 + + +def test_scanner_invalidates_cache_when_source_changes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """文件大小或修改时间变化后重新解析，不沿用旧能力集合。""" + plugin_dir = _write_plugin( + tmp_path, + "ChangedPlugin", + "from app.sdk.browser import launch_browser_context\n", + ) + source_path = plugin_dir / "__init__.py" + parse_calls = 0 + original_parse = resource_imports.ast.parse + + def count_parse(*args, **kwargs): + nonlocal parse_calls + parse_calls += 1 + return original_parse(*args, **kwargs) + + monkeypatch.setattr(resource_imports.ast, "parse", count_parse) + + assert scan_plugin_resource_imports("ChangedPlugin", plugin_dir) == () + source_path.write_text( + "from cloakbrowser.browser import launch_persistent_context_async\n", + encoding="utf-8", + ) + assert scan_plugin_resource_imports("ChangedPlugin", plugin_dir) == ( + "host.display", + ) + assert parse_calls == 2 + + +def test_scanner_invalidates_equal_size_source_with_preserved_mtime( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """等长热更新即使保留 mtime，也不能复用替换前的导入结论。""" + plugin_dir = _write_plugin(tmp_path, "ReplacedPlugin", "import cloakbrowser\n") + source_path = plugin_dir / "__init__.py" + original_stat = source_path.stat() + parse_calls = 0 + original_parse = resource_imports.ast.parse + + def count_parse(*args, **kwargs): + nonlocal parse_calls + parse_calls += 1 + return original_parse(*args, **kwargs) + + monkeypatch.setattr(resource_imports.ast, "parse", count_parse) + + assert scan_plugin_resource_imports("ReplacedPlugin", plugin_dir) == ( + "host.display", + ) + source_path.write_text("import cloakbrowsex\n", encoding="utf-8") + os.utime( + source_path, + ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns), + ) + + assert scan_plugin_resource_imports("ReplacedPlugin", plugin_dir) == () + assert parse_calls == 2 + + +def test_scanner_conservatively_prepares_resources_for_invalid_source( + tmp_path: Path, +) -> None: + """未被导入的残留语法文件不得阻断插件，但必须准备全部资源。""" + plugin_dir = _write_plugin( + tmp_path, + "BrokenPlugin", + "from app.sdk.browser import launch_browser_context\n", + ) + (plugin_dir / "unused.py").write_text( + "from cloakbrowser import (\n", + encoding="utf-8", + ) + + assert scan_plugin_resource_imports("BrokenPlugin", plugin_dir) == ("host.display",) + + +def test_scanner_conservatively_prepares_resources_for_read_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """源码读取失败时按全部资源准备，不能降级为空资源集合。""" + plugin_dir = _write_plugin( + tmp_path, + "UnreadablePlugin", + "from app.sdk.browser import launch_browser_context\n", + ) + + original_open = resource_imports.tokenize.open + + def guarded_open(path: Path): + if Path(path).parent == plugin_dir: + raise OSError("fixture read failure") + return original_open(path) + + monkeypatch.setattr(resource_imports.tokenize, "open", guarded_open) + + assert scan_plugin_resource_imports("UnreadablePlugin", plugin_dir) == ( + "host.display", + ) + + +def test_scanner_conservatively_prepares_resources_for_walk_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """目录遍历失败时准备全部资源，后续导入仍由 Python loader 判断。""" + plugin_dir = _write_plugin(tmp_path, "WalkErrorPlugin", "plugin_name = 'ok'\n") + + def fail_walk(_path: Path, _pattern: str): + raise OSError("fixture walk failure") + + monkeypatch.setattr(Path, "rglob", fail_walk) + + assert scan_plugin_resource_imports("WalkErrorPlugin", plugin_dir) == ( + "host.display", + ) + + +def test_scanner_honors_python_source_encoding_cookie(tmp_path: Path) -> None: + """合法的非 UTF-8 Python 源码按 PEP 263 声明解析。""" + plugin_dir = tmp_path / "encodedplugin" + plugin_dir.mkdir() + (plugin_dir / "__init__.py").write_bytes( + "# -*- coding: latin-1 -*-\n# café\nimport cloakbrowser\n".encode("latin-1") + ) + + assert scan_plugin_resource_imports("EncodedPlugin", plugin_dir) == ( + "host.display", + ) + + +def _fake_plugin_module(module_name: str) -> ModuleType: + """构造满足 PluginManager 类发现合同的内存模块。""" + module = ModuleType(module_name) + plugin_type = type( + module_name.rsplit(".", maxsplit=1)[-1].title(), + (), + { + "init_plugin": lambda self, _config: None, + "plugin_name": "Fixture", + }, + ) + setattr(module, plugin_type.__name__, plugin_type) + return module + + +def test_plugin_preparer_runs_before_import_in_non_debug_and_isolates_failures( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """扫描或资源失败只阻止对应插件，后续插件仍按准备后导入的顺序加载。""" + plugins_root = tmp_path / "app" / "plugins" + for plugin_id in ("scanfailed", "resourcefailed", "healthy"): + _write_plugin(plugins_root, plugin_id, "plugin_name = 'Fixture'\n") + + events: list[tuple[str, str]] = [] + + def prepare(*, plugin_id: str, plugin_dir: Path) -> None: + assert plugin_dir.name == plugin_id + events.append(("prepare", plugin_id)) + if plugin_id == "scanfailed": + raise PluginResourceImportScanError("fixture scan failure") + if plugin_id == "resourcefailed": + raise RuntimeError("fixture resource activation failure") + + def import_plugin(module_name: str) -> ModuleType: + plugin_id = module_name.rsplit(".", maxsplit=1)[-1] + assert events[-1] == ("prepare", plugin_id) + events.append(("import", plugin_id)) + return _fake_plugin_module(module_name) + + monkeypatch.setattr( + plugin_manager_module, + "settings", + SimpleNamespace(ROOT_PATH=tmp_path, DEBUG=False), + ) + monkeypatch.setattr( + plugin_manager_module, + "_legacy_plugin_import_preparer", + prepare, + ) + monkeypatch.setattr( + plugin_manager_module, + "_legacy_import_scanner", + lambda **_kwargs: None, + ) + monkeypatch.setattr(importlib, "import_module", import_plugin) + + plugins = PluginManager._load_selective_plugins( + None, + ["ScanFailed", "ResourceFailed", "Healthy"], + lambda plugin_type: hasattr(plugin_type, "init_plugin"), + ) + + assert [plugin.__name__ for plugin in plugins] == ["Healthy"] + assert ("prepare", "scanfailed") in events + assert ("prepare", "resourcefailed") in events + assert ("prepare", "healthy") in events + assert ("import", "scanfailed") not in events + assert ("import", "resourcefailed") not in events + assert ("import", "healthy") in events + + +def test_startup_preparer_activates_scanner_results_generically( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """组合根逐项激活扫描结果，并使用稳定的旧插件导入原因。""" + events: list[tuple[str, str, str]] = [] + plugin_dir = _write_plugin(tmp_path, "LegacyPlugin", "import cloakbrowser\n") + monkeypatch.setattr( + plugins_initializer, + "scan_plugin_resource_imports", + lambda plugin_id, path: ( + events.append(("scan", plugin_id, path.name)) + or ("host.display", "fixture.resource") + ), + ) + monkeypatch.setattr( + plugins_initializer, + "acquire_managed_resource", + lambda capability_id, *, reason: events.append( + ("acquire", capability_id, reason) + ), + ) + + plugins_initializer._prepare_legacy_plugin_import( + plugin_id="LegacyPlugin", + plugin_dir=plugin_dir, + ) + + assert events == [ + ("scan", "LegacyPlugin", "legacyplugin"), + ("acquire", "host.display", "legacy_plugin_import"), + ("acquire", "fixture.resource", "legacy_plugin_import"), + ] diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 6feb4d568..9e92576df 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -366,7 +366,6 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict: for name, method_name in ( ("ModuleManager", "shutdown"), ("EventManager", "stop"), - ("DisplayHelper", "stop"), ("DohHelper", "shutdown"), ("ThreadHelper", "shutdown"), ("RedisHelper", "close"), @@ -381,11 +380,24 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict: key = name.removesuffix("Helper").removesuffix("Manager").lower() dependencies[key] = getattr(instance, method_name) - for name in ("stop_message", "stop_frontend", "clear_temp"): + for name in ( + "close_browser_sessions", + "stop_message", + "stop_frontend", + "clear_temp", + ): dependency = MagicMock() monkeypatch.setattr(modules_initializer, name, dependency) dependencies[name] = dependency + stop_managed_resources = AsyncMock() + monkeypatch.setattr( + modules_initializer, + "stop_managed_resources", + stop_managed_resources, + ) + dependencies["stop_managed_resources"] = stop_managed_resources + async_redis = MagicMock() async_redis.close = AsyncMock() monkeypatch.setattr( @@ -400,6 +412,23 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict: return dependencies +def test_browser_sessions_close_before_managed_resources(monkeypatch) -> None: + """显示等宿主资源必须晚于浏览器会话释放，避免存活上下文失去依赖。""" + calls: list[str] = [] + monkeypatch.setattr(modules_initializer, "stop_agent", AsyncMock()) + dependencies = _patch_module_shutdown_dependencies(monkeypatch) + dependencies["close_browser_sessions"].side_effect = lambda: calls.append("browser") + + async def stop_resources() -> None: + calls.append("resources") + + dependencies["stop_managed_resources"].side_effect = stop_resources + + asyncio.run(modules_initializer.stop_modules()) + + assert calls == ["browser", "resources"] + + def test_shared_http_close_waits_for_real_lru_eviction(monkeypatch): """最终 HTTP 关闭必须等待真实 LRU 淘汰任务并消费其异常""" diff --git a/tests/test_managed_resources.py b/tests/test_managed_resources.py new file mode 100644 index 000000000..0ba671c09 --- /dev/null +++ b/tests/test_managed_resources.py @@ -0,0 +1,339 @@ +"""Managed Resource 与 Capability Runtime 的集成合同测试。""" + +from __future__ import annotations + +import asyncio +import subprocess +import sys +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import ModuleType +from unittest.mock import MagicMock + +import pytest + +from app.runtime.capabilities.errors import ( + CapabilityOperationError, + CapabilityRuntimeClosedError, +) +from app.runtime.capabilities.runtime import CapabilityRuntime +from app.runtime.extensions.managed_resource_adapter import ( + AsyncManagedResourceAdapter, + SyncManagedResourceAdapter, + build_managed_resource_registry, +) +from app.runtime import managed_resources as managed_resource_facade +from app.runtime.managed_resources import ( + MANAGED_RESOURCE_ASYNC_KIND, + MANAGED_RESOURCE_SYNC_KIND, + acquire_managed_resource, + acquire_managed_resource_async, + configure_managed_resource_runtime, + managed_resource_observations, + managed_resource_snapshot, + shutdown_managed_resource_runtime, +) + + +PROJECT_ROOT = Path(__file__).parents[1] + + +@pytest.fixture(autouse=True) +def isolate_managed_resource_facade(monkeypatch: pytest.MonkeyPatch) -> None: + """每个用例使用独立 Runtime，避免不可逆关闭态泄漏到后续测试。""" + monkeypatch.setattr( + managed_resource_facade, + "_managed_resource_runtime", + None, + ) + + +def _write_manifest( + root: Path, *, capability_id: str, kind: str, entrypoint: str +) -> None: + """写入一个最小 on-first-use 托管资源声明。""" + resource_dir = root / capability_id.replace(".", "_") + resource_dir.mkdir(parents=True) + (resource_dir / "capability.toml").write_text( + "\n".join( + ( + "schema_version = 1", + f'id = "{capability_id}"', + f'kind = "{kind}"', + f'entrypoint = "{entrypoint}"', + "depends_on = []", + "", + "[metadata]", + f'name = "{capability_id}"', + "", + "[activation]", + 'policy = "on_first_use"', + "watch = []", + "", + ) + ), + encoding="utf-8", + ) + + +def _runtime(root: Path) -> CapabilityRuntime: + """构造同时支持同步与异步资源的测试 Runtime。""" + registry = build_managed_resource_registry((root,)) + return CapabilityRuntime( + registry, + adapters={ + MANAGED_RESOURCE_SYNC_KIND: SyncManagedResourceAdapter(), + MANAGED_RESOURCE_ASYNC_KIND: AsyncManagedResourceAdapter(), + }, + ) + + +def test_sync_managed_resource_is_single_flight( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """并发首用只能发布一个同步资源实例。""" + module_name = "fixture_sync_managed_resource" + module = ModuleType(module_name) + + class SyncResource: + """记录同步资源的创建、启动和停止次数。""" + + instances: list["SyncResource"] = [] + + def __init__(self) -> None: + self.started = 0 + self.stopped = 0 + type(self).instances.append(self) + + def start(self) -> None: + self.started += 1 + + def stop(self) -> None: + self.stopped += 1 + + module.SyncResource = SyncResource + monkeypatch.setitem(sys.modules, module_name, module) + _write_manifest( + tmp_path, + capability_id="fixture.sync", + kind=MANAGED_RESOURCE_SYNC_KIND, + entrypoint=f"{module_name}:SyncResource", + ) + runtime = _runtime(tmp_path) + configure_managed_resource_runtime(runtime) + + barrier = threading.Barrier(8) + + def activate() -> SyncResource: + barrier.wait(timeout=2) + return acquire_managed_resource("fixture.sync", reason="test") + + with ThreadPoolExecutor(max_workers=8) as executor: + resources = list(executor.map(lambda _index: activate(), range(8))) + + assert len({id(resource) for resource in resources}) == 1 + assert len(SyncResource.instances) == 1 + assert SyncResource.instances[0].started == 1 + assert managed_resource_snapshot("fixture.sync").generation == 1 + assert [ + observation.outcome + for observation in managed_resource_observations("fixture.sync") + if observation.operation == "activate" + ] == ["started", "succeeded"] + + asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown")) + + assert SyncResource.instances[0].stopped == 1 + with pytest.raises(CapabilityRuntimeClosedError): + acquire_managed_resource("fixture.sync", reason="after_shutdown") + + +def test_async_managed_resource_uses_async_adapter( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """异步资源通过异步 Runtime 入口启动并关闭。""" + module_name = "fixture_async_managed_resource" + module = ModuleType(module_name) + + class AsyncResource: + """记录异步资源生命周期调用。""" + + instances: list["AsyncResource"] = [] + + def __init__(self) -> None: + self.events: list[str] = [] + type(self).instances.append(self) + + async def start(self) -> None: + self.events.append("start") + + async def stop(self) -> None: + self.events.append("stop") + + module.AsyncResource = AsyncResource + monkeypatch.setitem(sys.modules, module_name, module) + _write_manifest( + tmp_path, + capability_id="fixture.async", + kind=MANAGED_RESOURCE_ASYNC_KIND, + entrypoint=f"{module_name}:AsyncResource", + ) + runtime = _runtime(tmp_path) + configure_managed_resource_runtime(runtime) + + async def exercise() -> AsyncResource: + resource = await acquire_managed_resource_async( + "fixture.async", + reason="test", + ) + await shutdown_managed_resource_runtime(reason="test_shutdown") + return resource + + resource = asyncio.run(exercise()) + + assert resource.events == ["start", "stop"] + assert AsyncResource.instances == [resource] + + +def test_failed_start_is_cleaned_before_explicit_retry( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """启动失败的候选必须先清理，显式 retry 才能发布下一代资源。""" + module_name = "fixture_retry_managed_resource" + module = ModuleType(module_name) + + class RetryResource: + """首个候选启动失败，后续候选正常启动。""" + + instances: list["RetryResource"] = [] + + def __init__(self) -> None: + self.events: list[str] = [] + self.fail_start = not type(self).instances + type(self).instances.append(self) + + def start(self) -> None: + self.events.append("start") + if self.fail_start: + raise RuntimeError("start failed") + + def stop(self) -> None: + self.events.append("stop") + + module.RetryResource = RetryResource + monkeypatch.setitem(sys.modules, module_name, module) + _write_manifest( + tmp_path, + capability_id="fixture.retry", + kind=MANAGED_RESOURCE_SYNC_KIND, + entrypoint=f"{module_name}:RetryResource", + ) + configure_managed_resource_runtime(_runtime(tmp_path)) + + with pytest.raises(CapabilityOperationError, match="start failed"): + acquire_managed_resource( + "fixture.retry", + reason="first_use", + retry=False, + ) + + resource = acquire_managed_resource( + "fixture.retry", + reason="retry", + retry=True, + ) + + assert RetryResource.instances[0].events == ["start", "stop"] + assert resource is RetryResource.instances[1] + assert resource.events == ["start"] + + asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown")) + + assert resource.events == ["start", "stop"] + + +def test_shutdown_does_not_materialize_unused_resource( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """关闭未激活 Runtime 时不得构造资源或调用 start。""" + module_name = "fixture_unused_managed_resource" + module = ModuleType(module_name) + + class UnusedResource: + """任何实例化都表示关闭路径发生反向激活。""" + + def __init__(self) -> None: + raise AssertionError("unused resource must not be materialized") + + def start(self) -> None: + raise AssertionError("unused resource must not start") + + def stop(self) -> None: + raise AssertionError("unused resource must not stop") + + module.UnusedResource = UnusedResource + monkeypatch.setitem(sys.modules, module_name, module) + _write_manifest( + tmp_path, + capability_id="fixture.unused", + kind=MANAGED_RESOURCE_SYNC_KIND, + entrypoint=f"{module_name}:UnusedResource", + ) + configure_managed_resource_runtime(_runtime(tmp_path)) + + asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown")) + + +def test_startup_initializer_discovers_manifest_without_importing_resource() -> None: + """启动装配只能读取声明，不得提前导入或构造虚拟显示实现。""" + script = """ +import asyncio +import sys +from app.startup.managed_resources_initializer import ( + init_managed_resources, + stop_managed_resources, +) + +runtime = init_managed_resources() +assert runtime.get_running("host.display") is None +assert "app.adapters.system.display.resource" not in sys.modules +assert "pyvirtualdisplay" not in sys.modules +asyncio.run(stop_managed_resources()) +assert "app.adapters.system.display.resource" not in sys.modules +assert "pyvirtualdisplay" not in sys.modules +""" + result = subprocess.run( + [sys.executable, "-c", script], + cwd=PROJECT_ROOT, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + + +def test_startup_shutdown_without_init_does_not_build_registry(monkeypatch) -> None: + """未执行启动装配时，关闭入口不得通过发现声明反向初始化 Runtime。""" + from app.startup import managed_resources_initializer + + build_registry = MagicMock(side_effect=AssertionError("must not discover")) + monkeypatch.setattr( + managed_resources_initializer, + "_managed_resource_runtime", + None, + ) + monkeypatch.setattr( + managed_resources_initializer, + "build_managed_resource_registry", + build_registry, + ) + + asyncio.run(managed_resources_initializer.stop_managed_resources()) + + build_registry.assert_not_called() diff --git a/tests/test_plugin_sdk.py b/tests/test_plugin_sdk.py index b20437765..9a24f4e71 100644 --- a/tests/test_plugin_sdk.py +++ b/tests/test_plugin_sdk.py @@ -1,4 +1,8 @@ import importlib +import subprocess +import sys +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock from app.sdk.cache import Cache, cached from app.sdk.config import settings @@ -12,6 +16,9 @@ from app.sdk.utilities import StringUtils as UtilityStringUtils from app.sdk.utilities import decrypt, encrypt +PROJECT_ROOT = Path(__file__).parents[1] + + def test_sdk_exports_canonical_plugin_interfaces(): """SDK 应复用 canonical 对象，不复制实现或制造第二套单例。""" from app.domain.context import MediaInfo as CanonicalMediaInfo @@ -68,3 +75,58 @@ def test_legacy_common_crypto_aliases_round_trip(): passphrase = b"0123456789abcdef" assert legacy_decrypt(legacy_encrypt(message, passphrase), passphrase) == message + + +def test_browser_sdk_import_is_provider_free(): + """仅导入浏览器 SDK 不得加载浏览器或虚拟显示实现。""" + script = """ +import sys +import app.sdk.browser + +for name in ( + "cloakbrowser", + "pyvirtualdisplay", + "app.adapters.network.browser", + "app.adapters.system.display.resource", +): + assert name not in sys.modules, name +""" + result = subprocess.run( + [sys.executable, "-c", script], + cwd=PROJECT_ROOT, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + + +def test_browser_sdk_delegates_sync_and_async_launch(monkeypatch): + """SDK 只转发浏览器参数，不复制宿主生命周期实现。""" + from app.sdk import browser as browser_sdk + from app.adapters.network import browser as browser_adapter + + sync_context = object() + async_context = object() + sync_launch = MagicMock(return_value=sync_context) + async_launch = AsyncMock(return_value=async_context) + monkeypatch.setattr(browser_adapter, "launch_browser_context", sync_launch) + monkeypatch.setattr(browser_adapter, "launch_browser_context_async", async_launch) + + assert browser_sdk.launch_browser_context(headless=False, locale="zh-CN") is sync_context + + async def run_async(): + return await browser_sdk.launch_browser_context_async( + headless=True, + timezone="Asia/Shanghai", + ) + + import asyncio + + assert asyncio.run(run_async()) is async_context + sync_launch.assert_called_once_with(headless=False, locale="zh-CN") + async_launch.assert_awaited_once_with( + headless=True, + timezone="Asia/Shanghai", + )