feat: Enhance CLI and testing capabilities

- Added a new script entry point `astrbot-sdk` in `pyproject.toml`.
- Introduced `has_waiter` method in `SessionWaiterManager` to check for existing waiters.
- Updated `cli.py` to improve error handling and added context to error messages.
- Implemented local development support in `cli.py` with a new `dev` command for running plugins against a mock core.
- Created a new testing module `astrbot_sdk.testing` with utilities for local development and plugin testing.
- Added comprehensive tests for the new testing module and CLI commands.
- Improved compatibility and error messaging for plugin loading failures.

ARCHITECTURE.md

@@ -58,6 +58,8 @@ src-new/
 │   ├── events.py                    # MessageEvent / PlainTextResult
 │   ├── errors.py                    # AstrBotError / ErrorCodes
 │   ├── star.py                      # Star 基类与 handler 收集
+│   ├── cli.py                       # astr / astrbot-sdk CLI 入口
+│   ├── testing.py                   # 本地开发与测试 harness
 │   ├── compat.py                    # 旧顶层兼容重导出
 │   ├── _legacy_api.py               # LegacyContext / LegacyStar / CommandComponent
 │   ├── _legacy_llm.py               # legacy LLM/tool 兼容辅助
@@ -273,7 +275,32 @@ from astrbot.core.utils.session_waiter import session_waiter
 
 只有在需要兼容现有旧插件时才应继续使用这些路径；新插件应直接使用 v4 顶层入口。
 
-## 9. 测试与维护约定
+## 9. 本地开发与测试
+
+当前仓库已经提供一条受控的本地开发路径：
+
+- CLI：`astr dev --local` 与 `astrbot-sdk dev --local`
+- 稳定测试入口：`astrbot_sdk.testing`
+
+`astrbot_sdk.testing` 当前公开的稳定面包括：
+
+- `PluginHarness`
+- `LocalRuntimeConfig`
+- `MockPeer`
+- `MockCapabilityRouter`
+- `InMemoryDB`
+- `InMemoryMemory`
+- `StdoutPlatformSink`
+- `RecordedSend`
+
+设计约束：
+
+- 本地 harness 复用真实的 `load_plugin()`、`HandlerDispatcher`、`CapabilityDispatcher`、`_legacy_runtime.py` 与 `_session_waiter.py`
+- `dev --local` 使用进程内 mock core，而不是重新发明一套并行 runtime
+- 同一次 `interactive` 会话会复用同一个 dispatcher / waiter manager / in-memory db / in-memory memory
+- `astrbot_sdk.testing` 是插件测试依赖的公开 API，minor 版本内保持兼容稳定
+
+## 10. 测试与维护约定
 
 - 当前主测试目录是 `tests_v4/`，覆盖 protocol、runtime、clients、compat facade、legacy plugin integration、top-level imports 与 integration flows。
 - 文档维护规则：
@@ -281,7 +308,7 @@ from astrbot.core.utils.session_waiter import session_waiter
   - compat 支持级别变化时，同时更新本文档、`CLAUDE.md` / `AGENTS.md` 备注以及相关契约测试。
   - `refactor.md` 不再承载现状；出现冲突时，一律以本文档和代码/测试为准。
 
-## 10. 当前建议的后续演进方向
+## 11. 当前建议的后续演进方向
 
 1. 继续把 runtime 对 compat 的认知收口到 `_legacy_runtime.py`。
 2. 继续拆薄 `_legacy_api.py`，让 `LegacyContext` 更偏向 facade 和 orchestration。

pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
 
 [project.scripts]
 astr = "astrbot_sdk.cli:cli"
+astrbot-sdk = "astrbot_sdk.cli:cli"
 
 [tool.setuptools]
 package-dir = {"" = "src-new"}

src-new/astrbot/core/platform/register.py

