feat: 更新运行时模块，增强 Peer 和 HandlerDispatcher 功能，优化插件加载和请求处理逻辑，添加新测试用例

AGENTS.md

@@ -12,6 +12,8 @@
 - 2026-03-13: `MemoryClient.get()` is part of the supported v4 client surface and must stay in sync with `CapabilityRouter` built-ins. The client method existed while the router forgot to register `memory.get`, which caused a real runtime gap hidden by API shape alone.
 - 2026-03-13: When checking whether a peer has finished remote initialization, avoid `getattr(mock, "remote_peer")` style probes in code that may receive `MagicMock` peers. `MagicMock` fabricates truthy child attributes, so `CapabilityProxy` should read explicit state from `peer.__dict__` or another concrete storage location instead of treating arbitrary attribute access as initialization.
 - 2026-03-13: The repository has no legacy `src/astrbot_sdk/protocol` package to migrate file-for-file. `src-new/astrbot_sdk/protocol` is a v4-native protocol layer; compare it against legacy JSON-RPC behavior in `src/astrbot_sdk/runtime/*` and the maintained migration tests, not against a nonexistent old package tree.
+- 2026-03-13: `load_plugin()` must not blindly `getattr()` every name from `dir(instance)` during handler discovery. Real plugins may expose properties or descriptors with side effects or exceptions; inspect attributes statically first, and only bind names that actually carry handler metadata.
+- 2026-03-13: In `Peer`, “remote initialized” and “transport still alive” are separate states. Waiting for initialization must fail when the connection closes first, and malformed inbound protocol messages should actively fail pending calls instead of leaving futures/streams hanging.
 
 # 开发命令
 

CLAUDE.md

@@ -12,6 +12,8 @@
 - 2026-03-13: `MemoryClient.get()` is part of the supported v4 client surface and must stay in sync with `CapabilityRouter` built-ins. The client method existed while the router forgot to register `memory.get`, which caused a real runtime gap hidden by API shape alone.
 - 2026-03-13: When checking whether a peer has finished remote initialization, avoid `getattr(mock, "remote_peer")` style probes in code that may receive `MagicMock` peers. `MagicMock` fabricates truthy child attributes, so `CapabilityProxy` should read explicit state from `peer.__dict__` or another concrete storage location instead of treating arbitrary attribute access as initialization.
 - 2026-03-13: The repository has no legacy `src/astrbot_sdk/protocol` package to migrate file-for-file. `src-new/astrbot_sdk/protocol` is a v4-native protocol layer; compare it against legacy JSON-RPC behavior in `src/astrbot_sdk/runtime/*` and the maintained migration tests, not against a nonexistent old package tree.
+- 2026-03-13: `load_plugin()` must not blindly `getattr()` every name from `dir(instance)` during handler discovery. Real plugins may expose properties or descriptors with side effects or exceptions; inspect attributes statically first, and only bind names that actually carry handler metadata.
+- 2026-03-13: In `Peer`, “remote initialized” and “transport still alive” are separate states. Waiting for initialization must fail when the connection closes first, and malformed inbound protocol messages should actively fail pending calls instead of leaving futures/streams hanging.
 
 # 开发命令
 

src-new/astrbot_sdk/runtime/__init__.py

@@ -110,6 +110,17 @@
         - 支持流式能力 (stream_handler)
         - 内置能力：llm.chat, memory.*, db.*, platform.*
         - 支持自定义能力注册
