From 731fa6d5bd9fcd7b5e346bf697bc22291500991f Mon Sep 17 00:00:00 2001 From: whatevertogo Date: Sat, 14 Mar 2026 22:25:51 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E7=83=AD=E9=87=8D=E8=BD=BD=E5=8A=9F=E8=83=BD=EF=BC=8C=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E6=96=87=E4=BB=B6=E5=8F=98=E6=9B=B4=E6=97=B6=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E9=87=8D=E6=96=B0=E5=8A=A0=E8=BD=BD=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 1 + CLAUDE.md | 1 + src-new/astrbot_sdk/cli.py | 293 +++++++++++++++++++++++++- src-new/astrbot_sdk/runtime/loader.py | 25 +++ tests_v4/test_testing_module.py | 90 ++++++++ 5 files changed, 403 insertions(+), 7 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 8289db732..29c3c9122 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -72,3 +72,4 @@ old文件夹是兼容旧插件的测试，旧插件全部放进old文件夹 - 2026-03-14: grouped worker / grouped env 路径不要再复制单 worker 的 compat 生命周期和 legacy runtime 绑定逻辑。优先复用 `_legacy_runtime.py` 里的 `bind_legacy_runtime_contexts()`、`run_legacy_worker_startup_hooks()`、`run_legacy_worker_shutdown_hooks()` 以及 `resolve_plugin_lifecycle_hook()`，否则很容易出现“普通 worker 测试通过，但真正的 grouped subprocess 路径在运行时 NameError/行为漂移”的回归。 - 2026-03-14: `inspect.getmembers(module, inspect.isclass)` 会按属性名排序，所以 legacy `main.py` 组件发现若要保留声明顺序，必须遍历 `module.__dict__`；只删除后面的 `.sort()` 仍然不够。 - 2026-03-14: 如果仓库正在收敛为纯 v4 SDK，删除 compat 文件前必须先移除或延迟隔离所有公开入口里的 `_legacy_*` import。`testing.py` 或 `cli.py` 里残留对 `_legacy_runtime` 的 eager import，会让 `import astrbot_sdk.testing` 和 `python -m astrbot_sdk --help` 在运行前直接失败，而仅检查 site-packages 安装态的 CLI smoke test 很容易漏掉这类回归。 +- 2026-03-14: 本地 `dev --watch` 或任何同一路径插件重复加载场景，不能只依赖 `import_string()` 的“跨插件模块根冲突”清理。即使模块仍属于当前插件目录，`sys.modules` 也会让 `load_plugin()` 复用旧代码；热重载前必须先按插件目录清理模块缓存。 diff --git a/CLAUDE.md b/CLAUDE.md index 87eb1d79d..855ef0e27 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,6 +63,7 @@ - 2026-03-14: If the repo is being simplified to a pure v4 SDK, remove or lazily isolate every `_legacy_*` import from public entrypoints before deleting the compat files. Leaving `testing.py` or `cli.py` with eager `_legacy_runtime` imports makes `import astrbot_sdk.testing` and `python -m astrbot_sdk --help` fail immediately, and install-only entrypoint tests can miss that regression. - 2026-03-14: `MemoryClient` 新增 `save_with_ttl()` / `get_many()` / `delete_many()` / `stats()` 方法，对应的 protocol schema 和 CapabilityRouter 处理器已同步添加。测试实现中 TTL 仅记录不实际过期，实际过期由后端实现。 - 2026-03-14: `SupervisorRuntime._register_plugin_capability()` 改进冲突处理：保留命名空间冲突直接跳过，非保留命名空间冲突自动添加插件名前缀（如 `plugin_name.capability_name`）解决。 +- 2026-03-14: 本地 `dev --watch`/重复加载同一路径插件时，不能只依赖 `import_string()` 的跨插件根包冲突清理。即使缓存模块仍然属于当前插件目录，`sys.modules` 也会让 `load_plugin()` 继续复用旧代码；热重载前必须先按插件目录清掉已加载模块缓存。 # 开发命令 diff --git a/src-new/astrbot_sdk/cli.py b/src-new/astrbot_sdk/cli.py index 73ac96228..58df39f51 100644 --- a/src-new/astrbot_sdk/cli.py +++ b/src-new/astrbot_sdk/cli.py @@ -8,6 +8,7 @@ import sys import typing import zipfile from collections.abc import Coroutine +from dataclasses import dataclass, field from pathlib import Path from textwrap import dedent from typing import Any @@ -38,6 +39,7 @@ BUILD_EXCLUDED_DIRS = { BUILD_EXCLUDED_FILES = { ".astrbot-worker-state.json", } +WATCH_POLL_INTERVAL_SECONDS = 0.5 class _CliPluginValidationError(RuntimeError): @@ -52,6 +54,25 @@ class _CliPluginExecutionError(RuntimeError): """CLI 侧的本地开发插件执行失败。""" +@dataclass(slots=True) +class _PluginTreeWatcher: + plugin_dir: Path + snapshot: dict[str, tuple[int, int]] = field(init=False, default_factory=dict) + + def __post_init__(self) -> None: + self.snapshot = _snapshot_watch_files(self.plugin_dir) + + def poll_changes(self) -> list[str]: + current = _snapshot_watch_files(self.plugin_dir) + changed = sorted( + path + for path in set(self.snapshot) | set(current) + if self.snapshot.get(path) != current.get(path) + ) + self.snapshot = current + return changed + + def setup_logger(verbose: bool = False) -> None: """初始化 CLI 使用的日志配置。""" logger.remove() @@ -168,16 +189,248 @@ def _render_cli_error( click.echo(f"{key}: {value}", err=True) +def _render_nonfatal_dev_error( + exc: Exception, + *, + context: dict[str, Any] | None = None, +) -> None: + exit_code, error_code, hint = _classify_cli_exception(exc) + _render_cli_error( + error_code=error_code, + message=str(exc), + hint=hint, + context=context, + ) + if exit_code == EXIT_UNEXPECTED: + logger.exception("watch 模式收到未分类异常") + + +def _iter_watch_files(plugin_dir: Path) -> typing.Iterator[Path]: + root = plugin_dir.resolve() + for path in sorted(root.rglob("*")): + if path.is_dir(): + continue + relative = path.relative_to(root) + if any(part in BUILD_EXCLUDED_DIRS for part in relative.parts[:-1]): + continue + if relative.name in BUILD_EXCLUDED_FILES: + continue + if path.suffix in {".pyc", ".pyo"}: + continue + yield path + + +def _snapshot_watch_files(plugin_dir: Path) -> dict[str, tuple[int, int]]: + root = plugin_dir.resolve() + snapshot: dict[str, tuple[int, int]] = {} + for path in _iter_watch_files(root): + try: + stat = path.stat() + except FileNotFoundError: + continue + snapshot[path.relative_to(root).as_posix()] = ( + stat.st_mtime_ns, + stat.st_size, + ) + return snapshot + + +def _format_watch_changes(changes: list[str], *, limit: int = 5) -> str: + if not changes: + return "未知文件" + preview = changes[:limit] + text = ", ".join(preview) + if len(changes) > limit: + text += f" 等 {len(changes)} 个文件" + return text + + +class _ReloadableLocalDevRunner: + def __init__( + self, + *, + plugin_dir: Path, + state: dict[str, Any], + plugin_load_error: type[Exception], + plugin_execution_error: type[Exception], + local_runtime_config, + plugin_harness, + stdout_platform_sink, + ) -> None: + self.plugin_dir = plugin_dir + self.state = state + self._plugin_load_error = plugin_load_error + self._plugin_execution_error = plugin_execution_error + self._local_runtime_config = local_runtime_config + self._plugin_harness = plugin_harness + self._stdout_platform_sink = stdout_platform_sink + self._harness = None + self._lock = asyncio.Lock() + + async def close(self) -> None: + async with self._lock: + await self._stop_harness() + + async def reload(self) -> bool: + async with self._lock: + await self._stop_harness() + harness = self._plugin_harness( + self._local_runtime_config( + plugin_dir=self.plugin_dir, + session_id=str(self.state["session_id"]), + user_id=str(self.state["user_id"]), + platform=str(self.state["platform"]), + group_id=typing.cast(str | None, self.state["group_id"]), + event_type=str(self.state["event_type"]), + ), + platform_sink=self._stdout_platform_sink(stream=sys.stdout), + ) + try: + await harness.start() + except self._plugin_load_error as exc: + _render_nonfatal_dev_error( + _CliPluginLoadError(str(exc)), + context={"plugin_dir": self.plugin_dir}, + ) + return False + except self._plugin_execution_error as exc: + _render_nonfatal_dev_error( + _CliPluginExecutionError(str(exc)), + context={"plugin_dir": self.plugin_dir}, + ) + return False + self._harness = harness + return True + + async def dispatch_text(self, text: str) -> bool: + async with self._lock: + if self._harness is None: + click.echo("当前插件未成功加载，等待下一次文件变更后重试。") + return False + try: + await self._harness.dispatch_text( + text, + session_id=str(self.state["session_id"]), + user_id=str(self.state["user_id"]), + platform=str(self.state["platform"]), + group_id=typing.cast(str | None, self.state["group_id"]), + event_type=str(self.state["event_type"]), + ) + except (self._plugin_load_error, self._plugin_execution_error) as exc: + _render_nonfatal_dev_error( + _CliPluginExecutionError(str(exc)), + context={"plugin_dir": self.plugin_dir}, + ) + return False + except Exception as exc: + _render_nonfatal_dev_error( + exc, + context={"plugin_dir": self.plugin_dir}, + ) + return False + return True + + async def _stop_harness(self) -> None: + if self._harness is None: + return + try: + await self._harness.stop() + finally: + self._harness = None + + +async def _run_local_dev_watch( + *, + runner: _ReloadableLocalDevRunner, + event_text: str | None, + interactive: bool, + watch_poll_interval: float, + max_watch_reloads: int | None = None, +) -> None: + watcher = _PluginTreeWatcher(runner.plugin_dir) + reload_count = 0 + + async def reload_and_maybe_rerun(*, announce: str | None) -> None: + if announce: + click.echo(announce) + if not await runner.reload(): + return + if event_text is not None: + await runner.dispatch_text(event_text) + + async def watch_loop(stop_event: asyncio.Event) -> None: + nonlocal reload_count + while not stop_event.is_set(): + await asyncio.sleep(watch_poll_interval) + changes = watcher.poll_changes() + if not changes: + continue + await reload_and_maybe_rerun( + announce=( + f"检测到文件变更，重新加载插件：{_format_watch_changes(changes)}" + ) + ) + reload_count += 1 + if max_watch_reloads is not None and reload_count >= max_watch_reloads: + stop_event.set() + return + + stop_event = asyncio.Event() + watch_task: asyncio.Task[None] | None = None + try: + await reload_and_maybe_rerun( + announce=( + "watch 模式已启动，监听插件目录变更。" + if event_text is not None + else "watch 模式已启动，监听插件目录变更并按需热重载。" + ) + ) + if max_watch_reloads == 0: + return + watch_task = asyncio.create_task(watch_loop(stop_event)) + if interactive: + click.echo( + "本地交互模式已启动。可用命令：/session /user /platform /group /private /event /exit" + ) + while not stop_event.is_set(): + line = await asyncio.to_thread(sys.stdin.readline) + if not line: + break + text = line.strip() + if not text: + continue + if _handle_dev_meta_command(text, runner.state): + if text in {"/exit", "/quit"}: + break + continue + await runner.dispatch_text(text) + stop_event.set() + return + await stop_event.wait() + finally: + stop_event.set() + if watch_task is not None: + watch_task.cancel() + try: + await watch_task + except asyncio.CancelledError: + pass + await runner.close() + + async def _run_local_dev( *, plugin_dir: Path, event_text: str | None, interactive: bool, + watch: bool, session_id: str, user_id: str, platform: str, group_id: str | None, event_type: str, + watch_poll_interval: float = WATCH_POLL_INTERVAL_SECONDS, + max_watch_reloads: int | None = None, ) -> None: from .testing import ( LocalRuntimeConfig, @@ -187,6 +440,32 @@ async def _run_local_dev( _PluginLoadError, ) + state = { + "session_id": session_id, + "user_id": user_id, + "platform": platform, + "group_id": group_id, + "event_type": event_type, + } + if watch: + runner = _ReloadableLocalDevRunner( + plugin_dir=plugin_dir, + state=state, + plugin_load_error=_PluginLoadError, + plugin_execution_error=_PluginExecutionError, + local_runtime_config=LocalRuntimeConfig, + plugin_harness=PluginHarness, + stdout_platform_sink=StdoutPlatformSink, + ) + await _run_local_dev_watch( + runner=runner, + event_text=event_text, + interactive=interactive, + watch_poll_interval=watch_poll_interval, + max_watch_reloads=max_watch_reloads, + ) + return + sink = StdoutPlatformSink(stream=sys.stdout) harness = PluginHarness( LocalRuntimeConfig( @@ -199,13 +478,6 @@ async def _run_local_dev( ), platform_sink=sink, ) - state = { - "session_id": session_id, - "user_id": user_id, - "platform": platform, - "group_id": group_id, - "event_type": event_type, - } try: async with harness: if interactive: @@ -565,6 +837,11 @@ def build(plugin_dir: Path, output_dir: Path | None) -> None: ) @click.option("--event-text", type=str, help="Single message text to dispatch") @click.option("--interactive", is_flag=True, help="Read follow-up messages from stdin") +@click.option( + "--watch", + is_flag=True, + help="Reload the local harness when plugin files change", +) @click.option("--session-id", default="local-session", show_default=True) @click.option("--user-id", default="local-user", show_default=True) @click.option("--platform", "platform_name", default="test", show_default=True) @@ -576,6 +853,7 @@ def dev( standalone_mode: bool, event_text: str | None, interactive: bool, + watch: bool, session_id: str, user_id: str, platform_name: str, @@ -594,6 +872,7 @@ def dev( plugin_dir=plugin_dir, event_text=event_text, interactive=interactive, + watch=watch, session_id=session_id, user_id=user_id, platform=platform_name, diff --git a/src-new/astrbot_sdk/runtime/loader.py b/src-new/astrbot_sdk/runtime/loader.py index 9a5418388..0669578fa 100644 --- a/src-new/astrbot_sdk/runtime/loader.py +++ b/src-new/astrbot_sdk/runtime/loader.py @@ -57,6 +57,7 @@ import inspect import json import os import re +import shutil import sys from dataclasses import dataclass, field from importlib import import_module @@ -540,6 +541,8 @@ def load_plugin(plugin: PluginSpec) -> LoadedPlugin: plugin_path = str(plugin.plugin_dir) if plugin_path not in sys.path: sys.path.insert(0, plugin_path) + _purge_plugin_bytecode(plugin.plugin_dir) + _purge_plugin_modules(plugin.plugin_dir) instances: list[Any] = [] handlers: list[LoadedHandler] = [] @@ -625,6 +628,28 @@ def _module_belongs_to_plugin(module: Any, plugin_dir: Path) -> bool: ) +def _purge_plugin_modules(plugin_dir: Path) -> None: + plugin_root = plugin_dir.resolve() + for module_name, module in list(sys.modules.items()): + if module is None: + continue + if _module_belongs_to_plugin(module, plugin_root): + sys.modules.pop(module_name, None) + + +def _purge_plugin_bytecode(plugin_dir: Path) -> None: + plugin_root = plugin_dir.resolve() + for path in plugin_root.rglob("*"): + try: + if path.is_dir() and path.name == "__pycache__": + shutil.rmtree(path, ignore_errors=True) + continue + if path.is_file() and path.suffix in {".pyc", ".pyo"}: + path.unlink(missing_ok=True) + except OSError: + continue + + def _purge_module_root(root_name: str) -> None: for module_name in list(sys.modules): if module_name == root_name or module_name.startswith(f"{root_name}."): diff --git a/tests_v4/test_testing_module.py b/tests_v4/test_testing_module.py index 200fd9bec..e6101613f 100644 --- a/tests_v4/test_testing_module.py +++ b/tests_v4/test_testing_module.py @@ -1,5 +1,7 @@ from __future__ import annotations +import asyncio +import io import os import subprocess import sys @@ -40,6 +42,19 @@ def test_cli_help_works_from_source_tree() -> None: assert "Usage" in process.stdout +def test_dev_help_lists_watch_option() -> None: + process = subprocess.run( + [sys.executable, "-m", "astrbot_sdk", "dev", "--help"], + capture_output=True, + text=True, + check=False, + env=_source_env(), + ) + + assert process.returncode == 0, process.stderr + assert "--watch" in process.stdout + + @pytest.mark.asyncio async def test_plugin_harness_dispatches_sample_plugin() -> None: from astrbot_sdk.testing import LocalRuntimeConfig, PluginHarness @@ -68,3 +83,78 @@ async def test_plugin_harness_supports_metadata_and_http_commands() -> None: assert any( "已注册 API，当前共 1 个" in (record.text or "") for record in api_records ) + + +def _write_watch_plugin(plugin_dir: Path, *, reply_text: str) -> None: + plugin_dir.mkdir(parents=True, exist_ok=True) + (plugin_dir / "requirements.txt").write_text("", encoding="utf-8") + (plugin_dir / "plugin.yaml").write_text( + "\n".join( + [ + "name: watch_demo", + "display_name: Watch Demo", + "author: test", + "version: 0.1.0", + "runtime:", + ' python: "3.13"', + "components:", + " - class: main:WatchDemo", + ] + ), + encoding="utf-8", + ) + (plugin_dir / "main.py").write_text( + "\n".join( + [ + "from astrbot_sdk import Context, MessageEvent, Star, on_command", + "", + "class WatchDemo(Star):", + ' @on_command("hello")', + " async def hello(self, event: MessageEvent, ctx: Context) -> None:", + f' await event.reply("{reply_text}")', + "", + ] + ), + encoding="utf-8", + ) + + +@pytest.mark.asyncio +async def test_run_local_dev_watch_reloads_on_file_change( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from astrbot_sdk.cli import _run_local_dev + + plugin_dir = tmp_path / "watch-plugin" + _write_watch_plugin(plugin_dir, reply_text="v1") + + stdout = io.StringIO() + monkeypatch.setattr(sys, "stdout", stdout) + + task = asyncio.create_task( + _run_local_dev( + plugin_dir=plugin_dir, + event_text="hello", + interactive=False, + watch=True, + session_id="local-session", + user_id="local-user", + platform="test", + group_id=None, + event_type="message", + watch_poll_interval=0.05, + max_watch_reloads=1, + ) + ) + + await asyncio.sleep(0.2) + _write_watch_plugin(plugin_dir, reply_text="v2") + + await asyncio.wait_for(task, timeout=3.0) + + output = stdout.getvalue() + assert "watch 模式已启动" in output + assert "检测到文件变更" in output + assert "[text][local-session] v1" in output + assert "[text][local-session] v2" in output