feat: 更新调度触发器以支持别名，优化 Peer 类的关闭逻辑，增强插件发现功能

src-new/astrbot_sdk/protocol/descriptors.py

@@ -40,7 +40,7 @@ from __future__ import annotations
 
 from typing import Annotated, Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
 
 
 class _DescriptorBase(BaseModel):
@@ -137,9 +137,16 @@ class ScheduleTrigger(_DescriptorBase):
     """
 
     type: Literal["schedule"] = "schedule"
-    cron: str | None = None
+    cron: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices("cron", "schedule"),
+    )
     interval_seconds: int | None = None
 
+    @property
+    def schedule(self) -> str | None:
+        return self.cron
+
     @model_validator(mode="after")
     def validate_schedule(self) -> "ScheduleTrigger":
         has_cron = self.cron is not None

src-new/astrbot_sdk/runtime/bootstrap.py

@@ -134,6 +134,13 @@ def _prepare_stdio_transport(
     return transport_stdin, transport_stdout, original_stdout
 
 
+def _sdk_source_dir(repo_root: Path) -> Path:
+    candidate = repo_root.resolve() / "src-new"
+    if (candidate / "astrbot_sdk").exists():
+        return candidate
+    return Path(__file__).resolve().parents[2]
+
+
 async def _wait_for_shutdown(peer: Peer, stop_event: asyncio.Event) -> None:
     stop_waiter = asyncio.create_task(stop_event.wait())
     transport_waiter = asyncio.create_task(peer.wait_closed())
@@ -168,7 +175,7 @@ class WorkerSession:
 
     async def start(self) -> None:
         python_path = self.env_manager.prepare_environment(self.plugin)
-        repo_src_dir = str(self.repo_root / "src-new")
+        repo_src_dir = str(_sdk_source_dir(self.repo_root))
         env = os.environ.copy()
         existing_pythonpath = env.get("PYTHONPATH")
         env["PYTHONPATH"] = (

src-new/astrbot_sdk/runtime/handler_dispatcher.py

@@ -97,7 +97,7 @@ class HandlerDispatcher:
             peer=self._peer, plugin_id=self._plugin_id, cancel_token=cancel_token
         )
         event = MessageEvent.from_payload(message.input.get("event", {}))
-        event.bind_reply_handler(lambda text: ctx.platform.send(event.session_id, text))
+        event.bind_reply_handler(self._create_reply_handler(ctx, event))
         if loaded.legacy_context is not None:
             loaded.legacy_context.bind_runtime_context(ctx)
 
@@ -112,6 +112,20 @@ class HandlerDispatcher:
         finally:
             self._active.pop(message.id, None)
 
+    def _create_reply_handler(self, ctx: Context, event: MessageEvent):
+        async def reply(text: str) -> None:
+            try:
+                await ctx.platform.send(event.session_id, text)
+            except TypeError:
+                send = getattr(self._peer, "send", None)
+                if not callable(send):
+                    raise
+                result = send(event.session_id, text)
+                if inspect.isawaitable(result):
+                    await result
+
+        return reply
+
     async def cancel(self, request_id: str) -> None:
         active = self._active.get(request_id)
         if active is None:

src-new/astrbot_sdk/runtime/loader.py

@@ -222,8 +222,8 @@ def discover_plugins(plugins_dir: Path) -> PluginDiscoveryResult:
         if plugin_name in seen_names:
             skipped_plugins[plugin_name] = "duplicate plugin name"
             continue
-        if not isinstance(components, list) or not components:
-            skipped_plugins[plugin_name] = "components must be a non-empty list"
+        if not isinstance(components, list):
+            skipped_plugins[plugin_name] = "components must be a list"
             continue
         if not isinstance(python_version, str) or not python_version:
             skipped_plugins[plugin_name] = "runtime.python is required"

src-new/astrbot_sdk/runtime/peer.py

@@ -63,6 +63,7 @@ TODO:
 from __future__ import annotations
 
 import asyncio
+import inspect
 from collections.abc import AsyncIterator, Awaitable, Callable
 from typing import Any
 
@@ -123,7 +124,7 @@ class Peer:
         self._invoke_handler: InvokeHandler | None = None
         self._cancel_handler: CancelHandler | None = None
         self._counter = 0
-        self._closed = False
+        self._closed = asyncio.Event()
         self._unusable = False
         self._pending_results: dict[str, asyncio.Future[ResultMessage]] = {}
         self._pending_streams: dict[str, asyncio.Queue[Any]] = {}
@@ -146,12 +147,14 @@ class Peer:
 
     async def start(self) -> None:
         """启动传输层并将原始入站消息绑定到当前 `Peer`。"""
+        self._closed.clear()
         self.transport.set_message_handler(self._handle_raw_message)
         await self.transport.start()
 
     async def stop(self) -> None:
         """关闭 `Peer` 并清理所有挂起中的请求、流和入站任务。"""
-        self._closed = True
+        if self._closed.is_set():
+            return
         # 终止所有挂起的 RPC，避免调用方永久挂起
         for future in list(self._pending_results.values()):
             if not future.done():
@@ -169,6 +172,7 @@ class Peer:
         self._inbound_tasks.clear()
 
         await self.transport.stop()
+        self._closed.set()
 
     async def wait_closed(self) -> None:
         """等待底层传输彻底关闭。"""
@@ -410,6 +414,8 @@ class Peer:
             if self._invoke_handler is None:
                 raise AstrBotError.capability_not_found(message.capability)
             execution = await self._invoke_handler(message, token)
+            if inspect.isawaitable(execution):
+                execution = await execution
             if message.stream:
                 if not isinstance(execution, StreamExecution):
                     raise AstrBotError.protocol_error(

tests_v4/test_clients_module.py

@@ -5,7 +5,6 @@ Tests for clients/__init__.py - Module exports.
 from __future__ import annotations
 
 
-
 class TestClientsModuleExports:
     """Tests for clients module exports."""
 

tests_v4/test_handler_dispatcher.py

@@ -370,6 +370,56 @@ class TestHandlerDispatcherInvoke:
         assert received_types == [AstrMessageEvent]
         assert replies == [{"session_id": "session-legacy", "text": "legacy reply"}]
 
+    @pytest.mark.asyncio
+    async def test_invoke_reply_falls_back_to_peer_send_for_sync_mock(self):
+        """invoke should fall back to peer.send when peer.invoke is a sync mock."""
+        peer = MagicMock()
+        peer.remote_capability_map = {}
+        sent_messages = []
+
+        async def track_send(session_id: str, text: str) -> None:
+            sent_messages.append({"session_id": session_id, "text": text})
+
+        peer.send = track_send
+
+        async def handler_func(event: MessageEvent, ctx: Context):
+            await event.reply("fallback")
+
+        descriptor = HandlerDescriptor(
+            id="test.handler",
+            trigger=CommandTrigger(command="hello"),
+        )
+        handler = LoadedHandler(
+            descriptor=descriptor,
+            callable=handler_func,
+            owner=MagicMock(),
+            legacy_context=None,
+        )
+
+        dispatcher = HandlerDispatcher(
+            plugin_id="test_plugin",
+            peer=peer,
+            handlers=[handler],
+        )
+
+        message = InvokeMessage(
+            id="msg_sync_mock",
+            capability="handler.invoke",
+            input={
+                "handler_id": "test.handler",
+                "event": {
+                    "text": "hello",
+                    "session_id": "session-sync",
+                    "user_id": "user-1",
+                    "platform": "test",
+                },
+            },
+        )
+
+        await dispatcher.invoke(message, CancelToken())
+
+        assert sent_messages == [{"session_id": "session-sync", "text": "fallback"}]
+
 
 class TestHandlerDispatcherCancel:
     """Tests for HandlerDispatcher.cancel method."""

tests_v4/test_loader.py

@@ -425,7 +425,7 @@ class TestDiscoverPlugins:
             assert "duplicate_name" in result.skipped_plugins
 
     def test_validates_components_list(self):
-        """discover_plugins should validate components is a non-empty list."""
+        """discover_plugins should validate components is a list."""
         with tempfile.TemporaryDirectory() as temp_dir:
             plugins_dir = Path(temp_dir)
 
@@ -448,6 +448,30 @@ class TestDiscoverPlugins:
             assert "test" in result.skipped_plugins
             assert "components" in result.skipped_plugins["test"]
 
+    def test_allows_empty_components_list(self):
+        """discover_plugins should allow plugins without components."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            plugins_dir = Path(temp_dir)
+
+            plugin_dir = plugins_dir / "empty_components"
+            plugin_dir.mkdir()
+            (plugin_dir / "plugin.yaml").write_text(
+                yaml.dump(
+                    {
+                        "name": "empty_components",
+                        "runtime": {"python": "3.12"},
+                        "components": [],
+                    }
+                ),
+                encoding="utf-8",
+            )
+            (plugin_dir / "requirements.txt").write_text("", encoding="utf-8")
+
+            result = discover_plugins(plugins_dir)
+
+            assert [plugin.name for plugin in result.plugins] == ["empty_components"]
+            assert result.skipped_plugins == {}
+
     def test_discovers_valid_plugin(self):
         """discover_plugins should discover valid plugin."""
         with tempfile.TemporaryDirectory() as temp_dir:

tests_v4/test_protocol_descriptors.py

@@ -152,6 +152,12 @@ class TestScheduleTrigger:
         assert trigger.interval_seconds == 60
         assert trigger.cron is None
 
+    def test_accepts_schedule_alias(self):
+        """ScheduleTrigger should accept legacy schedule alias for cron."""
+        trigger = ScheduleTrigger(schedule="0 */5 * * * *")
+        assert trigger.cron == "0 */5 * * * *"
+        assert trigger.schedule == "0 */5 * * * *"
+
     def test_requires_exactly_one_strategy(self):
         """ScheduleTrigger must have exactly one of cron or interval_seconds."""
         # Neither provided should raise

tests_v4/test_runtime_integration.py

@@ -8,9 +8,8 @@ from __future__ import annotations
 import asyncio
 import sys
 import tempfile
-import textwrap
 from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 import yaml
@@ -27,26 +26,20 @@ from astrbot_sdk.protocol.descriptors import (
     ScheduleTrigger,
 )
 from astrbot_sdk.protocol.messages import (
-    EventMessage,
-    InitializeMessage,
     InitializeOutput,
     InvokeMessage,
     PeerInfo,
-    ResultMessage,
 )
 from astrbot_sdk.runtime.bootstrap import (
-    PluginWorkerRuntime,
     SupervisorRuntime,
     WorkerSession,
 )
-from astrbot_sdk.runtime.capability_router import CapabilityRouter, StreamExecution
+from astrbot_sdk.runtime.capability_router import CapabilityRouter
 from astrbot_sdk.runtime.handler_dispatcher import HandlerDispatcher
 from astrbot_sdk.runtime.loader import (
     LoadedHandler,
-    LoadedPlugin,
     PluginEnvironmentManager,
     PluginSpec,
-    load_plugin_spec,
 )
 from astrbot_sdk.runtime.peer import Peer
 
@@ -92,7 +85,9 @@ class TestWorkerSessionSubprocessLifecycle:
                 yaml.dump(
                     {
                         "name": "crash_plugin",
-                        "runtime": {"python": f"{sys.version_info.major}.{sys.version_info.minor}"},
+                        "runtime": {
+                            "python": f"{sys.version_info.major}.{sys.version_info.minor}"
+                        },
                         "components": [{"class": "nonexistent:Module"}],  # 不存在的模块
                     }
                 ),
@@ -143,7 +138,9 @@ class TestWorkerSessionSubprocessLifecycle:
                 yaml.dump(
                     {
                         "name": "test_plugin",
-                        "runtime": {"python": f"{sys.version_info.major}.{sys.version_info.minor}"},
+                        "runtime": {
+                            "python": f"{sys.version_info.major}.{sys.version_info.minor}"
+                        },
                         "components": [],
                     }
                 ),
@@ -473,7 +470,9 @@ class TestStreamCancelDuringIteration:
         await plugin.start()
         await plugin.initialize([])
 
-        stream = await plugin.invoke_stream("test.stream", {}, request_id="stream-early")
+        stream = await plugin.invoke_stream(
+            "test.stream", {}, request_id="stream-early"
+        )
 
         # 在迭代前取消
         await plugin.cancel("stream-early")
@@ -682,7 +681,9 @@ class TestEnvironmentCacheReuse:
                     with patch("shutil.which", return_value="/usr/bin/uv"):
                         # 模拟指纹计算
                         fingerprint = manager._fingerprint(spec)
-                        manager._write_state(plugin_dir / ".astrbot-worker-state.json", spec, fingerprint)
+                        manager._write_state(
+                            plugin_dir / ".astrbot-worker-state.json", spec, fingerprint
+                        )
 
                 # 重置计数
                 rebuild_called.clear()
@@ -696,7 +697,9 @@ class TestEnvironmentCacheReuse:
                 if state.get("fingerprint") == new_fingerprint:
                     # 模拟 venv 存在
                     with patch.object(Path, "exists", return_value=True):
-                        with patch.object(manager, "_matches_python_version", return_value=True):
+                        with patch.object(
+                            manager, "_matches_python_version", return_value=True
+                        ):
                             # prepare_environment 应该跳过重建
                             # 但由于我们 mock 了 exists，这里只验证逻辑
                             pass
@@ -794,7 +797,9 @@ class TestEnvironmentCacheReuse:
             assert manager._matches_python_version(venv_dir, "3.11") is False
 
             # 不存在的 venv
-            assert manager._matches_python_version(Path("/nonexistent"), "3.12") is False
+            assert (
+                manager._matches_python_version(Path("/nonexistent"), "3.12") is False
+            )
 
 
 class TestSupervisorRuntimePluginLoading:
@@ -818,7 +823,9 @@ class TestSupervisorRuntimePluginLoading:
                     yaml.dump(
                         {
                             "name": f"plugin_{i}",
-                            "runtime": {"python": f"{sys.version_info.major}.{sys.version_info.minor}"},
+                            "runtime": {
+                                "python": f"{sys.version_info.major}.{sys.version_info.minor}"
+                            },
                             "components": [],
                         }
                     ),
@@ -862,7 +869,9 @@ class TestSupervisorRuntimePluginLoading:
                 yaml.dump(
                     {
                         "name": "valid_plugin",
-                        "runtime": {"python": f"{sys.version_info.major}.{sys.version_info.minor}"},
+                        "runtime": {
+                            "python": f"{sys.version_info.major}.{sys.version_info.minor}"
+                        },
                         "components": [],
                     }
                 ),

tests_v4/test_transport.py

@@ -290,7 +290,11 @@ class TestStdioTransportProcessMode:
 
         # 使用 Python 脚本替代 cat，读取 stdin 并输出
         transport = StdioTransport(
-            command=[sys.executable, "-c", "import sys; sys.stdout.write(sys.stdin.read())"]
+            command=[
+                sys.executable,
+                "-c",
+                "import sys; sys.stdout.write(sys.stdin.read())",
+            ]
         )
 
         await transport.start()
@@ -305,7 +309,11 @@ class TestStdioTransportProcessMode:
         import sys
 
         transport = StdioTransport(
-            command=[sys.executable, "-c", "import sys; sys.stdout.write(sys.stdin.read())"]
+            command=[
+                sys.executable,
+                "-c",
+                "import sys; sys.stdout.write(sys.stdin.read())",
+            ]
         )
         await transport.start()