fix: 修复 LegacyConversationManager 和 Peer 的异步处理逻辑，确保正确处理取消和初始化事件

AGENTS.md

@@ -5,6 +5,8 @@
 - 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.
 - 2026-03-12: Legacy `src/astrbot_sdk/api/event/filter.py` exported a much larger decorator surface than `src-new/astrbot_sdk/api/event/filter.py`. Current compat coverage is enough for `command` / `regex` / `permission` and the exercised migration tests, but it is not a full drop-in replacement for every historical filter helper.
 - 2026-03-13: Transport-pair startup tests for `SupervisorRuntime` must start a real peer on the opposite transport and provide an `initialize` response. Wiring only the supervisor side drops or captures the outgoing initialize message without replying, and `Peer.initialize()` then waits forever.
+- 2026-03-13: `CapabilityRouter.register(..., stream_handler=...)` uses the signature `(request_id, payload, cancel_token)`, not the peer-level `(message, token)`. Reusing the peer handler shape in router tests causes an immediate failed stream event before the test logic runs.
+- 2026-03-13: `Peer` had an early-cancel race for inbound invokes: if `CancelMessage` arrived before the invoke task executed its first line, the task could be cancelled before sending any terminal event, leaving the caller's stream iterator waiting forever. Preserve a per-request start event and pre-check the cancel token at the top of `_handle_invoke`.
 
 # 开发命令
 

CLAUDE.md

@@ -5,6 +5,8 @@
 - 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.
 - 2026-03-12: Legacy `src/astrbot_sdk/api/event/filter.py` exported a much larger decorator surface than `src-new/astrbot_sdk/api/event/filter.py`. Current compat coverage is enough for `command` / `regex` / `permission` and the exercised migration tests, but it is not a full drop-in replacement for every historical filter helper.
 - 2026-03-13: Transport-pair startup tests for `SupervisorRuntime` must start a real peer on the opposite transport and provide an `initialize` response. Wiring only the supervisor side drops or captures the outgoing initialize message without replying, and `Peer.initialize()` then waits forever.
+- 2026-03-13: `CapabilityRouter.register(..., stream_handler=...)` uses the signature `(request_id, payload, cancel_token)`, not the peer-level `(message, token)`. Reusing the peer handler shape in router tests causes an immediate failed stream event before the test logic runs.
+- 2026-03-13: `Peer` had an early-cancel race for inbound invokes: if `CancelMessage` arrived before the invoke task executed its first line, the task could be cancelled before sending any terminal event, leaving the caller's stream iterator waiting forever. Preserve a per-request start event and pre-check the cancel token at the top of `_handle_invoke`.
 
 # 开发命令
 

src-new/astrbot_sdk/_legacy_api.py

@@ -305,9 +305,7 @@ class LegacyConversationManager:
         Deprecated:
             请使用 update_conversation() 的 title 参数。
         """
-        await self.update_conversation(
-            unified_msg_origin, conversation_id, title=title
-        )
+        await self.update_conversation(unified_msg_origin, conversation_id, title=title)
 
     async def update_conversation_persona_id(
         self,

src-new/astrbot_sdk/protocol/legacy_adapter.py

@@ -173,7 +173,8 @@ class LegacyAdapter:
                         self._coerce_error_payload(error)
                     ),
                 )
-            return EventMessage(id=request_id, phase="completed")
+            # completed phase 需要 output 字段，提供空字典作为默认值
+            return EventMessage(id=request_id, phase="completed", output={"done": True})
 
         if method == "cancel":
             return CancelMessage(

src-new/astrbot_sdk/runtime/peer.py

@@ -65,7 +65,9 @@ class Peer:
         self._unusable = False
         self._pending_results: dict[str, asyncio.Future[ResultMessage]] = {}
         self._pending_streams: dict[str, asyncio.Queue[Any]] = {}
-        self._inbound_tasks: dict[str, tuple[asyncio.Task[None], CancelToken]] = {}
+        self._inbound_tasks: dict[
+            str, tuple[asyncio.Task[None], CancelToken, asyncio.Event]
+        ] = {}
         self._remote_initialized = asyncio.Event()
 
     def set_initialize_handler(self, handler: InitializeHandler) -> None:
@@ -99,7 +101,7 @@ class Peer:
         self._pending_streams.clear()
 
         # 取消所有入站任务
-        for task, token in list(self._inbound_tasks.values()):
+        for task, token, _started in list(self._inbound_tasks.values()):
             token.cancel()
             task.cancel()
         self._inbound_tasks.clear()
@@ -287,9 +289,10 @@ class Peer:
             await self._handle_initialize(message)
             return
         if isinstance(message, InvokeMessage):
-            task = asyncio.create_task(self._handle_invoke(message))
             token = CancelToken()
-            self._inbound_tasks[message.id] = (task, token)
+            started = asyncio.Event()
+            task = asyncio.create_task(self._handle_invoke(message, token, started))
+            self._inbound_tasks[message.id] = (task, token, started)
             task.add_done_callback(
                 lambda _task, request_id=message.id: self._inbound_tasks.pop(
                     request_id, None
@@ -332,11 +335,16 @@ class Peer:
         )
         self._remote_initialized.set()
 
-    async def _handle_invoke(self, message: InvokeMessage) -> None:
+    async def _handle_invoke(
+        self,
+        message: InvokeMessage,
+        token: CancelToken,
+        started: asyncio.Event,
+    ) -> None:
         """处理远端发起的能力调用，并按流式或非流式协议返回结果。"""
-        active = self._inbound_tasks.get(message.id)
-        token = active[1] if active is not None else CancelToken()
         try:
+            started.set()
+            token.raise_if_cancelled()
             if self._invoke_handler is None:
                 raise AstrBotError.capability_not_found(message.capability)
             execution = await self._invoke_handler(message, token)
@@ -382,11 +390,12 @@ class Peer:
         inbound = self._inbound_tasks.get(message.id)
         if inbound is None:
             return
-        task, token = inbound
+        task, token, started = inbound
         token.cancel()
         if self._cancel_handler is not None:
             await self._cancel_handler(message.id)
-        task.cancel()
+        if started.is_set():
+            task.cancel()
 
     async def _handle_result(self, message: ResultMessage) -> None:
         """处理非流式结果消息并唤醒等待中的调用方。"""