Consolidate controlled legacy facade compatibility

AGENTS.md

@@ -27,6 +27,7 @@
 - 2026-03-13: On Windows, `.gitignore` matching is case-insensitive. A broad entry like `astrBot/` will also ignore `src-new/astrbot/...`, which can silently hide real compat alias packages from `git status`. Keep that ignore anchored as `/astrBot/` if it is only meant for a root-level scratch checkout.
 - 2026-03-13: The reference checkout under `astrBot/astrbot/api` exposes a broader old plugin-facing package surface than the current `src-new/astrbot` alias package. When improving package-name compatibility, compare against those public `api/*` modules as a set instead of only patching the one import path hit by the latest migrated plugin.
 - 2026-03-13: Some real legacy plugins call `asyncio.create_task()` during object construction. Calling `load_plugin()` outside a running event loop can therefore produce false-negative compat failures even though the real worker path is fine. External compat smoke tests should load plugins under an active loop, and "real compatibility" claims should preferably exercise `SupervisorRuntime` + worker + a real handler invocation instead of import-only checks.
+- 2026-03-13: Legacy package-name compatibility now has an explicit contract: keep `src-new/astrbot` as a controlled facade for old `astrbot.api.*` and selected `astrbot.core.*` paths, not a wholesale copy of the old application tree. Guard that facade with the checked-in import matrix and the external plugin matrix in `tests_v4/external_plugin_matrix.json`; do not claim compat from `load_plugin()` alone.
 
 # 开发命令
 

CLAUDE.md

@@ -27,6 +27,7 @@
 - 2026-03-13: On Windows, `.gitignore` matching is case-insensitive. A broad entry like `astrBot/` will also ignore `src-new/astrbot/...`, which can silently hide real compat alias packages from `git status`. Keep that ignore anchored as `/astrBot/` if it is only meant for a root-level scratch checkout.
 - 2026-03-13: The reference checkout under `astrBot/astrbot/api` exposes a broader old plugin-facing package surface than the current `src-new/astrbot` alias package. When improving package-name compatibility, compare against those public `api/*` modules as a set instead of only patching the one import path hit by the latest migrated plugin.
 - 2026-03-13: Some real legacy plugins call `asyncio.create_task()` during object construction. Calling `load_plugin()` outside a running event loop can therefore produce false-negative compat failures even though the real worker path is fine. External compat smoke tests should load plugins under an active loop, and "real compatibility" claims should preferably exercise `SupervisorRuntime` + worker + a real handler invocation instead of import-only checks.
+- 2026-03-13: Treat `src-new/astrbot` as a controlled legacy facade, not as a mirror of the old `astrBot/` tree. The compat contract is the checked-in public import matrix plus the external plugin matrix in `tests_v4/external_plugin_matrix.json`; when a new deep-path shim is proposed, require both an import assertion and a real supervisor/worker plugin case before growing the facade.
 
 # 开发命令
 

COMPATIBILITY_MATRIX.md

@@ -0,0 +1,63 @@
+# AstrBot 兼容矩阵
+
+`src-new/astrbot_sdk` 是 v4 真源，`src-new/astrbot` 只承担旧插件兼容门面。
+本文件记录当前兼容合同，避免把整个旧 `astrBot/core` 重新搬回新架构。
+
+## 边界
+
+| 级别 | 路径 | 策略 |
+| --- | --- | --- |
+| 一级 | `astrbot.api.*` | 优先做真实兼容 |
+| 二级 | `astrbot.core.*` 常见深路径 | 只有真实插件命中时才补薄 shim |
+| 三级 | 旧应用内部系统 | 不做树级复刻 |
+
+## 当前兼容面
+
+| 模块/路径 | 状态 | 说明 |
+| --- | --- | --- |
+| `astrbot.api` | 真实兼容 | 根入口、常见子模块可导入 |
+| `astrbot.api.all` | 真实兼容 | 聚合入口对齐旧公开面 |
+| `astrbot.api.event/filter/star/platform/provider/util` | 真实兼容 | 高频插件入口已收敛到 `src-new/astrbot` facade |
+| `astrbot.api.message_components` | 真实兼容 | 旧消息组件导入路径可用 |
+| `astrbot.core` | 导入兼容 | `AstrBotConfig`、`sp`、`logger`、`html_renderer` 门面可导入 |
+| `astrbot.core.config.*` | 导入兼容 | 当前只对齐 `AstrBotConfig` |
+| `astrbot.core.message.components` | 真实兼容 | 走 v4 消息组件 compat 实现 |
+| `astrbot.core.message.message_event_result` | 真实兼容 | 走 v4 事件结果 compat 实现 |
+| `astrbot.core.utils.session_waiter` | 真实兼容 | 已接上真实 follow-up message 路由 |
+| `astrbot.core.platform.*` | 导入兼容 / 部分真实兼容 | 高频模型与事件路径可导入，平台适配器注册仍 loud-fail |
+
+## 兼容合同测试
+
+以下合同由仓库内测试显式守护：
+
+- [tests_v4/test_compatibility_contract.py](/d:/GitObjectsOwn/astrbot-sdk/tests_v4/test_compatibility_contract.py)
+  - 一级：`astrbot.api`、`astrbot.api.all`、`astrbot.api.message_components`、`astrbot.api.event`、`astrbot.api.event.filter`、`astrbot.api.star`、`astrbot.api.platform`、`astrbot.api.provider`、`astrbot.api.util`
+  - 二级：`astrbot.core`、`astrbot.core.config.*`、`astrbot.core.message.*`、`astrbot.core.platform.*`、`astrbot.core.utils.session_waiter`
+- [tests_v4/test_external_plugin_smoke.py](/d:/GitObjectsOwn/astrbot-sdk/tests_v4/test_external_plugin_smoke.py)
+  - 外部真实插件矩阵必须走 `SupervisorRuntime -> Worker -> handler.invoke` 真链路
+  - 不以单独 `load_plugin()` 成功替代运行时兼容结论
+
+## 显式未支持
+
+以下能力仍保持 loud-fail，不伪造旧执行链：
+
+- `astrbot.api.agent`
+- `astrbot.api` / `astrbot.core` 的旧 html 渲染系统
+- `register_platform_adapter`
+- 旧 LLM hook / plugin hook / result decorate hook 的完整执行链
+
+## 真实插件矩阵
+
+矩阵清单位于 [tests_v4/external_plugin_matrix.json](/d:/GitObjectsOwn/astrbot-sdk/tests_v4/external_plugin_matrix.json)。
+当前标准是：
+
+1. 可加载
+2. 可初始化
+3. 至少一个代表命令在 `SupervisorRuntime -> Worker -> handler.invoke` 真链路下通过
+
+已纳入矩阵：
+
+- `astrbot_plugin_hapi_connector`
+- `astrbot_plugin_endfield`
+
+不 vendoring 第三方源码；测试时按需 clone。

src-new/astrbot/api/__init__.py

@@ -1,10 +1,8 @@
 """旧版 ``astrbot.api`` 导入路径兼容入口。"""
 
-from loguru import logger
-
-from astrbot_sdk._shared_preferences import sp
+from astrbot import logger
+from astrbot.core import AstrBotConfig, html_renderer, sp
 from astrbot_sdk.api import (
-    AstrBotConfig,
     components,
     event,
     message,
@@ -22,12 +20,6 @@ def agent(*args, **kwargs):
     )
 
 
