fix(cli): 优化消息接收逻辑，用 finalize 机制替代延迟响应

- CLIMessageEvent: 用 finalize() 替代 _delayed_response() 延迟机制，
  管道完成后统一返回响应，解决工具调用响应截断问题
- CLIMessageEvent: 添加 send_streaming() 支持，采用收集后一次性发送策略
- SocketClientHandler: 超时从30s增加到120s，处理 finalize 返回 None 的情况
- PipelineScheduler: 管道完成后调用 event.finalize()（鸭子类型）
- CLIAdapter: 添加 get_stats()/unified_webhook() 兼容 CLIConfig 数据类
- PlatformManager: 安全获取平台ID，兼容 dict 和 dataclass 类型
- PlatformRoute: 兼容 CLIConfig 的 webhook_uuid 获取方式
- MessageConverter: 补充 raw_message 字段

Author: YukiRa1n

astrbot/core/pipeline/scheduler.py

@@ -85,4 +85,8 @@ async def execute(self, event: AstrMessageEvent) -> None:
         if isinstance(event, WebChatMessageEvent | WecomAIBotMessageEvent):
             await event.send(None)
 
+        # 通知事件管道已完成（鸭子类型，供需要收集完整响应的适配器使用）
+        if hasattr(event, "finalize"):
+            await event.finalize()
+
         logger.debug("pipeline 执行完毕。")

astrbot/core/platform/manager.py

@@ -292,6 +292,16 @@ async def terminate(self) -> None:
     def get_insts(self):
         return self.platform_insts
 
+    @staticmethod
+    def _get_platform_id(inst) -> str:
+        """安全获取平台ID，兼容dict和dataclass类型的config"""
+        config = getattr(inst, "config", None)
+        if config is None:
+            return "unknown"
+        if isinstance(config, dict):
+            return config.get("id", "unknown")
+        return getattr(config, "platform_id", getattr(config, "id", "unknown"))
+
     def get_all_stats(self) -> dict:
         """获取所有平台的统计信息
 
@@ -317,7 +327,7 @@ def get_all_stats(self) -> dict:
                 logger.warning(f"获取平台统计信息失败: {e}")
                 stats_list.append(
                     {
-                        "id": getattr(inst, "config", {}).get("id", "unknown"),
+                        "id": self._get_platform_id(inst),
                         "type": "unknown",
                         "status": "unknown",
                         "error_count": 0,

astrbot/core/platform/sources/cli/cli_adapter.py

@@ -195,6 +195,39 @@ def meta(self) -> PlatformMetadata:
         """获取平台元数据"""
         return self.metadata
 
+    def unified_webhook(self) -> bool:
+        """CLI不使用webhook"""
+        return False
+
+    def get_stats(self) -> dict:
+        """获取平台统计信息（兼容CLIConfig数据类）"""
+        meta = self.meta()
+        meta_info = {
+            "id": meta.id,
+            "name": meta.name,
+            "display_name": meta.adapter_display_name or meta.name,
+            "description": meta.description,
+            "support_streaming_message": meta.support_streaming_message,
+            "support_proactive_message": meta.support_proactive_message,
+        }
+        return {
+            "id": meta.id or self.config.platform_id,
+            "type": meta.name,
+            "display_name": meta.adapter_display_name or meta.name,
+            "status": self._status.value,
+            "started_at": self._started_at.isoformat() if self._started_at else None,
+            "error_count": len(self._errors),
+            "last_error": {
+                "message": self.last_error.message,
+                "timestamp": self.last_error.timestamp.isoformat(),
+                "traceback": self.last_error.traceback,
+            }
+            if self.last_error
+            else None,
+            "unified_webhook": False,
+            "meta": meta_info,
+        }
+
     async def terminate(self) -> None:
         """终止平台运行"""
         self._running = False

astrbot/core/platform/sources/cli/cli_event.py

@@ -6,6 +6,7 @@
 """
 
 import asyncio
+from collections.abc import AsyncGenerator
 from typing import Any
 
 from astrbot import logger
@@ -21,11 +22,9 @@ class CLIMessageEvent(AstrMessageEvent):
     """CLI消息事件
 
     处理命令行模拟器的消息事件。
+    Socket模式下收集管道中所有send()调用的消息，在管道完成(finalize)后统一返回。
     """
 
