feat(4.0-P7): 插件 dev_mode 文件监听，新增 PluginManager.reload

- plugins.dev_mode 开启后每秒比较本地插件目录的文件指纹，变化时重新加载该插件，包括上次加载失败的
- PluginManager.reload(id) 重新发现并重载单个插件，忽略配置快照

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"
     )

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]:

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()