Add unit tests for protocol messages and transport implementations

- Created `test_protocol_messages.py` to cover tests for protocol message models including ErrorPayload, PeerInfo, InitializeMessage, ResultMessage, InvokeMessage, EventMessage, CancelMessage, and the parse_message function.
- Implemented validation tests to ensure required fields and serialization behavior.
- Added `test_transport.py` to test the Transport base class and its implementations: StdioTransport and WebSocketTransport.
- Included tests for lifecycle methods, message handling, and error conditions in transport classes.

tests_v4/test_capability_router.py

@@ -0,0 +1,820 @@
+"""
+Tests for runtime/capability_router.py - CapabilityRouter implementation.
+"""
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from astrbot_sdk.context import CancelToken
+from astrbot_sdk.errors import AstrBotError
+from astrbot_sdk.protocol.descriptors import CapabilityDescriptor
+from astrbot_sdk.runtime.capability_router import (
+    CAPABILITY_NAME_PATTERN,
+    RESERVED_CAPABILITY_PREFIXES,
+    CallHandler,
+    FinalizeHandler,
+    StreamExecution,
+    StreamHandler,
+    _CapabilityRegistration,
+)
+from astrbot_sdk.runtime.capability_router import CapabilityRouter
+
+
+class TestStreamExecution:
+    """Tests for StreamExecution dataclass."""
+
+    def test_init(self):
+        """StreamExecution should store iterator and finalize."""
+        async def gen():
+            yield {"text": "a"}
+
+        def fin(chunks):
+            return {"count": len(chunks)}
+
+        execution = StreamExecution(iterator=gen(), finalize=fin)
+        assert execution.iterator is not None
+        assert execution.finalize is fin
+
+
+class TestCapabilityRegistration:
+    """Tests for _CapabilityRegistration dataclass."""
+
+    def test_init(self):
+        """_CapabilityRegistration should store all fields."""
+        descriptor = CapabilityDescriptor(name="test.cap", description="Test")
+        call_handler = AsyncMock()
+
+        reg = _CapabilityRegistration(
+            descriptor=descriptor,
+            call_handler=call_handler,
+            exposed=True,
+        )
+
+        assert reg.descriptor == descriptor
+        assert reg.call_handler == call_handler
+        assert reg.stream_handler is None
+        assert reg.finalize is None
+        assert reg.exposed is True
+
+    def test_defaults(self):
+        """_CapabilityRegistration should have correct defaults."""
+        descriptor = CapabilityDescriptor(name="test.cap", description="Test")
+
+        reg = _CapabilityRegistration(descriptor=descriptor)
+
+        assert reg.call_handler is None
+        assert reg.stream_handler is None
+        assert reg.finalize is None
+        assert reg.exposed is True
+
+
+class TestCapabilityNamePattern:
+    """Tests for capability name validation pattern."""
+
+    def test_valid_names(self):
+        """Valid capability names should match pattern."""
+        valid_names = [
+            "llm.chat",
+            "db.get",
+            "memory.search",
+            "platform.send",
+            "a.b",
+            "my_module.my_method",
+            "ns123.method456",
+        ]
+        for name in valid_names:
+            assert CAPABILITY_NAME_PATTERN.fullmatch(name), f"{name} should be valid"
+
+    def test_invalid_names(self):
+        """Invalid capability names should not match pattern."""
+        invalid_names = [
+            "llm",  # No dot
+            "LLM.chat",  # Uppercase
+            "llm.Chat",  # Uppercase method
+            "llm.chat.extra",  # Too many parts
+            "1llm.chat",  # Starts with number
+            "llm.1chat",  # Method starts with number
+            ".chat",  # Empty namespace
+            "llm.",  # Empty method
+            "llm-chat",  # Hyphen instead of dot
+        ]
+        for name in invalid_names:
+            assert not CAPABILITY_NAME_PATTERN.fullmatch(name), f"{name} should be invalid"
+
+
+class TestReservedCapabilityPrefixes:
+    """Tests for reserved capability prefixes."""
+
+    def test_reserved_prefixes(self):
+        """Reserved prefixes should be defined."""
+        assert "handler." in RESERVED_CAPABILITY_PREFIXES
+        assert "system." in RESERVED_CAPABILITY_PREFIXES
+        assert "internal." in RESERVED_CAPABILITY_PREFIXES
+
+    def test_reserved_names_are_detected(self):
+        """Reserved names should be detected by startswith."""
+        reserved_names = [
+            "handler.demo",
+            "system.health",
+            "internal.trace",
+        ]
+        for name in reserved_names:
+            assert any(
+                name.startswith(prefix) for prefix in RESERVED_CAPABILITY_PREFIXES
+            ), f"{name} should be reserved"
+
+
+class TestCapabilityRouterInit:
+    """Tests for CapabilityRouter initialization."""
+
+    def test_init_creates_empty_registrations(self):
+        """CapabilityRouter should start with empty registrations."""
+        router = CapabilityRouter()
+        assert router._registrations == {}
+        assert router.db_store == {}
+        assert router.memory_store == {}
+        assert router.sent_messages == []
+
+    def test_init_registers_builtin_capabilities(self):
+        """CapabilityRouter should register built-in capabilities on init."""
+        router = CapabilityRouter()
+        descriptors = router.descriptors()
+
+        capability_names = [d.name for d in descriptors]
+
+        # LLM capabilities
+        assert "llm.chat" in capability_names
+        assert "llm.chat_raw" in capability_names
+        assert "llm.stream_chat" in capability_names
+
+        # Memory capabilities
+        assert "memory.search" in capability_names
+        assert "memory.save" in capability_names
+        assert "memory.delete" in capability_names
+
+        # DB capabilities
+        assert "db.get" in capability_names
+        assert "db.set" in capability_names
+        assert "db.delete" in capability_names
+        assert "db.list" in capability_names
+
+        # Platform capabilities
+        assert "platform.send" in capability_names
+        assert "platform.send_image" in capability_names
+        assert "platform.get_members" in capability_names
+
+
+class TestCapabilityRouterRegister:
+    """Tests for CapabilityRouter.register method."""
+
+    def test_register_adds_capability(self):
+        """register should add capability to registrations."""
+        router = CapabilityRouter()
+        descriptor = CapabilityDescriptor(name="test.cap", description="Test")
+
+        router.register(descriptor)
+
+        assert "test.cap" in router._registrations
+        assert router._registrations["test.cap"].descriptor == descriptor
+
+    def test_register_with_handlers(self):
+        """register should store handlers."""
+        router = CapabilityRouter()
+        descriptor = CapabilityDescriptor(name="test.cap", description="Test")
+
+        async def call_handler(req_id, payload, token):
+            return {"result": "ok"}
+
+        async def stream_handler(req_id, payload, token):
+            yield {"chunk": 1}
+
+        def finalize(chunks):
+            return {"count": len(chunks)}
+
+        router.register(
+            descriptor,
+            call_handler=call_handler,
+            stream_handler=stream_handler,
+            finalize=finalize,
+        )
+
+        reg = router._registrations["test.cap"]
+        assert reg.call_handler == call_handler
+        assert reg.stream_handler == stream_handler
+        assert reg.finalize == finalize
+
+    def test_register_invalid_name_raises(self):
+        """register should reject invalid capability names."""
+        router = CapabilityRouter()
+
+        with pytest.raises(ValueError, match="capability 名称必须匹配"):
+            router.register(CapabilityDescriptor(name="invalid", description="Bad"))
+
+    def test_register_reserved_name_raises(self):
+        """register should reject reserved names for exposed registrations."""
+        router = CapabilityRouter()
+
+        with pytest.raises(ValueError, match="保留 capability"):
+            router.register(
+                CapabilityDescriptor(name="handler.demo", description="Reserved")
+            )
+
+    def test_register_reserved_name_allowed_for_internal(self):
+        """register should allow reserved names for internal (exposed=False)."""
+        router = CapabilityRouter()
+
+        # Should not raise
+        router.register(
+            CapabilityDescriptor(name="handler.internal", description="Internal"),
+            exposed=False,
+        )
+
+        # Should not appear in descriptors
+        names = [d.name for d in router.descriptors()]
+        assert "handler.internal" not in names
+
+    def test_descriptors_only_returns_exposed(self):
+        """descriptors should only return exposed capabilities."""
+        router = CapabilityRouter()
+
+        router.register(
+            CapabilityDescriptor(name="exposed.cap", description="Exposed"),
+            exposed=True,
+        )
+        router.register(
+            CapabilityDescriptor(name="hidden.cap", description="Hidden"),
+            exposed=False,
+        )
+
+        names = [d.name for d in router.descriptors()]
+        assert "exposed.cap" in names
+        assert "hidden.cap" not in names
+
+
+class TestCapabilityRouterExecute:
+    """Tests for CapabilityRouter.execute method."""
+
+    @pytest.mark.asyncio
+    async def test_execute_calls_handler(self):
+        """execute should call the registered handler."""
+        router = CapabilityRouter()
+        router.register(
+            CapabilityDescriptor(name="test.cap", description="Test"),
+            call_handler=AsyncMock(return_value={"result": "ok"}),
+        )
+
+        token = CancelToken()
+        result = await router.execute(
+            "test.cap",
+            {},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert result == {"result": "ok"}
+
+    @pytest.mark.asyncio
+    async def test_execute_validates_input_schema(self):
+        """execute should validate input against schema."""
+        router = CapabilityRouter()
+        router.register(
+            CapabilityDescriptor(
+                name="test.cap",
+                description="Test",
+                input_schema={
+                    "type": "object",
+                    "properties": {"name": {"type": "string"}},
+                    "required": ["name"],
+                },
+            ),
+            call_handler=AsyncMock(return_value={}),
+        )
+
+        token = CancelToken()
+
+        # Missing required field
+        with pytest.raises(AstrBotError, match="缺少必填字段"):
+            await router.execute(
+                "test.cap",
+                {},
+                stream=False,
+                cancel_token=token,
+                request_id="req-1",
+            )
+
+    @pytest.mark.asyncio
+    async def test_execute_validates_output_schema(self):
+        """execute should validate output against schema."""
+        router = CapabilityRouter()
+        router.register(
+            CapabilityDescriptor(
+                name="test.cap",
+                description="Test",
+                output_schema={
+                    "type": "object",
+                    "properties": {"result": {"type": "string"}},
+                    "required": ["result"],
+                },
+            ),
+            call_handler=AsyncMock(return_value={}),  # Missing required field
+        )
+
+        token = CancelToken()
+
+        with pytest.raises(AstrBotError, match="缺少必填字段"):
+            await router.execute(
+                "test.cap",
+                {},
+                stream=False,
+                cancel_token=token,
+                request_id="req-1",
+            )
+
+    @pytest.mark.asyncio
+    async def test_execute_missing_capability_raises(self):
+        """execute should raise for unknown capability."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        with pytest.raises(AstrBotError, match="capability_not_found"):
+            await router.execute(
+                "unknown.cap",
+                {},
+                stream=False,
+                cancel_token=token,
+                request_id="req-1",
+            )
+
+    @pytest.mark.asyncio
+    async def test_execute_stream_returns_stream_execution(self):
+        """execute with stream=True should return StreamExecution."""
+        router = CapabilityRouter()
+
+        async def stream_handler(req_id, payload, token):
+            yield {"chunk": 1}
+            yield {"chunk": 2}
+
+        router.register(
+            CapabilityDescriptor(
+                name="test.stream",
+                description="Test",
+                supports_stream=True,
+            ),
+            stream_handler=stream_handler,
+        )
+
+        token = CancelToken()
+        result = await router.execute(
+            "test.stream",
+            {},
+            stream=True,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert isinstance(result, StreamExecution)
+
+    @pytest.mark.asyncio
+    async def test_execute_stream_without_handler_raises(self):
+        """execute with stream=True and no stream_handler should raise."""
+        router = CapabilityRouter()
+        router.register(
+            CapabilityDescriptor(name="test.cap", description="Test"),
+            call_handler=AsyncMock(return_value={}),
+        )
+
+        token = CancelToken()
+
+        with pytest.raises(AstrBotError, match="不支持 stream=true"):
+            await router.execute(
+                "test.cap",
+                {},
+                stream=True,
+                cancel_token=token,
+                request_id="req-1",
+            )
+
+    @pytest.mark.asyncio
+    async def test_execute_call_without_handler_raises(self):
+        """execute without stream and no call_handler should raise."""
+        router = CapabilityRouter()
+        router.register(
+            CapabilityDescriptor(
+                name="test.stream_only",
+                description="Stream only",
+                supports_stream=True,
+            ),
+            stream_handler=AsyncMock(),
+        )
+
+        token = CancelToken()
+
+        with pytest.raises(AstrBotError, match="只能以 stream=true 调用"):
+            await router.execute(
+                "test.stream_only",
+                {},
+                stream=False,
+                cancel_token=token,
+                request_id="req-1",
+            )
+
+
+class TestBuiltinLlmCapabilities:
+    """Tests for built-in LLM capabilities."""
+
+    @pytest.mark.asyncio
+    async def test_llm_chat(self):
+        """llm.chat should return echo response."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        result = await router.execute(
+            "llm.chat",
+            {"prompt": "hello"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert result["text"] == "Echo: hello"
+
+    @pytest.mark.asyncio
+    async def test_llm_chat_raw(self):
+        """llm.chat_raw should return full response."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        result = await router.execute(
+            "llm.chat_raw",
+            {"prompt": "test"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert result["text"] == "Echo: test"
+        assert "usage" in result
+        assert "finish_reason" in result
+        assert "tool_calls" in result
+
+    @pytest.mark.asyncio
+    async def test_llm_stream_chat(self):
+        """llm.stream_chat should yield characters."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        result = await router.execute(
+            "llm.stream_chat",
+            {"prompt": "hi"},
+            stream=True,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert isinstance(result, StreamExecution)
+
+        chunks = []
+        async for chunk in result.iterator:
+            chunks.append(chunk)
+
+        # Should yield each character
+        text = "".join(c.get("text", "") for c in chunks)
+        assert text == "Echo: hi"
+
+
+class TestBuiltinMemoryCapabilities:
+    """Tests for built-in memory capabilities."""
+
+    @pytest.mark.asyncio
+    async def test_memory_save_and_search(self):
+        """memory.save and memory.search should work together."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        # Save
+        await router.execute(
+            "memory.save",
+            {"key": "test_key", "value": {"data": "test_value"}},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        # Search
+        result = await router.execute(
+            "memory.search",
+            {"query": "test"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-2",
+        )
+
+        assert len(result["items"]) == 1
+        assert result["items"][0]["key"] == "test_key"
+
+    @pytest.mark.asyncio
+    async def test_memory_delete(self):
+        """memory.delete should remove saved memory."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        # Save
+        await router.execute(
+            "memory.save",
+            {"key": "to_delete", "value": {"data": "value"}},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        # Delete
+        await router.execute(
+            "memory.delete",
+            {"key": "to_delete"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-2",
+        )
+
+        # Search should return empty
+        result = await router.execute(
+            "memory.search",
+            {"query": "to_delete"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-3",
+        )
+
+        assert len(result["items"]) == 0
+
+    @pytest.mark.asyncio
+    async def test_memory_save_invalid_value(self):
+        """memory.save should reject non-object value."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        with pytest.raises(AstrBotError, match="value 必须是 object"):
+            await router.execute(
+                "memory.save",
+                {"key": "test", "value": "not_an_object"},
+                stream=False,
+                cancel_token=token,
+                request_id="req-1",
+            )
+
+
+class TestBuiltinDbCapabilities:
+    """Tests for built-in DB capabilities."""
+
+    @pytest.mark.asyncio
+    async def test_db_set_and_get(self):
+        """db.set and db.get should work together."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        # Set
+        await router.execute(
+            "db.set",
+            {"key": "test_key", "value": {"data": "test_value"}},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        # Get
+        result = await router.execute(
+            "db.get",
+            {"key": "test_key"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-2",
+        )
+
+        assert result["value"] == {"data": "test_value"}
+
+    @pytest.mark.asyncio
+    async def test_db_get_missing_key(self):
+        """db.get should return None for missing key."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        result = await router.execute(
+            "db.get",
+            {"key": "nonexistent"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert result["value"] is None
+
+    @pytest.mark.asyncio
+    async def test_db_delete(self):
+        """db.delete should remove stored value."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        # Set
+        await router.execute(
+            "db.set",
+            {"key": "to_delete", "value": {"data": "value"}},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        # Delete
+        await router.execute(
+            "db.delete",
+            {"key": "to_delete"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-2",
+        )
+
+        # Get should return None
+        result = await router.execute(
+            "db.get",
+            {"key": "to_delete"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-3",
+        )
+
+        assert result["value"] is None
+
+    @pytest.mark.asyncio
+    async def test_db_list(self):
+        """db.list should return keys."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        # Set multiple keys
+        await router.execute(
+            "db.set",
+            {"key": "prefix_a", "value": {}},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+        await router.execute(
+            "db.set",
+            {"key": "prefix_b", "value": {}},
+            stream=False,
+            cancel_token=token,
+            request_id="req-2",
+        )
+        await router.execute(
+            "db.set",
+            {"key": "other", "value": {}},
+            stream=False,
+            cancel_token=token,
+            request_id="req-3",
+        )
+
+        # List all
+        result = await router.execute(
+            "db.list",
+            {},
+            stream=False,
+            cancel_token=token,
+            request_id="req-4",
+        )
+
+        assert "prefix_a" in result["keys"]
+        assert "prefix_b" in result["keys"]
+        assert "other" in result["keys"]
+
+        # List with prefix
+        result = await router.execute(
+            "db.list",
+            {"prefix": "prefix_"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-5",
+        )
+
+        assert "prefix_a" in result["keys"]
+        assert "prefix_b" in result["keys"]
+        assert "other" not in result["keys"]
+
+    @pytest.mark.asyncio
+    async def test_db_set_invalid_value(self):
+        """db.set should reject non-object value."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        with pytest.raises(AstrBotError, match="value 必须是 object"):
+            await router.execute(
+                "db.set",
+                {"key": "test", "value": "not_an_object"},
+                stream=False,
+                cancel_token=token,
+                request_id="req-1",
+            )
+
+
+class TestBuiltinPlatformCapabilities:
+    """Tests for built-in platform capabilities."""
+
+    @pytest.mark.asyncio
+    async def test_platform_send(self):
+        """platform.send should store message."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        result = await router.execute(
+            "platform.send",
+            {"session": "session-1", "text": "Hello"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert "message_id" in result
+        assert len(router.sent_messages) == 1
+        assert router.sent_messages[0]["session"] == "session-1"
+        assert router.sent_messages[0]["text"] == "Hello"
+
+    @pytest.mark.asyncio
+    async def test_platform_send_image(self):
+        """platform.send_image should store image message."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        result = await router.execute(
+            "platform.send_image",
+            {"session": "session-1", "image_url": "http://example.com/image.png"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert "message_id" in result
+        assert len(router.sent_messages) == 1
+        assert router.sent_messages[0]["image_url"] == "http://example.com/image.png"
+
+    @pytest.mark.asyncio
+    async def test_platform_get_members(self):
+        """platform.get_members should return mock members."""
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        result = await router.execute(
+            "platform.get_members",
+            {"session": "session-1"},
+            stream=False,
+            cancel_token=token,
+            request_id="req-1",
+        )
+
+        assert len(result["members"]) == 2
+        assert result["members"][0]["user_id"] == "session-1:member-1"
+
+
+class TestValidateSchema:
+    """Tests for _validate_schema method."""
+
+    @pytest.mark.asyncio
+    async def test_none_schema_passes(self):
+        """_validate_schema with None schema should pass."""
+        router = CapabilityRouter()
+
+        # Should not raise
+        router._validate_schema(None, {"any": "data"})
+
+    @pytest.mark.asyncio
+    async def test_non_object_payload_raises(self):
+        """_validate_schema should reject non-object when schema type is object."""
+        router = CapabilityRouter()
+
+        with pytest.raises(AstrBotError, match="输入必须是 object"):
+            router._validate_schema({"type": "object"}, "not an object")
+
+    @pytest.mark.asyncio
+    async def test_missing_required_field_raises(self):
+        """_validate_schema should reject missing required fields."""
+        router = CapabilityRouter()
+
+        with pytest.raises(AstrBotError, match="缺少必填字段"):
+            router._validate_schema(
+                {"type": "object", "required": ["name"]},
+                {},
+            )
+
+    @pytest.mark.asyncio
+    async def test_none_required_field_raises(self):
+        """_validate_schema should reject None required fields."""
+        router = CapabilityRouter()
+
+        with pytest.raises(AstrBotError, match="缺少必填字段"):
+            router._validate_schema(
+                {"type": "object", "required": ["name"]},
+                {"name": None},
+            )

tests_v4/test_protocol_descriptors.py

@@ -0,0 +1,341 @@
+"""
+Tests for protocol/descriptors.py - Descriptor models.
+"""
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from astrbot_sdk.protocol.descriptors import (
+    CapabilityDescriptor,
+    CommandTrigger,
+    EventTrigger,
+    HandlerDescriptor,
+    MessageTrigger,
+    Permissions,
+    ScheduleTrigger,
+)
+
+
+class TestPermissions:
+    """Tests for Permissions model."""
+
+    def test_default_values(self):
+        """Permissions should have default values."""
+        perms = Permissions()
+        assert perms.require_admin is False
+        assert perms.level == 0
+
+    def test_custom_values(self):
+        """Permissions should accept custom values."""
+        perms = Permissions(require_admin=True, level=5)
+        assert perms.require_admin is True
+        assert perms.level == 5
+
+    def test_model_dump(self):
+        """Permissions should serialize correctly."""
+        perms = Permissions(require_admin=True, level=10)
+        data = perms.model_dump()
+        assert data == {"require_admin": True, "level": 10}
+
+    def test_extra_fields_forbidden(self):
+        """Permissions should forbid extra fields."""
+        with pytest.raises(ValidationError):
+            Permissions(require_admin=True, unknown_field="value")
+
+
+class TestCommandTrigger:
+    """Tests for CommandTrigger model."""
+
+    def test_required_command(self):
+        """CommandTrigger requires command field."""
+        trigger = CommandTrigger(command="hello")
+        assert trigger.type == "command"
+        assert trigger.command == "hello"
+        assert trigger.aliases == []
+        assert trigger.description is None
+
+    def test_with_aliases_and_description(self):
+        """CommandTrigger should accept aliases and description."""
+        trigger = CommandTrigger(
+            command="hello",
+            aliases=["hi", "hey"],
+            description="Say hello",
+        )
+        assert trigger.command == "hello"
+        assert trigger.aliases == ["hi", "hey"]
+        assert trigger.description == "Say hello"
+
+    def test_type_literal(self):
+        """CommandTrigger type should always be 'command'."""
+        trigger = CommandTrigger(command="test")
+        assert trigger.type == "command"
+
+    def test_extra_fields_forbidden(self):
+        """CommandTrigger should forbid extra fields."""
+        with pytest.raises(ValidationError):
+            CommandTrigger(command="test", extra="field")
+
+
+class TestMessageTrigger:
+    """Tests for MessageTrigger model."""
+
+    def test_default_values(self):
+        """MessageTrigger should have default values."""
+        trigger = MessageTrigger()
+        assert trigger.type == "message"
+        assert trigger.regex is None
+        assert trigger.keywords == []
+        assert trigger.platforms == []
+
+    def test_with_regex(self):
+        """MessageTrigger should accept regex pattern."""
+        trigger = MessageTrigger(regex=r"^hello.*$")
+        assert trigger.regex == r"^hello.*$"
+
+    def test_with_keywords(self):
+        """MessageTrigger should accept keywords."""
+        trigger = MessageTrigger(keywords=["hello", "hi"])
+        assert trigger.keywords == ["hello", "hi"]
+
+    def test_with_platforms(self):
+        """MessageTrigger should accept platforms."""
+        trigger = MessageTrigger(platforms=["wechat", "qq"])
+        assert trigger.platforms == ["wechat", "qq"]
+
+    def test_with_all_fields(self):
+        """MessageTrigger should accept all fields."""
+        trigger = MessageTrigger(
+            regex=r"test",
+            keywords=["keyword"],
+            platforms=["platform"],
+        )
+        assert trigger.regex == "test"
+        assert trigger.keywords == ["keyword"]
+        assert trigger.platforms == ["platform"]
+
+
+class TestEventTrigger:
+    """Tests for EventTrigger model."""
+
+    def test_required_event_type(self):
+        """EventTrigger requires event_type field."""
+        trigger = EventTrigger(event_type="message")
+        assert trigger.type == "event"
+        assert trigger.event_type == "message"
+
+    def test_numeric_event_type(self):
+        """EventTrigger should accept numeric string event_type."""
+        trigger = EventTrigger(event_type="3")
+        assert trigger.event_type == "3"
+
+    def test_type_literal(self):
+        """EventTrigger type should always be 'event'."""
+        trigger = EventTrigger(event_type="custom")
+        assert trigger.type == "event"
+
+
+class TestScheduleTrigger:
+    """Tests for ScheduleTrigger model."""
+
+    def test_with_cron(self):
+        """ScheduleTrigger should accept cron expression."""
+        trigger = ScheduleTrigger(cron="0 9 * * *")
+        assert trigger.type == "schedule"
+        assert trigger.cron == "0 9 * * *"
+        assert trigger.interval_seconds is None
+
+    def test_with_interval_seconds(self):
+        """ScheduleTrigger should accept interval_seconds."""
+        trigger = ScheduleTrigger(interval_seconds=60)
+        assert trigger.interval_seconds == 60
+        assert trigger.cron is None
+
+    def test_requires_exactly_one_strategy(self):
+        """ScheduleTrigger must have exactly one of cron or interval_seconds."""
+        # Neither provided should raise
+        with pytest.raises(ValidationError) as exc_info:
+            ScheduleTrigger()
+        assert "必须且只能有一个非 null" in str(exc_info.value)
+
+        # Both provided should raise
+        with pytest.raises(ValidationError) as exc_info:
+            ScheduleTrigger(cron="* * * * *", interval_seconds=10)
+        assert "必须且只能有一个非 null" in str(exc_info.value)
+
+    def test_valid_cron_expressions(self):
+        """ScheduleTrigger should accept various cron expressions."""
+        trigger1 = ScheduleTrigger(cron="*/5 * * * *")
+        assert trigger1.cron == "*/5 * * * *"
+
+        trigger2 = ScheduleTrigger(cron="0 0 1 1 *")
+        assert trigger2.cron == "0 0 1 1 *"
+
+    def test_valid_intervals(self):
+        """ScheduleTrigger should accept various intervals."""
+        trigger1 = ScheduleTrigger(interval_seconds=30)
+        assert trigger1.interval_seconds == 30
+
+        trigger2 = ScheduleTrigger(interval_seconds=3600)
+        assert trigger2.interval_seconds == 3600
+
+
+class TestHandlerDescriptor:
+    """Tests for HandlerDescriptor model."""
+
+    def test_required_id_and_trigger(self):
+        """HandlerDescriptor requires id and trigger."""
+        trigger = CommandTrigger(command="hello")
+        handler = HandlerDescriptor(id="test.handler", trigger=trigger)
+        assert handler.id == "test.handler"
+        assert handler.trigger == trigger
+        assert handler.priority == 0
+        assert handler.permissions.require_admin is False
+
+    def test_with_priority_and_permissions(self):
+        """HandlerDescriptor should accept priority and permissions."""
+        trigger = CommandTrigger(command="admin")
+        perms = Permissions(require_admin=True, level=5)
+        handler = HandlerDescriptor(
+            id="admin.handler",
+            trigger=trigger,
+            priority=10,
+            permissions=perms,
+        )
+        assert handler.priority == 10
+        assert handler.permissions.require_admin is True
+        assert handler.permissions.level == 5
+
+    def test_with_event_trigger(self):
+        """HandlerDescriptor should work with EventTrigger."""
+        trigger = EventTrigger(event_type="message")
+        handler = HandlerDescriptor(id="event.handler", trigger=trigger)
+        assert handler.trigger.type == "event"
+        assert handler.trigger.event_type == "message"
+
+    def test_with_schedule_trigger(self):
+        """HandlerDescriptor should work with ScheduleTrigger."""
+        trigger = ScheduleTrigger(cron="0 9 * * *")
+        handler = HandlerDescriptor(id="scheduled.handler", trigger=trigger)
+        assert handler.trigger.type == "schedule"
+        assert handler.trigger.cron == "0 9 * * *"
+
+    def test_model_dump(self):
+        """HandlerDescriptor should serialize correctly."""
+        trigger = CommandTrigger(command="test", aliases=["t"])
+        perms = Permissions(require_admin=True, level=5)
+        handler = HandlerDescriptor(
+            id="test.handler",
+            trigger=trigger,
+            priority=10,
+            permissions=perms,
+        )
+        data = handler.model_dump()
+        assert data["id"] == "test.handler"
+        assert data["priority"] == 10
+        assert data["trigger"]["type"] == "command"
+        assert data["trigger"]["command"] == "test"
+        assert data["permissions"]["require_admin"] is True
+
+    def test_extra_fields_forbidden(self):
+        """HandlerDescriptor should forbid extra fields."""
+        trigger = CommandTrigger(command="test")
+        with pytest.raises(ValidationError):
+            HandlerDescriptor(id="test", trigger=trigger, extra="field")
+
+
+class TestCapabilityDescriptor:
+    """Tests for CapabilityDescriptor model."""
+
+    def test_required_name_and_description(self):
+        """CapabilityDescriptor requires name and description."""
+        cap = CapabilityDescriptor(name="llm.chat", description="Chat with LLM")
+        assert cap.name == "llm.chat"
+        assert cap.description == "Chat with LLM"
+        assert cap.input_schema is None
+        assert cap.output_schema is None
+        assert cap.supports_stream is False
+        assert cap.cancelable is False
+
+    def test_with_schemas(self):
+        """CapabilityDescriptor should accept input/output schemas."""
+        cap = CapabilityDescriptor(
+            name="db.query",
+            description="Query database",
+            input_schema={"type": "object", "properties": {"sql": {"type": "string"}}},
+            output_schema={"type": "array"},
+        )
+        assert cap.input_schema["type"] == "object"
+        assert cap.output_schema["type"] == "array"
+
+    def test_with_stream_and_cancelable(self):
+        """CapabilityDescriptor should accept stream and cancelable flags."""
+        cap = CapabilityDescriptor(
+            name="llm.stream",
+            description="Stream chat",
+            supports_stream=True,
+            cancelable=True,
+        )
+        assert cap.supports_stream is True
+        assert cap.cancelable is True
+
+    def test_model_dump(self):
+        """CapabilityDescriptor should serialize correctly."""
+        cap = CapabilityDescriptor(
+            name="test.cap",
+            description="Test capability",
+            supports_stream=True,
+        )
+        data = cap.model_dump()
+        assert data["name"] == "test.cap"
+        assert data["description"] == "Test capability"
+        assert data["supports_stream"] is True
+
+    def test_extra_fields_forbidden(self):
+        """CapabilityDescriptor should forbid extra fields."""
+        with pytest.raises(ValidationError):
+            CapabilityDescriptor(
+                name="test",
+                description="Test",
+                extra="field",
+            )
+
+
+class TestTriggerDiscriminator:
+    """Tests for Trigger type discriminator."""
+
+    def test_command_trigger_discrimination(self):
+        """CommandTrigger should be correctly discriminated."""
+        handler = HandlerDescriptor(
+            id="cmd.handler",
+            trigger={"type": "command", "command": "test"},
+        )
+        assert isinstance(handler.trigger, CommandTrigger)
+        assert handler.trigger.command == "test"
+
+    def test_message_trigger_discrimination(self):
+        """MessageTrigger should be correctly discriminated."""
+        handler = HandlerDescriptor(
+            id="msg.handler",
+            trigger={"type": "message", "keywords": ["hello"]},
+        )
+        assert isinstance(handler.trigger, MessageTrigger)
+        assert handler.trigger.keywords == ["hello"]
+
+    def test_event_trigger_discrimination(self):
+        """EventTrigger should be correctly discriminated."""
+        handler = HandlerDescriptor(
+            id="evt.handler",
+            trigger={"type": "event", "event_type": "message"},
+        )
+        assert isinstance(handler.trigger, EventTrigger)
+        assert handler.trigger.event_type == "message"
+
+    def test_schedule_trigger_discrimination(self):
+        """ScheduleTrigger should be correctly discriminated."""
+        handler = HandlerDescriptor(
+            id="sched.handler",
+            trigger={"type": "schedule", "cron": "0 9 * * *"},
+        )
+        assert isinstance(handler.trigger, ScheduleTrigger)
+        assert handler.trigger.cron == "0 9 * * *"

tests_v4/test_protocol_legacy_adapter.py

@@ -0,0 +1,768 @@
+"""
+Tests for protocol/legacy_adapter.py - Legacy protocol adapter.
+"""
+from __future__ import annotations
+
+import pytest
+
+from astrbot_sdk.protocol.descriptors import EventTrigger, HandlerDescriptor, Permissions
+from astrbot_sdk.protocol.legacy_adapter import (
+    LEGACY_ADAPTER_MESSAGE_EVENT,
+    LEGACY_CONTEXT_CAPABILITY,
+    LEGACY_HANDSHAKE_METADATA_KEY,
+    LEGACY_JSONRPC_VERSION,
+    LEGACY_PLUGIN_KEYS_METADATA_KEY,
+    LegacyAdapter,
+    LegacyErrorData,
+    LegacyErrorResponse,
+    LegacyRequest,
+    LegacySuccessResponse,
+    cancel_to_legacy_request,
+    event_to_legacy_notification,
+    initialize_to_legacy_handshake_response,
+    invoke_to_legacy_request,
+    legacy_message_to_v4,
+    legacy_request_to_invoke,
+    legacy_response_to_message,
+    parse_legacy_message,
+    result_to_legacy_response,
+)
+from astrbot_sdk.protocol.messages import (
+    CancelMessage,
+    ErrorPayload,
+    EventMessage,
+    InitializeMessage,
+    InvokeMessage,
+    PeerInfo,
+    ResultMessage,
+)
+
+
+class TestLegacyRequest:
+    """Tests for LegacyRequest model."""
+
+    def test_default_values(self):
+        """LegacyRequest should have default values."""
+        req = LegacyRequest(method="test_method")
+        assert req.jsonrpc == LEGACY_JSONRPC_VERSION
+        assert req.id is None
+        assert req.method == "test_method"
+        assert req.params == {}
+
+    def test_with_all_fields(self):
+        """LegacyRequest should accept all fields."""
+        req = LegacyRequest(
+            id="req_001",
+            method="handshake",
+            params={"key": "value"},
+        )
+        assert req.id == "req_001"
+        assert req.method == "handshake"
+        assert req.params["key"] == "value"
+
+
+class TestLegacySuccessResponse:
+    """Tests for LegacySuccessResponse model."""
+
+    def test_with_result(self):
+        """LegacySuccessResponse should accept result."""
+        resp = LegacySuccessResponse(id="req_001", result={"status": "ok"})
+        assert resp.jsonrpc == LEGACY_JSONRPC_VERSION
+        assert resp.id == "req_001"
+        assert resp.result["status"] == "ok"
+
+
+class TestLegacyErrorResponse:
+    """Tests for LegacyErrorResponse model."""
+
+    def test_with_error(self):
+        """LegacyErrorResponse should accept error."""
+        error = LegacyErrorData(code=-32000, message="Server error")
+        resp = LegacyErrorResponse(id="req_001", error=error)
+        assert resp.id == "req_001"
+        assert resp.error.code == -32000
+        assert resp.error.message == "Server error"
+
+
+class TestLegacyErrorData:
+    """Tests for LegacyErrorData model."""
+
+    def test_default_code(self):
+        """LegacyErrorData should have default code."""
+        error = LegacyErrorData(message="Error")
+        assert error.code == -32000
+        assert error.message == "Error"
+        assert error.data is None
+
+    def test_with_data(self):
+        """LegacyErrorData should accept data."""
+        error = LegacyErrorData(
+            code=-32600,
+            message="Invalid Request",
+            data={"details": "Missing field"},
+        )
+        assert error.code == -32600
+        assert error.data["details"] == "Missing field"
+
+
+class TestParseLegacyMessage:
+    """Tests for parse_legacy_message function."""
+
+    def test_parse_request(self):
+        """parse_legacy_message should parse LegacyRequest."""
+        payload = {"jsonrpc": "2.0", "id": "1", "method": "test", "params": {}}
+        msg = parse_legacy_message(payload)
+        assert isinstance(msg, LegacyRequest)
+        assert msg.method == "test"
+
+    def test_parse_request_from_json(self):
+        """parse_legacy_message should parse request from JSON string."""
+        json_str = '{"jsonrpc": "2.0", "id": "1", "method": "handshake"}'
+        msg = parse_legacy_message(json_str)
+        assert isinstance(msg, LegacyRequest)
+        assert msg.method == "handshake"
+
+    def test_parse_request_from_bytes(self):
+        """parse_legacy_message should parse request from bytes."""
+        json_bytes = b'{"jsonrpc": "2.0", "method": "call_handler"}'
+        msg = parse_legacy_message(json_bytes)
+        assert isinstance(msg, LegacyRequest)
+        assert msg.method == "call_handler"
+
+    def test_parse_success_response(self):
+        """parse_legacy_message should parse LegacySuccessResponse."""
+        payload = {"jsonrpc": "2.0", "id": "1", "result": {"status": "ok"}}
+        msg = parse_legacy_message(payload)
+        assert isinstance(msg, LegacySuccessResponse)
+        assert msg.result["status"] == "ok"
+
+    def test_parse_error_response(self):
+        """parse_legacy_message should parse LegacyErrorResponse."""
+        payload = {
+            "jsonrpc": "2.0",
+            "id": "1",
+            "error": {"code": -32000, "message": "Error"},
+        }
+        msg = parse_legacy_message(payload)
+        assert isinstance(msg, LegacyErrorResponse)
+        assert msg.error.message == "Error"
+
+    def test_parse_unknown_raises(self):
+        """parse_legacy_message should raise for unknown type."""
+        with pytest.raises(ValueError) as exc_info:
+            parse_legacy_message({"jsonrpc": "2.0", "unknown": "field"})
+        assert "未知" in str(exc_info.value)
+
+    def test_pass_through_legacy_message(self):
+        """parse_legacy_message should pass through already-parsed messages."""
+        req = LegacyRequest(method="test")
+        result = parse_legacy_message(req)
+        assert result is req
+
+
+class TestLegacyAdapterInit:
+    """Tests for LegacyAdapter initialization."""
+
+    def test_default_values(self):
+        """LegacyAdapter should have default values."""
+        adapter = LegacyAdapter()
+        assert adapter.protocol_version == "1.0"
+        assert adapter.legacy_peer_name == "legacy-peer"
+        assert adapter.legacy_peer_role == "plugin"
+        assert adapter.legacy_peer_version is None
+
+    def test_custom_values(self):
+        """LegacyAdapter should accept custom values."""
+        adapter = LegacyAdapter(
+            protocol_version="2.0",
+            legacy_peer_name="custom-peer",
+            legacy_peer_role="core",
+            legacy_peer_version="1.5.0",
+        )
+        assert adapter.protocol_version == "2.0"
+        assert adapter.legacy_peer_name == "custom-peer"
+        assert adapter.legacy_peer_role == "core"
+        assert adapter.legacy_peer_version == "1.5.0"
+
+
+class TestLegacyAdapterTrackHandler:
+    """Tests for LegacyAdapter.track_handler method."""
+
+    def test_track_handler(self):
+        """track_handler should store handler name by request ID."""
+        adapter = LegacyAdapter()
+        adapter.track_handler("req_001", "module.handler")
+        assert adapter._handler_names_by_request_id["req_001"] == "module.handler"
+
+    def test_track_handler_empty_id(self):
+        """track_handler should not store for empty request ID."""
+        adapter = LegacyAdapter()
+        adapter.track_handler("", "module.handler")
+        assert "" not in adapter._handler_names_by_request_id
+
+
+class TestLegacyAdapterHandshake:
+    """Tests for LegacyAdapter handshake handling."""
+
+    def test_legacy_request_to_handshake(self):
+        """legacy_request_to_message should convert handshake request."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(id="req_001", method="handshake", params={})
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, InitializeMessage)
+        assert msg.protocol_version == "1.0"
+        assert msg.peer.name == "legacy-peer"
+        assert msg.peer.role == "plugin"
+        assert msg.metadata.get("legacy_handshake") is True
+
+    def test_build_legacy_handshake_request(self):
+        """build_legacy_handshake_request should create handshake request."""
+        adapter = LegacyAdapter()
+        result = adapter.build_legacy_handshake_request("req_001")
+
+        assert result["jsonrpc"] == LEGACY_JSONRPC_VERSION
+        assert result["id"] == "req_001"
+        assert result["method"] == "handshake"
+
+
+class TestLegacyAdapterCallHandler:
+    """Tests for LegacyAdapter call_handler handling."""
+
+    def test_legacy_request_to_call_handler(self):
+        """legacy_request_to_message should convert call_handler request."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            id="req_001",
+            method="call_handler",
+            params={
+                "handler_full_name": "module.handler",
+                "event": {"type": "message"},
+                "args": {"key": "value"},
+            },
+        )
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, InvokeMessage)
+        assert msg.capability == "handler.invoke"
+        assert msg.input["handler_id"] == "module.handler"
+        assert msg.input["event"]["type"] == "message"
+
+    def test_call_handler_tracks_handler(self):
+        """call_handler should track handler name."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            id="req_001",
+            method="call_handler",
+            params={"handler_full_name": "test.handler"},
+        )
+        adapter.legacy_request_to_message(req)
+        assert adapter._handler_names_by_request_id["req_001"] == "test.handler"
+
+
+class TestLegacyAdapterContextFunction:
+    """Tests for LegacyAdapter call_context_function handling."""
+
+    def test_legacy_request_to_context_function(self):
+        """legacy_request_to_message should convert call_context_function."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            id="req_001",
+            method="call_context_function",
+            params={"name": "get_user", "args": {"user_id": 123}},
+        )
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, InvokeMessage)
+        assert msg.capability == LEGACY_CONTEXT_CAPABILITY
+        assert msg.input["name"] == "get_user"
+        assert msg.input["args"]["user_id"] == 123
+
+
+class TestLegacyAdapterStreamMethods:
+    """Tests for LegacyAdapter stream handling."""
+
+    def test_handler_stream_start(self):
+        """legacy_request_to_message should convert handler_stream_start."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            method="handler_stream_start",
+            params={"id": "stream_001", "handler_full_name": "module.handler"},
+        )
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, EventMessage)
+        assert msg.phase == "started"
+
+    def test_handler_stream_update(self):
+        """legacy_request_to_message should convert handler_stream_update."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            method="handler_stream_update",
+            params={
+                "id": "stream_001",
+                "handler_full_name": "module.handler",
+                "data": {"text": "chunk"},
+            },
+        )
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, EventMessage)
+        assert msg.phase == "delta"
+        assert msg.data["text"] == "chunk"
+
+    def test_handler_stream_end_completed(self):
+        """legacy_request_to_message should convert handler_stream_end (completed)."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            method="handler_stream_end",
+            params={"id": "stream_001", "handler_full_name": "module.handler"},
+        )
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, EventMessage)
+        assert msg.phase == "completed"
+
+    def test_handler_stream_end_failed(self):
+        """legacy_request_to_message should convert handler_stream_end (failed)."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            method="handler_stream_end",
+            params={
+                "id": "stream_001",
+                "handler_full_name": "module.handler",
+                "error": {"message": "Something went wrong"},
+            },
+        )
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, EventMessage)
+        assert msg.phase == "failed"
+        assert msg.error is not None
+        assert msg.error.message == "Something went wrong"
+
+
+class TestLegacyAdapterCancel:
+    """Tests for LegacyAdapter cancel handling."""
+
+    def test_legacy_request_to_cancel(self):
+        """legacy_request_to_message should convert cancel request."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            id="req_001",
+            method="cancel",
+            params={"reason": "user_cancelled"},
+        )
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, CancelMessage)
+        assert msg.reason == "user_cancelled"
+
+
+class TestLegacyAdapterGenericMethod:
+    """Tests for LegacyAdapter generic method handling."""
+
+    def test_unknown_method_becomes_invoke(self):
+        """Unknown methods should become InvokeMessage with capability=method."""
+        adapter = LegacyAdapter()
+        req = LegacyRequest(
+            id="req_001",
+            method="custom.capability",
+            params={"key": "value"},
+        )
+        msg = adapter.legacy_request_to_message(req)
+
+        assert isinstance(msg, InvokeMessage)
+        assert msg.capability == "custom.capability"
+        assert msg.input["key"] == "value"
+
+
+class TestLegacyAdapterResponseHandling:
+    """Tests for LegacyAdapter response handling."""
+
+    def test_success_response_to_result(self):
+        """legacy_response_to_message should convert success response."""
+        adapter = LegacyAdapter()
+        resp = LegacySuccessResponse(id="req_001", result={"status": "ok"})
+        msg = adapter.legacy_response_to_message(resp)
+
+        assert isinstance(msg, ResultMessage)
+        assert msg.success is True
+        assert msg.output["status"] == "ok"
+
+    def test_error_response_to_result(self):
+        """legacy_error_to_result should convert error response."""
+        adapter = LegacyAdapter()
+        error = LegacyErrorData(code=-32000, message="Server error")
+        resp = LegacyErrorResponse(id="req_001", error=error)
+        msg = adapter.legacy_error_to_result(resp)
+
+        assert isinstance(msg, ResultMessage)
+        assert msg.success is False
+        assert msg.error.code == "legacy_rpc_error"
+        assert msg.error.message == "Server error"
+
+    def test_handshake_response_to_initialize(self):
+        """legacy_response_to_message should detect handshake response."""
+        adapter = LegacyAdapter()
+        resp = LegacySuccessResponse(
+            id="req_001",
+            result={
+                "module.path": {
+                    "name": "test-plugin",
+                    "version": "1.0.0",
+                    "handlers": [],
+                }
+            },
+        )
+        msg = adapter.legacy_response_to_message(resp)
+
+        assert isinstance(msg, InitializeMessage)
+        assert msg.peer.name == "test-plugin"
+        assert msg.peer.version == "1.0.0"
+
+
+class TestLegacyAdapterV4ToLegacy:
+    """Tests for LegacyAdapter V4 to legacy conversion."""
+
+    def test_initialize_to_legacy_handshake_response(self):
+        """initialize_to_legacy_handshake_response should convert InitializeMessage."""
+        adapter = LegacyAdapter()
+        init_msg = InitializeMessage(
+            id="msg_001",
+            protocol_version="1.0",
+            peer=PeerInfo(name="test", role="plugin", version="1.0.0"),
+            handlers=[],
+            metadata={
+                "plugin_id": "test-plugin",
+                "display_name": "Test Plugin",
+            },
+        )
+        result = adapter.initialize_to_legacy_handshake_response(init_msg)
+
+        assert result["jsonrpc"] == LEGACY_JSONRPC_VERSION
+        assert result["id"] == "msg_001"
+        assert "result" in result
+
+    def test_initialize_with_legacy_payload(self):
+        """initialize_to_legacy_handshake_response should preserve legacy payload."""
+        adapter = LegacyAdapter()
+        legacy_payload = {
+            "module.path": {
+                "name": "test-plugin",
+                "version": "1.0.0",
+                "handlers": [],
+            }
+        }
+        init_msg = InitializeMessage(
+            id="msg_001",
+            protocol_version="1.0",
+            peer=PeerInfo(name="test", role="plugin", version="1.0.0"),
+            handlers=[],
+            metadata={LEGACY_HANDSHAKE_METADATA_KEY: legacy_payload},
+        )
+        result = adapter.initialize_to_legacy_handshake_response(init_msg)
+
+        assert result["result"] == legacy_payload
+
+    def test_invoke_to_legacy_request_handler(self):
+        """invoke_to_legacy_request should convert handler.invoke."""
+        adapter = LegacyAdapter()
+        invoke_msg = InvokeMessage(
+            id="msg_001",
+            capability="handler.invoke",
+            input={
+                "handler_id": "module.handler",
+                "event": {"type": "message"},
+                "args": {},
+            },
+        )
+        result = adapter.invoke_to_legacy_request(invoke_msg)
+
+        assert result["method"] == "call_handler"
+        assert result["params"]["handler_full_name"] == "module.handler"
+
+    def test_invoke_to_legacy_request_context_function(self):
+        """invoke_to_legacy_request should convert context function."""
+        adapter = LegacyAdapter()
+        invoke_msg = InvokeMessage(
+            id="msg_001",
+            capability=LEGACY_CONTEXT_CAPABILITY,
+            input={"name": "get_user", "args": {}},
+        )
+        result = adapter.invoke_to_legacy_request(invoke_msg)
+
+        assert result["method"] == "call_context_function"
+        assert result["params"]["name"] == "get_user"
+
+    def test_invoke_to_legacy_request_generic(self):
+        """invoke_to_legacy_request should convert generic capability."""
+        adapter = LegacyAdapter()
+        invoke_msg = InvokeMessage(
+            id="msg_001",
+            capability="custom.capability",
+            input={"key": "value"},
+        )
+        result = adapter.invoke_to_legacy_request(invoke_msg)
+
+        assert result["method"] == "custom.capability"
+        assert result["params"]["key"] == "value"
+
+    def test_result_to_legacy_response_success(self):
+        """result_to_legacy_response should convert success result."""
+        adapter = LegacyAdapter()
+        result_msg = ResultMessage(
+            id="msg_001",
+            success=True,
+            output={"status": "ok"},
+        )
+        result = adapter.result_to_legacy_response(result_msg)
+
+        assert "result" in result
+        assert result["result"]["status"] == "ok"
+
+    def test_result_to_legacy_response_error(self):
+        """result_to_legacy_response should convert error result."""
+        adapter = LegacyAdapter()
+        result_msg = ResultMessage(
+            id="msg_001",
+            success=False,
+            error=ErrorPayload(code="error", message="Failed"),
+        )
+        result = adapter.result_to_legacy_response(result_msg)
+
+        assert "error" in result
+        assert result["error"]["message"] == "Failed"
+
+    def test_event_to_legacy_notification_started(self):
+        """event_to_legacy_notification should convert started event."""
+        adapter = LegacyAdapter()
+        adapter.track_handler("msg_001", "module.handler")
+        event_msg = EventMessage(id="msg_001", phase="started")
+        result = adapter.event_to_legacy_notification(event_msg)
+
+        assert result["method"] == "handler_stream_start"
+        assert result["params"]["handler_full_name"] == "module.handler"
+
+    def test_event_to_legacy_notification_delta(self):
+        """event_to_legacy_notification should convert delta event."""
+        adapter = LegacyAdapter()
+        event_msg = EventMessage(
+            id="msg_001",
+            phase="delta",
+            data={"text": "chunk"},
+        )
+        result = adapter.event_to_legacy_notification(event_msg)
+
+        assert result["method"] == "handler_stream_update"
+        assert result["params"]["data"]["text"] == "chunk"
+
+    def test_event_to_legacy_notification_completed(self):
+        """event_to_legacy_notification should convert completed event."""
+        adapter = LegacyAdapter()
+        event_msg = EventMessage(id="msg_001", phase="completed")
+        result = adapter.event_to_legacy_notification(event_msg)
+
+        assert result["method"] == "handler_stream_end"
+
+    def test_event_to_legacy_notification_failed(self):
+        """event_to_legacy_notification should convert failed event."""
+        adapter = LegacyAdapter()
+        event_msg = EventMessage(
+            id="msg_001",
+            phase="failed",
+            error=ErrorPayload(code="error", message="Failed"),
+        )
+        result = adapter.event_to_legacy_notification(event_msg)
+
+        assert result["method"] == "handler_stream_end"
+        assert result["params"]["error"]["message"] == "Failed"
+
+    def test_cancel_to_legacy_request(self):
+        """cancel_to_legacy_request should convert cancel message."""
+        adapter = LegacyAdapter()
+        cancel_msg = CancelMessage(id="msg_001", reason="user_request")
+        result = adapter.cancel_to_legacy_request(cancel_msg)
+
+        assert result["method"] == "cancel"
+        assert result["params"]["reason"] == "user_request"
+
+
+class TestLegacyAdapterHandlerDescriptors:
+    """Tests for LegacyAdapter handler descriptor conversion."""
+
+    def test_legacy_handlers_to_descriptors(self):
+        """_legacy_handlers_to_descriptors should convert handlers."""
+        adapter = LegacyAdapter()
+        payload = {
+            "module.path": {
+                "handlers": [
+                    {
+                        "handler_full_name": "module.handler",
+                        "event_type": "3",
+                        "extras_configs": {
+                            "priority": 10,
+                            "require_admin": True,
+                            "level": 5,
+                        },
+                    }
+                ]
+            }
+        }
+        handlers = adapter._legacy_handlers_to_descriptors(payload)
+
+        assert len(handlers) == 1
+        assert handlers[0].id == "module.handler"
+        assert handlers[0].priority == 10
+        assert handlers[0].permissions.require_admin is True
+        assert handlers[0].permissions.level == 5
+
+    def test_descriptor_to_legacy_handler(self):
+        """_descriptor_to_legacy_handler should convert HandlerDescriptor."""
+        descriptor = HandlerDescriptor(
+            id="module.handler",
+            trigger=EventTrigger(event_type="3"),
+            priority=10,
+            permissions=Permissions(require_admin=True, level=5),
+        )
+        result = LegacyAdapter._descriptor_to_legacy_handler(descriptor)
+
+        assert result["handler_full_name"] == "module.handler"
+        assert result["event_type"] == 3
+        assert result["extras_configs"]["priority"] == 10
+        assert result["extras_configs"]["require_admin"] is True
+
+
+class TestLegacyAdapterHelpers:
+    """Tests for LegacyAdapter helper methods."""
+
+    def test_request_id_with_value(self):
+        """_request_id should return string value."""
+        result = LegacyAdapter._request_id("req_001", "fallback")
+        assert result == "req_001"
+
+    def test_request_id_with_none(self):
+        """_request_id should return fallback for None."""
+        result = LegacyAdapter._request_id(None, "fallback")
+        assert result == "fallback"
+
+    def test_request_id_with_empty_string(self):
+        """_request_id should return fallback for empty string."""
+        result = LegacyAdapter._request_id("", "fallback")
+        assert result == "fallback"
+
+    def test_as_dict_with_dict(self):
+        """_as_dict should pass through dict."""
+        result = LegacyAdapter._as_dict({"key": "value"}, field_name="data")
+        assert result == {"key": "value"}
+
+    def test_as_dict_with_none(self):
+        """_as_dict should return empty dict for None."""
+        result = LegacyAdapter._as_dict(None, field_name="data")
+        assert result == {}
+
+    def test_as_dict_with_other(self):
+        """_as_dict should wrap other values."""
+        result = LegacyAdapter._as_dict("value", field_name="data")
+        assert result == {"data": "value"}
+
+    def test_looks_like_handshake_payload_valid(self):
+        """_looks_like_handshake_payload should detect valid payload."""
+        payload = {"module.path": {"handlers": []}}
+        assert LegacyAdapter._looks_like_handshake_payload(payload) is True
+
+    def test_looks_like_handshake_payload_invalid(self):
+        """_looks_like_handshake_payload should reject invalid payload."""
+        assert LegacyAdapter._looks_like_handshake_payload({}) is False
+        assert LegacyAdapter._looks_like_handshake_payload({"key": "value"}) is False
+        assert LegacyAdapter._looks_like_handshake_payload(None) is False
+
+
+class TestLegacyConvenienceFunctions:
+    """Tests for module-level convenience functions."""
+
+    def test_legacy_message_to_v4(self):
+        """legacy_message_to_v4 should convert legacy message."""
+        payload = {"jsonrpc": "2.0", "method": "handshake"}
+        msg = legacy_message_to_v4(payload)
+        assert isinstance(msg, InitializeMessage)
+
+    def test_legacy_request_to_invoke(self):
+        """legacy_request_to_invoke should convert to InvokeMessage."""
+        payload = {
+            "jsonrpc": "2.0",
+            "id": "1",
+            "method": "custom.capability",
+            "params": {},
+        }
+        msg = legacy_request_to_invoke(payload)
+        assert isinstance(msg, InvokeMessage)
+        assert msg.capability == "custom.capability"
+
+    def test_legacy_request_to_invoke_non_invoke_raises(self):
+        """legacy_request_to_invoke should raise for non-invoke messages."""
+        payload = {"jsonrpc": "2.0", "method": "handshake"}
+        with pytest.raises(ValueError, match="不能直接映射为 invoke"):
+            legacy_request_to_invoke(payload)
+
+    def test_legacy_response_to_message(self):
+        """legacy_response_to_message should convert response."""
+        payload = {"jsonrpc": "2.0", "id": "1", "result": {"status": "ok"}}
+        msg = legacy_response_to_message(payload)
+        assert isinstance(msg, ResultMessage)
+
+    def test_initialize_to_legacy_handshake_response(self):
+        """initialize_to_legacy_handshake_response should convert."""
+        msg = InitializeMessage(
+            id="msg_001",
+            protocol_version="1.0",
+            peer=PeerInfo(name="test", role="plugin"),
+            handlers=[],
+        )
+        result = initialize_to_legacy_handshake_response(msg)
+        assert result["jsonrpc"] == LEGACY_JSONRPC_VERSION
+
+    def test_invoke_to_legacy_request(self):
+        """invoke_to_legacy_request should convert."""
+        msg = InvokeMessage(id="msg_001", capability="test.cap", input={})
+        result = invoke_to_legacy_request(msg)
+        assert result["method"] == "test.cap"
+
+    def test_result_to_legacy_response(self):
+        """result_to_legacy_response should convert."""
+        msg = ResultMessage(id="msg_001", success=True, output={"ok": True})
+        result = result_to_legacy_response(msg)
+        assert result["result"]["ok"] is True
+
+    def test_event_to_legacy_notification(self):
+        """event_to_legacy_notification should convert."""
+        msg = EventMessage(id="msg_001", phase="started")
+        result = event_to_legacy_notification(msg, handler_full_name="test.handler")
+        assert result["method"] == "handler_stream_start"
+
+    def test_cancel_to_legacy_request(self):
+        """cancel_to_legacy_request should convert."""
+        msg = CancelMessage(id="msg_001", reason="test")
+        result = cancel_to_legacy_request(msg)
+        assert result["method"] == "cancel"
+
+
+class TestLegacyConstants:
+    """Tests for legacy adapter constants."""
+
+    def test_jsonrpc_version(self):
+        """LEGACY_JSONRPC_VERSION should be 2.0."""
+        assert LEGACY_JSONRPC_VERSION == "2.0"
+
+    def test_context_capability(self):
+        """LEGACY_CONTEXT_CAPABILITY should be internal capability."""
+        assert LEGACY_CONTEXT_CAPABILITY == "internal.legacy.call_context_function"
+
+    def test_message_event(self):
+        """LEGACY_ADAPTER_MESSAGE_EVENT should be 3."""
+        assert LEGACY_ADAPTER_MESSAGE_EVENT == 3
+
+    def test_metadata_keys(self):
+        """Metadata keys should be defined."""
+        assert LEGACY_HANDSHAKE_METADATA_KEY == "legacy_handshake_payload"
+        assert LEGACY_PLUGIN_KEYS_METADATA_KEY == "legacy_plugin_keys"

tests_v4/test_protocol_messages.py

@@ -0,0 +1,509 @@
+"""
+Tests for protocol/messages.py - Protocol message models.
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+from pydantic import ValidationError
+
+from astrbot_sdk.protocol.descriptors import CommandTrigger, HandlerDescriptor
+from astrbot_sdk.protocol.messages import (
+    CancelMessage,
+    ErrorPayload,
+    EventMessage,
+    InitializeMessage,
+    InitializeOutput,
+    InvokeMessage,
+    PeerInfo,
+    ResultMessage,
+    parse_message,
+)
+
+
+class TestErrorPayload:
+    """Tests for ErrorPayload model."""
+
+    def test_required_code_and_message(self):
+        """ErrorPayload requires code and message."""
+        error = ErrorPayload(code="test_error", message="Test error occurred")
+        assert error.code == "test_error"
+        assert error.message == "Test error occurred"
+        assert error.hint == ""
+        assert error.retryable is False
+
+    def test_with_all_fields(self):
+        """ErrorPayload should accept all fields."""
+        error = ErrorPayload(
+            code="server_error",
+            message="Internal server error",
+            hint="Try again later",
+            retryable=True,
+        )
+        assert error.code == "server_error"
+        assert error.message == "Internal server error"
+        assert error.hint == "Try again later"
+        assert error.retryable is True
+
+    def test_model_dump(self):
+        """ErrorPayload should serialize correctly."""
+        error = ErrorPayload(
+            code="not_found",
+            message="Resource not found",
+            hint="Check the ID",
+        )
+        data = error.model_dump()
+        assert data == {
+            "code": "not_found",
+            "message": "Resource not found",
+            "hint": "Check the ID",
+            "retryable": False,
+        }
+
+    def test_extra_fields_forbidden(self):
+        """ErrorPayload should forbid extra fields."""
+        with pytest.raises(ValidationError):
+            ErrorPayload(code="test", message="test", extra="field")
+
+
+class TestPeerInfo:
+    """Tests for PeerInfo model."""
+
+    def test_required_name_and_role(self):
+        """PeerInfo requires name and role."""
+        peer = PeerInfo(name="test-plugin", role="plugin")
+        assert peer.name == "test-plugin"
+        assert peer.role == "plugin"
+        assert peer.version is None
+
+    def test_with_version(self):
+        """PeerInfo should accept version."""
+        peer = PeerInfo(name="my-plugin", role="plugin", version="1.0.0")
+        assert peer.version == "1.0.0"
+
+    def test_role_must_be_valid(self):
+        """PeerInfo role must be 'plugin' or 'core'."""
+        peer1 = PeerInfo(name="p1", role="plugin")
+        assert peer1.role == "plugin"
+
+        peer2 = PeerInfo(name="p2", role="core")
+        assert peer2.role == "core"
+
+        with pytest.raises(ValidationError):
+            PeerInfo(name="p3", role="invalid")
+
+    def test_model_dump(self):
+        """PeerInfo should serialize correctly."""
+        peer = PeerInfo(name="test", role="plugin", version="2.0.0")
+        data = peer.model_dump()
+        assert data == {"name": "test", "role": "plugin", "version": "2.0.0"}
+
+    def test_extra_fields_forbidden(self):
+        """PeerInfo should forbid extra fields."""
+        with pytest.raises(ValidationError):
+            PeerInfo(name="test", role="plugin", extra="field")
+
+
+class TestInitializeMessage:
+    """Tests for InitializeMessage model."""
+
+    def test_required_fields(self):
+        """InitializeMessage requires id, protocol_version, and peer."""
+        peer = PeerInfo(name="test", role="plugin")
+        msg = InitializeMessage(
+            id="msg_001",
+            protocol_version="1.0",
+            peer=peer,
+        )
+        assert msg.type == "initialize"
+        assert msg.id == "msg_001"
+        assert msg.protocol_version == "1.0"
+        assert msg.peer == peer
+        assert msg.handlers == []
+        assert msg.metadata == {}
+
+    def test_with_handlers(self):
+        """InitializeMessage should accept handlers."""
+        peer = PeerInfo(name="test", role="plugin")
+        handler = HandlerDescriptor(
+            id="test.handler",
+            trigger=CommandTrigger(command="hello"),
+        )
+        msg = InitializeMessage(
+            id="msg_002",
+            protocol_version="1.0",
+            peer=peer,
+            handlers=[handler],
+        )
+        assert len(msg.handlers) == 1
+        assert msg.handlers[0].id == "test.handler"
+
+    def test_with_metadata(self):
+        """InitializeMessage should accept metadata."""
+        peer = PeerInfo(name="test", role="plugin")
+        msg = InitializeMessage(
+            id="msg_003",
+            protocol_version="1.0",
+            peer=peer,
+            metadata={"author": "test", "version": "1.0.0"},
+        )
+        assert msg.metadata["author"] == "test"
+        assert msg.metadata["version"] == "1.0.0"
+
+    def test_model_dump_json(self):
+        """InitializeMessage should serialize to JSON correctly."""
+        peer = PeerInfo(name="test", role="plugin", version="1.0.0")
+        handler = HandlerDescriptor(
+            id="test.handler",
+            trigger=CommandTrigger(command="hello"),
+        )
+        msg = InitializeMessage(
+            id="msg_004",
+            protocol_version="1.0",
+            peer=peer,
+            handlers=[handler],
+            metadata={"key": "value"},
+        )
+        json_str = msg.model_dump_json()
+        data = json.loads(json_str)
+        assert data["type"] == "initialize"
+        assert data["id"] == "msg_004"
+        assert data["peer"]["name"] == "test"
+        assert len(data["handlers"]) == 1
+
+
+class TestInitializeOutput:
+    """Tests for InitializeOutput model."""
+
+    def test_required_peer(self):
+        """InitializeOutput requires peer."""
+        peer = PeerInfo(name="core", role="core")
+        output = InitializeOutput(peer=peer)
+        assert output.peer == peer
+        assert output.capabilities == []
+        assert output.metadata == {}
+
+    def test_with_capabilities(self):
+        """InitializeOutput should accept capabilities."""
+        from astrbot_sdk.protocol.descriptors import CapabilityDescriptor
+
+        peer = PeerInfo(name="core", role="core")
+        cap = CapabilityDescriptor(name="llm.chat", description="Chat capability")
+        output = InitializeOutput(peer=peer, capabilities=[cap])
+        assert len(output.capabilities) == 1
+        assert output.capabilities[0].name == "llm.chat"
+
+    def test_with_metadata(self):
+        """InitializeOutput should accept metadata."""
+        peer = PeerInfo(name="core", role="core")
+        output = InitializeOutput(peer=peer, metadata={"session": "abc"})
+        assert output.metadata["session"] == "abc"
+
+
+class TestResultMessage:
+    """Tests for ResultMessage model."""
+
+    def test_success_result(self):
+        """ResultMessage for success case."""
+        msg = ResultMessage(id="msg_001", success=True, output={"text": "ok"})
+        assert msg.type == "result"
+        assert msg.id == "msg_001"
+        assert msg.success is True
+        assert msg.output["text"] == "ok"
+        assert msg.error is None
+        assert msg.kind is None
+
+    def test_error_result(self):
+        """ResultMessage for error case."""
+        error = ErrorPayload(code="not_found", message="Resource not found")
+        msg = ResultMessage(id="msg_002", success=False, error=error)
+        assert msg.success is False
+        assert msg.error.code == "not_found"
+        assert msg.error.message == "Resource not found"
+
+    def test_with_kind(self):
+        """ResultMessage should accept kind."""
+        msg = ResultMessage(
+            id="msg_003",
+            kind="initialize_result",
+            success=True,
+        )
+        assert msg.kind == "initialize_result"
+
+    def test_default_output(self):
+        """ResultMessage should have empty dict as default output."""
+        msg = ResultMessage(id="msg_004", success=True)
+        assert msg.output == {}
+
+
+class TestInvokeMessage:
+    """Tests for InvokeMessage model."""
+
+    def test_required_fields(self):
+        """InvokeMessage requires id and capability."""
+        msg = InvokeMessage(id="msg_001", capability="llm.chat")
+        assert msg.type == "invoke"
+        assert msg.id == "msg_001"
+        assert msg.capability == "llm.chat"
+        assert msg.input == {}
+        assert msg.stream is False
+
+    def test_with_input(self):
+        """InvokeMessage should accept input payload."""
+        msg = InvokeMessage(
+            id="msg_002",
+            capability="db.get",
+            input={"key": "user:123"},
+        )
+        assert msg.input["key"] == "user:123"
+
+    def test_with_stream(self):
+        """InvokeMessage should accept stream flag."""
+        msg = InvokeMessage(
+            id="msg_003",
+            capability="llm.stream",
+            input={"prompt": "hello"},
+            stream=True,
+        )
+        assert msg.stream is True
+
+    def test_model_dump(self):
+        """InvokeMessage should serialize correctly."""
+        msg = InvokeMessage(
+            id="msg_004",
+            capability="test.cap",
+            input={"data": "value"},
+            stream=True,
+        )
+        data = msg.model_dump()
+        assert data["type"] == "invoke"
+        assert data["capability"] == "test.cap"
+        assert data["input"] == {"data": "value"}
+        assert data["stream"] is True
+
+
+class TestEventMessage:
+    """Tests for EventMessage model."""
+
+    def test_started_phase(self):
+        """EventMessage with started phase."""
+        msg = EventMessage(id="msg_001", phase="started")
+        assert msg.type == "event"
+        assert msg.phase == "started"
+        assert msg.data == {}
+        assert msg.output == {}
+        assert msg.error is None
+
+    def test_delta_phase(self):
+        """EventMessage with delta phase and data."""
+        msg = EventMessage(
+            id="msg_002",
+            phase="delta",
+            data={"text": "chunk"},
+        )
+        assert msg.phase == "delta"
+        assert msg.data["text"] == "chunk"
+
+    def test_completed_phase(self):
+        """EventMessage with completed phase."""
+        msg = EventMessage(
+            id="msg_003",
+            phase="completed",
+            output={"result": "done"},
+        )
+        assert msg.phase == "completed"
+        assert msg.output["result"] == "done"
+
+    def test_failed_phase(self):
+        """EventMessage with failed phase."""
+        error = ErrorPayload(code="runtime_error", message="Failed")
+        msg = EventMessage(
+            id="msg_004",
+            phase="failed",
+            error=error,
+        )
+        assert msg.phase == "failed"
+        assert msg.error.code == "runtime_error"
+
+    def test_invalid_phase(self):
+        """EventMessage should reject invalid phase."""
+        with pytest.raises(ValidationError):
+            EventMessage(id="msg_005", phase="invalid")
+
+
+class TestCancelMessage:
+    """Tests for CancelMessage model."""
+
+    def test_default_reason(self):
+        """CancelMessage should have default reason."""
+        msg = CancelMessage(id="msg_001")
+        assert msg.type == "cancel"
+        assert msg.id == "msg_001"
+        assert msg.reason == "user_cancelled"
+
+    def test_custom_reason(self):
+        """CancelMessage should accept custom reason."""
+        msg = CancelMessage(id="msg_002", reason="timeout")
+        assert msg.reason == "timeout"
+
+    def test_model_dump(self):
+        """CancelMessage should serialize correctly."""
+        msg = CancelMessage(id="msg_003", reason="user_request")
+        data = msg.model_dump()
+        assert data == {
+            "type": "cancel",
+            "id": "msg_003",
+            "reason": "user_request",
+        }
+
+
+class TestParseMessage:
+    """Tests for parse_message function."""
+
+    def test_parse_initialize_from_dict(self):
+        """parse_message should parse InitializeMessage from dict."""
+        data = {
+            "type": "initialize",
+            "id": "msg_001",
+            "protocol_version": "1.0",
+            "peer": {"name": "test", "role": "plugin"},
+        }
+        msg = parse_message(data)
+        assert isinstance(msg, InitializeMessage)
+        assert msg.id == "msg_001"
+        assert msg.peer.name == "test"
+
+    def test_parse_result_from_dict(self):
+        """parse_message should parse ResultMessage from dict."""
+        data = {
+            "type": "result",
+            "id": "msg_002",
+            "success": True,
+            "output": {"text": "ok"},
+        }
+        msg = parse_message(data)
+        assert isinstance(msg, ResultMessage)
+        assert msg.success is True
+
+    def test_parse_invoke_from_dict(self):
+        """parse_message should parse InvokeMessage from dict."""
+        data = {
+            "type": "invoke",
+            "id": "msg_003",
+            "capability": "test.cap",
+            "input": {"key": "value"},
+        }
+        msg = parse_message(data)
+        assert isinstance(msg, InvokeMessage)
+        assert msg.capability == "test.cap"
+
+    def test_parse_event_from_dict(self):
+        """parse_message should parse EventMessage from dict."""
+        data = {
+            "type": "event",
+            "id": "msg_004",
+            "phase": "delta",
+            "data": {"text": "chunk"},
+        }
+        msg = parse_message(data)
+        assert isinstance(msg, EventMessage)
+        assert msg.phase == "delta"
+
+    def test_parse_cancel_from_dict(self):
+        """parse_message should parse CancelMessage from dict."""
+        data = {
+            "type": "cancel",
+            "id": "msg_005",
+            "reason": "user_request",
+        }
+        msg = parse_message(data)
+        assert isinstance(msg, CancelMessage)
+        assert msg.reason == "user_request"
+
+    def test_parse_from_json_string(self):
+        """parse_message should parse from JSON string."""
+        json_str = '{"type": "invoke", "id": "msg_006", "capability": "test"}'
+        msg = parse_message(json_str)
+        assert isinstance(msg, InvokeMessage)
+        assert msg.capability == "test"
+
+    def test_parse_from_bytes(self):
+        """parse_message should parse from bytes."""
+        json_bytes = b'{"type": "result", "id": "msg_007", "success": true}'
+        msg = parse_message(json_bytes)
+        assert isinstance(msg, ResultMessage)
+        assert msg.success is True
+
+    def test_parse_unknown_type_raises(self):
+        """parse_message should raise for unknown type."""
+        with pytest.raises(ValueError) as exc_info:
+            parse_message({"type": "unknown"})
+        assert "未知消息类型" in str(exc_info.value)
+
+    def test_roundtrip_serialize_deserialize(self):
+        """Message should survive serialize/deserialize roundtrip."""
+        original = InitializeMessage(
+            id="msg_008",
+            protocol_version="1.0",
+            peer=PeerInfo(name="test", role="plugin", version="1.0.0"),
+            handlers=[
+                HandlerDescriptor(
+                    id="test.handler",
+                    trigger=CommandTrigger(command="hello"),
+                )
+            ],
+            metadata={"key": "value"},
+        )
+        json_str = original.model_dump_json()
+        parsed = parse_message(json_str)
+        assert isinstance(parsed, InitializeMessage)
+        assert parsed.id == original.id
+        assert parsed.peer.name == original.peer.name
+        assert len(parsed.handlers) == 1
+
+
+class TestMessageExtraForbidden:
+    """Tests for extra field rejection across all message types."""
+
+    def test_initialize_extra_forbidden(self):
+        """InitializeMessage should reject extra fields."""
+        with pytest.raises(ValidationError):
+            InitializeMessage(
+                id="msg_001",
+                protocol_version="1.0",
+                peer={"name": "test", "role": "plugin"},
+                extra="field",
+            )
+
+    def test_result_extra_forbidden(self):
+        """ResultMessage should reject extra fields."""
+        with pytest.raises(ValidationError):
+            ResultMessage(
+                id="msg_001",
+                success=True,
+                extra="field",
+            )
+
+    def test_invoke_extra_forbidden(self):
+        """InvokeMessage should reject extra fields."""
+        with pytest.raises(ValidationError):
+            InvokeMessage(
+                id="msg_001",
+                capability="test",
+                extra="field",
+            )
+
+    def test_event_extra_forbidden(self):
+        """EventMessage should reject extra fields."""
+        with pytest.raises(ValidationError):
+            EventMessage(
+                id="msg_001",
+                phase="started",
+                extra="field",
+            )
+
+    def test_cancel_extra_forbidden(self):
+        """CancelMessage should reject extra fields."""
+        with pytest.raises(ValidationError):
+            CancelMessage(id="msg_001", extra="field")

tests_v4/test_transport.py

@@ -0,0 +1,447 @@
+"""
+Tests for runtime/transport.py - Transport implementations.
+"""
+from __future__ import annotations
+
+import asyncio
+from io import StringIO
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from astrbot_sdk.runtime.transport import (
+    MessageHandler,
+    StdioTransport,
+    Transport,
+    WebSocketClientTransport,
+    WebSocketServerTransport,
+)
+
+
+class TestTransportBase:
+    """Tests for Transport base class."""
+
+    def test_init_sets_handler_none(self):
+        """Transport should initialize with _handler as None."""
+        transport = Transport()
+        assert transport._handler is None
+
+    def test_set_message_handler(self):
+        """set_message_handler should store handler."""
+        transport = Transport()
+        handler = MagicMock()
+        transport.set_message_handler(handler)
+        assert transport._handler is handler
+
+    def test_start_not_implemented(self):
+        """Transport.start should raise NotImplementedError."""
+        transport = Transport()
+        with pytest.raises(NotImplementedError):
+            asyncio.get_event_loop().run_until_complete(transport.start())
+
+    def test_stop_not_implemented(self):
+        """Transport.stop should raise NotImplementedError."""
+        transport = Transport()
+        with pytest.raises(NotImplementedError):
+            asyncio.get_event_loop().run_until_complete(transport.stop())
+
+    def test_send_not_implemented(self):
+        """Transport.send should raise NotImplementedError."""
+        transport = Transport()
+        with pytest.raises(NotImplementedError):
+            asyncio.get_event_loop().run_until_complete(transport.send("test"))
+
+    @pytest.mark.asyncio
+    async def test_wait_closed(self):
+        """wait_closed should wait for _closed event."""
+        transport = Transport()
+        transport._closed.set()
+        # Should return immediately since _closed is already set
+        await transport.wait_closed()
+
+    @pytest.mark.asyncio
+    async def test_dispatch_calls_handler(self):
+        """_dispatch should call handler with payload."""
+        transport = Transport()
+        handler = AsyncMock()
+        transport.set_message_handler(handler)
+        await transport._dispatch("test payload")
+        handler.assert_called_once_with("test payload")
+
+    @pytest.mark.asyncio
+    async def test_dispatch_without_handler(self):
+        """_dispatch should work without handler."""
+        transport = Transport()
+        # Should not raise
+        await transport._dispatch("test payload")
+
+
+class TestStdioTransportInit:
+    """Tests for StdioTransport initialization."""
+
+    def test_default_init(self):
+        """StdioTransport should initialize with default values."""
+        transport = StdioTransport()
+        assert transport._stdin is None
+        assert transport._stdout is None
+        assert transport._command is None
+        assert transport._cwd is None
+        assert transport._env is None
+        assert transport._process is None
+        assert transport._reader_task is None
+
+    def test_with_custom_streams(self):
+        """StdioTransport should accept custom stdin/stdout."""
+        stdin = StringIO()
+        stdout = StringIO()
+        transport = StdioTransport(stdin=stdin, stdout=stdout)
+        assert transport._stdin is stdin
+        assert transport._stdout is stdout
+
+    def test_with_command(self):
+        """StdioTransport should accept command for subprocess."""
+        transport = StdioTransport(command=["python", "-m", "module"])
+        assert transport._command == ["python", "-m", "module"]
+
+    def test_with_cwd_and_env(self):
+        """StdioTransport should accept cwd and env."""
+        transport = StdioTransport(cwd="/tmp", env={"VAR": "value"})
+        assert transport._cwd == "/tmp"
+        assert transport._env == {"VAR": "value"}
+
+
+class TestStdioTransportFileMode:
+    """Tests for StdioTransport in file mode (no subprocess)."""
+
+    @pytest.mark.asyncio
+    async def test_start_without_command(self):
+        """start() without command should use stdin/stdout."""
+        transport = StdioTransport()
+        with patch("sys.stdin"), patch("sys.stdout"):
+            await transport.start()
+            assert transport._reader_task is not None
+            await transport.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_cancels_reader_task(self):
+        """stop() should cancel reader task."""
+        transport = StdioTransport()
+        with patch("sys.stdin"), patch("sys.stdout"):
+            await transport.start()
+            task = transport._reader_task
+            await transport.stop()
+            assert task.cancelled() or task.done()
+
+    @pytest.mark.asyncio
+    async def test_send_without_process(self):
+        """send() without process should write to stdout."""
+        stdout = MagicMock()
+        stdout.write = MagicMock()
+        stdout.flush = MagicMock()
+        transport = StdioTransport(stdout=stdout)
+
+        with patch("sys.stdin"):
+            await transport.start()
+
+        await transport.send("test message")
+
+        # Should have written the message with newline
+        stdout.write.assert_called_once_with("test message\n")
+        stdout.flush.assert_called_once()
+
+        await transport.stop()
+
+    @pytest.mark.asyncio
+    async def test_send_adds_newline_if_missing(self):
+        """send() should add newline if not present."""
+        stdout = MagicMock()
+        stdout.write = MagicMock()
+        stdout.flush = MagicMock()
+        transport = StdioTransport(stdout=stdout)
+
+        with patch("sys.stdin"):
+            await transport.start()
+
+        await transport.send("message")
+        stdout.write.assert_called_once_with("message\n")
+
+        await transport.stop()
+
+    @pytest.mark.asyncio
+    async def test_send_preserves_existing_newline(self):
+        """send() should not add extra newline."""
+        stdout = MagicMock()
+        stdout.write = MagicMock()
+        stdout.flush = MagicMock()
+        transport = StdioTransport(stdout=stdout)
+
+        with patch("sys.stdin"):
+            await transport.start()
+
+        await transport.send("message\n")
+        stdout.write.assert_called_once_with("message\n")
+
+        await transport.stop()
+
+    @pytest.mark.asyncio
+    async def test_send_raises_without_stdout(self):
+        """send() should raise if stdout is None."""
+        transport = StdioTransport(stdout=None)
+        transport._stdout = None
+
+        with pytest.raises(RuntimeError, match="stdout"):
+            await transport.send("test")
+
+
+class TestStdioTransportProcessMode:
+    """Tests for StdioTransport in subprocess mode."""
+
+    @pytest.mark.asyncio
+    async def test_start_with_command_creates_process(self):
+        """start() with command should create subprocess."""
+        transport = StdioTransport(command=["echo", "test"])
+
+        await transport.start()
+        assert transport._process is not None
+        assert transport._reader_task is not None
+
+        await transport.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_terminates_process(self):
+        """stop() should terminate the subprocess."""
+        transport = StdioTransport(command=["sleep", "100"])
+
+        await transport.start()
+        process = transport._process
+        assert process is not None
+
+        await transport.stop()
+        assert process.returncode is not None
+
+    @pytest.mark.asyncio
+    async def test_send_to_process(self):
+        """send() should write to process stdin."""
+        transport = StdioTransport(command=["cat"])
+
+        await transport.start()
+        # Should not raise
+        await transport.send("test data")
+
+        await transport.stop()
+
+    @pytest.mark.asyncio
+    async def test_send_raises_if_process_stdin_none(self):
+        """send() should raise if process stdin is None."""
+        transport = StdioTransport(command=["cat"])
+        await transport.start()
+
+        # Manually set stdin to None to simulate error condition
+        if transport._process:
+            transport._process.stdin = None  # type: ignore
+
+        with pytest.raises(RuntimeError, match="stdin"):
+            await transport.send("test")
+
+        await transport.stop()
+
+
+class TestWebSocketServerTransportInit:
+    """Tests for WebSocketServerTransport initialization."""
+
+    def test_default_init(self):
+        """WebSocketServerTransport should have default values."""
+        transport = WebSocketServerTransport()
+        assert transport._host == "127.0.0.1"
+        assert transport._port == 8765
+        assert transport._path == "/"
+        assert transport._heartbeat == 30.0
+        assert transport._app is None
+        assert transport._ws is None
+
+    def test_custom_values(self):
+        """WebSocketServerTransport should accept custom values."""
+        transport = WebSocketServerTransport(
+            host="0.0.0.0",
+            port=9000,
+            path="/ws",
+            heartbeat=60.0,
+        )
+        assert transport._host == "0.0.0.0"
+        assert transport._port == 9000
+        assert transport._path == "/ws"
+        assert transport._heartbeat == 60.0
+
+    def test_port_property_returns_actual_port(self):
+        """port property should return actual port after start."""
+        transport = WebSocketServerTransport(port=8765)
+        # Before start, should return configured port
+        assert transport.port == 8765
+
+    def test_url_property(self):
+        """url property should return WebSocket URL."""
+        transport = WebSocketServerTransport(host="localhost", port=8080, path="/ws")
+        assert transport.url == "ws://localhost:8080/ws"
+
+
+class TestWebSocketServerTransportLifecycle:
+    """Tests for WebSocketServerTransport lifecycle."""
+
+    @pytest.mark.asyncio
+    async def test_start_creates_app(self):
+        """start() should create aiohttp app."""
+        transport = WebSocketServerTransport(port=0)
+        await transport.start()
+
+        assert transport._app is not None
+        assert transport._runner is not None
+        assert transport._site is not None
+
+        await transport.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_closes_websocket(self):
+        """stop() should close the WebSocket."""
+        transport = WebSocketServerTransport(port=0)
+        await transport.start()
+        await transport.stop()
+
+        assert transport._ws is None
+        assert transport._runner is None
+
+    @pytest.mark.asyncio
+    async def test_send_waits_for_connection(self):
+        """send() should wait for WebSocket connection."""
+        transport = WebSocketServerTransport(port=0)
+        await transport.start()
+
+        # Mock connected state
+        transport._connected.set()
+        transport._ws = MagicMock()
+        transport._ws.closed = False
+        transport._ws.send_str = AsyncMock()
+
+        await transport.send("test")
+        transport._ws.send_str.assert_called_once_with("test")
+
+        await transport.stop()
+
+    @pytest.mark.asyncio
+    async def test_send_raises_if_not_connected(self):
+        """send() should raise if WebSocket not connected."""
+        transport = WebSocketServerTransport(port=0, heartbeat=0)
+        await transport.start()
+
+        # Set timeout to 0 for immediate failure
+        with pytest.raises((RuntimeError, asyncio.TimeoutError)):
+            await transport.send("test")
+
+        await transport.stop()
+
+
+class TestWebSocketClientTransportInit:
+    """Tests for WebSocketClientTransport initialization."""
+
+    def test_required_url(self):
+        """WebSocketClientTransport requires url."""
+        transport = WebSocketClientTransport(url="ws://localhost:8080")
+        assert transport._url == "ws://localhost:8080"
+        assert transport._heartbeat == 30.0
+
+    def test_custom_heartbeat(self):
+        """WebSocketClientTransport should accept custom heartbeat."""
+        transport = WebSocketClientTransport(url="ws://localhost:8080", heartbeat=60.0)
+        assert transport._heartbeat == 60.0
+
+
+class TestWebSocketClientTransportLifecycle:
+    """Tests for WebSocketClientTransport lifecycle."""
+
+    @pytest.mark.asyncio
+    async def test_start_creates_session(self):
+        """start() should create aiohttp session and connect."""
+        server = WebSocketServerTransport(port=0)
+        await server.start()
+
+        client = WebSocketClientTransport(url=server.url)
+        await client.start()
+
+        assert client._session is not None
+        assert client._ws is not None
+
+        await client.stop()
+        await server.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_closes_session_and_websocket(self):
+        """stop() should close session and WebSocket."""
+        server = WebSocketServerTransport(port=0)
+        await server.start()
+
+        client = WebSocketClientTransport(url=server.url)
+        await client.start()
+        await client.stop()
+
+        assert client._session is None
+        assert client._ws is None
+
+        await server.stop()
+
+    @pytest.mark.asyncio
+    async def test_send_after_start(self):
+        """send() should work after start()."""
+        server = WebSocketServerTransport(port=0)
+        await server.start()
+
+        client = WebSocketClientTransport(url=server.url)
+        await client.start()
+
+        # Should not raise
+        await client.send("test message")
+
+        await client.stop()
+        await server.stop()
+
+    @pytest.mark.asyncio
+    async def test_send_raises_if_not_connected(self):
+        """send() should raise if WebSocket not connected."""
+        client = WebSocketClientTransport(url="ws://localhost:99999")
+
+        with pytest.raises(RuntimeError, match="尚未连接"):
+            await client.send("test")
+
+
+class TestTransportIntegration:
+    """Integration tests for transport pairs."""
+
+    @pytest.mark.asyncio
+    async def test_websocket_client_server_communication(self):
+        """WebSocket client and server should communicate."""
+        server = WebSocketServerTransport(port=0)
+        client = WebSocketClientTransport(url="ws://invalid")
+
+        received_messages = []
+
+        async def handle_message(payload: str):
+            received_messages.append(payload)
+
+        server.set_message_handler(handle_message)
+
+        await server.start()
+
+        # Create new client with correct URL
+        client = WebSocketClientTransport(url=server.url)
+        await client.start()
+
+        # Wait for connection
+        await asyncio.sleep(0.1)
+
+        await client.send("hello from client")
+
+        # Wait for message to be received
+        await asyncio.sleep(0.1)
+
+        assert "hello from client" in received_messages
+
+        await client.stop()
+        await server.stop()