-    # 延迟配置
-    INITIAL_DELAY = 5.0  # 首次发送延迟
-    EXTENDED_DELAY = 10.0  # 后续发送延迟
     MAX_BUFFER_SIZE = 100  # 缓冲区最大消息组件数
 
     def __init__(
@@ -48,29 +47,21 @@ def __init__(
         self.output_queue = output_queue
         self.response_future = response_future
 
-        # 多次回复收集
+        # 多次回复收集（Socket模式）
         self.send_buffer = None
-        self._response_delay_task = None
-        self._response_delay = self.INITIAL_DELAY
 
     async def send(self, message_chain: MessageChain) -> dict[str, Any]:
         """发送消息到CLI"""
-        # 调用父类方法以设置 _has_send_oper 标志
-        # 这告诉 ProcessStage 已经有发送操作，避免触发 LLM
         await super().send(message_chain)
 
-        # Socket模式：收集多次回复
+        # Socket模式：收集所有回复到buffer，等待finalize()统一返回
         if self.response_future is not None and not self.response_future.done():
-            # 使用 ImageProcessor 预处理图片（避免临时文件被删除）
             ImageProcessor.preprocess_chain(message_chain)
 
-            # 收集多次回复到buffer
             if not self.send_buffer:
                 self.send_buffer = message_chain
-                self._response_delay = self.INITIAL_DELAY
                 logger.debug("[CLI] First send: buffer initialized")
             else:
-                # 检查缓冲区大小限制
                 current_size = len(self.send_buffer.chain)
                 new_size = len(message_chain.chain)
                 if current_size + new_size > self.MAX_BUFFER_SIZE:
@@ -80,47 +71,61 @@ async def send(self, message_chain: MessageChain) -> dict[str, Any]:
                         new_size,
                         self.MAX_BUFFER_SIZE,
                     )
-                    # 只添加能容纳的部分
                     available = self.MAX_BUFFER_SIZE - current_size
                     if available > 0:
                         self.send_buffer.chain.extend(message_chain.chain[:available])
                 else:
                     self.send_buffer.chain.extend(message_chain.chain)
-                self._response_delay = self.EXTENDED_DELAY
                 logger.debug(
                     "[CLI] Appended to buffer, total: %d", len(self.send_buffer.chain)
                 )
-
-            # 重置延迟任务
-            if self._response_delay_task and not self._response_delay_task.done():
-                self._response_delay_task.cancel()
-
-            self._response_delay_task = asyncio.create_task(self._delayed_response())
         else:
-            # 其他模式：直接放入输出队列
+            # 非Socket模式或future已完成：直接放入输出队列
             await self.output_queue.put(message_chain)
 
         return {"success": True}
 
+    async def send_streaming(
+        self,
+        generator: AsyncGenerator[MessageChain, None],
+        use_fallback: bool = False,
+    ) -> None:
+        """处理流式LLM响应
+
+        CLI不支持真正的流式输出，采用收集后一次性发送的策略。
+        与aiocqhttp的非fallback模式一致。
+        """
+        buffer = None
+        async for chain in generator:
+            if not buffer:
+                buffer = chain
+            else:
+                buffer.chain.extend(chain.chain)
+
+        if not buffer:
+            return
+
+        buffer.squash_plain()
+        await self.send(buffer)
+        await super().send_streaming(generator, use_fallback)
+
     async def reply(self, message_chain: MessageChain) -> dict[str, Any]:
         """回复消息"""
         return await self.send(message_chain)
 
-    async def _delayed_response(self) -> None:
-        """延迟响应：收集所有回复后统一返回"""
-        try:
-            await asyncio.sleep(self._response_delay)
+    async def finalize(self) -> None:
+        """管道完成后调用，将收集的所有回复统一返回给Socket客户端。
 
-            if self.response_future and not self.response_future.done():
+        由PipelineScheduler.execute()在所有阶段执行完毕后调用。
+        """
+        if self.response_future and not self.response_future.done():
+            if self.send_buffer:
                 self.response_future.set_result(self.send_buffer)
                 logger.debug(
-                    "[CLI] Delayed response set, %d components",
+                    "[CLI] Pipeline done, response set with %d components",
                     len(self.send_buffer.chain),
                 )
-
-        except asyncio.CancelledError:
-            pass
-        except Exception as e:
-            logger.error("[CLI] Delayed response error: %s", e)
-            if self.response_future and not self.response_future.done():
-                self.response_future.set_exception(e)
+            else:
+                # 管道完成但没有任何发送操作（如被白名单/频率限制拦截）
+                self.response_future.set_result(None)
+                logger.debug("[CLI] Pipeline done, no response to send")

astrbot/core/platform/sources/cli/handlers/socket_handler.py

@@ -12,6 +12,7 @@
 from typing import TYPE_CHECKING
 
 from astrbot import logger
+from astrbot.core.message.message_event_result import MessageChain
 
 from ..interfaces import IHandler, IMessageConverter, ISessionManager, ITokenValidator
 from ..message.response_builder import ResponseBuilder
@@ -34,7 +35,7 @@ class SocketClientHandler:
 
     RECV_BUFFER_SIZE = 4096
     MAX_REQUEST_SIZE = 1024 * 1024  # 1MB 最大请求大小
-    RESPONSE_TIMEOUT = 30.0
+    RESPONSE_TIMEOUT = 120.0
 
     def __init__(
         self,
@@ -181,14 +182,13 @@ async def _process_message(self, message_text: str, request_id: str) -> str:
             message_chain = await asyncio.wait_for(
                 response_future, timeout=self.RESPONSE_TIMEOUT
             )
+            if message_chain is None:
+                # 管道完成但没有产生任何回复（被白名单/频率限制等拦截）
+                return ResponseBuilder.build_success(
+                    MessageChain([]), request_id
+                )
             return ResponseBuilder.build_success(message_chain, request_id)
         except asyncio.TimeoutError:
-            # 超时时取消延迟响应任务，防止资源泄露
-            if (
-                hasattr(message_event, "_response_delay_task")
-                and message_event._response_delay_task
-            ):
-                message_event._response_delay_task.cancel()
             return ResponseBuilder.build_error("Request timeout", request_id, "TIMEOUT")
 
     async def _get_logs(self, request: dict, request_id: str) -> str:

astrbot/core/platform/sources/cli/message/converter.py

@@ -71,4 +71,6 @@ def convert(
             nickname=self.user_nickname,
         )
 
+        message.raw_message = None
+
         return message

astrbot/dashboard/routes/platform.py

@@ -81,7 +81,9 @@ def _find_platform_by_uuid(self, webhook_uuid: str) -> Platform | None:
             平台适配器实例，未找到则返回 None
         """
         for platform in self.platform_manager.platform_insts:
-            if platform.config.get("webhook_uuid") == webhook_uuid:
+            config = platform.config
+            uuid_val = config.get("webhook_uuid") if isinstance(config, dict) else getattr(config, "webhook_uuid", None)
+            if uuid_val == webhook_uuid:
                 if platform.unified_webhook():
                     return platform
         return None