diff --git a/src/bus/message.rs b/src/bus/message.rs index b1e5518..08b611c 100644 --- a/src/bus/message.rs +++ b/src/bus/message.rs @@ -425,6 +425,8 @@ pub enum OutboundEventKind { StreamDelta, /// 流式结束信号 StreamEnd, + /// 智能体执行完全结束（不再有后续工具调用或 LLM 迭代） + ExecutionCompleted, } impl OutboundMessage { @@ -629,7 +631,32 @@ impl OutboundMessage { message_id: None, } } - + + /// 构造执行完成信号 + pub fn execution_completed( + channel: impl Into, + chat_id: impl Into, + session_id: Option, + metadata: HashMap, + ) -> Self { + Self { + channel: channel.into(), + chat_id: chat_id.into(), + session_id, + content: String::new(), + reply_to: None, + media: Vec::new(), + metadata, + event_kind: OutboundEventKind::ExecutionCompleted, + role: "assistant".to_string(), + tool_call_id: None, + tool_name: None, + tool_arguments: None, + reasoning_content: None, + message_id: None, + } + } + pub fn from_chat_message( channel: &str, chat_id: &str, diff --git a/src/channels/feishu.rs b/src/channels/feishu.rs index 6c75bf9..6d7484f 100644 --- a/src/channels/feishu.rs +++ b/src/channels/feishu.rs @@ -2461,7 +2461,7 @@ impl Channel for FeishuChannel { } async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError> { - if matches!(msg.event_kind, OutboundEventKind::ToolResult | OutboundEventKind::ToolPending | OutboundEventKind::StreamDelta | OutboundEventKind::StreamEnd) + if matches!(msg.event_kind, OutboundEventKind::ToolResult | OutboundEventKind::ToolPending | OutboundEventKind::StreamDelta | OutboundEventKind::StreamEnd | OutboundEventKind::ExecutionCompleted) || msg.metadata.get("is_subagent_event").map(|v| v == "true").unwrap_or(false) { return Ok(()); diff --git a/src/channels/wechat.rs b/src/channels/wechat.rs index 7df128e..2664cf9 100644 --- a/src/channels/wechat.rs +++ b/src/channels/wechat.rs @@ -315,6 +315,7 @@ impl Channel for WechatChannel { | OutboundEventKind::ToolCall | OutboundEventKind::StreamDelta | OutboundEventKind::StreamEnd + | OutboundEventKind::ExecutionCompleted ) || msg.metadata.get("is_subagent_event").map(|v| v == "true").unwrap_or(false) { return Ok(()); diff --git a/src/gateway/processor.rs b/src/gateway/processor.rs index 87d0284..f70a9d4 100644 --- a/src/gateway/processor.rs +++ b/src/gateway/processor.rs @@ -387,6 +387,25 @@ impl InboundProcessor { self.cancel_manager.remove_by_topic(topic_id).await; } + // 发送执行完成信号，通知前端可以停止 loading 状态 + // 无论成功还是失败都发送，确保前端状态正确 + let mut completion_metadata = inbound.forwarded_metadata.clone(); + if let Some(ref topic_id) = current_topic { + completion_metadata.insert("topic_id".to_string(), topic_id.clone()); + } + if let Err(error) = self + .bus + .publish_outbound(OutboundMessage::execution_completed( + channel, + chat_id, + Some(session_id), + completion_metadata, + )) + .await + { + tracing::error!(error = %error, "Failed to publish execution_completed"); + } + Ok(()) } } diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index 7e93086..33e9414 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -295,6 +295,13 @@ pub enum WsOutbound { #[serde(default, skip_serializing_if = "Option::is_none")] topic_id: Option, }, + #[serde(rename = "execution_completed")] + ExecutionCompleted { + #[serde(default, skip_serializing_if = "Option::is_none")] + topic_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + timestamp: Option, + }, #[serde(rename = "todo_list")] TodoList { todos: Vec, diff --git a/src/protocol/ws_adapter.rs b/src/protocol/ws_adapter.rs index a85111b..06575b9 100644 --- a/src/protocol/ws_adapter.rs +++ b/src/protocol/ws_adapter.rs @@ -194,6 +194,10 @@ pub(crate) fn ws_outbound_from_outbound_message(message: &OutboundMessage) -> Ve subagent_task_id: message.metadata.get("subagent_task_id").cloned(), topic_id: message.metadata.get("topic_id").cloned(), }], + OutboundEventKind::ExecutionCompleted => vec![WsOutbound::ExecutionCompleted { + topic_id: message.metadata.get("topic_id").cloned(), + timestamp: Some(crate::protocol::now_timestamp()), + }], } } diff --git a/web/src/hooks/useChat.ts b/web/src/hooks/useChat.ts index 71de632..43b3c4d 100644 --- a/web/src/hooks/useChat.ts +++ b/web/src/hooks/useChat.ts @@ -29,6 +29,7 @@ import type { ChannelList, StreamDelta, StreamEnd, + ExecutionCompleted, WsInbound, } from '../types/protocol' @@ -658,7 +659,7 @@ export function useChat(): UseChatReturn { }, ] }) - setIsLoading(false) + // 注意：stream_delta 期间不设置 isLoading=false，智能体仍在生成 if (msg.user_message_id) applyUserMessageId(msg.user_message_id) break } @@ -668,6 +669,15 @@ export function useChat(): UseChatReturn { break } + case 'execution_completed': { + // 智能体执行完全结束（不再有后续工具调用或 LLM 迭代） + const msg = message as ExecutionCompleted + // 按 topic_id 隔离：只处理当前话题的完成信号 + if (msg.topic_id && msg.topic_id !== selectedTopicRef.current) return + setIsLoading(false) + break + } + case 'assistant_response': { const msg = message as AssistantResponse // 按 topic_id 隔离：如果消息属于其他话题则丢弃 @@ -692,7 +702,7 @@ export function useChat(): UseChatReturn { } return [...prev, newMsg] }) - setIsLoading(false) + // 注意：assistant_response 不设置 isLoading=false，智能体可能还会调用工具继续迭代 // 当前话题无描述时，可能刚触发了异步生成，标记需要刷新 const currentTopic = topicsRef.current.find(t => t.id === selectedTopicRef.current) diff --git a/web/src/types/protocol.ts b/web/src/types/protocol.ts index 546751f..6dece06 100644 --- a/web/src/types/protocol.ts +++ b/web/src/types/protocol.ts @@ -278,6 +278,12 @@ export interface StreamEnd { topic_id?: string } +export interface ExecutionCompleted { + type: 'execution_completed' + topic_id?: string + timestamp?: number +} + export type WsOutbound = | AssistantResponse | ToolCall @@ -287,6 +293,7 @@ export type WsOutbound = | TaskStarted | StreamDelta | StreamEnd + | ExecutionCompleted | SessionEstablished | SessionCreated | SessionList