From decc95481a545f4b7b742b985d55017a1ba55ad0 Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Thu, 8 Oct 2026 14:23:06 +0200 Subject: [PATCH] =?UTF-8?q?feat(4.0-P7):=20=E6=8F=92=E4=BB=B6=20dev=5Fmode?= =?UTF-8?q?=20=E6=96=87=E4=BB=B6=E7=9B=91=E5=90=AC=EF=BC=8C=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=20PluginManager.reload?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - plugins.dev_mode 开启后每秒比较本地插件目录的文件指纹，变化时重新加载该插件，包括上次加载失败的 - PluginManager.reload(id) 重新发现并重载单个插件，忽略配置快照 --- backend/src/module/models/config.py | 4 + backend/src/module/plugin/manager.py | 80 +++++++++++++ backend/src/test/test_plugin_devmode.py | 151 ++++++++++++++++++++++++ 3 files changed, 235 insertions(+) create mode 100644 backend/src/test/test_plugin_devmode.py diff --git a/backend/src/module/models/config.py b/backend/src/module/models/config.py index 2ea89877..9e7d2725 100644 --- a/backend/src/module/models/config.py +++ b/backend/src/module/models/config.py @@ -412,6 +412,10 @@ class Plugins(BaseModel): allow_unsigned: bool = Field( default=False, description="Allow loading unsigned (local / pip) plugins" ) + dev_mode: bool = Field( + default=False, + description="Reload local plugins automatically when their files change", + ) enabled: dict[str, bool] = Field( default_factory=dict, description="Per-plugin enable switch, keyed by id" ) diff --git a/backend/src/module/plugin/manager.py b/backend/src/module/plugin/manager.py index 9539c8ec..dc10df53 100644 --- a/backend/src/module/plugin/manager.py +++ b/backend/src/module/plugin/manager.py @@ -4,6 +4,7 @@ import asyncio import inspect import json import logging +import os from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from pathlib import Path @@ -25,9 +26,27 @@ logger = logging.getLogger(__name__) PluginState = Literal["active", "disabled", "error"] DEFAULT_SETUP_TIMEOUT = 30.0 +DEFAULT_WATCH_INTERVAL = 1.0 Discover = Callable[[], tuple[list[PluginCandidate], list[DiscoveryError]]] +Fingerprint = tuple[tuple[str, int, int], ...] + + +def _fingerprint(root: Path) -> Fingerprint: + """目录下所有文件的（相对路径、修改时间、大小），用于判断插件源码是否变过。""" + stamps = [] + for dirpath, dirs, names in os.walk(root): + dirs[:] = [d for d in dirs if d != "__pycache__"] + for name in names: + path = Path(dirpath, name) + try: + stat = path.stat() + except OSError: # 编辑器保存时临时文件可能刚好消失 + continue + stamps.append((str(path.relative_to(root)), stat.st_mtime_ns, stat.st_size)) + return tuple(sorted(stamps)) + @dataclass(frozen=True) class PluginStatus: @@ -63,8 +82,12 @@ class PluginManager: discover_fn: Discover = discover, setup_timeout: float = DEFAULT_SETUP_TIMEOUT, data_root: Path = PLUGIN_DATA_ROOT, + watch_interval: float = DEFAULT_WATCH_INTERVAL, ) -> None: self._settings = settings_obj + self._watch_interval = watch_interval + self._watcher: asyncio.Task | None = None + self._stamps: dict[str, Fingerprint] = {} self._discover = discover_fn self._setup_timeout = setup_timeout self._data_root = data_root @@ -96,6 +119,7 @@ class PluginManager: for candidate in self._candidates.values(): if self._blocked_reason(candidate) is None: await self._activate(candidate) + self._sync_watcher() logger.info( "[Plugin] %d 个插件已启用，共发现 %d 个", len(self._active), @@ -104,6 +128,10 @@ class PluginManager: await self._notify_change() async def stop(self) -> None: + if self._watcher is not None: + self._watcher.cancel() + await asyncio.gather(self._watcher, return_exceptions=True) + self._watcher = None for task in list(self._pending): task.cancel() await asyncio.gather(*self._pending, return_exceptions=True) @@ -132,8 +160,60 @@ class PluginManager: if failed is not None and failed[0] == self._snapshot(plugin_id): continue await self._activate(candidate) + self._sync_watcher() await self._notify_change() + async def reload(self, plugin_id: str) -> None: + """从磁盘重新发现并重新加载一个插件（含上次加载失败的），忽略配置快照。 + + 用于 dev_mode 下源码变更，以及安装 / 升级插件之后。 + """ + async with self._lock: + await self._deactivate(plugin_id) + self._failed.pop(plugin_id, None) + self._rediscover() + self._probe_schemas() + candidate = self._candidates.get(plugin_id) + if candidate is not None and self._blocked_reason(candidate) is None: + await self._activate(candidate) + await self._notify_change() + + # ------------------------------------------------------------ dev_mode + + def _sync_watcher(self) -> None: + """按 ``plugins.dev_mode`` 启停文件监听。""" + if self._settings.plugins.dev_mode and self._watcher is None: + # 同步取基线：监听任务启动前发生的修改也能被发现 + self._stamps = {i: _fingerprint(r) for i, r in self._watched()} + self._watcher = asyncio.create_task(self._watch()) + elif not self._settings.plugins.dev_mode and self._watcher is not None: + self._watcher.cancel() + self._watcher = None + + def _watched(self) -> list[tuple[str, Path]]: + """被监听的插件：应当运行的本地插件（包括加载失败、等待修复的）。""" + return [ + (plugin_id, candidate.root) + for plugin_id, candidate in self._candidates.items() + if candidate.source == "local" + and candidate.root is not None + and self._blocked_reason(candidate) is None + ] + + async def _watch(self) -> None: + # ponytail: 轮询修改时间，插件目录很大时改用 watchfiles + while True: + await asyncio.sleep(self._watch_interval) + try: + for plugin_id, root in self._watched(): + stamp = await asyncio.to_thread(_fingerprint, root) + if self._stamps.setdefault(plugin_id, stamp) != stamp: + self._stamps[plugin_id] = stamp + logger.info("[Plugin:%s] 源码已变更，重新加载", plugin_id) + await self.reload(plugin_id) + except Exception: + logger.exception("[Plugin] dev_mode 文件监听出错") + # ------------------------------------------------------------ status def statuses(self) -> list[PluginStatus]: diff --git a/backend/src/test/test_plugin_devmode.py b/backend/src/test/test_plugin_devmode.py new file mode 100644 index 00000000..3fce2a5e --- /dev/null +++ b/backend/src/test/test_plugin_devmode.py @@ -0,0 +1,151 @@ +"""dev_mode：本地插件文件变更后自动重载。""" + +import asyncio +import shutil +from collections.abc import Callable +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from module.models.config import Plugins +from module.plugin.loader import discover +from module.plugin.manager import PluginManager +from module.plugin.registry import ExtensionPoint + +MANIFEST = """ +[plugin] +id = "dev" +name = "Dev" +version = "1.0.0" +sdk = ">=0.1,<1" +entry = "dev:DevPlugin" +""" + +SOURCE = """ +from ab_sdk import Plugin, provider + + +class DevPlugin(Plugin): + @provider("rename_strategy", id="dev") + def strategy(self): + return "{value}" +""" + + +def write_source(plugin_dir: Path, value: str) -> None: + (plugin_dir / "dev.py").write_text(SOURCE.format(value=value)) + + +@pytest.fixture +def plugin_dir(tmp_path): + path = tmp_path / "local" / "dev" + path.mkdir(parents=True) + (path / "plugin.toml").write_text(MANIFEST) + write_source(path, "v1") + return path + + +def make_manager(tmp_path: Path, **plugins_conf) -> PluginManager: + manager = PluginManager( + SimpleNamespace( + plugins=Plugins(allow_unsigned=True, enabled={"dev": True}, **plugins_conf) + ), + discover_fn=lambda: discover( + builtin_root=tmp_path / "none", + local_root=tmp_path / "local", + entry_point_group="ab-test-none", + ), + data_root=tmp_path / "data", + watch_interval=0.02, + ) + manager.registry.declare(ExtensionPoint("rename_strategy", "provider")) + return manager + + +def strategy(manager: PluginManager) -> str | None: + entry = manager.registry.providers("rename_strategy").get("dev") + return entry.factory() if entry else None + + +async def eventually(check: Callable[[], bool], timeout: float = 3.0) -> None: + async with asyncio.timeout(timeout): + while not check(): + await asyncio.sleep(0.02) + + +async def test_dev_mode_reloads_local_plugin_when_source_changes(tmp_path, plugin_dir): + manager = make_manager(tmp_path, dev_mode=True) + await manager.start() + assert strategy(manager) == "v1" + + write_source(plugin_dir, "version-2") + + await eventually(lambda: strategy(manager) == "version-2") + await manager.stop() + + +async def test_dev_mode_recovers_plugin_that_failed_to_load(tmp_path, plugin_dir): + write_source(plugin_dir, "v1") + (plugin_dir / "dev.py").write_text("def broken(:\n") + manager = make_manager(tmp_path, dev_mode=True) + await manager.start() + assert manager.statuses()[0].state == "error" + + write_source(plugin_dir, "fixed") + + await eventually(lambda: strategy(manager) == "fixed") + assert manager.statuses()[0].state == "active" + await manager.stop() + + +async def test_changes_are_ignored_without_dev_mode(tmp_path, plugin_dir): + manager = make_manager(tmp_path) + await manager.start() + + write_source(plugin_dir, "version-2") + await asyncio.sleep(0.2) + + assert strategy(manager) == "v1" + await manager.stop() + + +async def test_watcher_follows_dev_mode_setting(tmp_path, plugin_dir): + manager = make_manager(tmp_path) + await manager.start() + manager._settings.plugins.dev_mode = True + await manager.apply_settings() + + write_source(plugin_dir, "version-2") + + await eventually(lambda: strategy(manager) == "version-2") + manager._settings.plugins.dev_mode = False + await manager.apply_settings() + write_source(plugin_dir, "version-3") + await asyncio.sleep(0.2) + assert strategy(manager) == "version-2" + await manager.stop() + + +async def test_reload_picks_up_new_code_without_config_change(tmp_path, plugin_dir): + manager = make_manager(tmp_path) + await manager.start() + write_source(plugin_dir, "version-2") + + await manager.reload("dev") + + assert strategy(manager) == "version-2" + await manager.stop() + + +async def test_reload_of_removed_plugin_deactivates_it(tmp_path, plugin_dir): + manager = make_manager(tmp_path) + await manager.start() + assert strategy(manager) == "v1" + shutil.rmtree(plugin_dir) + + await manager.reload("dev") + + assert strategy(manager) is None + assert manager.statuses() == [] + await manager.stop()