+
+`runtime` 负责把协议、传输、插件加载和生命周期管理拼成一条完整执行链：
+
+- `Transport`: 只负责字符串级别收发
+- `Peer`: 负责协议消息、请求关联、流式事件和取消
+- `CapabilityRouter`: 核心侧能力注册与路由
+- `HandlerDispatcher`: 插件侧 handler 调用适配
+- `loader` / `bootstrap`: 插件发现、Worker 启动和 Supervisor 编排
+
+设计上，legacy 兼容只出现在加载与分发边界；`Transport` 和 `Peer` 不直接携带
+旧版业务语义。
 """
 
 from .bootstrap import (

src-new/astrbot_sdk/runtime/bootstrap.py

@@ -73,6 +73,14 @@ Supervisor 管理多个 Worker 进程，Worker 运行单个插件。
 信号处理：
     - SIGTERM: 设置 stop_event，触发优雅关闭
     - SIGINT: 设置 stop_event，触发优雅关闭
+
+这层负责把 `loader`、`Peer`、`CapabilityRouter` 和 `HandlerDispatcher` 串起来：
+
+- `SupervisorRuntime`: 启动多个插件 Worker，并把所有 handler 暴露给上游 Core
+- `WorkerSession`: Supervisor 侧对单个 Worker 的会话包装
+- `PluginWorkerRuntime`: Worker 进程内的插件加载与 handler 执行
+
+当前实现会在 Worker 连接关闭时清理对应 handler，但不会自动重启或重连。
 """
 
 from __future__ import annotations
@@ -415,6 +423,13 @@ class SupervisorRuntime:
         # 从 loaded_plugins 中移除
         if plugin_name in self.loaded_plugins:
             self.loaded_plugins.remove(plugin_name)
+        stale_requests = [
+            request_id
+            for request_id, active_session in self.active_requests.items()
+            if active_session is session
+        ]
+        for request_id in stale_requests:
+            self.active_requests.pop(request_id, None)
         logger.warning("插件 {} worker 连接已关闭，已清理相关 handlers", plugin_name)
 
     async def stop(self) -> None:

src-new/astrbot_sdk/runtime/handler_dispatcher.py

@@ -1,63 +1,8 @@
-"""处理器分发模块。
+"""插件侧 handler 调度器。
 
-定义 HandlerDispatcher 类，负责将能力调用分发到具体的处理器函数。
-支持参数注入、流式执行、错误处理和生命周期回调。
-
-核心职责：
-    - 根据处理器 ID 查找处理器
-    - 构建处理器参数（支持类型注解注入）
-    - 执行处理器并处理结果
-    - 处理异步生成器流式结果
-    - 统一的错误处理
-
-参数注入优先级：
-    1. 按类型注解注入（支持 Optional[Type]）
-    2. 按参数名注入（兼容无类型注解）
-    3. 从 legacy_args 注入（命令参数等）
-
-支持的注入类型：
-    - MessageEvent: 消息事件
-    - Context: 运行时上下文
-
-与旧版对比：
-    旧版 HandlerExecutor:
-        - 从 star_handlers_registry 获取处理器
-        - 直接调用 handler(event, **args)
-        - 无参数注入支持
-        - 通过 JSON-RPC notification 发送流式结果
-        - 错误通过 JSON-RPC error 响应
-
-    新版 HandlerDispatcher:
-        - 从 LoadedHandler 映射获取处理器
-        - 支持类型注解注入 (MessageEvent, Context)
-        - 支持参数名注入 (event, ctx, context)
-        - 支持 legacy_args 注入
-        - 支持 Optional[Type] 类型
-        - 支持默认值
-        - 统一的错误处理和 on_error 回调
-
-处理器签名兼容：
-    # 旧版签名
-    def handler(event: AstrMessageEvent) -> str:
-        return "result"
-
-    # 新版签名（类型注入）
-    async def handler(event: MessageEvent, ctx: Context) -> None:
-        await event.reply("result")
-
-    # 新版签名（名字注入）
-    async def handler(event, ctx) -> None:
-        await ctx.platform.send(event.session_id, "result")
-
-    # 流式处理器
-    async def streaming_handler(event: MessageEvent):
-        yield "chunk 1"
-        yield "chunk 2"
-
-结果处理：
-    - PlainTextResult: 调用 event.reply()
-    - str: 调用 event.reply()
-    - dict with "text": 调用 event.reply(str(item["text"]))
+`HandlerDispatcher` 把运行时收到的 `handler.invoke` 请求转成真实 Python 调用。
+它的职责只包括参数注入、legacy 返回值兼容和错误回调；不负责 handler 发现或
+远端能力路由。
 """
 
 from __future__ import annotations

src-new/astrbot_sdk/runtime/loader.py

@@ -1,82 +1,18 @@
-"""插件加载模块。
+"""插件发现、环境准备和组件加载。
 
-定义插件发现、环境管理和加载的核心逻辑。
-支持新旧两种 Star 组件的兼容加载。
+`loader` 是 runtime 与插件代码之间的边界层，负责三件事：
 
-核心概念：
-    PluginSpec: 插件规范，描述插件的基本信息
-    PluginDiscoveryResult: 插件发现结果，包含成功和跳过的插件
-    PluginEnvironmentManager: 插件虚拟环境管理器
-    LoadedHandler: 加载后的处理器，包含描述符和可调用对象
-    LoadedPlugin: 加载后的插件，包含处理器和实例
+- 从 `plugin.yaml` 解析出可运行的 `PluginSpec`
+- 用 `uv` 为插件准备独立环境
+- 把组件实例和 handler 元数据整理成 `LoadedPlugin`
 
-插件发现流程：
-    1. 扫描 plugins_dir 下的子目录
-    2. 检查 plugin.yaml 和 requirements.txt
-    3. 解析 manifest_data 获取插件信息
-    4. 验证必要字段（name, components, runtime.python）
-    5. 返回 PluginDiscoveryResult
-
-环境管理流程：
-    1. 检查 .venv 目录是否存在
-    2. 检查 Python 版本是否匹配
-    3. 检查指纹是否变化（requirements 内容）
-    4. 必要时重建虚拟环境
-    5. 使用 uv 安装依赖
-
-插件加载流程：
-    1. 将插件目录添加到 sys.path
-    2. 遍历 components 列表
-    3. 动态导入组件类
-    4. 判断是否为新版 Star
-    5. 创建实例（新版直接实例化，旧版传入 legacy_context）
-    6. 扫描处理器方法
-    7. 构建 HandlerDescriptor
-
-新旧 Star 组件兼容：
-    新版 Star:
-        - 继承自 Star 基类
-        - __astrbot_is_new_star__ 返回 True
-        - 无参构造函数
-        - 通过 @handler 装饰器注册处理器
-
-    旧版 Star:
-        - 不继承或 __astrbot_is_new_star__ 返回 False
-        - 需要 legacy_context 参数
-        - 通过 @xxx_handler 装饰器注册处理器
-        - 使用 extras_configs 传递配置
-
-与旧版对比：
-    旧版 StarManager:
-        - 通过 plugin.yaml 发现插件
-        - 动态导入组件类并实例化
-        - 注册到 star_handlers_registry
-        - 使用 functools.partial 绑定实例
-        - 无环境管理
-        - 无指纹缓存
-
-    新版 loader.py:
-        - PluginSpec 描述插件规范
-        - PluginEnvironmentManager 管理虚拟环境
-        - load_plugin() 加载并解析组件
-        - LoadedHandler 封装处理器和描述符
-        - 支持新旧 Star 组件兼容
-        - 支持环境指纹缓存
-
-plugin.yaml 格式：
-    name: my_plugin
-    author: author_name
-    desc: Plugin description
-    version: 1.0.0
-    runtime:
-        python: "3.11"
-    components:
-        - class: my_plugin.main:MyComponent
+legacy 兼容也集中放在这里，尤其是“同一插件共享一个 `LegacyContext`”这一旧语义。
 """
 
 from __future__ import annotations
 
 import json
+import inspect
 import os
 import re
 import shutil
@@ -158,6 +94,25 @@ def _iter_handler_names(instance: Any) -> list[str]:
     return list(dir(instance))
 
 
+def _resolve_handler_candidate(instance: Any, name: str) -> tuple[Any, Any] | None:
+    """解析 handler 名称，避免在扫描阶段触发无关 descriptor 副作用。"""
+    try:
+        raw = inspect.getattr_static(instance, name)
+    except AttributeError:
+        return None
+
+    candidates = [raw]
+    wrapped = getattr(raw, "__func__", None)
+    if wrapped is not None:
+        candidates.append(wrapped)
+
+    for candidate in candidates:
+        meta = get_handler_meta(candidate)
+        if meta is not None and meta.trigger is not None:
+            return getattr(instance, name), meta
+    return None
+
+
 def load_plugin_spec(plugin_dir: Path) -> PluginSpec:
     plugin_dir = plugin_dir.resolve()
     manifest_path = plugin_dir / "plugin.yaml"
@@ -387,11 +342,10 @@ def load_plugin(plugin: PluginSpec) -> LoadedPlugin:
                     setattr(instance, "context", legacy_context)
         instances.append(instance)
         for name in _iter_handler_names(instance):
-            bound = getattr(instance, name)
-            func = getattr(bound, "__func__", bound)
-            meta = get_handler_meta(func)
-            if meta is None or meta.trigger is None:
+            resolved = _resolve_handler_candidate(instance, name)
+            if resolved is None:
                 continue
