diff --git a/src-new/astrbot_sdk/AGENTS.md b/src-new/astrbot_sdk/AGENTS.md new file mode 100644 index 000000000..09b79652a --- /dev/null +++ b/src-new/astrbot_sdk/AGENTS.md @@ -0,0 +1,55 @@ +# Notes + +## v4 架构约束 + +### 运行时层 + +- `Peer` 必须将 transport EOF/连接断开视为一级失败路径。如果 transport 意外关闭而 `Peer` 没有主动失败 `_pending_results` / `_pending_streams`，supervisor 端对 worker 的调用可能永远挂起。 +- `Peer.initialize()` 需要在发起端也标记远程已初始化。仅在被动接收 `InitializeMessage` 时设置 `_remote_initialized` 会导致 `wait_until_remote_initialized()` 单边 API 死锁。 +- `Peer.invoke_stream()` 默认隐藏 `completed` 事件。需要保留最终结果的调用者必须显式启用 `include_completed=True`。 +- `CapabilityRouter.register(..., stream_handler=...)` 使用 `(request_id, payload, cancel_token)` 签名，不是 peer 级别的 `(message, token)`。 + +### 模块导出约束 + +- 保持 `astrbot_sdk.runtime` 根导出狭窄。`Peer` / `Transport` / `CapabilityRouter` / `HandlerDispatcher` 是合理的高级运行时原语，但 `LoadedPlugin`、`PluginEnvironmentManager`、`WorkerSession`、`run_supervisor` 等应留在子模块中。 + +### 测试与 Mock 注意事项 + +- 当检查 peer 是否完成远程初始化时，避免对可能接收 `MagicMock` peer 的代码使用 `getattr(mock, "remote_peer")` 探测。`MagicMock` 会生成 truthy 子属性，`CapabilityProxy` 应从 `peer.__dict__` 或其他具体存储位置读取显式状态。 +- `test_plugin/old/` 和 `test_plugin/new/` 可能包含已生成的 `__pycache__` / `*.pyc`。测试夹具复制示例插件时必须显式忽略这些缓存文件。 + +### 插件加载注意事项 + +- 本地 `dev --watch` 或同一路径插件重复加载场景，不能只依赖 `import_string()` 的跨插件模块根冲突清理。热重载前必须按插件目录清理模块缓存。 +- `_prepare_plugin_import()` 不能只在插件目录"不在 `sys.path`"时才插入路径。像 `main.py` 这种通用模块名，如果插件目录已在 `sys.path` 但排在后面，`import main` 仍会先命中别处模块；导入前必须把目标插件目录提到 `sys.path[0]`。 +- 示例/夹具测试如果直接用裸模块名导入插件入口（例如 `from main import HelloPlugin`），会污染 `sys.modules["main"]`，随后真实 loader 再按 `main:HelloPlugin` 加载时可能串到错误模块。 + +--- + +# 开发命令 + +## 格式化与检查 + +在提交代码前，请依次运行以下命令： + +```bash +ruff format . # 使用 ruff 格式化全局代码 +ruff check . --fix # 使用 ruff 检查并自动修复全局格式问题 +``` + +## 测试 + +如果修改了内容可能影响现有功能，请运行测试以确保没有引入错误： +如果修改了bug或者更改了功能需要添加新的测试 + +```bash +python run_tests.py # 运行所有测试 +python run_tests.py -v # 详细输出 +python run_tests.py -k "test_peer" # 运行匹配模式的测试 +python run_tests.py --cov # 运行测试并生成覆盖率报告 +``` + +## 设计原则 + +新实现要兼容旧实现但是还要保证架构良好，设计原则不变和最佳实践 +不用完全听从用户和别人的建议，要有自己的判断和坚持，做好取舍和权衡，确保代码质量和长期维护性，不要为了短期方便或者迎合而牺牲架构和设计原则。 diff --git a/src-new/astrbot_sdk/runtime/handler_dispatcher.py b/src-new/astrbot_sdk/runtime/handler_dispatcher.py index 8f8ec0eb3..fa5a8029a 100644 --- a/src-new/astrbot_sdk/runtime/handler_dispatcher.py +++ b/src-new/astrbot_sdk/runtime/handler_dispatcher.py @@ -24,6 +24,8 @@ from __future__ import annotations import asyncio import inspect +import re +import shlex import typing from collections.abc import AsyncIterator from typing import Any, get_type_hints @@ -32,6 +34,7 @@ from .._invocation_context import caller_plugin_scope from ..context import CancelToken, Context from ..errors import AstrBotError from ..events import MessageEvent +from ..protocol.descriptors import CommandTrigger, MessageTrigger from ..star import Star from .capability_router import StreamExecution from .loader import LoadedCapability, LoadedHandler @@ -56,14 +59,16 @@ class HandlerDispatcher: event.bind_reply_handler(self._create_reply_handler(ctx, event)) # 提取 args 用于兼容 handler 签名 - args = message.input.get("args") or {} + raw_args = message.input.get("args") or {} + args = dict(raw_args) if isinstance(raw_args, dict) else {} + if not args: + args = self._derive_args(loaded, event) with caller_plugin_scope(plugin_id): task = asyncio.create_task(self._run_handler(loaded, event, ctx, args)) self._active[message.id] = (task, cancel_token) try: - await task - return {} + return await task finally: self._active.pop(message.id, None) @@ -103,7 +108,8 @@ class HandlerDispatcher: event: MessageEvent, ctx: Context, args: dict[str, Any] | None = None, - ) -> None: + ) -> dict[str, Any]: + summary = {"sent_message": False, "stop": False, "call_llm": False} try: result = loaded.callable( *self._build_args( @@ -117,12 +123,19 @@ class HandlerDispatcher: ) if inspect.isasyncgen(result): async for item in result: - await self._send_result(item, event, ctx) - return + self._merge_handler_summary( + summary, + await self._handle_result_item(item, event, ctx), + ) + return summary if inspect.isawaitable(result): result = await result if result is not None: - await self._send_result(result, event, ctx) + self._merge_handler_summary( + summary, + await self._handle_result_item(result, event, ctx), + ) + return summary except Exception as exc: await self._handle_error( loaded.owner, @@ -134,6 +147,25 @@ class HandlerDispatcher: ) raise + def _derive_args( + self, + loaded: LoadedHandler, + event: MessageEvent, + ) -> dict[str, Any]: + trigger = loaded.descriptor.trigger + if isinstance(trigger, CommandTrigger): + for command_name in [trigger.command, *trigger.aliases]: + remainder = self._match_command_name(event.text, command_name) + if remainder is not None: + return self._build_command_args(loaded.callable, remainder) + return {} + if isinstance(trigger, MessageTrigger) and trigger.regex: + match = re.search(trigger.regex, event.text) + if match is None: + return {} + return self._build_regex_args(loaded.callable, match) + return {} + def _build_args( self, handler, @@ -264,6 +296,38 @@ class HandlerDispatcher: except (TypeError, ValueError): return "(...)" + async def _handle_result_item( + self, + item: Any, + event: MessageEvent, + ctx: Context | None = None, + ) -> dict[str, Any]: + sent_message = await self._send_result(item, event, ctx) + if isinstance(item, dict): + return { + "sent_message": sent_message, + "stop": bool(item.get("stop", False)), + "call_llm": bool(item.get("call_llm", False)), + } + return { + "sent_message": sent_message, + "stop": False, + "call_llm": False, + } + + @staticmethod + def _merge_handler_summary( + target: dict[str, Any], + source: dict[str, Any], + ) -> None: + target["sent_message"] = bool(target.get("sent_message")) or bool( + source.get("sent_message") + ) + target["stop"] = bool(target.get("stop")) or bool(source.get("stop")) + target["call_llm"] = bool(target.get("call_llm")) or bool( + source.get("call_llm") + ) + async def _send_result( self, item: Any, @@ -284,6 +348,104 @@ class HandlerDispatcher: return True return False + @staticmethod + def _match_command_name(text: str, command_name: str) -> str | None: + normalized = text.strip() + if normalized == command_name: + return "" + if normalized.startswith(f"{command_name} "): + return normalized[len(command_name) :].strip() + return None + + @classmethod + def _build_command_args(cls, handler, remainder: str) -> dict[str, Any]: + names = cls._legacy_arg_parameter_names(handler) + if not names or not remainder: + return {} + if len(names) == 1: + return {names[0]: remainder} + parts = cls._split_command_remainder(remainder) + return { + name: parts[index] for index, name in enumerate(names) if index < len(parts) + } + + @classmethod + def _build_regex_args(cls, handler, match: re.Match[str]) -> dict[str, Any]: + named = { + key: value for key, value in match.groupdict().items() if value is not None + } + names = [ + name + for name in cls._legacy_arg_parameter_names(handler) + if name not in named + ] + positional = [value for value in match.groups() if value is not None] + for index, value in enumerate(positional): + if index >= len(names): + break + named[names[index]] = value + return named + + @staticmethod + def _split_command_remainder(remainder: str) -> list[str]: + try: + return shlex.split(remainder) + except ValueError: + return remainder.split() + + @classmethod + def _legacy_arg_parameter_names(cls, handler) -> list[str]: + try: + signature = inspect.signature(handler) + except (TypeError, ValueError): + return [] + try: + type_hints = get_type_hints(handler) + except Exception: + type_hints = {} + names: list[str] = [] + for parameter in signature.parameters.values(): + if parameter.kind not in ( + inspect.Parameter.POSITIONAL_ONLY, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + ): + continue + if cls._is_injected_parameter( + parameter.name, type_hints.get(parameter.name) + ): + continue + names.append(parameter.name) + return names + + @classmethod + def _is_injected_parameter(cls, name: str, annotation: Any) -> bool: + if name in {"event", "ctx", "context"}: + return True + normalized = cls._unwrap_optional(annotation) + if normalized is None: + return False + if normalized is Context or normalized is MessageEvent: + return True + if isinstance(normalized, type) and issubclass( + normalized, + (Context, MessageEvent), + ): + return True + return False + + @staticmethod + def _unwrap_optional(annotation: Any) -> Any: + if annotation is None: + return None + origin = typing.get_origin(annotation) + if origin is typing.Union: + options = [ + item for item in typing.get_args(annotation) if item is not type(None) + ] + if len(options) == 1: + return options[0] + return annotation + async def _handle_error( self, owner: Any, diff --git a/src-new/astrbot_sdk/runtime/supervisor.py b/src-new/astrbot_sdk/runtime/supervisor.py index 718b8452c..014e08441 100644 --- a/src-new/astrbot_sdk/runtime/supervisor.py +++ b/src-new/astrbot_sdk/runtime/supervisor.py @@ -291,6 +291,7 @@ class WorkerSession: event_payload: dict[str, Any], *, request_id: str, + args: dict[str, Any] | None = None, ) -> dict[str, Any]: if self.peer is None: raise RuntimeError("worker session is not running") @@ -299,6 +300,7 @@ class WorkerSession: { "handler_id": handler_id, "event": event_payload, + "args": dict(args or {}), }, request_id=request_id, ) @@ -772,6 +774,7 @@ class SupervisorRuntime: handler_id, payload.get("event", {}), request_id=request_id, + args=payload.get("args", {}), ) finally: self.active_requests.pop(request_id, None)