Expand new and legacy sample plugin coverage

test_plugin/new/commands/hello.py

@@ -1,20 +1,32 @@
-"""V4 sample plugin used by integration tests."""
+"""V4 sample plugin used by integration tests.
+
+This fixture intentionally exercises the currently supported public v4 API:
+- top-level decorators: on_command/on_message/on_event/on_schedule/require_admin
+- Context clients: llm, memory, db, platform
+- MessageEvent helpers: reply/plain_result/target/to_payload
+- plugin-provided capabilities: normal + stream
+"""
 
 from __future__ import annotations
 
+import asyncio
+
 from astrbot_sdk import (
     Context,
     MessageEvent,
     Star,
     on_command,
+    on_event,
     on_message,
+    on_schedule,
     provide_capability,
+    require_admin,
 )
 from astrbot_sdk.context import CancelToken
 
 
 class HelloPlugin(Star):
-    """Small but representative v4 plugin fixture."""
+    """Representative v4 plugin fixture."""
 
     @on_command("hello", aliases=["hi"], description="发送问候消息")
     async def hello(self, event: MessageEvent, ctx: Context) -> None:
@@ -26,22 +38,50 @@ class HelloPlugin(Star):
             chunks.append(chunk)
         await event.reply("".join(chunks))
 
-    @on_command("remember", description="保存一条记忆并回读")
-    async def remember(self, event: MessageEvent, ctx: Context) -> None:
+    @on_command("raw", description="调用 llm.chat_raw 并返回结构化信息")
+    async def raw(self, event: MessageEvent, ctx: Context):
+        response = await ctx.llm.chat_raw(
+            event.text,
+            system="be concise",
+            history=[{"role": "user", "content": "history"}],
+        )
+        payload = event.to_payload()
+        ctx.logger.info("raw handler for {}", payload.get("session_id"))
+        return event.plain_result(
+            f"raw={response.text}|finish={response.finish_reason}|"
+            f"cancelled={ctx.cancel_token.cancelled}"
+        )
+
+    @on_command("remember", description="覆盖 memory/db 全量基本操作")
+    async def remember(self, event: MessageEvent, ctx: Context):
         await ctx.memory.save(
             "demo:last_message",
             {"user_id": event.user_id or "", "text": event.text},
+            source="fixture",
         )
         remembered = await ctx.memory.get("demo:last_message") or {}
+        searched = await ctx.memory.search("fixture")
+        await ctx.memory.delete("demo:last_message")
+
         await ctx.db.set("demo:last_session", event.session_id)
+        session_value = await ctx.db.get("demo:last_session")
         keys = await ctx.db.list("demo:")
-        await event.reply(
-            f"Memory saved for {remembered.get('user_id', 'unknown')} with {len(keys)} keys"
+        await ctx.db.delete("demo:last_session")
+
+        return event.plain_result(
+            f"remembered={remembered.get('user_id', 'unknown')}|"
+            f"searched={len(searched)}|session={session_value}|keys={len(keys)}"
         )
 
-    @on_message(regex=r"^ping$")
-    async def ping(self, event: MessageEvent) -> None:
-        await event.reply("pong")
+    @on_command("platforms", description="覆盖 platform 相关 API")
+    async def platforms(self, event: MessageEvent, ctx: Context) -> None:
+        target = event.target or event.session_id
+        members = await ctx.platform.get_members(target)
+        await ctx.platform.send_image(target, "https://example.com/demo.png")
+        await ctx.platform.send(
+            target,
+            f"members={len(members)} first={members[0]['user_id'] if members else 'none'}",
+        )
 
     @on_command("announce", description="发送一条富消息链")
     async def announce(self, event: MessageEvent, ctx: Context) -> None:
@@ -53,6 +93,23 @@ class HelloPlugin(Star):
             ],
         )
 
+    @require_admin
+    @on_command("secure", description="测试 require_admin")
+    async def secure(self, event: MessageEvent):
+        return event.plain_result(f"secure:{event.user_id or 'unknown'}")
+
+    @on_message(regex=r"^ping$", keywords=["ping"], platforms=["test"])
+    async def ping(self, event: MessageEvent):
+        return event.plain_result("pong")
+
+    @on_event("group_join")
+    async def on_group_join(self, event: MessageEvent, ctx: Context) -> None:
+        ctx.logger.info("event handler observed {}", event.text)
+
+    @on_schedule(interval_seconds=60)
+    async def heartbeat(self, ctx: Context) -> None:
+        await ctx.db.set("demo:last_schedule", {"status": "ok"})
+
     @provide_capability(
         "demo.echo",
         description="回显输入文本",
@@ -77,7 +134,46 @@ class HelloPlugin(Star):
         cancel_token: CancelToken,
     ) -> dict[str, str]:
         cancel_token.raise_if_cancelled()
+        await ctx.db.set(
+            "demo:capability_echo",
+            {"text": str(payload.get("text", ""))},
+        )
+        stored = await ctx.db.get("demo:capability_echo") or {}
         return {
-            "echo": str(payload.get("text", "")),
+            "echo": str(stored.get("text", "")),
             "plugin_id": ctx.plugin_id,
         }
+
+    @provide_capability(
+        "demo.stream",
+        description="流式回显输入文本",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        output_schema={
+            "type": "object",
+            "properties": {
+                "items": {
+                    "type": "array",
+                    "items": {"type": "object"},
+                }
+            },
+            "required": ["items"],
+        },
+        supports_stream=True,
+        cancelable=True,
+    )
+    async def stream_capability(
+        self,
+        payload: dict[str, object],
+        ctx: Context,
+        cancel_token: CancelToken,
+    ):
+        text = str(payload.get("text", ""))
+        await ctx.db.set("demo:last_stream", {"text": text})
+        for char in text:
+            cancel_token.raise_if_cancelled()
+            await asyncio.sleep(0)
+            yield {"text": char}

test_plugin/old/commands/hello.py

@@ -1,103 +1,233 @@
-"""旧版插件兼容测试 - 使用 astrbot_sdk 旧版 API。
+"""旧版插件兼容测试夹具。
 
-测试覆盖：
-- CommandComponent 继承
-- filter.command 装饰器
-- LegacyContext 功能 (conversation_manager, send_message, db)
-- AstrMessageEvent 和 MessageChain
-- 消息组件 (Plain, At, Image)
+这个样例故意覆盖当前仍被兼容层支持的旧 API：
+- CommandComponent / AstrMessageEvent / MessageChain
+- filter.command / regex / permission / permission_type /
+  event_message_type / platform_adapter_type
+- LegacyContext 的 conversation_manager / llm_generate / tool_loop_agent /
+  send_message / put|get|delete_kv_data / call_context_function
+- 消息组件及其旧字段别名/工厂方法
 """
 
+from __future__ import annotations
+
+from astrbot_sdk import provide_capability
 from astrbot_sdk.api.components.command import CommandComponent
 from astrbot_sdk.api.event import AstrMessageEvent, filter
-from astrbot_sdk.api.event.filter import EventMessageType, PlatformAdapterType
-from astrbot_sdk.api.star.context import Context
+from astrbot_sdk.api.event.filter import (
+    ADMIN,
+    EventMessageType,
+    PermissionType,
+    PlatformAdapterType,
+)
 from astrbot_sdk.api.message import MessageChain
-from astrbot_sdk.api.message_components import Plain, At, Image
+from astrbot_sdk.api.message_components import (
+    At,
+    AtAll,
+    Face,
+    File,
+    Image,
+    Node,
+    Plain,
+    Record,
+    Reply,
+    Video,
+)
+from astrbot_sdk.api.star.context import Context
 from loguru import logger
 
 
+class CompatHelper:
+    __compat_component_name__ = "CompatHelper"
+
+    async def shout(self, text: str) -> str:
+        return text.upper()
+
+
 class HelloCommand(CommandComponent):
     """测试旧版 CommandComponent 兼容性。"""
 
     def __init__(self, context: Context):
         self.context = context
+        self.context._register_component(CompatHelper())
 
-    @filter.command("hello")
+    @filter.command("hello", alias={"hi"}, priority=10, desc="问候命令")
     async def hello(self, event: AstrMessageEvent):
         """基本命令测试。"""
         ret = await self.context.conversation_manager.new_conversation("hello")
-        logger.info(f"New conversation created: {ret}")
+        logger.info("New conversation created: {}", ret)
         yield event.plain_result(f"Hello, Astrbot! Created conversation ID: {ret}")
 
     @filter.command("echo")
     async def echo(self, event: AstrMessageEvent):
-        """测试消息获取。"""
+        """测试消息获取和 extra 状态。"""
         text = event.get_message_str()
-        yield event.plain_result(f"Echo: {text}")
+        event.set_extra("last_echo", text)
+        extra = event.get_extra("last_echo")
+        event.clear_extra()
+        group = await event.get_group()
+        yield event.plain_result(
+            f"Echo: {text}|sender={event.get_sender_id()}|sender_name={event.get_sender_name()}|"
+            f"platform={event.get_platform_name()}:{event.get_platform_id()}|"
+            f"type={event.get_message_type().value}|messages={len(event.get_messages())}|"
+            f"self={event.get_self_id()}|private={event.is_private_chat()}|"
+            f"wake={event.is_wake_up()}|group={group is not None}|extra={extra}"
+        )
 
     @filter.command("chain")
     async def test_chain(self, event: AstrMessageEvent):
-        """测试 MessageChain 构建和发送。"""
-        # 构建消息链
+        """测试 MessageChain 构建、发送与 react。"""
         chain = (
             MessageChain()
-            .message("Hello! ")
+            .message("Hello")
+            .message(" ")
             .at("user", "12345")
+            .at_all()
             .message(" check this image: ")
             .url_image("https://example.com/test.png")
+            .file_image("C:/tmp/test.png")
+            .base64_image("base64://fixture")
+            .use_t2i(True)
+            .squash_plain()
         )
 
-        # 测试 to_payload 和 is_plain_text_only
-        payload = chain.to_payload()
-        is_plain = chain.is_plain_text_only()
-
-        logger.info(f"Chain payload: {payload}, is_plain_text_only: {is_plain}")
-
-        yield event.plain_result(f"Chain sent with {len(payload)} components")
+        await event.send(chain)
+        await event.react(":thumbsup:")
+        yield event.plain_result(
+            f"Chain sent with {len(chain.to_payload())} components and t2i={chain.use_t2i_}"
+        )
 
     @filter.command("db")
     async def test_db(self, event: AstrMessageEvent):
         """测试 KV 数据库操作。"""
-        # 写入数据
         await self.context.put_kv_data("test_key", {"value": "test_data"})
-
-        # 读取数据
         data = await self.context.get_kv_data("test_key")
-        logger.info(f"Got data from db: {data}")
+        await self.context.delete_kv_data("test_key")
+        deleted = await self.context.get_kv_data("test_key", "missing")
+        logger.info("Got data from db: {}", data)
+        yield event.plain_result(f"DB test: stored={data} deleted={deleted}")
 
-        yield event.plain_result(f"DB test: stored and retrieved {data}")
+    @filter.command("conversation")
+    async def test_conversation(self, event: AstrMessageEvent):
+        """测试会话管理器和 call_context_function。"""
+        umo = f"platform:{event.get_session_id()}"
+        cid = await self.context.conversation_manager.new_conversation(
+            unified_msg_origin=umo,
+            title="Compat Chat",
+            persona_id="assistant",
+        )
+        await self.context.conversation_manager.add_message_pair(cid, "hello", "world")
+        await self.context.conversation_manager.update_conversation(
+            unified_msg_origin=umo,
+            conversation_id=cid,
+            title="Compat Chat Updated",
+        )
+        await self.context.conversation_manager.update_conversation_title(umo, "Compat")
+        await self.context.conversation_manager.update_conversation_persona_id(
+            umo,
+            "compat-persona",
+        )
+        current_id = await self.context.conversation_manager.get_curr_conversation_id(
+            umo
+        )
+        conv = await self.context.conversation_manager.get_conversation(umo, cid)
+        all_conversations = await self.context.conversation_manager.get_conversations(
+            umo
+        )
+        helper_result = await self.context.call_context_function(
+            "CompatHelper.shout",
+            {"text": "compat"},
+        )
+        await self.context.conversation_manager.switch_conversation(umo, cid)
+        await self.context.conversation_manager.delete_conversation(umo, cid)
+        yield event.plain_result(
+            f"conversation={current_id}|messages={len((conv or {}).get('content', []))}|"
+            f"all={len(all_conversations)}|helper={helper_result['data']}"
+        )
+
+    @filter.command("ai")
+    async def test_ai(self, event: AstrMessageEvent):
+        """测试旧版 AI compat 入口。"""
+        llm_resp = await self.context.llm_generate(
+            chat_provider_id="provider-demo",
+            prompt="legacy hello",
+            contexts=[{"role": "user", "content": "hi"}],
+        )
+        agent_resp = await self.context.tool_loop_agent(
+            chat_provider_id="provider-demo",
+            prompt="legacy hello",
+            tools=[{"name": "search"}],
+            max_steps=3,
+        )
+        yield event.plain_result(f"LLM:{llm_resp.text}|AGENT:{agent_resp.text}")
+
+    @filter.command("sendmsg")
+    async def test_send_message(self, event: AstrMessageEvent):
+        """测试 LegacyContext.send_message。"""
+        chain = (
+            MessageChain()
+            .message("compat send ")
+            .at("legacy-user", "10001")
+            .url_image("https://example.com/send.png")
+        )
+        await self.context.send_message(event.get_session_id(), chain)
+        yield event.plain_result("send_message invoked")
 
     @filter.command("components")
     async def test_components(self, event: AstrMessageEvent):
-        """测试消息组件。"""
-        # 测试 Plain
-        plain = Plain(text="Hello")
+        """测试消息组件和 chain_result/image_result。"""
+        components = [
+            Plain(text="Hello"),
+            At(qq="123", name="legacy_user"),
+            AtAll(),
+            Image.fromURL("https://example.com/img.png"),
+            Image.fromFileSystem("C:/tmp/local.png"),
+            Record.fromFileSystem("C:/tmp/sound.wav"),
+            Video.fromURL("https://example.com/video.mp4"),
+            File(name="demo.txt", file="https://example.com/demo.txt"),
+            Reply(id="reply-1"),
+            Node(uin="10001", name="node_user", content=[Plain(text="node")]),
+            Face(id=1),
+        ]
+        component_dicts = [component.to_dict() for component in components]
+        logger.info("Components: {}", component_dicts)
+        yield event.chain_result(components)
+        yield event.image_result("https://example.com/components.png")
 
-        # 测试 At
-        at = At(user_id="123", user_name="test_user")
-
-        # 测试 Image
-        image = Image.fromURL("https://example.com/img.png")
-
-        # 测试 to_dict
-        plain_dict = plain.to_dict()
-        at_dict = at.to_dict()
-        image_dict = image.to_dict()
-
-        logger.info(f"Plain: {plain_dict}, At: {at_dict}, Image: {image_dict}")
-
-        yield event.plain_result("Components created: Plain, At, Image")
+    @filter.command("state")
+    async def test_event_state(self, event: AstrMessageEvent):
+        """测试事件状态方法。"""
+        result = event.make_result().message("state-ok")
+        event.set_result(result)
+        before_stop = event.is_stopped()
+        event.stop_event()
+        after_stop = event.is_stopped()
+        event.continue_event()
+        after_continue = event.is_stopped()
+        event.should_call_llm(True)
+        stored = event.get_result()
+        event.clear_result()
+        yield event.plain_result(
+            f"before={before_stop}|after_stop={after_stop}|after_continue={after_continue}|"
+            f"call_llm={event.call_llm}|stored={stored is not None}"
+        )
 
     @filter.regex(r"^ping.*")
     async def ping_regex(self, event: AstrMessageEvent):
         """测试正则匹配。"""
         yield event.plain_result("Pong from regex!")
 
-    @filter.permission("admin")
+    @filter.permission(ADMIN)
+    @filter.command("admin")
     async def admin_only(self, event: AstrMessageEvent):
-        """测试权限过滤 (应该被跳过，因为没有 require_admin)。"""
-        yield event.plain_result("Admin command executed")
+        """测试 permission(ADMIN)。"""
+        yield event.plain_result(f"Admin command executed by {event.is_admin()}")
+
+    @filter.permission_type(PermissionType.ADMIN)
+    @filter.command("admin_type")
+    async def admin_type_only(self, event: AstrMessageEvent):
+        """测试 permission_type(PermissionType.ADMIN)。"""
+        yield event.plain_result(f"Admin type command executed by {event.is_admin()}")
 
     @filter.event_message_type(EventMessageType.GROUP_MESSAGE)
     async def group_only(self, event: AstrMessageEvent):
@@ -108,3 +238,30 @@ class HelloCommand(CommandComponent):
     async def cqhttp_only(self, event: AstrMessageEvent):
         """测试平台过滤。"""
         yield event.plain_result("CQHttp platform detected")
+
+    @provide_capability(
+        "compat.echo",
+        description="使用 legacy Context 回显输入文本",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        output_schema={
+            "type": "object",
+            "properties": {
+                "echo": {"type": "string"},
+                "plugin_id": {"type": "string"},
+            },
+            "required": ["echo", "plugin_id"],
+        },
+    )
+    async def echo_capability(self, payload: dict):
+        """测试旧版插件 capability 能走 LegacyContext。"""
+        text = str(payload.get("text", ""))
+        await self.context.put_kv_data("compat_capability", {"text": text})
+        stored = await self.context.get_kv_data("compat_capability", {})
+        return {
+            "echo": str(stored.get("text", "")),
+            "plugin_id": self.context.plugin_id,
+        }

tests_v4/test_legacy_plugin_integration.py

@@ -496,10 +496,35 @@ class TestRealTestPlugin:
             assert loaded.plugin.name == "astrbot_plugin_helloworld"
             assert len(loaded.instances) == 1
             assert len(loaded.handlers) >= 1
+            assert [item.descriptor.name for item in loaded.capabilities] == [
+                "compat.echo"
+            ]
 
-            # 验证处理器
-            handler_ids = [h.descriptor.id for h in loaded.handlers]
-            assert any("hello" in hid for hid in handler_ids)
+            command_names = {
+                handler.descriptor.trigger.command
+                for handler in loaded.handlers
+                if isinstance(handler.descriptor.trigger, CommandTrigger)
+            }
+            assert {
+                "admin",
+                "admin_type",
+                "ai",
+                "chain",
+                "components",
+                "conversation",
+                "db",
+                "echo",
+                "hello",
+                "sendmsg",
+                "state",
+            } <= command_names
+
+            regex_triggers = [
+                handler.descriptor.trigger
+                for handler in loaded.handlers
+                if isinstance(handler.descriptor.trigger, MessageTrigger)
+            ]
+            assert any(trigger.regex == r"^ping.*" for trigger in regex_triggers)
 
             # 验证实例类型
             instance = loaded.instances[0]

tests_v4/test_new_plugin_integration.py

@@ -7,7 +7,12 @@ from pathlib import Path
 
 import pytest
 
-from astrbot_sdk.protocol.descriptors import CommandTrigger, MessageTrigger
+from astrbot_sdk.protocol.descriptors import (
+    CommandTrigger,
+    EventTrigger,
+    MessageTrigger,
+    ScheduleTrigger,
+)
 from astrbot_sdk.runtime.loader import load_plugin, load_plugin_spec
 
 
@@ -49,22 +54,59 @@ class TestRealNewTestPlugin:
                 for handler in loaded.handlers
                 if isinstance(handler.descriptor.trigger, MessageTrigger)
             ]
+            event_triggers = [
+                handler.descriptor.trigger
+                for handler in loaded.handlers
+                if isinstance(handler.descriptor.trigger, EventTrigger)
+            ]
+            schedule_triggers = [
+                handler.descriptor.trigger
+                for handler in loaded.handlers
+                if isinstance(handler.descriptor.trigger, ScheduleTrigger)
+            ]
+            handler_map = {
+                handler.descriptor.id: handler.descriptor for handler in loaded.handlers
+            }
 
             assert {trigger.command for trigger in command_triggers} == {
                 "announce",
                 "hello",
+                "platforms",
+                "raw",
                 "remember",
+                "secure",
             }
             hello_trigger = next(
                 trigger for trigger in command_triggers if trigger.command == "hello"
             )
             assert "hi" in hello_trigger.aliases
+            secure_descriptor = next(
+                descriptor
+                for descriptor in handler_map.values()
+                if getattr(descriptor.trigger, "command", None) == "secure"
+            )
+            assert secure_descriptor.permissions.require_admin is True
 
             assert len(message_triggers) == 1
             assert message_triggers[0].regex == r"^ping$"
+            assert message_triggers[0].keywords == ["ping"]
+            assert message_triggers[0].platforms == ["test"]
+
+            assert len(event_triggers) == 1
+            assert event_triggers[0].event_type == "group_join"
+
+            assert len(schedule_triggers) == 1
+            assert schedule_triggers[0].interval_seconds == 60
 
             capability_names = [item.descriptor.name for item in loaded.capabilities]
-            assert capability_names == ["demo.echo"]
+            assert capability_names == ["demo.echo", "demo.stream"]
+            stream_descriptor = next(
+                item.descriptor
+                for item in loaded.capabilities
+                if item.descriptor.name == "demo.stream"
+            )
+            assert stream_descriptor.supports_stream is True
+            assert stream_descriptor.cancelable is True
         finally:
             for path in paths_to_add:
                 if path in sys.path:

tests_v4/test_runtime.py

@@ -39,6 +39,13 @@ class RuntimeIntegrationTest(unittest.IsolatedAsyncioTestCase):
     async def asyncTearDown(self) -> None:
         await self.core.stop()
 
+    def _find_handler_id(self, command_name: str) -> str:
+        return next(
+            handler.id
+            for handler in self.core.remote_handlers
+            if getattr(handler.trigger, "command", None) == command_name
+        )
+
     async def test_supervisor_runs_v4_plugin_over_stdio_worker(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             plugins_root = Path(temp_dir) / "plugins"
@@ -53,11 +60,7 @@ class RuntimeIntegrationTest(unittest.IsolatedAsyncioTestCase):
             try:
                 await runtime.start()
                 await self.core.wait_until_remote_initialized()
-                handler_id = next(
-                    handler.id
-                    for handler in self.core.remote_handlers
-                    if getattr(handler.trigger, "command", None) == "hello"
-                )
+                handler_id = self._find_handler_id("hello")
 
                 await self.core.invoke(
                     "handler.invoke",
@@ -79,6 +82,77 @@ class RuntimeIntegrationTest(unittest.IsolatedAsyncioTestCase):
             finally:
                 await runtime.stop()
 
+    async def test_supervisor_runs_v4_plugin_client_commands(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_root = Path(temp_dir) / "plugins"
+            plugin_root = plugins_root / "v4_plugin"
+            shutil.copytree(sample_plugin_dir("new"), plugin_root)
+
+            runtime = SupervisorRuntime(
+                transport=self.right,
+                plugins_dir=plugins_root,
+                env_manager=FakeEnvManager(),
+            )
+            try:
+                await runtime.start()
+                await self.core.wait_until_remote_initialized()
+
+                for request_id, command_name in (
+                    ("call-v4-raw", "raw"),
+                    ("call-v4-remember", "remember"),
+                    ("call-v4-platforms", "platforms"),
+                ):
+                    await self.core.invoke(
+                        "handler.invoke",
+                        {
+                            "handler_id": self._find_handler_id(command_name),
+                            "event": {
+                                "text": command_name,
+                                "session_id": "session-v4-clients",
+                                "user_id": "user-1",
+                                "platform": "test",
+                            },
+                        },
+                        request_id=request_id,
+                    )
+
+                texts = [
+                    item.get("text")
+                    for item in runtime.capability_router.sent_messages
+                    if "text" in item
+                ]
+                self.assertTrue(
+                    any(text.startswith("raw=Echo: raw|finish=stop|") for text in texts)
+                )
+                self.assertTrue(
+                    any(
+                        text.startswith(
+                            "remembered=user-1|searched=1|session=session-v4-clients|keys=1"
+                        )
+                        for text in texts
+                    )
+                )
+                image_message = next(
+                    item
+                    for item in runtime.capability_router.sent_messages
+                    if item.get("image_url") == "https://example.com/demo.png"
+                )
+                self.assertEqual(image_message["session"], "session-v4-clients")
+                self.assertEqual(
+                    image_message["target"]["conversation_id"],
+                    "session-v4-clients",
+                )
+                self.assertTrue(
+                    any(
+                        item.get("text", "").startswith(
+                            "members=2 first=session-v4-clients:member-1"
+                        )
+                        for item in runtime.capability_router.sent_messages
+                    )
+                )
+            finally:
+                await runtime.stop()
+
     async def test_supervisor_exposes_real_v4_plugin_capability(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             plugins_root = Path(temp_dir) / "plugins"
@@ -99,6 +173,7 @@ class RuntimeIntegrationTest(unittest.IsolatedAsyncioTestCase):
                     for descriptor in self.core.remote_provided_capabilities
                 }
                 self.assertIn("demo.echo", capability_names)
+                self.assertIn("demo.stream", capability_names)
 
                 result = await self.core.invoke(
                     "demo.echo",
@@ -129,11 +204,7 @@ class RuntimeIntegrationTest(unittest.IsolatedAsyncioTestCase):
             try:
                 await runtime.start()
                 await self.core.wait_until_remote_initialized()
-                handler_id = next(
-                    handler.id
-                    for handler in self.core.remote_handlers
-                    if getattr(handler.trigger, "command", None) == "announce"
-                )
+                handler_id = self._find_handler_id("announce")
 
                 await self.core.invoke(
                     "handler.invoke",
@@ -162,6 +233,31 @@ class RuntimeIntegrationTest(unittest.IsolatedAsyncioTestCase):
             finally:
                 await runtime.stop()
 
+    async def test_supervisor_exposes_real_v4_stream_capability(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_root = Path(temp_dir) / "plugins"
+            plugin_root = plugins_root / "v4_plugin"
+            shutil.copytree(sample_plugin_dir("new"), plugin_root)
+
+            runtime = SupervisorRuntime(
+                transport=self.right,
+                plugins_dir=plugins_root,
+                env_manager=FakeEnvManager(),
+            )
+            try:
+                await runtime.start()
+                await self.core.wait_until_remote_initialized()
+
+                stream = await self.core.invoke_stream(
+                    "demo.stream",
+                    {"text": "abc"},
+                    request_id="call-v4-stream-capability",
+                )
+                chunks = [event.data["text"] async for event in stream]
+                self.assertEqual(chunks, ["a", "b", "c"])
+            finally:
+                await runtime.stop()
+
     async def test_supervisor_runs_compat_plugin(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             plugins_root = Path(temp_dir) / "plugins"
@@ -176,11 +272,7 @@ class RuntimeIntegrationTest(unittest.IsolatedAsyncioTestCase):
             try:
                 await runtime.start()
                 await self.core.wait_until_remote_initialized()
-                handler_id = next(
-                    handler.id
-                    for handler in self.core.remote_handlers
-                    if getattr(handler.trigger, "command", None) == "hello"
-                )
+                handler_id = self._find_handler_id("hello")
 
                 await self.core.invoke(
                     "handler.invoke",
@@ -202,3 +294,122 @@ class RuntimeIntegrationTest(unittest.IsolatedAsyncioTestCase):
                 self.assertIn("Created conversation ID", texts[0])
             finally:
                 await runtime.stop()
+
+    async def test_supervisor_runs_compat_plugin_extended_api_commands(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_root = Path(temp_dir) / "plugins"
+            plugin_root = plugins_root / "compat_plugin"
+            shutil.copytree(Path.cwd() / "test_plugin" / "old", plugin_root)
+
+            runtime = SupervisorRuntime(
+                transport=self.right,
+                plugins_dir=plugins_root,
+                env_manager=FakeEnvManager(),
+            )
+            try:
+                await runtime.start()
+                await self.core.wait_until_remote_initialized()
+
+                for request_id, command_name in (
+                    ("call-compat-ai", "ai"),
+                    ("call-compat-conversation", "conversation"),
+                    ("call-compat-sendmsg", "sendmsg"),
+                    ("call-compat-chain", "chain"),
+                    ("call-compat-components", "components"),
+                ):
+                    await self.core.invoke(
+                        "handler.invoke",
+                        {
+                            "handler_id": self._find_handler_id(command_name),
+                            "event": {
+                                "text": command_name,
+                                "session_id": "session-compat-extended",
+                                "user_id": "user-1",
+                                "platform": "test",
+                            },
+                        },
+                        request_id=request_id,
+                    )
+
+                texts = [
+                    item.get("text")
+                    for item in runtime.capability_router.sent_messages
+                    if "text" in item
+                ]
+                self.assertTrue(
+                    any(
+                        text.startswith(
+                            "LLM:Echo: legacy hello|AGENT:Echo: legacy hello"
+                        )
+                        for text in texts
+                    )
+                )
+                self.assertTrue(
+                    any(
+                        text.startswith("conversation=") and "|helper=COMPAT" in text
+                        for text in texts
+                    )
+                )
+                self.assertTrue(any(text == "send_message invoked" for text in texts))
+                chain_messages = [
+                    item
+                    for item in runtime.capability_router.sent_messages
+                    if "chain" in item
+                ]
+                self.assertTrue(
+                    any(
+                        any(
+                            component.get("type") == "At"
+                            and component.get("user_id") == "all"
+                            for component in item["chain"]
+                        )
+                        for item in chain_messages
+                    )
+                )
+                self.assertTrue(
+                    any(
+                        any(
+                            component.get("type") == "Node"
+                            for component in item["chain"]
+                        )
+                        for item in chain_messages
+                    )
+                )
+            finally:
+                await runtime.stop()
+
+    async def test_supervisor_exposes_compat_plugin_capability(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_root = Path(temp_dir) / "plugins"
+            plugin_root = plugins_root / "compat_plugin"
+            shutil.copytree(Path.cwd() / "test_plugin" / "old", plugin_root)
+
+            runtime = SupervisorRuntime(
+                transport=self.right,
+                plugins_dir=plugins_root,
+                env_manager=FakeEnvManager(),
+            )
+            try:
+                await runtime.start()
+                await self.core.wait_until_remote_initialized()
+
+                capability_names = {
+                    descriptor.name
+                    for descriptor in self.core.remote_provided_capabilities
+                }
+                self.assertIn("compat.echo", capability_names)
+
+                result = await self.core.invoke(
+                    "compat.echo",
+                    {"text": "legacy-capability"},
+                    request_id="call-compat-capability",
+                )
+                self.assertEqual(
+                    result,
+                    {
+                        "echo": "legacy-capability",
+                        "plugin_id": "astrbot_plugin_helloworld",
+                    },
+                )
+            finally:
+                await runtime.stop()