-def html_renderer(*args, **kwargs):
-    raise NotImplementedError(
-        "astrbot.api.html_renderer 在 v4 兼容层中尚未提供，请改用当前平台发送/渲染能力。"
-    )
-
-
 __all__ = [
     "AstrBotConfig",
     "agent",

src-new/astrbot/api/platform/__init__.py

@@ -1,5 +1,6 @@
 """旧版 ``astrbot.api.platform`` 导入路径兼容入口。"""
 
+from astrbot.core.message.components import *  # noqa: F403
 from astrbot_sdk.api.event import (
     AstrBotMessage,
     AstrMessageEvent,

src-new/astrbot/core/__init__.py

@@ -5,6 +5,37 @@ from loguru import logger
 from astrbot_sdk._shared_preferences import sp
 from astrbot_sdk.api.basic import AstrBotConfig
 
-from . import utils
+from . import config, message, platform, utils
 
-__all__ = ["AstrBotConfig", "logger", "sp", "utils"]
+
+class _HtmlRendererCompat:
+    """旧版 ``html_renderer`` 的导入占位。
+
+    v4 兼容层目前没有复刻旧 core 的整套 HTML 渲染系统。
+    保留符号用于导入兼容，真实调用时显式报错，避免静默伪兼容。
+    TODO: 后续如果需要，可以在这里实现一个基于当前平台能力的 HTML 渲染适配器。
+    """
+
+    def __call__(self, *args, **kwargs):
+        raise NotImplementedError(
+            "astrbot.core.html_renderer 在 v4 兼容层中尚未提供，请改用当前平台发送/渲染能力。"
+        )
+
+    def __getattr__(self, _name: str):
+        raise NotImplementedError(
+            "astrbot.core.html_renderer 在 v4 兼容层中尚未提供，请改用当前平台发送/渲染能力。"
+        )
+
+
+html_renderer = _HtmlRendererCompat()
+
+__all__ = [
+    "AstrBotConfig",
+    "config",
+    "html_renderer",
+    "logger",
+    "message",
+    "platform",
+    "sp",
+    "utils",
+]

src-new/astrbot/core/config/__init__.py

@@ -0,0 +1,5 @@
+"""旧版 ``astrbot.core.config`` 导入路径兼容入口。"""
+
+from astrbot_sdk.api.basic import AstrBotConfig
+
+__all__ = ["AstrBotConfig"]

src-new/astrbot/core/config/astrbot_config.py

@@ -0,0 +1,5 @@
+"""旧版 ``astrbot.core.config.astrbot_config`` 导入路径兼容入口。"""
+
+from astrbot_sdk.api.basic import AstrBotConfig
+
+__all__ = ["AstrBotConfig"]

src-new/astrbot/core/message/__init__.py

@@ -1,3 +1,4 @@
 """旧版 ``astrbot.core.message`` 导入路径兼容入口。"""
 
 from .components import *  # noqa: F403
+from .message_event_result import *  # noqa: F403

src-new/astrbot/core/message/message_event_result.py

@@ -0,0 +1,17 @@
+"""旧版 ``astrbot.core.message.message_event_result`` 导入路径兼容入口。"""
+
+from astrbot_sdk.api.event import (
+    EventResultType,
+    MessageChain,
+    MessageEventResult,
+    ResultContentType,
+)
+from astrbot_sdk.api.event.event_result import CommandResult
+
+__all__ = [
+    "CommandResult",
+    "EventResultType",
+    "MessageChain",
+    "MessageEventResult",
+    "ResultContentType",
+]

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

@@ -0,0 +1,26 @@
+"""旧版 ``astrbot.core.platform`` 导入路径兼容入口。"""
+
+from astrbot.core.message.components import *  # noqa: F403
+from astrbot_sdk.api.event import (
+    AstrBotMessage,
+    AstrMessageEvent,
+    Group,
+    MessageMember,
+    MessageType,
+)
+from astrbot_sdk.api.platform import PlatformMetadata
+
+
+class Platform:
+    """旧版平台适配器基类占位。"""
+
+
+__all__ = [
+    "AstrBotMessage",
+    "AstrMessageEvent",
+    "Group",
+    "MessageMember",
+    "MessageType",
+    "Platform",
+    "PlatformMetadata",
+]

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

@@ -0,0 +1,18 @@
+"""旧版 ``astrbot.core.platform.astr_message_event`` 导入路径兼容入口。"""
+
+from astrbot_sdk.api.event import (
+    AstrMessageEvent,
+    AstrMessageEventModel,
+    MessageSesion,
+    MessageSession,
+)
+
+if not hasattr(AstrMessageEvent, "bot"):
+    AstrMessageEvent.bot = None
+
+__all__ = [
+    "AstrMessageEvent",
+    "AstrMessageEventModel",
+    "MessageSesion",
+    "MessageSession",
+]

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

@@ -0,0 +1,5 @@
+"""旧版 ``astrbot.core.platform.astrbot_message`` 导入路径兼容入口。"""
+
+from astrbot_sdk.api.event import AstrBotMessage, Group, MessageMember, MessageType
+
+__all__ = ["AstrBotMessage", "Group", "MessageMember", "MessageType"]

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

@@ -0,0 +1,5 @@
+"""旧版 ``astrbot.core.platform.message_type`` 导入路径兼容入口。"""
+
+from astrbot_sdk.api.event import MessageType
+
+__all__ = ["MessageType"]

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

@@ -0,0 +1,8 @@
+"""旧版 ``astrbot.core.platform.platform`` 导入路径兼容入口。"""
+
+
+class Platform:
+    """旧版平台适配器基类占位。"""
+
+
+__all__ = ["Platform"]

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

@@ -0,0 +1,5 @@
+"""旧版 ``astrbot.core.platform.platform_metadata`` 导入路径兼容入口。"""
+
+from astrbot_sdk.api.platform import PlatformMetadata
+
+__all__ = ["PlatformMetadata"]

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

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

src-new/astrbot/core/platform/sources/__init__.py

@@ -0,0 +1 @@
+"""旧版 ``astrbot.core.platform.sources`` 导入路径兼容入口。"""

src-new/astrbot/core/platform/sources/aiocqhttp/__init__.py

@@ -0,0 +1,5 @@
+"""旧版 ``astrbot.core.platform.sources.aiocqhttp`` 导入路径兼容入口。"""
+
+from .aiocqhttp_message_event import AiocqhttpMessageEvent
+
+__all__ = ["AiocqhttpMessageEvent"]

src-new/astrbot/core/platform/sources/aiocqhttp/aiocqhttp_message_event.py

@@ -0,0 +1,7 @@
+"""旧版 aiocqhttp 事件类型的最小导入兼容入口。"""
+
+from astrbot.core.platform.astr_message_event import AstrMessageEvent
+
+AiocqhttpMessageEvent = AstrMessageEvent
+
+__all__ = ["AiocqhttpMessageEvent"]

tests_v4/external_plugin_matrix.json

@@ -0,0 +1,22 @@
+{
+  "cases": [
+    {
+      "name": "hapi_connector",
+      "repo": "https://github.com/LiJinHao999/astrbot_plugin_hapi_connector.git",
+      "command": "hapi",
+      "event_text": "hapi",
+      "expected_text": "HAPI",
+      "known_unsupported": []
+    },
+    {
+      "name": "endfield",
+      "repo": "https://github.com/Entropy-Increase-Team/astrbot_plugin_endfield.git",
+      "command": "zmd",
+      "event_text": "zmd",
+      "expected_text": "终末地协议终端",
+      "known_unsupported": [
+        "依赖 playwright 时会退回纯文本帮助输出"
+      ]
+    }
+  ]
+}

tests_v4/test_api_modules.py

@@ -254,6 +254,39 @@ class TestAstrbotImportAlias:
         assert Node(uin="1", content=[]).sender_id == "1"
         assert Nodes(nodes=[]).nodes == []
 
+    def test_legacy_astrbot_core_config_imports(self):
+        """astrbot.core.config old import paths should remain available."""
+        from astrbot.core.config import AstrBotConfig
+        from astrbot.core.config.astrbot_config import AstrBotConfig as ConfigModel
+
+        assert AstrBotConfig is ConfigModel
+
+    def test_legacy_astrbot_core_platform_imports(self):
+        """astrbot.core.platform old import paths should remain available."""
+        from astrbot.core.platform import (
+            AstrBotMessage,
+            AstrMessageEvent,
+            MessageType,
+            Platform,
+            PlatformMetadata,
+        )
+        from astrbot.core.platform.astr_message_event import MessageSession
+        from astrbot.core.platform.message_type import MessageType as MessageTypeModule
+        from astrbot.core.platform.platform_metadata import (
+            PlatformMetadata as MetaModule,
+        )
+        from astrbot.core.platform.register import register_platform_adapter
+        from astrbot.core.platform.sources.aiocqhttp import AiocqhttpMessageEvent
+
+        assert AstrBotMessage is not None
+        assert AstrMessageEvent is AiocqhttpMessageEvent
+        assert MessageSession is not None
+        assert MessageType is MessageTypeModule
+        assert PlatformMetadata is MetaModule
+        assert Platform is not None
+        with pytest.raises(NotImplementedError, match="register_platform_adapter"):
+            register_platform_adapter()
+
     def test_legacy_astrbot_event_filter_module_exports(self):
         """astrbot.api.event.filter should be importable from the old module path."""
         from astrbot.api.event.filter import EventMessageType, command, llm_tool

tests_v4/test_compatibility_contract.py

@@ -0,0 +1,57 @@
+"""Compatibility contract tests for the controlled ``astrbot`` facade."""
+
+from __future__ import annotations
+
+from importlib import import_module
+
+import pytest
+
+LEVEL_ONE_MODULES = [
+    "astrbot.api",
+    "astrbot.api.all",
+    "astrbot.api.message_components",
+    "astrbot.api.event",
+    "astrbot.api.event.filter",
+    "astrbot.api.star",
+    "astrbot.api.platform",
+    "astrbot.api.provider",
+    "astrbot.api.util",
+]
+
+LEVEL_TWO_MODULES = [
+    "astrbot.core",
+    "astrbot.core.config",
+    "astrbot.core.config.astrbot_config",
+    "astrbot.core.message",
+    "astrbot.core.message.components",
+    "astrbot.core.message.message_event_result",
+    "astrbot.core.platform",
+    "astrbot.core.platform.astr_message_event",
+    "astrbot.core.platform.astrbot_message",
+    "astrbot.core.platform.message_type",
+    "astrbot.core.platform.platform_metadata",
+    "astrbot.core.platform.register",
+    "astrbot.core.platform.sources.aiocqhttp",
+    "astrbot.core.utils",
+    "astrbot.core.utils.session_waiter",
+]
+
+
+@pytest.mark.parametrize("module_name", LEVEL_ONE_MODULES)
+def test_level_one_legacy_facade_modules_import(module_name: str):
+    """一级 compat 合同中的旧公开模块必须始终可导入。"""
+    assert import_module(module_name) is not None
+
+
+@pytest.mark.parametrize("module_name", LEVEL_TWO_MODULES)
+def test_level_two_legacy_facade_modules_import(module_name: str):
+    """二级 compat 合同中的高频深路径必须始终可导入。"""
+    assert import_module(module_name) is not None
+
+
+def test_level_two_html_renderer_stays_loud_fail():
+    """未实现的旧 HTML 渲染系统应保持显式失败，而不是静默伪兼容。"""
+    from astrbot.core import html_renderer
+
+    with pytest.raises(NotImplementedError, match="html_renderer"):
+        html_renderer()

tests_v4/test_external_plugin_smoke.py

@@ -1,7 +1,13 @@
 """可选的外部插件兼容 smoke 测试。
 
-默认不跑；手动验证真实外部插件时，设置
-``ASTRBOT_EXTERNAL_PLUGIN_REPO=https://...`` 即可启用。
+默认不跑；标准入口是仓库内的外部插件矩阵：
+
+- ``ASTRBOT_EXTERNAL_PLUGIN_CASES=all``
+- ``ASTRBOT_EXTERNAL_PLUGIN_CASES=hapi_connector,endfield``
+
+也保留单仓库 ad-hoc 模式：
+
+- ``ASTRBOT_EXTERNAL_PLUGIN_REPO=https://...``
 
 如果还希望验证真实 handler 调用，而不是仅验证可加载，可以额外设置：
 
@@ -17,6 +23,7 @@ import os
 import subprocess
 import tempfile
 import textwrap
+import json
 from pathlib import Path
 
 import pytest
@@ -35,6 +42,8 @@ EXTERNAL_PLUGIN_REPO_ENV = "ASTRBOT_EXTERNAL_PLUGIN_REPO"
 EXTERNAL_PLUGIN_COMMAND_ENV = "ASTRBOT_EXTERNAL_PLUGIN_COMMAND"
 EXTERNAL_PLUGIN_EVENT_TEXT_ENV = "ASTRBOT_EXTERNAL_PLUGIN_EVENT_TEXT"
 EXTERNAL_PLUGIN_EXPECT_TEXT_ENV = "ASTRBOT_EXTERNAL_PLUGIN_EXPECT_TEXT"
+EXTERNAL_PLUGIN_CASES_ENV = "ASTRBOT_EXTERNAL_PLUGIN_CASES"
+EXTERNAL_PLUGIN_MATRIX_PATH = Path(__file__).with_name("external_plugin_matrix.json")
 
 
 def _clone_external_plugin(
@@ -52,6 +61,143 @@ def _clone_external_plugin(
     )
 
 
+def _load_external_plugin_matrix() -> list[dict[str, str]]:
+    payload = json.loads(EXTERNAL_PLUGIN_MATRIX_PATH.read_text(encoding="utf-8"))
+    cases = payload.get("cases", [])
+    return [case for case in cases if isinstance(case, dict)]
+
+
+def _selected_matrix_cases() -> list[dict[str, str]]:
+    selector = os.getenv(EXTERNAL_PLUGIN_CASES_ENV, "").strip()
+    if not selector:
+        return []
+    cases = _load_external_plugin_matrix()
+    if selector.lower() == "all":
+        return cases
+    selected_names = {item.strip() for item in selector.split(",") if item.strip()}
+    return [case for case in cases if case.get("name") in selected_names]
+
+
+def _load_plugin_in_subprocess(
+    *,
+    project_root: Path,
+    clone_dir: Path,
+) -> subprocess.CompletedProcess[str]:
+    spec = load_plugin_spec(clone_dir)
+    manager = PluginEnvironmentManager(project_root)
+    python_path = manager.prepare_environment(spec)
+    script = textwrap.dedent(
+        f"""
+        import asyncio
+        import sys
+        from pathlib import Path
+
+        repo_root = Path({str(project_root)!r})
+        plugin_dir = Path({str(clone_dir)!r})
+        sys.path.insert(0, str((repo_root / "src-new").resolve()))
+
+        from astrbot_sdk.runtime.loader import load_plugin, load_plugin_spec
+
+        async def main():
+            spec = load_plugin_spec(plugin_dir)
+            loaded = load_plugin(spec)
+            print("PLUGIN", loaded.plugin.name)
+            print("HANDLERS", len(loaded.handlers))
+            print("CAPS", len(loaded.capabilities))
+
+        asyncio.run(main())
+        """
+    )
+    env = os.environ.copy()
+    env["PYTHONPATH"] = str((project_root / "src-new").resolve())
+    return subprocess.run(
+        [str(python_path), "-c", script],
+        check=False,
+        cwd=project_root,
+        env=env,
+        capture_output=True,
+        text=True,
+    )
+
+
+async def _run_runtime_command_smoke(
+    *,
+    project_root: Path,
+    plugin_root: Path,
+    command_name: str,
+    event_text: str,
+    expected_text: str | None,
+) -> None:
+    left, right = make_transport_pair()
+    core = Peer(
+        transport=left,
+        peer_info=PeerInfo(name="outer-core", role="core", version="v4"),
+    )
+    core.set_initialize_handler(
+        lambda _message: asyncio.sleep(
+            0,
+            result=InitializeOutput(
+                peer=PeerInfo(name="outer-core", role="core", version="v4"),
+                capabilities=[],
+                metadata={},
+            ),
+        )
+    )
+
+    runtime = SupervisorRuntime(
+        transport=right,
+        plugins_dir=plugin_root.parent,
+        env_manager=PluginEnvironmentManager(project_root),
+    )
+    await core.start()
+    try:
+        await runtime.start()
+        await core.wait_until_remote_initialized()
+
+        handler = next(
+            (
+                item
+                for item in core.remote_handlers
+                if getattr(item.trigger, "command", None) == command_name
+            ),
+            None,
+        )
+        assert handler is not None, (
+            f"command handler not found: {command_name}; "
+            f"available={[getattr(item.trigger, 'command', None) for item in core.remote_handlers]}"
+        )
+
+        await core.invoke(
+            "handler.invoke",
+            {
+                "handler_id": handler.id,
+                "event": {
+                    "text": event_text,
+                    "session_id": "external-smoke-session",
+                    "user_id": "user-1",
+                    "platform": "test",
+                },
+            },
+            request_id=f"external-runtime-command-{command_name}",
+        )
+
+        sent_messages = list(runtime.capability_router.sent_messages)
+        assert sent_messages, (
+            "external plugin command completed but did not emit any platform "
+            "message; this usually means the command path was not really exercised"
+        )
+
+        if expected_text is not None:
+            assert any(
+                expected_text in item.get("text", "")
+                for item in sent_messages
+                if "text" in item
+            ), sent_messages
+    finally:
+        await runtime.stop()
+        await core.stop()
+
+
 @pytest.mark.skipif(
     not os.getenv(EXTERNAL_PLUGIN_REPO_ENV),
     reason=f"set {EXTERNAL_PLUGIN_REPO_ENV} to enable external plugin smoke tests",
@@ -69,40 +215,8 @@ def test_external_plugin_load_smoke():
             clone_dir=clone_dir,
         )
 
-        spec = load_plugin_spec(clone_dir)
-        manager = PluginEnvironmentManager(project_root)
-        python_path = manager.prepare_environment(spec)
-        script = textwrap.dedent(
-            f"""
-            import asyncio
-            import sys
-            from pathlib import Path
-
-            repo_root = Path({str(project_root)!r})
-            plugin_dir = Path({str(clone_dir)!r})
-            sys.path.insert(0, str((repo_root / "src-new").resolve()))
-
-            from astrbot_sdk.runtime.loader import load_plugin, load_plugin_spec
-
-            async def main():
-                spec = load_plugin_spec(plugin_dir)
-                loaded = load_plugin(spec)
-                print("PLUGIN", loaded.plugin.name)
-                print("HANDLERS", len(loaded.handlers))
-                print("CAPS", len(loaded.capabilities))
-
-            asyncio.run(main())
-            """
-        )
-        env = os.environ.copy()
-        env["PYTHONPATH"] = str((project_root / "src-new").resolve())
-        result = subprocess.run(
-            [str(python_path), "-c", script],
-            check=False,
-            cwd=project_root,
-            env=env,
-            capture_output=True,
-            text=True,
+        result = _load_plugin_in_subprocess(
+            project_root=project_root, clone_dir=clone_dir
         )
 
         assert result.returncode == 0, (
@@ -139,71 +253,74 @@ async def test_external_plugin_runtime_command_smoke():
             clone_dir=plugin_root,
         )
 
-        left, right = make_transport_pair()
-        core = Peer(
-            transport=left,
-            peer_info=PeerInfo(name="outer-core", role="core", version="v4"),
-        )
-        core.set_initialize_handler(
-            lambda _message: asyncio.sleep(
-                0,
-                result=InitializeOutput(
-                    peer=PeerInfo(name="outer-core", role="core", version="v4"),
-                    capabilities=[],
-                    metadata={},
-                ),
-            )
+        await _run_runtime_command_smoke(
+            project_root=project_root,
+            plugin_root=plugin_root,
+            command_name=command_name,
+            event_text=event_text,
+            expected_text=expected_text,
         )
 
-        runtime = SupervisorRuntime(
-            transport=right,
-            plugins_dir=plugins_root,
-            env_manager=PluginEnvironmentManager(project_root),
-        )
-        await core.start()
-        try:
-            await runtime.start()
-            await core.wait_until_remote_initialized()
 
-            handler = next(
-                (
-                    item
-                    for item in core.remote_handlers
-                    if getattr(item.trigger, "command", None) == command_name
-                ),
-                None,
-            )
-            assert handler is not None, (
-                f"command handler not found: {command_name}; "
-                f"available={[getattr(item.trigger, 'command', None) for item in core.remote_handlers]}"
-            )
+@pytest.mark.skipif(
+    not os.getenv(EXTERNAL_PLUGIN_CASES_ENV),
+    reason=f"set {EXTERNAL_PLUGIN_CASES_ENV} to enable matrix external plugin smoke tests",
+)
+def test_external_plugin_matrix_load_smoke():
+    """按矩阵批量验证外部插件能在独立环境里完成真实加载。"""
+    project_root = Path(__file__).resolve().parent.parent
+    cases = _selected_matrix_cases()
+    assert cases, f"no matrix cases matched {EXTERNAL_PLUGIN_CASES_ENV}"
 
-            await core.invoke(
-                "handler.invoke",
-                {
-                    "handler_id": handler.id,
-                    "event": {
-                        "text": event_text,
-                        "session_id": "external-smoke-session",
-                        "user_id": "user-1",
-                        "platform": "test",
-                    },
-                },
-                request_id="external-runtime-command",
+    with tempfile.TemporaryDirectory(
+        prefix="astrbot-external-matrix-load-"
+    ) as temp_dir:
+        temp_root = Path(temp_dir)
+        for case in cases:
+            clone_dir = temp_root / case["name"]
+            _clone_external_plugin(
+                project_root=project_root,
+                repo_url=case["repo"],
+                clone_dir=clone_dir,
             )
-
-            sent_messages = list(runtime.capability_router.sent_messages)
-            assert sent_messages, (
-                "external plugin command completed but did not emit any platform "
-                "message; this usually means the command path was not really exercised"
+            result = _load_plugin_in_subprocess(
+                project_root=project_root,
+                clone_dir=clone_dir,
             )
+            assert result.returncode == 0, (
+                f"case={case['name']}\nstdout:\n{result.stdout}\n\nstderr:\n{result.stderr}"
+            )
+            assert "HANDLERS" in result.stdout
+            assert "PLUGIN" in result.stdout
 
-            if expected_text is not None:
-                assert any(
-                    expected_text in item.get("text", "")
-                    for item in sent_messages
-                    if "text" in item
-                ), sent_messages
-        finally:
-            await runtime.stop()
-            await core.stop()
+
+@pytest.mark.skipif(
+    not os.getenv(EXTERNAL_PLUGIN_CASES_ENV),
+    reason=f"set {EXTERNAL_PLUGIN_CASES_ENV} to enable matrix external plugin smoke tests",
+)
+@pytest.mark.asyncio
+async def test_external_plugin_matrix_runtime_smoke():
+    """按矩阵批量验证外部插件代表命令能走真实 supervisor/worker 链路。"""
+    project_root = Path(__file__).resolve().parent.parent
+    cases = _selected_matrix_cases()
+    assert cases, f"no matrix cases matched {EXTERNAL_PLUGIN_CASES_ENV}"
+
+    with tempfile.TemporaryDirectory(
+        prefix="astrbot-external-matrix-runtime-"
+    ) as temp_dir:
+        temp_root = Path(temp_dir)
+        for case in cases:
+            plugins_root = temp_root / f"{case['name']}_plugins"
+            plugin_root = plugins_root / case["name"]
+            _clone_external_plugin(
+                project_root=project_root,
+                repo_url=case["repo"],
+                clone_dir=plugin_root,
+            )
+            await _run_runtime_command_smoke(
+                project_root=project_root,
+                plugin_root=plugin_root,
+                command_name=case["command"],
+                event_text=case.get("event_text", case["command"]),
+                expected_text=case.get("expected_text"),
+            )