@@ -2,9 +2,6 @@
 旧版 ``astrbot.core.platform.register`` 导入路径兼容入口。
 TODO: 目前仅保留符号以兼容导入，后续如果需要，可以在这里实现一个基于当前平台能力的注册系统适配器。
 """
-
-
-
 def register_platform_adapter(*args, **kwargs):
     raise NotImplementedError(
         "astrbot.core.platform.register_platform_adapter() 尚未在 v4 兼容层实现。"

src-new/astrbot_sdk/_session_waiter.py

@@ -57,6 +57,9 @@ class SessionWaiterManager:
         self._waiters[key] = state
         return key
 
+    def has_waiter(self, event: Any) -> bool:
+        return self.session_key(event) in self._waiters
+
     def unregister(self, key: str, state: _SessionWaitState) -> None:
         if self._waiters.get(key) is state:
             self._waiters.pop(key, None)

src-new/astrbot_sdk/cli.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import sys
+import typing
 from collections.abc import Coroutine
 from pathlib import Path
 from typing import Any
@@ -11,7 +12,22 @@ from typing import Any
 import click
 from loguru import logger
 
+from .errors import AstrBotError
 from .runtime.bootstrap import run_plugin_worker, run_supervisor, run_websocket_server
+from .testing import (
+    LocalRuntimeConfig,
+    PluginHarness,
+    StdoutPlatformSink,
+    _PluginExecutionError,
+    _PluginLoadError,
+)
+
+EXIT_OK = 0
+EXIT_UNEXPECTED = 1
+EXIT_USAGE = 2
+EXIT_PLUGIN_LOAD = 3
+EXIT_RUNTIME = 4
+EXIT_PLUGIN_EXECUTION = 5
 
 
 def setup_logger(verbose: bool = False) -> None:
@@ -30,10 +46,170 @@ def _run_async_entrypoint(
     *,
     log_message: str,
     log_level: str = "info",
+    context: dict[str, Any] | None = None,
 ) -> None:
     log_method = getattr(logger, log_level)
     log_method(log_message)
-    asyncio.run(entrypoint)
+    try:
+        asyncio.run(entrypoint)
+    except Exception as exc:
+        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("CLI 异常退出")
+        raise SystemExit(exit_code) from exc
+
+
+def _classify_cli_exception(exc: Exception) -> tuple[int, str, str]:
+    if isinstance(exc, AstrBotError):
+        return (
+            EXIT_RUNTIME,
+            exc.code,
+            exc.hint or "请检查本地 mock core 与插件调用参数",
+        )
+    if isinstance(
+        exc,
+        (_PluginLoadError, FileNotFoundError, ImportError, ModuleNotFoundError),
+    ):
+        return (
+            EXIT_PLUGIN_LOAD,
+            "plugin_load_error",
+            "请检查插件目录、plugin.yaml、requirements.txt 和导入路径",
+        )
+    if isinstance(exc, LookupError):
+        return (
+            EXIT_RUNTIME,
+            "dispatch_error",
+            "请检查 handler 或 capability 是否已正确注册",
+        )
+    if isinstance(exc, _PluginExecutionError):
+        return (
+            EXIT_PLUGIN_EXECUTION,
+            "plugin_execution_error",
+            "请检查插件生命周期、handler 或 capability 的实现",
+        )
+    return (
+        EXIT_UNEXPECTED,
+        "unexpected_error",
+        "请查看详细日志，必要时使用 --verbose 重试",
+    )
+
+
+def _render_cli_error(
+    *,
+    error_code: str,
+    message: str,
+    hint: str = "",
+    context: dict[str, Any] | None = None,
+) -> None:
+    click.echo(f"Error[{error_code}]: {message}", err=True)
+    if hint:
+        click.echo(f"Suggestion: {hint}", err=True)
+    if not context:
+        return
+    for key, value in context.items():
+        click.echo(f"{key}: {value}", err=True)
+
+
+async def _run_local_dev(
+    *,
+    plugin_dir: Path,
+    event_text: str | None,
+    interactive: bool,
+    session_id: str,
+    user_id: str,
+    platform: str,
+    group_id: str | None,
+    event_type: str,
+) -> None:
+    sink = StdoutPlatformSink(stream=sys.stdout)
+    harness = PluginHarness(
+        LocalRuntimeConfig(
+            plugin_dir=plugin_dir,
+            session_id=session_id,
+            user_id=user_id,
+            platform=platform,
+            group_id=group_id,
+            event_type=event_type,
+        ),
+        platform_sink=sink,
+    )
+    state = {
+        "session_id": session_id,
+        "user_id": user_id,
+        "platform": platform,
+        "group_id": group_id,
+        "event_type": event_type,
+    }
+    async with harness:
+        if interactive:
+            click.echo(
+                "本地交互模式已启动。可用命令：/session <id> /user <id> /platform <name> /group <id> /private /event <type> /exit"
+            )
+            while True:
+                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, state):
+                    if text in {"/exit", "/quit"}:
+                        break
+                    continue
+                await harness.dispatch_text(
+                    text,
+                    session_id=str(state["session_id"]),
+                    user_id=str(state["user_id"]),
+                    platform=str(state["platform"]),
+                    group_id=typing.cast(str | None, state["group_id"]),
+                    event_type=str(state["event_type"]),
+                )
+            return
+        assert event_text is not None
+        await harness.dispatch_text(
+            event_text,
+            session_id=session_id,
+            user_id=user_id,
+            platform=platform,
+            group_id=group_id,
+            event_type=event_type,
+        )
+
+
+def _handle_dev_meta_command(command: str, state: dict[str, Any]) -> bool:
+    if command in {"/exit", "/quit"}:
+        return True
+    if command.startswith("/session "):
+        state["session_id"] = command.split(" ", 1)[1].strip()
+        click.echo(f"切换 session_id -> {state['session_id']}")
+        return True
+    if command.startswith("/user "):
+        state["user_id"] = command.split(" ", 1)[1].strip()
+        click.echo(f"切换 user_id -> {state['user_id']}")
+        return True
+    if command.startswith("/platform "):
+        state["platform"] = command.split(" ", 1)[1].strip()
+        click.echo(f"切换 platform -> {state['platform']}")
+        return True
+    if command.startswith("/group "):
+        state["group_id"] = command.split(" ", 1)[1].strip()
+        click.echo(f"切换 group_id -> {state['group_id']}")
+        return True
+    if command == "/private":
+        state["group_id"] = None
+        click.echo("已切换为私聊上下文")
+        return True
+    if command.startswith("/event "):
+        state["event_type"] = command.split(" ", 1)[1].strip()
+        click.echo(f"切换 event_type -> {state['event_type']}")
+        return True
+    return False
 
 
 @click.group()
@@ -58,6 +234,68 @@ def run(plugins_dir: Path) -> None:
     _run_async_entrypoint(
         run_supervisor(plugins_dir=plugins_dir),
         log_message=f"启动插件主管进程，插件目录：{plugins_dir}",
+        context={"plugins_dir": plugins_dir},
+    )
+
+
+@cli.command()
+@click.option(
+    "--plugin-dir",
+    required=True,
+    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
+    help="Plugin directory to run locally",
+)
+@click.option("--local", "local_mode", is_flag=True, help="Run against local mock core")
+@click.option(
+    "--standalone",
+    "standalone_mode",
+    is_flag=True,
+    help="Alias of --local for compatibility",
+)
+@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("--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)
+@click.option("--group-id", default=None)
+@click.option("--event-type", default="message", show_default=True)
+def dev(
+    plugin_dir: Path,
+    local_mode: bool,
+    standalone_mode: bool,
+    event_text: str | None,
+    interactive: bool,
+    session_id: str,
+    user_id: str,
+    platform_name: str,
+    group_id: str | None,
+    event_type: str,
+) -> None:
+    """Run a plugin against the local mock core for development."""
+    if not (local_mode or standalone_mode):
+        raise click.BadParameter("当前 dev 只支持 --local/--standalone 模式")
+    if interactive and event_text:
+        raise click.BadParameter("--interactive 与 --event-text 不能同时使用")
+    if not interactive and not event_text:
+        raise click.BadParameter("请提供 --event-text，或改用 --interactive")
+    _run_async_entrypoint(
+        _run_local_dev(
+            plugin_dir=plugin_dir,
+            event_text=event_text,
+            interactive=interactive,
+            session_id=session_id,
+            user_id=user_id,
+            platform=platform_name,
+            group_id=group_id,
+            event_type=event_type,
+        ),
+        log_message=f"启动本地开发模式：{plugin_dir}",
+        context={
+            "plugin_dir": plugin_dir,
+            "session_id": session_id,
+            "platform": platform_name,
+            "event_type": event_type,
+        },
     )
 
 
@@ -73,6 +311,7 @@ def worker(plugin_dir: Path) -> None:
         run_plugin_worker(plugin_dir=plugin_dir),
         log_message=f"启动插件工作进程：{plugin_dir}",
         log_level="debug",
+        context={"plugin_dir": plugin_dir},
     )
 
 
@@ -83,4 +322,5 @@ def websocket(port: int) -> None:
     _run_async_entrypoint(
         run_websocket_server(port=port),
         log_message=f"启动 WebSocket 服务器，端口：{port}",
+        context={"port": port},
     )

src-new/astrbot_sdk/testing.py

@@ -0,0 +1,855 @@
+"""本地开发与插件测试辅助。
+
+`astrbot_sdk.testing` 是面向插件作者的稳定开发入口：
+
+- `PluginHarness` 负责复用现有 loader / dispatcher / compat 执行链
+- `MockCapabilityRouter` 提供进程内 mock core 能力
+- `MockPeer` 让 `Context` 客户端继续走真实的 capability 调用路径
+- `StdoutPlatformSink` / `RecordedSend` 提供可观测的发送记录
+
+这个模块刻意不暴露 runtime 内部编排数据结构，只封装本地开发/测试真正
+需要的最小稳定面。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import re
+import shlex
+import typing
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, TextIO, get_type_hints
+
+from ._legacy_runtime import (
+    bind_legacy_runtime_contexts,
+    run_legacy_worker_shutdown_hooks,
+    run_legacy_worker_startup_hooks,
+)
+from .context import CancelToken, Context as RuntimeContext
+from .errors import AstrBotError
+from .events import MessageEvent
+from .protocol.descriptors import (
+    CommandTrigger,
+    EventTrigger,
+    MessageTrigger,
+    ScheduleTrigger,
+)
+from .protocol.messages import EventMessage, InvokeMessage, PeerInfo
+from .runtime.capability_router import CapabilityRouter, StreamExecution
+from .runtime.handler_dispatcher import CapabilityDispatcher, HandlerDispatcher
+from .runtime.loader import (
+    LoadedHandler,
+    LoadedPlugin,
+    PluginSpec,
+    load_plugin,
+    load_plugin_spec,
+)
+from .star import Star
+
+
+class _PluginLoadError(RuntimeError):
+    """本地 harness 初始化阶段的已知插件加载失败。"""
+
+
+class _PluginExecutionError(RuntimeError):
+    """本地 harness 执行插件代码时的已知插件异常。"""
+
+
+@dataclass(slots=True)
+class RecordedSend:
+    """结构化发送记录，供断言和本地调试输出复用。"""
+
+    kind: str
+    message_id: str
+    session_id: str
+    text: str | None = None
+    image_url: str | None = None
+    chain: list[dict[str, Any]] | None = None
+    target: dict[str, Any] | None = None
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @property
+    def session(self) -> str:
+        return self.session_id
+
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> "RecordedSend":
+        if "text" in payload:
+            kind = "text"
+        elif "image_url" in payload:
+            kind = "image"
+        elif "chain" in payload:
+            kind = "chain"
+        else:
+            kind = "unknown"
+        return cls(
+            kind=kind,
+            message_id=str(payload.get("message_id", "")),
+            session_id=str(payload.get("session", "")),
+            text=payload.get("text") if isinstance(payload.get("text"), str) else None,
+            image_url=(
+                payload.get("image_url")
+                if isinstance(payload.get("image_url"), str)
+                else None
+            ),
+            chain=(
+                [dict(item) for item in payload.get("chain", [])]
+                if isinstance(payload.get("chain"), list)
+                else None
+            ),
+            target=(
+                dict(payload.get("target"))
+                if isinstance(payload.get("target"), dict)
+                else None
+            ),
+            raw=dict(payload),
+        )
+
+
+class StdoutPlatformSink:
+    """把 platform.* 的发送结果同时写到终端与内存记录。"""
+
+    def __init__(self, stream: TextIO | None = None) -> None:
+        self._stream = stream
+        self.records: list[RecordedSend] = []
+
+    def record(self, item: RecordedSend) -> None:
+        self.records.append(item)
+        if self._stream is None:
+            return
+        self._stream.write(self._format(item) + "\n")
+        self._stream.flush()
+
+    def clear(self) -> None:
+        self.records.clear()
+
+    def _format(self, item: RecordedSend) -> str:
+        if item.kind == "text":
+            return f"[text][{item.session_id}] {item.text or ''}"
+        if item.kind == "image":
+            return f"[image][{item.session_id}] {item.image_url or ''}"
+        if item.kind == "chain":
+            count = len(item.chain or [])
+            return f"[chain][{item.session_id}] {count} components"
+        return f"[send][{item.session_id}] {item.raw}"
+
+
+class InMemoryDB:
+    """测试友好的 KV 视图，直接绑定到 mock router 的内存存储。"""
+
+    def __init__(self, store: dict[str, Any]) -> None:
+        self._store = store
+
+    def get(self, key: str, default: Any = None) -> Any:
+        return self._store.get(key, default)
+
+    def set(self, key: str, value: Any) -> None:
+        self._store[key] = value
+
+    def delete(self, key: str) -> None:
+        self._store.pop(key, None)
+
+    def list(self, prefix: str | None = None) -> list[str]:
+        keys = sorted(self._store.keys())
+        if prefix is None:
+            return keys
+        return [key for key in keys if key.startswith(prefix)]
+
+    def get_many(self, keys: list[str]) -> list[dict[str, Any]]:
+        return [{"key": key, "value": self._store.get(key)} for key in keys]
+
+    def set_many(self, items: list[dict[str, Any]]) -> None:
+        for item in items:
+            self.set(str(item.get("key", "")), item.get("value"))
+
+
+class InMemoryMemory:
+    """测试友好的 memory 视图，保持与 mock router 同步。"""
+
+    def __init__(self, store: dict[str, dict[str, Any]]) -> None:
+        self._store = store
+
+    def get(self, key: str, default: Any = None) -> Any:
+        return self._store.get(key, default)
+
+    def save(self, key: str, value: dict[str, Any]) -> None:
+        self._store[key] = dict(value)
+
+    def delete(self, key: str) -> None:
+        self._store.pop(key, None)
+
+    def search(self, query: str) -> list[dict[str, Any]]:
+        results: list[dict[str, Any]] = []
+        for key, value in self._store.items():
+            if query in key or query in str(value):
+                results.append({"key": key, "value": value})
+        return results
+
+
+class MockCapabilityRouter(CapabilityRouter):
+    """本地 mock core，直接复用已有的内建 capability 实现。"""
+
+    def __init__(self, *, platform_sink: StdoutPlatformSink | None = None) -> None:
+        self.platform_sink = platform_sink or StdoutPlatformSink()
+        super().__init__()
+        self.db = InMemoryDB(self.db_store)
+        self.memory = InMemoryMemory(self.memory_store)
+
+    async def execute(
+        self,
+        capability: str,
+        payload: dict[str, Any],
+        *,
+        stream: bool,
+        cancel_token,
+        request_id: str,
+    ) -> dict[str, Any] | StreamExecution:
+        before = len(self.sent_messages)
+        result = await super().execute(
+            capability,
+            payload,
+            stream=stream,
+            cancel_token=cancel_token,
+            request_id=request_id,
+        )
+        self._flush_platform_records(before)
+        return result
+
+    def _flush_platform_records(self, start_index: int) -> None:
+        for payload in self.sent_messages[start_index:]:
+            self.platform_sink.record(RecordedSend.from_payload(payload))
+
+
+class MockPeer:
+    """满足 `Context`/`CapabilityProxy` 需要的最小 peer。"""
+
+    def __init__(self, router: MockCapabilityRouter) -> None:
+        self._router = router
+        self._counter = 0
+        self.remote_peer = PeerInfo(
+            name="astrbot-local-core",
+            role="core",
+            version="local",
+        )
+        self.remote_capabilities = list(router.descriptors())
+        self.remote_capability_map = {
+            item.name: item for item in self.remote_capabilities
+        }
+        self.remote_handlers: list[Any] = []
+        self.remote_provided_capabilities: list[Any] = []
+        self.remote_metadata = {"mode": "local"}
+
+    async def invoke(
+        self,
+        capability: str,
+        payload: dict[str, Any],
+        *,
+        stream: bool = False,
+        request_id: str | None = None,
+    ) -> dict[str, Any]:
+        if stream:
+            raise ValueError("stream=True 请使用 invoke_stream()")
+        return typing.cast(
+            dict[str, Any],
+            await self._router.execute(
+                capability,
+                payload,
+                stream=False,
+                cancel_token=CancelToken(),
+                request_id=request_id or self._next_id(),
+            ),
+        )
+
+    async def invoke_stream(
+        self,
+        capability: str,
+        payload: dict[str, Any],
+        *,
+        request_id: str | None = None,
+        include_completed: bool = False,
+    ):
+        request_id = request_id or self._next_id()
+        execution = typing.cast(
+            StreamExecution,
+            await self._router.execute(
+                capability,
+                payload,
+                stream=True,
+                cancel_token=CancelToken(),
+                request_id=request_id,
+            ),
+        )
+
+        async def iterator():
+            yield EventMessage(id=request_id, phase="started")
+            chunks: list[dict[str, Any]] = []
+            async for chunk in execution.iterator:
+                if execution.collect_chunks:
+                    chunks.append(chunk)
+                yield EventMessage(id=request_id, phase="delta", data=chunk)
+            output = execution.finalize(chunks)
+            if include_completed:
+                yield EventMessage(id=request_id, phase="completed", output=output)
+
+        return iterator()
+
+    def _next_id(self) -> str:
+        self._counter += 1
+        return f"local_{self._counter:04d}"
+
+
+@dataclass(slots=True)
+class LocalRuntimeConfig:
+    """本地 harness 的稳定配置对象。"""
+
+    plugin_dir: Path
+    session_id: str = "local-session"
+    user_id: str = "local-user"
+    platform: str = "test"
+    group_id: str | None = None
+    event_type: str = "message"
+
+
+class PluginHarness:
+    """本地插件消息泵。
+
+    这里复用真实的 loader / dispatcher / compat 执行链，只负责：
+    - 在同一个事件循环里装配单插件运行时
+    - 维持本地 mock core 与发送记录
+    - 把后续消息持续送入同一个 dispatcher/session_waiter 图
+    """
+
+    def __init__(
+        self,
+        config: LocalRuntimeConfig,
+        *,
+        platform_sink: StdoutPlatformSink | None = None,
+    ) -> None:
+        self.config = config
+        self.platform_sink = platform_sink or StdoutPlatformSink()
+        self.router = MockCapabilityRouter(platform_sink=self.platform_sink)
+        self.peer = MockPeer(self.router)
+        self.plugin: PluginSpec | None = None
+        self.loaded_plugin: LoadedPlugin | None = None
+        self.dispatcher: HandlerDispatcher | None = None
+        self.capability_dispatcher: CapabilityDispatcher | None = None
+        self.lifecycle_context: RuntimeContext | None = None
+        self._request_counter = 0
+        self._started = False
+
+    async def __aenter__(self) -> "PluginHarness":
+        await self.start()
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        await self.stop()
+
+    @property
+    def sent_messages(self) -> list[RecordedSend]:
+        return list(self.platform_sink.records)
+
+    def clear_sent_messages(self) -> None:
+        self.platform_sink.clear()
+
+    async def start(self) -> None:
+        if self._started:
+            return
+        try:
+            self.plugin = load_plugin_spec(self.config.plugin_dir)
+            self.loaded_plugin = load_plugin(self.plugin)
+        except Exception as exc:  # pragma: no cover - 由 CLI/集成测试覆盖
+            raise _PluginLoadError(str(exc)) from exc
+        self.dispatcher = HandlerDispatcher(
+            plugin_id=self.plugin.name,
+            peer=self.peer,
+            handlers=self.loaded_plugin.handlers,
+        )
+        self.capability_dispatcher = CapabilityDispatcher(
+            plugin_id=self.plugin.name,
+            peer=self.peer,
+            capabilities=self.loaded_plugin.capabilities,
+        )
+        self.lifecycle_context = RuntimeContext(
+            peer=self.peer,
+            plugin_id=self.plugin.name,
+        )
+        bind_legacy_runtime_contexts(
+            [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities],
+            self.lifecycle_context,
+        )
+        try:
+            await self._run_lifecycle("on_start")
+            await run_legacy_worker_startup_hooks(
+                [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities],
+                context=self.lifecycle_context,
+                metadata=dict(self.plugin.manifest_data),
+            )
+        except AstrBotError:
+            raise
+        except Exception as exc:  # pragma: no cover - 由 CLI/集成测试覆盖
+            raise _PluginExecutionError(str(exc)) from exc
+        self._started = True
+
+    async def stop(self) -> None:
+        if (
+            not self._started
+            or self.loaded_plugin is None
+            or self.lifecycle_context is None
+        ):
+            return
+        try:
+            await run_legacy_worker_shutdown_hooks(
+                [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities],
+                context=self.lifecycle_context,
+                metadata=dict(self.plugin.manifest_data),
+            )
+            await self._run_lifecycle("on_stop")
+        finally:
+            self._started = False
+
+    async def dispatch_text(
+        self,
+        text: str,
+        *,
+        session_id: str | None = None,
+        user_id: str | None = None,
+        platform: str | None = None,
+        group_id: str | None = None,
+        event_type: str | None = None,
+        request_id: str | None = None,
+    ) -> list[RecordedSend]:
+        payload = self.build_event_payload(
+            text=text,
+            session_id=session_id,
+            user_id=user_id,
+            platform=platform,
+            group_id=group_id,
+            event_type=event_type,
+            request_id=request_id,
+        )
+        return await self.dispatch_event(payload, request_id=request_id)
+
+    async def dispatch_event(
+        self,
+        event_payload: dict[str, Any],
+        *,
+        request_id: str | None = None,
+    ) -> list[RecordedSend]:
+        await self.start()
+        assert self.loaded_plugin is not None
+        assert self.dispatcher is not None
+
+        start_index = len(self.platform_sink.records)
+        if self._has_waiter_for_event(event_payload):
+            carrier = (
+                self.loaded_plugin.handlers[0] if self.loaded_plugin.handlers else None
+            )
+            if carrier is None:
+                raise AstrBotError.invalid_input(
+                    "当前没有可用于承接 session_waiter 的 handler"
+                )
+            await self._invoke_handler(
+                carrier,
+                event_payload,
+                args={},
+                request_id=request_id,
+            )
+            await self._wait_for_followup_side_effects(
+                start_index=start_index,
+                event_payload=event_payload,
+            )
+            return self.platform_sink.records[start_index:]
+
+        matches = self._match_handlers(event_payload)
+        if not matches:
+            raise AstrBotError.invalid_input("未找到匹配的 handler")
+        for loaded, args in matches:
+            await self._invoke_handler(
+                loaded,
+                event_payload,
+                args=args,
+                request_id=request_id,
+            )
+        return self.platform_sink.records[start_index:]
+
+    async def invoke_capability(
+        self,
+        capability: str,
+        payload: dict[str, Any],
+        *,
+        request_id: str | None = None,
+        stream: bool = False,
+    ) -> dict[str, Any] | StreamExecution:
+        await self.start()
+        assert self.capability_dispatcher is not None
+        message = InvokeMessage(
+            id=request_id or self._next_request_id("cap"),
+            capability=capability,
+            input=dict(payload),
+            stream=stream,
+        )
+        try:
+            return await self.capability_dispatcher.invoke(message, CancelToken())
+        except AstrBotError:
+            raise
+        except Exception as exc:  # pragma: no cover - 由 CLI/集成测试覆盖
+            raise _PluginExecutionError(str(exc)) from exc
+
+    def build_event_payload(
+        self,
+        *,
+        text: str,
+        session_id: str | None = None,
+        user_id: str | None = None,
+        platform: str | None = None,
+        group_id: str | None = None,
+        event_type: str | None = None,
+        request_id: str | None = None,
+    ) -> dict[str, Any]:
+        session_value = session_id or self.config.session_id
+        group_value = group_id if group_id is not None else self.config.group_id
+        event_type_value = event_type or self.config.event_type
+        payload = {
+            "type": event_type_value,
+            "event_type": event_type_value,
+            "text": text,
+            "session_id": session_value,
+            "user_id": user_id or self.config.user_id,
+            "platform": platform or self.config.platform,
+            "group_id": group_value,
+            "raw": {
+                "trace_id": request_id or self._next_request_id("trace"),
+                "event_type": event_type_value,
+            },
+        }
+        if group_value:
+            payload["message_type"] = "group"
+        elif payload["user_id"]:
+            payload["message_type"] = "private"
+        else:
+            payload["message_type"] = "other"
+        return payload
+
+    async def _invoke_handler(
+        self,
+        loaded: LoadedHandler,
+        event_payload: dict[str, Any],
+        *,
+        args: dict[str, Any],
+        request_id: str | None = None,
+    ) -> None:
+        assert self.dispatcher is not None
+        message = InvokeMessage(
+            id=request_id or self._next_request_id("msg"),
+            capability="handler.invoke",
+            input={
+                "handler_id": loaded.descriptor.id,
+                "event": dict(event_payload),
+                "args": dict(args),
+            },
+        )
+        try:
+            await self.dispatcher.invoke(message, CancelToken())
+        except AstrBotError:
+            raise
+        except Exception as exc:  # pragma: no cover - 由 CLI/集成测试覆盖
+            raise _PluginExecutionError(str(exc)) from exc
+
+    async def _wait_for_followup_side_effects(
+        self,
+        *,
+        start_index: int,
+        event_payload: dict[str, Any],
+    ) -> None:
+        for _ in range(20):
+            if len(self.platform_sink.records) > start_index:
+                return
+            await asyncio.sleep(0)
+            if not self._has_waiter_for_event(event_payload):
+                return
+
+    async def _run_lifecycle(self, method_name: str) -> None:
+        assert self.loaded_plugin is not None
+        assert self.lifecycle_context is not None
+
+        for instance in self.loaded_plugin.instances:
+            hook = self._resolve_lifecycle_hook(instance, method_name)
+            if hook is None:
+                continue
+            args: list[Any] = []
+            try:
+                signature = inspect.signature(hook)
+            except (TypeError, ValueError):
+                signature = None
+            if signature is not None:
+                positional_params = [
+                    parameter
+                    for parameter in signature.parameters.values()
+                    if parameter.kind
+                    in (
+                        inspect.Parameter.POSITIONAL_ONLY,
+                        inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                    )
+                ]
+                if positional_params:
+                    args.append(self.lifecycle_context)
+            result = hook(*args)
+            if inspect.isawaitable(result):
+                await result
+
+    def _match_handlers(
+        self,
+        event_payload: dict[str, Any],
+    ) -> list[tuple[LoadedHandler, dict[str, Any]]]:
+        assert self.loaded_plugin is not None
+        ranked: list[tuple[int, int, LoadedHandler, dict[str, Any]]] = []
+        for index, loaded in enumerate(self.loaded_plugin.handlers):
+            args = self._match_handler(loaded, event_payload)
+            if args is None:
+                continue
+            ranked.append((loaded.descriptor.priority, index, loaded, args))
+        ranked.sort(key=lambda item: (-item[0], item[1]))
+        return [(loaded, args) for _priority, _index, loaded, args in ranked]
+
+    def _match_handler(
+        self,
+        loaded: LoadedHandler,
+        event_payload: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        trigger = loaded.descriptor.trigger
+        if isinstance(trigger, CommandTrigger):
+            return self._match_command_trigger(loaded, trigger, event_payload)
+        if isinstance(trigger, MessageTrigger):
+            return self._match_message_trigger(loaded, trigger, event_payload)
+        if isinstance(trigger, EventTrigger):
+            current_type = str(
+                event_payload.get("event_type")
+                or event_payload.get("type")
+                or "message"
+            )
+            if current_type != trigger.event_type:
+                return None
+            return {}
+        if isinstance(trigger, ScheduleTrigger):
+            return None
+        return None
+
+    def _match_command_trigger(
+        self,
+        loaded: LoadedHandler,
+        trigger: CommandTrigger,
+        event_payload: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        if not self._passes_trigger_constraints(
+            trigger.platforms, trigger.message_types, event_payload
+        ):
+            return None
+        text = str(event_payload.get("text", "")).strip()
+        for command_name in [trigger.command, *trigger.aliases]:
+            if not command_name:
+                continue
+            match = self._match_command_name(text, command_name)
+            if match is None:
+                continue
+            return self._build_command_args(loaded.callable, match)
+        return None
+
+    def _match_message_trigger(
+        self,
+        loaded: LoadedHandler,
+        trigger: MessageTrigger,
+        event_payload: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        if not self._passes_trigger_constraints(
+            trigger.platforms, trigger.message_types, event_payload
+        ):
+            return None
+        text = str(event_payload.get("text", ""))
+        if trigger.regex:
+            match = re.search(trigger.regex, text)
+            if match is None:
+                return None
+            return self._build_regex_args(loaded.callable, match)
+        if trigger.keywords and not any(
+            keyword in text for keyword in trigger.keywords
+        ):
+            return None
+        return {}
+
+    def _passes_trigger_constraints(
+        self,
+        platforms: list[str],
+        message_types: list[str],
+        event_payload: dict[str, Any],
+    ) -> bool:
+        platform = str(event_payload.get("platform", ""))
+        if platforms and platform not in platforms:
+            return False
+        if not message_types:
+            return True
+        current_message_type = self._message_type_name(event_payload)
+        return current_message_type in message_types
+
+    def _has_waiter_for_event(self, event_payload: dict[str, Any]) -> bool:
+        assert self.dispatcher is not None
+        probe_event = MessageEvent.from_payload(
+            event_payload,
+            context=self.lifecycle_context,
+        )
+        return self.dispatcher._session_waiters.has_waiter(probe_event)
+
+    @staticmethod
+    def _message_type_name(event_payload: dict[str, Any]) -> str:
+        explicit = str(event_payload.get("message_type", "")).lower()
+        if explicit in {"group", "private", "other"}:
+            return explicit
+        if event_payload.get("group_id"):
+            return "group"
+        if event_payload.get("user_id"):
+            return "private"
+        return "other"
+
+    @staticmethod
+    def _match_command_name(text: str, command_name: str) -> str | None:
+        if text == command_name:
+            return ""
+        if text.startswith(f"{command_name} "):
+            return text[len(command_name) :].strip()
+        return None
+
+    def _build_command_args(self, handler, remainder: str) -> dict[str, Any]:
+        names = self._legacy_arg_parameter_names(handler)
+        if not names or not remainder:
+            return {}
+        if len(names) == 1:
+            return {names[0]: remainder}
+        tokens = self._split_command_remainder(remainder)
+        if not tokens:
+            return {}
+        values: dict[str, Any] = {}
+        for index, name in enumerate(names):
+            if index >= len(tokens):
+                break
+            if index == len(names) - 1:
+                values[name] = " ".join(tokens[index:])
+                break
+            values[name] = tokens[index]
+        return values
+
+    def _build_regex_args(self, handler, match: re.Match[str]) -> dict[str, Any]:
+        named = {
+            key: value for key, value in match.groupdict().items() if value is not None
+        }
+        names = [
+            name
+            for name in self._legacy_arg_parameter_names(handler)
+            if name not in named
+        ]
+        positional = [value for value in match.groups() if value is not None]
+        for index, value in enumerate(positional):
+            if index >= len(names):
+                break
+            named[names[index]] = value
+        return named
+
+    @staticmethod
+    def _split_command_remainder(remainder: str) -> list[str]:
+        try:
+            return shlex.split(remainder)
+        except ValueError:
+            return remainder.split()
+
+    @staticmethod
+    def _resolve_lifecycle_hook(instance: Any, method_name: str):
+        hook = getattr(instance, method_name, None)
+        marker = getattr(instance.__class__, "__astrbot_is_new_star__", None)
+        is_new_star = True
+        if callable(marker):
+            is_new_star = bool(marker())
+
+        if hook is not None and callable(hook):
+            bound_func = getattr(hook, "__func__", hook)
+            star_default = getattr(Star, method_name, None)
+            if star_default is None or bound_func is not star_default:
+                return hook
+
+        if not is_new_star:
+            alias = {"on_start": "initialize", "on_stop": "terminate"}.get(method_name)
+            if alias is not None:
+                legacy_hook = getattr(instance, alias, None)
+                if legacy_hook is not None and callable(legacy_hook):
+                    return legacy_hook
+
+        if hook is not None and callable(hook):
+            return hook
+        return None
+
+    def _legacy_arg_parameter_names(self, handler) -> list[str]:
+        try:
+            signature = inspect.signature(handler)
+        except (TypeError, ValueError):
+            return []
+        try:
+            type_hints = get_type_hints(handler)
+        except Exception:
+            type_hints = {}
+        names: list[str] = []
+        for parameter in signature.parameters.values():
+            if parameter.kind not in (
+                inspect.Parameter.POSITIONAL_ONLY,
+                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+            ):
+                continue
+            if self._is_injected_parameter(
+                parameter.name, type_hints.get(parameter.name)
+            ):
+                continue
+            names.append(parameter.name)
+        return names
+
+    def _is_injected_parameter(self, name: str, annotation: Any) -> bool:
+        if name in {"event", "ctx", "context"}:
+            return True
+        normalized = self._unwrap_optional(annotation)
+        if normalized is None:
+            return False
+        if normalized is RuntimeContext:
+            return True
+        if normalized is MessageEvent:
+            return True
+        if isinstance(normalized, type) and issubclass(
+            normalized, (RuntimeContext, MessageEvent)
+        ):
+            return True
+        return False
+
+    @staticmethod
+    def _unwrap_optional(annotation: Any) -> Any:
+        if annotation is None:
+            return None
+        origin = typing.get_origin(annotation)
+        if origin is typing.Union:
+            options = [
+                item for item in typing.get_args(annotation) if item is not type(None)
+            ]
+            if len(options) == 1:
+                return options[0]
+        return annotation
+
+    def _next_request_id(self, prefix: str) -> str:
+        self._request_counter += 1
+        return f"{prefix}_{self._request_counter:04d}"
+
+
+__all__ = [
+    "InMemoryDB",
+    "InMemoryMemory",
+    "LocalRuntimeConfig",
+    "MockCapabilityRouter",
+    "MockPeer",
+    "PluginHarness",
+    "RecordedSend",
+    "StdoutPlatformSink",
+]

tests_v4/test_testing_module.py

@@ -0,0 +1,132 @@
+"""Tests for the public local-dev/testing helpers."""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+import pytest
+
+import astrbot_sdk.testing as testing_module
+from astrbot_sdk.testing import (
+    LocalRuntimeConfig,
+    MockCapabilityRouter,
+    MockPeer,
+    PluginHarness,
+)
+
+
+class TestTestingModule:
+    """Tests for `astrbot_sdk.testing` exports and behavior."""
+
+    def test_public_all_matches_stable_testing_surface(self):
+        """testing.__all__ should stay aligned with the documented stable helper API."""
+        assert testing_module.__all__ == [
+            "InMemoryDB",
+            "InMemoryMemory",
+            "LocalRuntimeConfig",
+            "MockCapabilityRouter",
+            "MockPeer",
+            "PluginHarness",
+            "RecordedSend",
+            "StdoutPlatformSink",
+        ]
+
+    @pytest.mark.asyncio
+    async def test_mock_peer_stream_emits_event_messages(self):
+        """MockPeer.invoke_stream should behave like a peer-level event stream."""
+        router = MockCapabilityRouter()
+        peer = MockPeer(router)
+
+        stream = await peer.invoke_stream(
+            "llm.stream_chat",
+            {"prompt": "hi"},
+            include_completed=True,
+        )
+        phases = []
+        chunks = []
+        async for event in stream:
+            phases.append(event.phase)
+            if event.phase == "delta":
+                chunks.append(event.data["text"])
+
+        assert phases[0] == "started"
+        assert phases[-1] == "completed"
+        assert "".join(chunks) == "Echo: hi"
+
+    @pytest.mark.asyncio
+    async def test_plugin_harness_dispatches_v4_sample_plugin(self):
+        """PluginHarness should run the maintained v4 sample against the local mock core."""
+        harness = PluginHarness(
+            LocalRuntimeConfig(plugin_dir=Path("test_plugin/new")),
+        )
+
+        async with harness:
+            records = await harness.dispatch_text("hello")
+
+        assert [item.text for item in records if item.kind == "text"] == [
+            "Echo: hello",
+            "Echo: stream",
+        ]
+
+    @pytest.mark.asyncio
+    async def test_plugin_harness_can_invoke_plugin_capabilities(self):
+        """Harness should expose plugin-provided capabilities for local assertions."""
+        harness = PluginHarness(
+            LocalRuntimeConfig(plugin_dir=Path("test_plugin/new")),
+        )
+
+        async with harness:
+            result = await harness.invoke_capability("demo.echo", {"text": "abc"})
+
+        assert result == {
+            "echo": "abc",
+            "plugin_id": "astrbot_plugin_v4demo",
+        }
+
+    @pytest.mark.asyncio
+    async def test_plugin_harness_reuses_session_waiter_across_followups(
+        self,
+        tmp_path: Path,
+    ):
+        """Follow-up messages from the same session should be routed into the active waiter."""
+        plugin_dir = tmp_path / "legacy_waiter"
+        plugin_dir.mkdir()
+        (plugin_dir / "main.py").write_text(
+            """
+from astrbot.core.utils.session_waiter import SessionController, session_waiter
+from astrbot_sdk.api.components.command import CommandComponent
+from astrbot_sdk.api.event import AstrMessageEvent, filter
+from astrbot_sdk.api.message import MessageChain
+
+
+class WaiterPlugin(CommandComponent):
+    @filter.command("ask")
+    async def ask(self, event: AstrMessageEvent):
+        await event.send(MessageChain().message("请输入确认内容"))
+
+        @session_waiter(timeout=0.2)
+        async def waiter(controller: SessionController, ev: AstrMessageEvent):
+            await ev.send(MessageChain().message(f"收到:{ev.message_str}"))
+            controller.stop()
+
+        await waiter(event)
+""".strip(),
+            encoding="utf-8",
+        )
+
+        harness = PluginHarness(
+            LocalRuntimeConfig(plugin_dir=plugin_dir, platform="test"),
+        )
+
+        async with harness:
+            first = asyncio.create_task(harness.dispatch_text("ask"))
+            await asyncio.sleep(0.05)
+            follow_up = await harness.dispatch_text("确认")
+            await first
+
+        assert [item.text for item in follow_up if item.kind == "text"] == ["收到:确认"]
+        assert [item.text for item in harness.sent_messages if item.kind == "text"] == [
+            "请输入确认内容",
+            "收到:确认",
+        ]

tests_v4/test_top_level_modules.py

@@ -37,6 +37,7 @@ from astrbot_sdk.runtime.transport import (
     WebSocketServerTransport,
 )
 from astrbot_sdk.star import Star
+from astrbot_sdk.testing import _PluginLoadError
 
 TOP_LEVEL_MODULES = [
     "astrbot_sdk",
@@ -49,6 +50,7 @@ TOP_LEVEL_MODULES = [
     "astrbot_sdk.events",
     "astrbot_sdk.runtime",
     "astrbot_sdk.star",
+    "astrbot_sdk.testing",
 ]
 
 
@@ -214,6 +216,85 @@ class TestCliModule:
         entrypoint_mock.assert_called_once_with(**kwargs)
         asyncio_run_mock.assert_called_once_with(sentinel)
 
+    def test_dev_command_delegates_to_local_runtime(self):
+        """dev --local should delegate to the local harness entrypoint."""
+        runner = CliRunner()
+        sentinel = object()
+
+        with (
+            patch(
+                "astrbot_sdk.cli._run_local_dev",
+                new=Mock(return_value=sentinel),
+            ) as dev_mock,
+            patch("astrbot_sdk.cli.asyncio.run") as asyncio_run_mock,
+        ):
+            result = runner.invoke(
+                cli,
+                [
+                    "dev",
+                    "--plugin-dir",
+                    "test_plugin/new",
+                    "--local",
+                    "--event-text",
+                    "hello",
+                ],
+            )
+
+        assert result.exit_code == 0
+        dev_mock.assert_called_once_with(
+            plugin_dir=Path("test_plugin/new"),
+            event_text="hello",
+            interactive=False,
+            session_id="local-session",
+            user_id="local-user",
+            platform="test",
+            group_id=None,
+            event_type="message",
+        )
+        asyncio_run_mock.assert_called_once_with(sentinel)
+
+    def test_dev_command_requires_local_mode(self):
+        """dev should reject invocations that do not opt into local mode."""
+        runner = CliRunner()
+
+        result = runner.invoke(
+            cli,
+            [
+                "dev",
+                "--plugin-dir",
+                "test_plugin/new",
+                "--event-text",
+                "hello",
+            ],
+        )
+
+        assert result.exit_code == 2
+        assert "--local/--standalone" in result.output
+
+    def test_dev_command_maps_plugin_load_errors_to_exit_code_3(self):
+        """Known plugin load failures should render a friendly error and exit code 3."""
+        runner = CliRunner()
+
+        async def fail(*args, **kwargs):
+            raise _PluginLoadError("missing plugin")
+
+        with patch("astrbot_sdk.cli._run_local_dev", new=fail):
+            result = runner.invoke(
+                cli,
+                [
+                    "dev",
+                    "--plugin-dir",
+                    "missing-plugin",
+                    "--local",
+                    "--event-text",
+                    "hello",
+                ],
+            )
+
+        assert result.exit_code == 3
+        assert "Error[plugin_load_error]" in result.output
+        assert "Suggestion:" in result.output
+
     def test_main_module_invokes_cli_entrypoint(self):
         """Running astrbot_sdk.__main__ as a script should call cli()."""
         cli_mock = Mock()