feat: Enhance HTTP API registration with plugin namespace validation

- Added validation to ensure HTTP routes and handler capabilities belong to the current plugin namespace in the SDK.
- Updated tests to reflect changes in API registration routes, ensuring they include the plugin ID as a prefix.
- Introduced new error handling for invalid route and capability registrations.
- Refactored existing tests to accommodate the new route structure and added tests for new validation logic.
- Improved the structure of SDK record creation for better readability and maintainability.

Author: whatevertogo

astrbot-sdk/src/astrbot_sdk/_internal/testing_support.py

@@ -389,7 +389,7 @@ def __init__(self, router: MockCapabilityRouter) -> None:
             role="core",
             version="local",
         )
-        self.remote_capabilities = list(router.descriptors())
+        self.remote_capabilities = list(router.all_descriptors())
         self.remote_capability_map = {
             item.name: item for item in self.remote_capabilities
         }

astrbot-sdk/src/astrbot_sdk/runtime/capability_dispatcher.py

@@ -361,6 +361,11 @@ async def _run_capability(
                 )
             if inspect.isawaitable(result):
                 result = await result
+            if inspect.isasyncgen(result):
+                return StreamExecution(
+                    iterator=self._iterate_generator(result),
+                    finalize=lambda chunks: {"items": chunks},
+                )
             if isinstance(result, StreamExecution):
                 return result
             raise AstrBotError.protocol_error(

astrbot-sdk/src/astrbot_sdk/runtime/capability_router.py

@@ -647,6 +647,11 @@ def _emit_db_change(self, *, op: str, key: str, value: Any | None) -> None:
             queue.put_nowait(event)
 
     def descriptors(self) -> list[CapabilityDescriptor]:
+        return [
+            entry.descriptor for entry in self._registrations.values() if entry.exposed
+        ]
+
+    def all_descriptors(self) -> list[CapabilityDescriptor]:
         return [entry.descriptor for entry in self._registrations.values()]
 
     def contains(self, name: str) -> bool:

astrbot-sdk/src/astrbot_sdk/runtime/peer.py

@@ -454,6 +454,7 @@ async def invoke_stream(
         )
 
         async def iterator() -> AsyncIterator[EventMessage]:
+            terminal_received = False
             try:
                 while True:
                     item = await queue.get()
@@ -467,15 +468,30 @@ async def iterator() -> AsyncIterator[EventMessage]:
                         yield item
                         continue
                     if item.phase == "completed":
+                        terminal_received = True
                         if include_completed:
                             yield item
                         break
                     if item.phase == "failed":
+                        terminal_received = True
                         raise AstrBotError.from_payload(
                             item.error.model_dump() if item.error else {}
                         )
             finally:
                 self._pending_streams.pop(request_id, None)
+                if not terminal_received:
+                    try:
+                        await self.cancel(
+                            request_id,
+                            reason="consumer_closed_stream_early",
+                        )
+                    except Exception as exc:
+                        logger.debug(
+                            "Failed to cancel stream after consumer closed early: "
+                            "request_id=%s error=%s",
+                            request_id,
+                            exc,
+                        )
 
         return iterator()
 

astrbot-sdk/src/astrbot_sdk/runtime/supervisor.py

@@ -390,7 +390,7 @@ async def cancel(self, request_id: str) -> None:
     async def _handle_initialize(self, _message) -> InitializeOutput:
         return InitializeOutput(
             peer=PeerInfo(name="astrbot-supervisor", role="core", version="v4"),
-            capabilities=self.capability_router.descriptors(),
+            capabilities=self.capability_router.all_descriptors(),
             metadata={
                 "group_id": self.group_id,
                 "plugins": [plugin.name for plugin in self.plugins],

astrbot/core/sdk_bridge/dispatch_engine.py

@@ -365,6 +365,7 @@ async def dispatch_message_event(
                     args={},
                 )
                 if isinstance(output, dict):
+                    handler_result = extract_sdk_handler_result(output)
                     if "sdk_local_extras" in output:
                         self.bridge._persist_sdk_local_extras_from_handler(
                             overlay,
@@ -390,6 +391,15 @@ async def dispatch_message_event(
                                 event_result,
                                 result_payload,
                             )
+                    if handler_result["stop"]:
+                        event.stop_event()
+                    if handler_result["call_llm"]:
+                        overlay.requested_llm = True
+                        overlay.should_call_llm = True
+                    if handler_result["sent_message"]:
+                        event._has_send_oper = True
+                    if handler_result["sent_message"] or handler_result["stop"]:
+                        overlay.should_call_llm = False
             except Exception as exc:
                 logger.warning(
                     "SDK event handler failed: plugin=%s handler=%s error=%s",
@@ -468,6 +478,13 @@ async def dispatch_waiter_event(
             handler_result = extract_sdk_handler_result(
                 output if isinstance(output, dict) else {}
             )
+            if isinstance(output, dict) and "sdk_local_extras" in output:
+                self.bridge._persist_sdk_local_extras_from_handler(
+                    overlay,
+                    output.get("sdk_local_extras"),
+                    plugin_id=record.plugin_id,
+                    handler_id="__sdk_session_waiter__",
+                )
             result.executed_handlers.append(
                 {"plugin_id": record.plugin_id, "handler_id": "__sdk_session_waiter__"}
             )

astrbot/core/sdk_bridge/lifecycle_manager.py

@@ -74,7 +74,7 @@ async def reload_all(self, *, reset_restart_budget: bool = False) -> None:
                     reset_restart_budget=reset_restart_budget,
                 )
             self.bridge.refresh_command_compatibility_issues()
-            await self.bridge._refresh_native_platform_commands({"telegram"})
+            await self.bridge._refresh_native_platform_commands()
 
     async def reload_plugin(self, plugin_id: str) -> None:
         async with self._reload_lock:
@@ -90,7 +90,7 @@ async def reload_plugin(self, plugin_id: str) -> None:
                     reset_restart_budget=True,
                 )
                 self.bridge.refresh_command_compatibility_issues()
-                await self.bridge._refresh_native_platform_commands({"telegram"})
+                await self.bridge._refresh_native_platform_commands()
                 return
             raise ValueError(f"SDK plugin not found: {plugin_id}")
 
@@ -105,7 +105,7 @@ async def turn_off_plugin(self, plugin_id: str) -> None:
             record.failure_reason = ""
             self.bridge._set_disabled_override(plugin_id, disabled=True)
             self.bridge.refresh_command_compatibility_issues()
-            await self.bridge._refresh_native_platform_commands({"telegram"})
+            await self.bridge._refresh_native_platform_commands()
 
     async def turn_on_plugin(self, plugin_id: str) -> None:
         async with self._reload_lock:
@@ -122,7 +122,7 @@ async def turn_on_plugin(self, plugin_id: str) -> None:
                     reset_restart_budget=True,
                 )
                 self.bridge.refresh_command_compatibility_issues()
-                await self.bridge._refresh_native_platform_commands({"telegram"})
+                await self.bridge._refresh_native_platform_commands()
                 return
             raise ValueError(f"SDK plugin not found: {plugin_id}")
 

astrbot/core/sdk_bridge/registry_manager.py

@@ -6,6 +6,12 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+from astrbot_sdk._internal.plugin_ids import (
+    capability_belongs_to_plugin,
+    http_route_belongs_to_plugin,
+    plugin_capability_prefix,
+    plugin_http_route_root,
+)
 from astrbot_sdk.errors import AstrBotError
 
 from astrbot.core import logger
@@ -268,6 +274,8 @@ def register_http_api(
             raise AstrBotError.invalid_input(
                 "http.register_api requires handler_capability"
             )
+        self._validate_http_route_namespace(normalized_route, plugin_id)
+        self._validate_http_handler_namespace(handler_capability, plugin_id)
         self.bridge._ensure_http_route_available(
             plugin_id=plugin_id,
             route=normalized_route,
@@ -297,6 +305,32 @@ def register_http_api(
             handler_capability,
         )
 
+    @staticmethod
+    def _validate_http_route_namespace(route: str, plugin_id: str) -> None:
+        if http_route_belongs_to_plugin(route, plugin_id):
+            return
+        route_root = plugin_http_route_root(plugin_id)
+        raise AstrBotError.invalid_input(
+            "http.register_api requires route to use the current plugin namespace: "
+            f"route={route!r}, plugin_id={plugin_id!r}, expected={route_root!r} "
+            f"or {route_root + '/...'}"
+        )
+
+    @staticmethod
+    def _validate_http_handler_namespace(
+        handler_capability: str,
+        plugin_id: str,
+    ) -> None:
+        if capability_belongs_to_plugin(handler_capability, plugin_id):
+            return
+        expected_prefix = plugin_capability_prefix(plugin_id)
+        raise AstrBotError.invalid_input(
+            "http.register_api requires handler_capability to belong to the current "
+            "plugin: "
+            f"capability={handler_capability!r}, plugin_id={plugin_id!r}, "
+            f"expected_prefix={expected_prefix!r}"
+        )
+
     def unregister_http_api(
         self,
         *,

astrbot/core/sdk_bridge/request_runtime.py

@@ -289,7 +289,10 @@ def set_result_payload_on_overlay(
             else None
         )
         overlay.result_is_set = True
-        self.set_overlay_stop_state(overlay, stopped=False)
+        self.set_overlay_stop_state(
+            overlay,
+            stopped=bool(normalized_payload.get("stop", False)),
+        )
 
     def sync_overlay_payload_from_result_object(
         self,
@@ -494,10 +497,13 @@ def legacy_result_to_sdk_payload(
             if isinstance(result.chain, MessageChain)
             else result.chain
         )
-        return {
+        payload = {
             "type": "chain" if chain else "empty",
             "chain": SdkRequestRuntime.components_to_sdk_payload(chain),
         }
+        if result.is_stopped():
+            payload["stop"] = True
+        return payload
 
     @staticmethod
     def components_to_sdk_payload(
@@ -738,6 +744,10 @@ def apply_sdk_result_payload(
         result.chain = updated.chain
         result.use_t2i_ = updated.use_t2i_
         result.type = updated.type
+        if bool(payload.get("stop", False)):
+            result.stop_event()
+        else:
+            result.continue_event()
         return result
 
     def get_effective_result(

tests/test_sdk/unit/_context_api_roundtrip.py

@@ -1,3 +1,4 @@
+# ruff: noqa: E402
 from __future__ import annotations
 
 import asyncio
@@ -45,7 +46,14 @@ def install(name: str, attrs: dict[str, object]) -> None:
 _install_optional_dependency_stubs()
 
 from astrbot_sdk._internal.invocation_context import current_caller_plugin_id
+from astrbot_sdk._internal.plugin_ids import (
+    capability_belongs_to_plugin,
+    http_route_belongs_to_plugin,
+    plugin_capability_prefix,
+    plugin_http_route_root,
+)
 from astrbot_sdk.context import Context
+from astrbot_sdk.errors import AstrBotError
 from astrbot_sdk.message.components import component_to_payload_sync
 from astrbot_sdk.runtime._streaming import StreamExecution
 
@@ -432,6 +440,21 @@ def register_http_api(
     ) -> None:
         normalized_route = self._normalize_route(route)
         normalized_methods = self._normalize_methods(methods)
+        if not http_route_belongs_to_plugin(normalized_route, plugin_id):
+            route_root = plugin_http_route_root(plugin_id)
+            raise AstrBotError.invalid_input(
+                "http.register_api requires route to use the current plugin "
+                f"namespace: route={normalized_route!r}, plugin_id={plugin_id!r}, "
+                f"expected={route_root!r} or {route_root + '/...'}"
+            )
+        if not capability_belongs_to_plugin(str(handler_capability), plugin_id):
+            expected_prefix = plugin_capability_prefix(plugin_id)
+            raise AstrBotError.invalid_input(
+                "http.register_api requires handler_capability to belong to the "
+                "current plugin: "
+                f"capability={handler_capability!r}, plugin_id={plugin_id!r}, "
+                f"expected_prefix={expected_prefix!r}"
+            )
         existing = [
             item
             for item in self.http_routes.get(plugin_id, [])
@@ -1055,7 +1078,7 @@ def __init__(self, bridge: CoreCapabilityBridge) -> None:
         self._request_counter = 0
         self.remote_peer = object()
         self.remote_capability_map = {
-            descriptor.name: descriptor for descriptor in bridge.descriptors()
+            descriptor.name: descriptor for descriptor in bridge.all_descriptors()
         }
 
     def _next_request_id(self) -> str: