Merge branch 'refact1/refactsome' of https://github.com/united-pooh/astrbot-sdk into refact1/refactsome

.gitignore

@@ -7,6 +7,7 @@ __pycache__/
 *.pyd
 *.so
 .pytest_cache/
+pytest-cache-files-*/
 .mypy_cache/
 .ruff_cache/
 .coverage
@@ -22,6 +23,13 @@ wheels/
 .eggs/
 pip-wheel-metadata/
 
+#
+fork-docs/
+tmp/
+openspec/
+scripts/
+docs/zh/reference
+
 # Virtual environments
 .venv/
 venv/

src-new/astrbot_sdk/clients/db.py

@@ -18,14 +18,11 @@
     - 数据永久存储，除非用户显式删除
     - 值类型支持任意 JSON 数据
     - 支持前缀查询键列表
-
-TODO:
-    - 缺少批量操作支持 (set_many, get_many)
-    - 缺少数据变更事件通知
 """
 
 from __future__ import annotations
 
+from collections.abc import AsyncIterator, Mapping, Sequence
 from typing import Any
 
 from ._proxy import CapabilityProxy
@@ -108,3 +105,67 @@ class DBClient:
         if not isinstance(keys, (list, tuple)):
             return []
         return [str(item) for item in keys]
+
+    async def get_many(self, keys: Sequence[str]) -> dict[str, Any | None]:
+        """批量获取多个键的值。
+
+        Args:
+            keys: 要读取的键列表
+
+        Returns:
+            一个 dict，key 为键名，value 为对应值（不存在则为 None）
+
+        示例:
+            values = await ctx.db.get_many(["user:1", "user:2"])
+            if values["user:1"] is None:
+                print("user:1 missing")
+        """
+        output = await self._proxy.call("db.get_many", {"keys": list(keys)})
+        items = output.get("items")
+        if not isinstance(items, (list, tuple)):
+            return {}
+        result: dict[str, Any | None] = {}
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            key = item.get("key")
+            if not isinstance(key, str):
+                continue
+            result[key] = item.get("value")
+        return result
+
+    async def set_many(
+        self, items: Mapping[str, Any] | Sequence[tuple[str, Any]]
+    ) -> None:
+        """批量写入多个键值对。
+
+        Args:
+            items: 键值对集合（dict 或二元组序列）
+
+        示例:
+            await ctx.db.set_many({"user:1": {"name": "a"}, "user:2": {"name": "b"}})
+        """
+        if isinstance(items, Mapping):
+            pairs = list(items.items())
+        else:
+            pairs = list(items)
+
+        payload_items: list[dict[str, Any]] = [
+            {"key": str(key), "value": value} for key, value in pairs
+        ]
+        await self._proxy.call("db.set_many", {"items": payload_items})
+
+    def watch(self, prefix: str | None = None) -> AsyncIterator[dict[str, Any]]:
+        """订阅 KV 变更事件（流式）。
+
+        Args:
+            prefix: 键前缀过滤；None 表示订阅所有键
+
+        Yields:
+            变更事件 dict：{"op": "set"|"delete", "key": str, "value": Any|None}
+
+        示例:
+            async for event in ctx.db.watch("user:"):
+                print(event["op"], event["key"])
+        """
+        return self._proxy.stream("db.watch", {"prefix": prefix})

src-new/astrbot_sdk/protocol/descriptors.py

@@ -127,6 +127,37 @@ DB_LIST_OUTPUT_SCHEMA = _object_schema(
     required=("keys",),
     keys={"type": "array", "items": {"type": "string"}},
 )
+DB_GET_MANY_INPUT_SCHEMA = _object_schema(
+    required=("keys",),
+    keys={"type": "array", "items": {"type": "string"}},
+)
+DB_GET_MANY_OUTPUT_SCHEMA = _object_schema(
+    required=("items",),
+    items={
+        "type": "array",
+        "items": _object_schema(
+            required=("key", "value"),
+            key={"type": "string"},
+            value=_nullable({}),
+        ),
+    },
+)
+DB_SET_MANY_INPUT_SCHEMA = _object_schema(
+    required=("items",),
+    items={
+        "type": "array",
+        "items": _object_schema(
+            required=("key", "value"),
+            key={"type": "string"},
+            value={},
+        ),
+    },
+)
+DB_SET_MANY_OUTPUT_SCHEMA = _object_schema()
+DB_WATCH_INPUT_SCHEMA = _object_schema(
+    prefix=_nullable({"type": "string"}),
+)
+DB_WATCH_OUTPUT_SCHEMA = _object_schema()
 SESSION_REF_SCHEMA = _object_schema(
     required=("conversation_id",),
     conversation_id={"type": "string"},
@@ -218,6 +249,18 @@ BUILTIN_CAPABILITY_SCHEMAS: dict[str, dict[str, JSONSchema]] = {
         "input": DB_LIST_INPUT_SCHEMA,
         "output": DB_LIST_OUTPUT_SCHEMA,
     },
+    "db.get_many": {
+        "input": DB_GET_MANY_INPUT_SCHEMA,
+        "output": DB_GET_MANY_OUTPUT_SCHEMA,
+    },
+    "db.set_many": {
+        "input": DB_SET_MANY_INPUT_SCHEMA,
+        "output": DB_SET_MANY_OUTPUT_SCHEMA,
+    },
+    "db.watch": {
+        "input": DB_WATCH_INPUT_SCHEMA,
+        "output": DB_WATCH_OUTPUT_SCHEMA,
+    },
     "platform.send": {
         "input": PLATFORM_SEND_INPUT_SCHEMA,
         "output": PLATFORM_SEND_OUTPUT_SCHEMA,
@@ -491,10 +534,16 @@ __all__ = [
     "DB_DELETE_OUTPUT_SCHEMA",
     "DB_GET_INPUT_SCHEMA",
     "DB_GET_OUTPUT_SCHEMA",
+    "DB_GET_MANY_INPUT_SCHEMA",
+    "DB_GET_MANY_OUTPUT_SCHEMA",
     "DB_LIST_INPUT_SCHEMA",
     "DB_LIST_OUTPUT_SCHEMA",
     "DB_SET_INPUT_SCHEMA",
     "DB_SET_OUTPUT_SCHEMA",
+    "DB_SET_MANY_INPUT_SCHEMA",
+    "DB_SET_MANY_OUTPUT_SCHEMA",
+    "DB_WATCH_INPUT_SCHEMA",
+    "DB_WATCH_OUTPUT_SCHEMA",
     "EventTrigger",
     "HandlerDescriptor",
     "JSONSchema",

src-new/astrbot_sdk/runtime/capability_router.py

@@ -109,6 +109,7 @@ CAPABILITY_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*$")
 class StreamExecution:
     iterator: AsyncIterator[dict[str, Any]]
     finalize: FinalizeHandler
+    collect_chunks: bool = True
 
 
 StreamHandler = Callable[
@@ -134,8 +135,18 @@ class CapabilityRouter:
         self.db_store: dict[str, Any] = {}
         self.memory_store: dict[str, dict[str, Any]] = {}
         self.sent_messages: list[dict[str, Any]] = []
+        self._db_watch_subscriptions: dict[
+            str, tuple[str | None, asyncio.Queue[dict[str, Any]]]
+        ] = {}
         self._register_builtin_capabilities()
 
+    def _emit_db_change(self, *, op: str, key: str, value: Any | None) -> None:
+        event = {"op": op, "key": key, "value": value}
+        for prefix, queue in list(self._db_watch_subscriptions.values()):
+            if prefix is not None and not key.startswith(prefix):
+                continue
+            queue.put_nowait(event)
+
     def descriptors(self) -> list[CapabilityDescriptor]:
         return [
             entry.descriptor for entry in self._registrations.values() if entry.exposed
@@ -227,6 +238,7 @@ class CapabilityRouter:
         return StreamExecution(
             iterator=execution.iterator,
             finalize=validated_finalize,
+            collect_chunks=execution.collect_chunks,
         )
 
     def _register_builtin_capabilities(self) -> None:
@@ -329,13 +341,17 @@ class CapabilityRouter:
             _request_id: str, payload: dict[str, Any], _token
         ) -> dict[str, Any]:
             key = str(payload.get("key", ""))
-            self.db_store[key] = payload.get("value")
+            value = payload.get("value")
+            self.db_store[key] = value
+            self._emit_db_change(op="set", key=key, value=value)
             return {}
 
         async def db_delete(
             _request_id: str, payload: dict[str, Any], _token
         ) -> dict[str, Any]:
-            self.db_store.pop(str(payload.get("key", "")), None)
+            key = str(payload.get("key", ""))
+            self.db_store.pop(key, None)
+            self._emit_db_change(op="delete", key=key, value=None)
             return {}
 
         async def db_list(
@@ -347,6 +363,63 @@ class CapabilityRouter:
                 keys = [item for item in keys if item.startswith(prefix)]
             return {"keys": keys}
 
+        async def db_get_many(
+            _request_id: str, payload: dict[str, Any], _token
+        ) -> dict[str, Any]:
+            keys_payload = payload.get("keys")
+            if not isinstance(keys_payload, (list, tuple)):
+                raise AstrBotError.invalid_input("db.get_many 的 keys 必须是数组")
+            keys = [str(item) for item in keys_payload]
+            items = [{"key": key, "value": self.db_store.get(key)} for key in keys]
+            return {"items": items}
+
+        async def db_set_many(
+            _request_id: str, payload: dict[str, Any], _token
+        ) -> dict[str, Any]:
+            items_payload = payload.get("items")
+            if not isinstance(items_payload, (list, tuple)):
+                raise AstrBotError.invalid_input("db.set_many 的 items 必须是数组")
+            for entry in items_payload:
+                if not isinstance(entry, dict):
+                    raise AstrBotError.invalid_input(
+                        "db.set_many 的 items 必须是 object 数组"
+                    )
+                key = str(entry.get("key", ""))
+                value = entry.get("value")
+                self.db_store[key] = value
+                self._emit_db_change(op="set", key=key, value=value)
+            return {}
+
+        async def db_watch(
+            request_id: str, payload: dict[str, Any], _token
+        ) -> StreamExecution:
+            prefix = payload.get("prefix")
+            prefix_value: str | None
+            if isinstance(prefix, str):
+                prefix_value = prefix
+            elif prefix is None:
+                prefix_value = None
+            else:
+                raise AstrBotError.invalid_input(
+                    "db.watch 的 prefix 必须是 string 或 null"
+                )
+
+            queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
+            self._db_watch_subscriptions[request_id] = (prefix_value, queue)
+
+            async def iterator() -> AsyncIterator[dict[str, Any]]:
+                try:
+                    while True:
+                        yield await queue.get()
+                finally:
+                    self._db_watch_subscriptions.pop(request_id, None)
+
+            return StreamExecution(
+                iterator=iterator(),
+                finalize=lambda _chunks: {},
+                collect_chunks=False,
+            )
+
         async def platform_send(
             _request_id: str, payload: dict[str, Any], _token
         ) -> dict[str, Any]:
@@ -460,6 +533,23 @@ class CapabilityRouter:
             builtin_descriptor("db.list", "列出 KV"),
             call_handler=db_list,
         )
+        self.register(
+            builtin_descriptor("db.get_many", "批量读取 KV"),
+            call_handler=db_get_many,
+        )
+        self.register(
+            builtin_descriptor("db.set_many", "批量写入 KV"),
+            call_handler=db_set_many,
+        )
+        self.register(
+            builtin_descriptor(
+                "db.watch",
+                "订阅 KV 变更",
+                supports_stream=True,
+                cancelable=True,
+            ),
+            stream_handler=db_watch,
+        )
         self.register(
             builtin_descriptor("platform.send", "发送消息"),
             call_handler=platform_send,

src-new/astrbot_sdk/runtime/peer.py

@@ -507,9 +507,11 @@ class Peer:
                         "stream=true 必须返回 StreamExecution"
                     )
                 await self._send(EventMessage(id=message.id, phase="started"))
+                collect_chunks = execution.collect_chunks
                 chunks: list[dict[str, Any]] = []
                 async for chunk in execution.iterator:
-                    chunks.append(chunk)
+                    if collect_chunks:
+                        chunks.append(chunk)
                     await self._send(
                         EventMessage(id=message.id, phase="delta", data=chunk)
                     )

tests_v4/conftest.py

@@ -1,8 +1,8 @@
-"""Pytest shared fixtures and test bootstrap helpers."""
+"""Pytest共享的fixture和测试引导辅助函数。"""
 
-# ruff: noqa: E402
+# ruff: noqa: E402  # 忽略E402（模块导入顺序）警告
 
-# Test configuration
+# 测试配置
 import asyncio
 import sys
 import tempfile
@@ -13,89 +13,144 @@ from typing import Any
 
 import pytest
 
-# 将 src-new 加入路径 - 这使得测试可以运行，但不算"已安装"
+# 将src-new目录添加到Python路径 - 这使得测试可以运行，但不算作"已安装"的包
 SRC_NEW_PATH = str(Path(__file__).parent.parent / "src-new")
 sys.path.insert(0, SRC_NEW_PATH)
 
 
 # ============================================================
-# Async Configuration
+# 异步测试配置
 # ============================================================
 
 
 @pytest.fixture(scope="session")
 def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
-    """Create an event loop for async tests."""
-    policy = asyncio.get_event_loop_policy()
-    loop = policy.new_event_loop()
-    yield loop
-    loop.close()
+    """为异步测试创建事件循环。
+
+    这是一个会话级别的fixture，在整个测试会话期间只创建一次事件循环。
+    """
+    policy = asyncio.get_event_loop_policy()  # 获取当前事件循环策略
+    loop = policy.new_event_loop()  # 创建新的事件循环
+    yield loop  # 提供事件循环给测试使用
+    loop.close()  # 测试结束后关闭事件循环
 
 
 # ============================================================
-# Transport Fixtures
+# 传输层Fixture（用于模拟网络通信）
 # ============================================================
 
 
 class MemoryTransport:
-    """In-memory transport for testing peer communication."""
+    """用于测试对等通信的内存传输模拟。
+
+    这个类模拟了两个对等方之间的通信通道，所有消息都在内存中传递，
+    无需实际网络连接。
+    """
 
     def __init__(self) -> None:
-        self._closed = asyncio.Event()
-        self._message_handler = None
-        self.partner: MemoryTransport | None = None
+        self._closed = asyncio.Event()  # 用于跟踪传输是否已关闭
+        self._message_handler = None  # 消息处理函数
+        self.partner: MemoryTransport | None = None  # 通信伙伴
 
     def set_message_handler(self, handler) -> None:
+        """设置消息处理函数。
+
+        Args:
+            handler: 接收消息的异步函数
+        """
         self._message_handler = handler
 
     async def start(self) -> None:
+        """启动传输。
+
+        清除关闭状态，使传输可用。
+        """
         self._closed.clear()
 
     async def stop(self) -> None:
+        """停止传输。
+
+        设置关闭事件，表示传输已停止。
+        """
         self._closed.set()
 
     async def wait_closed(self) -> None:
+        """等待传输关闭。
+
+        阻塞直到传输完全关闭。
+        """
         await self._closed.wait()
 
     async def send(self, payload: str) -> None:
+        """发送消息给伙伴。
+
+        Args:
+            payload: 要发送的消息内容
+
+        Raises:
+            RuntimeError: 如果没有设置伙伴传输
+        """
         if self.partner is None:
             raise RuntimeError("MemoryTransport 未连接 partner")
         if self.partner._message_handler is not None:
-            await self.partner._message_handler(payload)
+            await self.partner._message_handler(payload)  # 将消息分发给伙伴的处理函数
 
     async def _dispatch(self, payload: str) -> None:
+        """内部方法：将消息分发给本地处理函数。
+
+        Args:
+            payload: 接收到的消息内容
+        """
         if self._message_handler is not None:
             await self._message_handler(payload)
 
 
 @pytest.fixture
 def transport_pair() -> tuple[MemoryTransport, MemoryTransport]:
-    """Create a connected pair of in-memory transports."""
+    """创建一对相互连接的内存传输实例。
+
+    返回的左右两个传输实例互为伙伴，可用于模拟两个对等方之间的通信。
+
+    Returns:
+        (left_transport, right_transport) 的元组
+    """
     left = MemoryTransport()
     right = MemoryTransport()
-    left.partner = right
-    right.partner = left
+    left.partner = right  # 左传输的伙伴是右传输
+    right.partner = left  # 右传输的伙伴是左传输
     return left, right
 
 
 # ============================================================
-# Mock/Fake Fixtures
+# 模拟/Fake Fixture（用于测试的假对象）
 # ============================================================
 
 
 class FakeEnvManager:
-    """Fake environment manager for testing."""
+    """用于测试的虚假环境管理器。
+
+    模拟真实的环境管理器行为，但不执行实际的环境准备操作。
+    """
 
     def prepare_environment(self, _plugin: Any) -> Path:
+        """模拟准备插件环境。
+
+        Args:
+            _plugin: 插件对象（未使用）
+
+        Returns:
+            返回当前Python解释器路径作为模拟的环境路径
+        """
         return Path(sys.executable)
 
 
 @pytest.fixture
 def fake_env_manager() -> FakeEnvManager:
-    """Provide a fake environment manager."""
+    """提供一个虚假的环境管理器fixture。"""
     return FakeEnvManager()
 
 
+# 导入需要使用的类型
 from astrbot_sdk.protocol.messages import InitializeOutput, PeerInfo
 from astrbot_sdk.runtime.capability_router import CapabilityRouter
 from astrbot_sdk.runtime.peer import Peer
@@ -103,66 +158,111 @@ from astrbot_sdk.runtime.peer import Peer
 
 @pytest.fixture
 async def core_peer(transport_pair: tuple[MemoryTransport, MemoryTransport]) -> Peer:
-    """Create a core peer with default handlers."""
-    left, _ = transport_pair
-    router = CapabilityRouter()
+    """创建一个配置了默认处理函数的核心对等方。
 
+    这个fixture创建并启动一个核心角色的对等方，设置了初始化和调用处理函数。
+
+    Args:
+        transport_pair: 传输对fixture
+
+    Returns:
+        已启动的核心对等方实例
+    """
+    left, _ = transport_pair  # 使用传输对中的左传输
+    router = CapabilityRouter()  # 创建能力路由器
+
+    # 创建核心对等方
     peer = Peer(
         transport=left,
         peer_info=PeerInfo(name="core", role="core", version="v4"),
     )
 
+    # 定义初始化处理函数
     async def init_handler(_message) -> InitializeOutput:
         return InitializeOutput(
             peer=PeerInfo(name="core", role="core", version="v4"),
-            capabilities=router.descriptors(),
+            capabilities=router.descriptors(),  # 获取路由器描述的能力列表
             metadata={},
         )
 
+    # 定义调用处理函数
     async def invoke_handler(message, token):
         return await router.execute(
-            message.capability,
-            message.input,
-            stream=message.stream,
-            cancel_token=token,
-            request_id=message.id,
+            message.capability,  # 要执行的能力
+            message.input,       # 输入参数
+            stream=message.stream,  # 是否流式输出
+            cancel_token=token,     # 取消令牌
+            request_id=message.id,  # 请求ID
         )
 
+    # 设置处理函数
     peer.set_initialize_handler(init_handler)
     peer.set_invoke_handler(invoke_handler)
 
-    await peer.start()
+    await peer.start()  # 启动对等方
     yield peer
-    await peer.stop()
+    await peer.stop()  # 测试结束后停止
 
 
 @pytest.fixture
 async def plugin_peer(transport_pair: tuple[MemoryTransport, MemoryTransport]) -> Peer:
-    """Create a plugin peer connected to core."""
-    _, right = transport_pair
+    """创建一个连接到核心的插件对等方。
 
+    这个fixture创建并启动一个插件角色的对等方。
+
+    Args:
+        transport_pair: 传输对fixture
+
+    Returns:
+        已启动的插件对等方实例
+    """
+    _, right = transport_pair  # 使用传输对中的右传输
+
+    # 创建插件对等方
     peer = Peer(
         transport=right,
         peer_info=PeerInfo(name="plugin", role="plugin", version="v4"),
     )
 
-    await peer.start()
+    await peer.start()  # 启动对等方
     yield peer
-    await peer.stop()
+    await peer.stop()  # 测试结束后停止
 
 
 @pytest.fixture
 def temp_plugin_dir() -> Generator[Path, None, None]:
-    """Create a temporary directory for plugin testing."""
+    """创建用于插件测试的临时目录。
+
+    这个fixture创建一个临时目录，并在测试结束后自动清理。
+
+    Yields:
+        临时目录的Path对象
+    """
     with tempfile.TemporaryDirectory() as temp_dir:
         yield Path(temp_dir)
 
 
 def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None:
-    """Helper to create a minimal test plugin."""
+    """辅助函数：创建一个最小的测试插件。
+
+    在指定目录创建插件所需的基本文件结构：
+    - commands/__init__.py
+    - commands/sample.py（包含测试命令）
+    - requirements.txt（空文件）
+    - plugin.yaml（插件配置文件）
+
+    Args:
+        plugin_root: 插件根目录
+        name: 插件名称，默认为"test_plugin"
+    """
+    # 创建commands目录和__init__.py文件
     (plugin_root / "commands").mkdir(parents=True, exist_ok=True)
     (plugin_root / "commands" / "__init__.py").write_text("", encoding="utf-8")
+
+    # 创建空的requirements.txt
     (plugin_root / "requirements.txt").write_text("", encoding="utf-8")
+
+    # 创建插件配置文件plugin.yaml
     (plugin_root / "plugin.yaml").write_text(
         textwrap.dedent(
             f"""\
@@ -173,7 +273,7 @@ def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None:
             author: tester
             version: 0.1.0
             runtime:
-              python: "{sys.version_info.major}.{sys.version_info.minor}"
+              python: "{sys.version_info.major}.{sys.version_info.minor}"  # 使用当前Python版本
             components:
               - class: commands.sample:TestPlugin
                 type: command
@@ -183,6 +283,8 @@ def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None:
         ),
         encoding="utf-8",
     )
+
+    # 创建测试命令文件sample.py
     (plugin_root / "commands" / "sample.py").write_text(
         textwrap.dedent(
             """\
@@ -190,9 +292,9 @@ def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None:
 
 
             class TestPlugin(Star):
-                @on_command("test")
+                @on_command("test")  # 注册test命令
                 async def test_cmd(self, event: MessageEvent, ctx: Context):
-                    await event.reply("test ok")
+                    await event.reply("test ok")  # 回复消息
             """
         ),
         encoding="utf-8",
@@ -201,7 +303,16 @@ def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None:
 
 @pytest.fixture
 def test_plugin(temp_plugin_dir: Path) -> Path:
-    """Create a test plugin and return its root directory."""
+    """创建一个测试插件并返回其根目录。
+
+    这个fixture使用create_test_plugin函数创建插件，并返回插件目录路径。
+
+    Args:
+        temp_plugin_dir: 临时插件目录fixture
+
+    Returns:
+        包含插件的目录路径
+    """
     plugin_root = temp_plugin_dir / "plugins" / "test_plugin"
-    create_test_plugin(plugin_root)
-    return temp_plugin_dir / "plugins"
+    create_test_plugin(plugin_root)  # 创建测试插件
+    return temp_plugin_dir / "plugins"

tests_v4/helpers.py

@@ -1,4 +1,4 @@
-from __future__ import annotations
+from __future__ import annotations  # 启用延迟类型注解求值，避免循环引用
 
 import asyncio
 from pathlib import Path
@@ -7,34 +7,91 @@ from astrbot_sdk.runtime.transport import Transport
 
 
 class MemoryTransport(Transport):
+    """基于内存的传输层实现，用于测试场景。
+
+    继承自Transport基类，模拟两个对等方之间的通信，所有消息在内存中传递，
+    无需实际网络连接。主要用于单元测试。
+    """
+
     def __init__(self) -> None:
-        super().__init__()
-        self.partner: "MemoryTransport | None" = None
+        """初始化内存传输实例。"""
+        super().__init__()  # 调用父类初始化方法
+        self.partner: "MemoryTransport | None" = None  # 通信伙伴，可以是对等的另一个MemoryTransport实例
 
     async def start(self) -> None:
-        self._closed.clear()
+        """启动传输。
+
+        通过清除关闭标志使传输变为可用状态。
+        """
+        self._closed.clear()  # 清除关闭事件
 
     async def stop(self) -> None:
-        self._closed.set()
+        """停止传输。
+
+        设置关闭标志，表示传输已停止。
+        """
+        self._closed.set()  # 设置关闭事件
 
     async def send(self, payload: str) -> None:
+        """发送消息给伙伴。
+
+        Args:
+            payload: 要发送的消息内容字符串
+
+        Raises:
+            RuntimeError: 如果没有设置伙伴传输（即self.partner为None）
+        """
         if self.partner is None:
             raise RuntimeError("MemoryTransport 未连接 partner")
+        # 将消息转发给伙伴的_dispatch方法进行处理
         await self.partner._dispatch(payload)
 
 
 def make_transport_pair() -> tuple[MemoryTransport, MemoryTransport]:
-    left = MemoryTransport()
-    right = MemoryTransport()
-    left.partner = right
-    right.partner = left
-    return left, right
+    """创建一对相互连接的内存传输实例。
+
+    工厂函数，用于创建两个互为通信伙伴的MemoryTransport实例，
+    简化测试设置过程。
+
+    Returns:
+        tuple[MemoryTransport, MemoryTransport]: 返回左右两个传输实例的元组，
+        它们已互相设置为伙伴关系
+    """
+    left = MemoryTransport()  # 创建左侧传输实例
+    right = MemoryTransport()  # 创建右侧传输实例
+    left.partner = right  # 设置左侧的伙伴为右侧
+    right.partner = left  # 设置右侧的伙伴为左侧
+    return left, right  # 返回配对的传输实例
 
 
 class FakeEnvManager:
+    """虚假的环境管理器，用于测试。
+
+    模拟真实环境管理器的行为，但不执行实际的环境准备操作，
+    主要用于需要环境管理器但又不希望产生副作用的测试场景。
+    """
+
     def prepare_environment(self, _plugin) -> Path:
+        """模拟准备插件环境的方法。
+
+        不实际创建虚拟环境，而是返回当前Python解释器路径作为模拟的环境路径。
+
+        Args:
+            _plugin: 插件对象参数，在模拟实现中未使用
+
+        Returns:
+            Path: 当前Python解释器的路径
+        """
+        # 动态导入sys模块并返回可执行文件路径
         return Path(__import__("sys").executable)
 
 
 async def drain_loop() -> None:
-    await asyncio.sleep(0.05)
+    """清空事件循环的辅助函数。
+
+    通过短暂的异步睡眠，让事件循环有机会处理所有已排队的任务和回调。
+    常用于测试中等待异步操作完成，确保所有待处理的事件被处理。
+
+    睡眠时间(0.05秒)足够短不会明显减慢测试，又足够长让事件循环有机会处理任务。
+    """
+    await asyncio.sleep(0.05)  # 暂停当前协程50毫秒，让出控制权给事件循环

tests_v4/test_capability_router.py

@@ -164,6 +164,9 @@ class TestCapabilityRouterInit:
         assert "db.set" in capability_names
         assert "db.delete" in capability_names
         assert "db.list" in capability_names
+        assert "db.get_many" in capability_names
+        assert "db.set_many" in capability_names
+        assert "db.watch" in capability_names
 
         # Platform capabilities
         assert "platform.send" in capability_names
@@ -182,6 +185,14 @@ class TestCapabilityRouterInit:
             assert descriptors[name].input_schema == schema["input"]
             assert descriptors[name].output_schema == schema["output"]
 
+    def test_db_watch_descriptor_supports_stream(self):
+        router = CapabilityRouter()
+        descriptors = {
+            descriptor.name: descriptor for descriptor in router.descriptors()
+        }
+
+        assert descriptors["db.watch"].supports_stream is True
+
 
 class TestCapabilityRouterRegister:
     """Tests for CapabilityRouter.register method."""
@@ -350,6 +361,85 @@ class TestCapabilityRouterExecute:
                 request_id="req-1",
             )
 
+
+class TestCapabilityRouterDBWatch:
+    """Router-level tests for db.watch behavior."""
+
+    @pytest.mark.asyncio
+    async def test_db_watch_receives_set_and_delete_events(self):
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        execution = await router.execute(
+            "db.watch",
+            {"prefix": None},
+            stream=True,
+            cancel_token=token,
+            request_id="watch-1",
+        )
+        assert isinstance(execution, StreamExecution)
+        assert execution.collect_chunks is False
+
+        await router.execute(
+            "db.set",
+            {"key": "a", "value": 1},
+            stream=False,
+            cancel_token=token,
+            request_id="set-1",
+        )
+        await router.execute(
+            "db.delete",
+            {"key": "a"},
+            stream=False,
+            cancel_token=token,
+            request_id="del-1",
+        )
+
+        event1 = await anext(execution.iterator)
+        event2 = await anext(execution.iterator)
+        assert event1 == {"op": "set", "key": "a", "value": 1}
+        assert event2 == {"op": "delete", "key": "a", "value": None}
+
+        close = getattr(execution.iterator, "aclose", None)
+        if close is not None:
+            await close()
+
+    @pytest.mark.asyncio
+    async def test_db_watch_prefix_filters_events(self):
+        router = CapabilityRouter()
+        token = CancelToken()
+
+        execution = await router.execute(
+            "db.watch",
+            {"prefix": "user:"},
+            stream=True,
+            cancel_token=token,
+            request_id="watch-2",
+        )
+        assert isinstance(execution, StreamExecution)
+
+        await router.execute(
+            "db.set",
+            {"key": "sys:1", "value": 1},
+            stream=False,
+            cancel_token=token,
+            request_id="set-2",
+        )
+        await router.execute(
+            "db.set",
+            {"key": "user:1", "value": {"ok": True}},
+            stream=False,
+            cancel_token=token,
+            request_id="set-3",
+        )
+
+        event = await anext(execution.iterator)
+        assert event == {"op": "set", "key": "user:1", "value": {"ok": True}}
+
+        close = getattr(execution.iterator, "aclose", None)
+        if close is not None:
+            await close()
+
     @pytest.mark.asyncio
     async def test_execute_missing_capability_raises(self):
         """execute should raise for unknown capability."""

tests_v4/test_db_client.py

@@ -237,3 +237,103 @@ class TestDBClientList:
 
         proxy.call.assert_called_once_with("db.list", {"prefix": None})
         assert result == []
+
+
+class TestDBClientGetMany:
+    """Tests for DBClient.get_many() method."""
+
+    @pytest.mark.asyncio
+    async def test_get_many_returns_mapping(self):
+        proxy = MagicMock(spec=CapabilityProxy)
+        proxy.call = AsyncMock(
+            return_value={
+                "items": [
+                    {"key": "a", "value": 1},
+                    {"key": "b", "value": None},
+                ]
+            }
+        )
+        client = DBClient(proxy)
+
+        result = await client.get_many(["a", "b"])
+
+        proxy.call.assert_called_once_with("db.get_many", {"keys": ["a", "b"]})
+        assert result == {"a": 1, "b": None}
+
+    @pytest.mark.asyncio
+    async def test_get_many_returns_empty_dict_for_malformed_items(self):
+        proxy = MagicMock(spec=CapabilityProxy)
+        proxy.call = AsyncMock(return_value={"items": "not-a-list"})
+        client = DBClient(proxy)
+
+        result = await client.get_many(["a"])
+
+        assert result == {}
+
+
+class TestDBClientSetMany:
+    """Tests for DBClient.set_many() method."""
+
+    @pytest.mark.asyncio
+    async def test_set_many_accepts_mapping(self):
+        proxy = MagicMock(spec=CapabilityProxy)
+        proxy.call = AsyncMock(return_value={})
+        client = DBClient(proxy)
+
+        await client.set_many({"a": 1, "b": 2})
+
+        proxy.call.assert_called_once_with(
+            "db.set_many",
+            {"items": [{"key": "a", "value": 1}, {"key": "b", "value": 2}]},
+        )
+
+    @pytest.mark.asyncio
+    async def test_set_many_accepts_sequence_pairs(self):
+        proxy = MagicMock(spec=CapabilityProxy)
+        proxy.call = AsyncMock(return_value={})
+        client = DBClient(proxy)
+
+        await client.set_many([("a", True), ("b", {"x": 1})])
+
+        proxy.call.assert_called_once_with(
+            "db.set_many",
+            {"items": [{"key": "a", "value": True}, {"key": "b", "value": {"x": 1}}]},
+        )
+
+
+class TestDBClientWatch:
+    """Tests for DBClient.watch() method."""
+
+    @pytest.mark.asyncio
+    async def test_watch_calls_proxy_stream_and_yields_events(self):
+        async def gen():
+            yield {"op": "set", "key": "a", "value": 1}
+            yield {"op": "delete", "key": "a", "value": None}
+
+        proxy = MagicMock(spec=CapabilityProxy)
+        proxy.stream = MagicMock(return_value=gen())
+        client = DBClient(proxy)
+
+        iterator = client.watch()
+
+        proxy.stream.assert_called_once_with("db.watch", {"prefix": None})
+        events = [event async for event in iterator]
+        assert events == [
+            {"op": "set", "key": "a", "value": 1},
+            {"op": "delete", "key": "a", "value": None},
+        ]
+
+    @pytest.mark.asyncio
+    async def test_watch_with_prefix(self):
+        async def gen():
+            yield {"op": "set", "key": "user:1", "value": {"ok": True}}
+
+        proxy = MagicMock(spec=CapabilityProxy)
+        proxy.stream = MagicMock(return_value=gen())
+        client = DBClient(proxy)
+
+        iterator = client.watch(prefix="user:")
+
+        proxy.stream.assert_called_once_with("db.watch", {"prefix": "user:"})
+        events = [event async for event in iterator]
+        assert events == [{"op": "set", "key": "user:1", "value": {"ok": True}}]