+            bound, meta = resolved
             handler_id = f"{plugin.name}:{instance.__class__.__module__}.{instance.__class__.__name__}.{name}"
             handlers.append(
                 LoadedHandler(

src-new/astrbot_sdk/runtime/peer.py

@@ -1,73 +1,15 @@
-"""协议对等端模块。
+"""运行时协议对等端。
 
-定义 Peer 类，封装双向传输通道上的消息收发、初始化握手、能力调用、
-流式事件转发与取消处理。这里的 peer 指"通信对端/本端"这一网络协议概念，
-而不是业务上的用户、群聊或会话对象。
+`Peer` 把 `Transport` 和 v4 协议消息模型接起来，负责：
 
-核心职责：
-    - 消息序列化/反序列化
-    - 初始化握手协议
-    - 能力调用（同步/流式）
-    - 取消处理
-    - 连接生命周期管理
-消息处理：
-    入站:
-        ResultMessage -> 唤醒等待的 Future
-        EventMessage -> 投递到流式队列
-        InitializeMessage -> 调用 initialize_handler
-        InvokeMessage -> 创建任务调用 invoke_handler
-        CancelMessage -> 取消对应的任务
+- 握手与远端元数据缓存
+- 请求 ID 关联
+- 非流式 / 流式调用分发
+- 取消传播
+- 连接异常时的统一收口
 
-    出站:
-        initialize() -> InitializeMessage
-        invoke() -> InvokeMessage(stream=False)
-        invoke_stream() -> InvokeMessage(stream=True)
-        cancel() -> CancelMessage
-    
-与旧版对比：
-    旧版 JSON-RPC:
-        - 分离的 JSONRPCClient 和 JSONRPCServer
-        - 通过 method 字段区分操作类型
-        - 使用 JSONRPCRequest/Response 消息类型
-        - 流式通过独立的 notification 实现
-        - 无统一的取消机制
-
-    新版 Peer:
-        - 统一的 Peer 抽象，既是客户端也是服务端
-        - 通过 type 字段区分消息类型
-        - 使用 InitializeMessage/InvokeMessage/EventMessage 等
-        - 流式通过 EventMessage(phase=delta) 实现
-        - 统一的 CancelMessage 取消机制
-
-使用示例：
-    # 作为客户端发起调用
-    peer = Peer(transport=transport, peer_info=PeerInfo(...))
-    await peer.start()
-    output = await peer.initialize(handlers)
-    result = await peer.invoke("llm.chat", {"prompt": "hello"})
-
-    # 作为服务端处理调用
-    peer.set_invoke_handler(my_handler)
-    await peer.start()
-
-消息处理流程：
-    入站消息:
-        ResultMessage -> 唤醒等待的 Future
-        EventMessage -> 投递到流式队列
-        InitializeMessage -> 调用 _initialize_handler
-        InvokeMessage -> 创建任务调用 _invoke_handler
-        CancelMessage -> 取消对应的任务
-
-    出站消息:
-        initialize() -> InitializeMessage
-        invoke() -> InvokeMessage(stream=False)
-        invoke_stream() -> InvokeMessage(stream=True)
-        cancel() -> CancelMessage
-
-取消机制：
-    - CancelToken 用于检查取消状态
-    - 入站任务在收到 CancelMessage 时被取消
-    - 早到取消：在任务执行前检查 cancel_token，避免竞态条件
+它本身不做业务路由，真正的执行逻辑交给 `CapabilityRouter` 或
+`HandlerDispatcher`。
 """
 
 from __future__ import annotations
@@ -158,6 +100,8 @@ class Peer:
     async def start(self) -> None:
         """启动传输层并将原始入站消息绑定到当前 `Peer`。"""
         self._closed.clear()
+        self._unusable = False
+        self._remote_initialized.clear()
         self.transport.set_message_handler(self._handle_raw_message)
         await self.transport.start()
 
@@ -194,10 +138,27 @@ class Peer:
         Args:
             timeout: 等待秒数。传入 `None` 表示无限等待。
         """
-        if timeout is None:
-            await self._remote_initialized.wait()
-            return
-        await asyncio.wait_for(self._remote_initialized.wait(), timeout=timeout)
+        init_waiter = asyncio.create_task(self._remote_initialized.wait())
+        closed_waiter = asyncio.create_task(self.wait_closed())
+        try:
+            done, pending = await asyncio.wait(
+                {init_waiter, closed_waiter},
+                timeout=timeout,
+                return_when=asyncio.FIRST_COMPLETED,
+            )
+            if not done:
+                raise TimeoutError()
+            if init_waiter in done:
+                return
+            raise AstrBotError.protocol_error("连接在初始化完成前关闭")
+        finally:
+            for task in (init_waiter, closed_waiter):
+                if not task.done():
+                    task.cancel()
+                    try:
+                        await task
+                    except asyncio.CancelledError:
+                        pass
 
     async def initialize(
         self,
@@ -354,30 +315,38 @@ class Peer:
 
     async def _handle_raw_message(self, payload: str) -> None:
         """解析原始消息并分发到对应的消息处理分支。"""
-        message = parse_message(payload)
-        if isinstance(message, ResultMessage):
-            await self._handle_result(message)
-            return
-        if isinstance(message, EventMessage):
-            await self._handle_event(message)
-            return
-        if isinstance(message, InitializeMessage):
-            await self._handle_initialize(message)
-            return
-        if isinstance(message, InvokeMessage):
-            token = CancelToken()
-            started = asyncio.Event()
-            task = asyncio.create_task(self._handle_invoke(message, token, started))
-            self._inbound_tasks[message.id] = (task, token, started)
-            task.add_done_callback(
-                lambda _task, request_id=message.id: self._inbound_tasks.pop(
-                    request_id, None
+        try:
+            message = parse_message(payload)
+            if isinstance(message, ResultMessage):
+                await self._handle_result(message)
+                return
+            if isinstance(message, EventMessage):
+                await self._handle_event(message)
+                return
+            if isinstance(message, InitializeMessage):
+                await self._handle_initialize(message)
+                return
+            if isinstance(message, InvokeMessage):
+                token = CancelToken()
+                started = asyncio.Event()
+                task = asyncio.create_task(self._handle_invoke(message, token, started))
+                self._inbound_tasks[message.id] = (task, token, started)
+                task.add_done_callback(
+                    lambda _task, request_id=message.id: self._inbound_tasks.pop(
+                        request_id, None
+                    )
                 )
-            )
-            return
-        if isinstance(message, CancelMessage):
-            await self._handle_cancel(message)
-            return
+                return
+            if isinstance(message, CancelMessage):
+                await self._handle_cancel(message)
+                return
+        except Exception as exc:
+            if isinstance(exc, AstrBotError):
+                error = exc
+            else:
+                error = AstrBotError.protocol_error(f"协议消息处理失败: {exc}")
+            await self._fail_connection(error)
+            raise error from exc
 
     async def _handle_initialize(self, message: InitializeMessage) -> None:
         """处理远端发起的初始化握手并返回握手结果。"""
@@ -543,6 +512,29 @@ class Peer:
         error = AstrBotError.cancelled()
         await self._send_error_result(message, error)
 
+    async def _fail_connection(self, error: AstrBotError) -> None:
+        """把连接标记为不可用，并让所有等待中的调用尽快失败。"""
+        if self._unusable:
+            return
+        self._unusable = True
+        self._remote_initialized.set()
+
+        for future in list(self._pending_results.values()):
+            if not future.done():
+                future.set_exception(error)
+        self._pending_results.clear()
+
+        for queue in list(self._pending_streams.values()):
+            await queue.put(error)
+        self._pending_streams.clear()
+
+        for task, token, _started in list(self._inbound_tasks.values()):
+            token.cancel()
+            task.cancel()
+        self._inbound_tasks.clear()
+
+        asyncio.create_task(self.stop())
+
     async def _send(self, message) -> None:
         """序列化协议消息并通过底层传输发送出去。"""
         await self.transport.send(message.model_dump_json(exclude_none=True))

src-new/astrbot_sdk/runtime/transport.py

@@ -60,6 +60,15 @@
     await transport.start()
     await transport.send(json_string)
     await transport.stop()
+
+`Transport` 只处理“字符串发出去 / 字符串收进来”这件事，不做协议解析，也不关心
+能力、handler 或 legacy 兼容。当前实现包括：
+
+- `StdioTransport`: 子进程或文件对象上的按行文本传输
+- `WebSocketServerTransport`: 单连接 WebSocket 服务端
+- `WebSocketClientTransport`: WebSocket 客户端
+
+自动重连、消息重放等策略不在这里实现，统一留给更上层编排。
 """
 
 from __future__ import annotations
@@ -83,6 +92,7 @@ class Transport(ABC):
         self._closed = asyncio.Event()
 
     def set_message_handler(self, handler: MessageHandler) -> None:
+        """注册收到原始字符串消息后的回调。"""
         self._handler = handler
 
     @abstractmethod
@@ -98,9 +108,11 @@ class Transport(ABC):
         raise NotImplementedError
 
     async def wait_closed(self) -> None:
+        """等待传输层进入关闭状态。"""
         await self._closed.wait()
 
     async def _dispatch(self, payload: str) -> None:
+        """把收到的原始载荷转交给上层处理器。"""
         if self._handler is not None:
             await self._handler(payload)
 

tests_v4/test_bootstrap.py

@@ -422,6 +422,34 @@ class TestSupervisorRuntimeMethods:
             assert "test.handler" not in runtime._handler_sources
             assert "test_plugin" not in runtime.loaded_plugins
 
+    @pytest.mark.asyncio
+    async def test_handle_worker_closed_removes_active_requests(self):
+        """_handle_worker_closed should drop in-flight requests owned by the worker."""
+        transport = MemoryTransport()
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            runtime = SupervisorRuntime(
+                transport=transport,
+                plugins_dir=Path(temp_dir),
+            )
+
+            mock_session = MagicMock()
+            mock_session.handlers = [
+                HandlerDescriptor(
+                    id="test.handler", trigger=CommandTrigger(command="test")
+                )
+            ]
+
+            runtime.worker_sessions["test_plugin"] = mock_session
+            runtime.handler_to_worker["test.handler"] = mock_session
+            runtime._handler_sources["test.handler"] = "test_plugin"
+            runtime.loaded_plugins.append("test_plugin")
+            runtime.active_requests["req-1"] = mock_session
+
+            runtime._handle_worker_closed("test_plugin")
+
+            assert "req-1" not in runtime.active_requests
+
     @pytest.mark.asyncio
     async def test_handle_worker_closed_unknown_plugin(self):
         """_handle_worker_closed should handle unknown plugin."""

tests_v4/test_loader.py

@@ -678,6 +678,60 @@ class TestLoadPlugin:
                 if str(plugin_dir) in sys.path:
                     sys.path.remove(str(plugin_dir))
 
+    def test_ignores_non_handler_descriptors_without_triggering_properties(self):
+        """load_plugin should not access unrelated properties during handler discovery."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugin_dir = Path(temp_dir)
+            manifest_path = plugin_dir / "plugin.yaml"
+            requirements_path = plugin_dir / "requirements.txt"
+
+            module_dir = plugin_dir / "mymodule"
+            module_dir.mkdir()
+            (module_dir / "__init__.py").write_text("", encoding="utf-8")
+            (module_dir / "component.py").write_text(
+                textwrap.dedent(
+                    """\
+                    from astrbot_sdk import Star, on_command
+
+
+                    class MyComponent(Star):
+                        @property
+                        def explode(self):
+                            raise RuntimeError("property should not be touched")
+
+                        @on_command("hello")
+                        async def hello_handler(self):
+                            pass
+                    """
+                ),
+                encoding="utf-8",
+            )
+
+            manifest_path.write_text(
+                yaml.dump(
+                    {
+                        "name": "safe_loader_plugin",
+                        "runtime": {"python": "3.12"},
+                        "components": [{"class": "mymodule.component:MyComponent"}],
+                    }
+                ),
+                encoding="utf-8",
+            )
+            requirements_path.write_text("", encoding="utf-8")
+
+            spec = load_plugin_spec(plugin_dir)
+
+            if str(plugin_dir) not in sys.path:
+                sys.path.insert(0, str(plugin_dir))
+
+            try:
+                loaded = load_plugin(spec)
+                assert len(loaded.instances) == 1
+                assert [handler.descriptor.id for handler in loaded.handlers]
+            finally:
+                if str(plugin_dir) in sys.path:
+                    sys.path.remove(str(plugin_dir))
+
     @pytest.mark.asyncio
     async def test_load_plugin_shares_legacy_context_between_components(self):
         """Legacy components in one plugin should share the same LegacyContext."""

tests_v4/test_peer.py

@@ -115,6 +115,30 @@ class PeerRuntimeTest(unittest.IsolatedAsyncioTestCase):
         await plugin.stop()
         await self.left.stop()
 
+    async def test_invalid_inbound_message_fails_pending_calls(self) -> None:
+        plugin = Peer(
+            transport=self.right,
+            peer_info=PeerInfo(name="plugin", role="plugin", version="v4"),
+        )
+        await self.left.start()
+        await plugin.start()
+
+        task = asyncio.create_task(
+            plugin.invoke("llm.chat", {"prompt": "bad"}, request_id="req-invalid")
+        )
+        await asyncio.sleep(0)
+
+        with self.assertRaises(AstrBotError) as raised_send:
+            await self.left.send("[]")
+        self.assertEqual(raised_send.exception.code, "protocol_error")
+
+        with self.assertRaises(AstrBotError) as raised_task:
+            await task
+        self.assertEqual(raised_task.exception.code, "protocol_error")
+
+        await asyncio.wait_for(plugin.wait_closed(), timeout=1.0)
+        await self.left.stop()
+
     async def test_cancel_waits_for_failed_terminal_event(self) -> None:
         descriptor = CapabilityDescriptor(
             name="slow.stream",
@@ -248,6 +272,24 @@ class PeerRuntimeTest(unittest.IsolatedAsyncioTestCase):
         self.assertTrue(core._closed)
         self.assertTrue(plugin._closed)
 
+    async def test_wait_until_remote_initialized_raises_if_connection_closes_first(
+        self,
+    ) -> None:
+        plugin = Peer(
+            transport=self.right,
+            peer_info=PeerInfo(name="plugin", role="plugin", version="v4"),
+        )
+
+        await self.left.start()
+        await plugin.start()
+        await plugin.stop()
+
+        with self.assertRaises(AstrBotError) as raised:
+            await plugin.wait_until_remote_initialized(timeout=None)
+        self.assertEqual(raised.exception.code, "protocol_error")
+
+        await self.left.stop()
+
 
 class CapabilityRouterContractTest(unittest.TestCase):
     def test_capability_names_must_match_namespace_method_format(self) -> None: