diff --git a/src-new/astrbot_sdk/protocol/descriptors.py b/src-new/astrbot_sdk/protocol/descriptors.py index f33412ccf..43994a61c 100644 --- a/src-new/astrbot_sdk/protocol/descriptors.py +++ b/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 diff --git a/src-new/astrbot_sdk/runtime/bootstrap.py b/src-new/astrbot_sdk/runtime/bootstrap.py index 72a9fc713..b8292f771 100644 --- a/src-new/astrbot_sdk/runtime/bootstrap.py +++ b/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"] = ( diff --git a/src-new/astrbot_sdk/runtime/handler_dispatcher.py b/src-new/astrbot_sdk/runtime/handler_dispatcher.py index 371edcb2d..fb9126221 100644 --- a/src-new/astrbot_sdk/runtime/handler_dispatcher.py +++ b/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: diff --git a/src-new/astrbot_sdk/runtime/loader.py b/src-new/astrbot_sdk/runtime/loader.py index ad7fa4c8c..55d0f842e 100644 --- a/src-new/astrbot_sdk/runtime/loader.py +++ b/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" diff --git a/src-new/astrbot_sdk/runtime/peer.py b/src-new/astrbot_sdk/runtime/peer.py index ebf3d34c4..b8f4e6ff0 100644 --- a/src-new/astrbot_sdk/runtime/peer.py +++ b/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( diff --git a/tests_v4/test_clients_module.py b/tests_v4/test_clients_module.py index 42c4b26f0..c378cdbf5 100644 --- a/tests_v4/test_clients_module.py +++ b/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.""" diff --git a/tests_v4/test_handler_dispatcher.py b/tests_v4/test_handler_dispatcher.py index 0712ebb67..8bcc45ffc 100644 --- a/tests_v4/test_handler_dispatcher.py +++ b/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.""" diff --git a/tests_v4/test_loader.py b/tests_v4/test_loader.py index a57fd7966..1d213a8d7 100644 --- a/tests_v4/test_loader.py +++ b/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: diff --git a/tests_v4/test_protocol_descriptors.py b/tests_v4/test_protocol_descriptors.py index e67b4a59d..127c80a51 100644 --- a/tests_v4/test_protocol_descriptors.py +++ b/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 diff --git a/tests_v4/test_runtime_integration.py b/tests_v4/test_runtime_integration.py index 5d5b6fb85..5597039b3 100644 --- a/tests_v4/test_runtime_integration.py +++ b/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": [], } ), diff --git a/tests_v4/test_transport.py b/tests_v4/test_transport.py index bda11edf0..56dad555a 100644 --- a/tests_v4/test_transport.py +++ b/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()