From c790ee1609dd76fda0674ba7252a66d6fb82dd03 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Wed, 5 Aug 2026 17:34:03 +0800 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20=E4=BF=AE=E5=A4=8D=E5=B7=B2?= =?UTF-8?q?=E5=AE=8C=E6=88=90=E5=AD=90=E6=99=BA=E8=83=BD=E4=BD=93=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=8D=A1=E5=9C=A8"=E5=AD=90=E6=99=BA=E8=83=BD?= =?UTF-8?q?=E4=BD=93=E6=AD=A3=E5=9C=A8=E6=89=A7=E8=A1=8C..."?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TaskStarted 补发逻辑改为基于 tool_result 存在性判断（替代 state==Running）， 并在补发事件中携带 tool_call_id 以实现前端精确匹配。 - TaskSession 新增 tool_call_id 字段，runtime.spawn 时从 parent_context 注入 - send_topic_history/send_task_messages 收集历史中已有 tool_result 的 tool_call_id 集合，仅对没有 tool_result 的任务补发 TaskStarted（不论 Running/Completed） - 补发时传入 task.tool_call_id，前端可精确匹配 task tool_call 卡片， 避免多并行子智能体场景下 fallback 串扰 - 移除未使用的 TaskSessionState 导入 修复场景：父智能体启动多个并行子智能体，其中一个完成但另一个仍在运行时 重载 topic，已完成的子智能体因 state!=Running 被跳过，导致其 tool_call 卡片永远收不到 TaskStarted，卡在"子智能体正在执行..."无法变为"查看实时进度"。 --- src/command/handlers/load_task_messages.rs | 1 + src/gateway/ws.rs | 123 ++++++++++++++------- src/tools/task/runtime.rs | 1 + src/tools/task/types.rs | 7 ++ 4 files changed, 92 insertions(+), 40 deletions(-) diff --git a/src/command/handlers/load_task_messages.rs b/src/command/handlers/load_task_messages.rs index 0d49c80..969daa6 100644 --- a/src/command/handlers/load_task_messages.rs +++ b/src/command/handlers/load_task_messages.rs @@ -188,6 +188,7 @@ fn reconstruct_task_from_db( updated_at: now, summary: None, error: None, + tool_call_id: None, })) } diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs index c9fa19e..64994c6 100644 --- a/src/gateway/ws.rs +++ b/src/gateway/ws.rs @@ -30,7 +30,6 @@ use crate::gateway::agent_factory::build_system_prompt_provider; use crate::protocol::{MediaSummary, WsInbound, WsOutbound, parse_inbound, serialize_outbound}; use crate::storage::persistent_session_id; use crate::tools::task::repository::TaskRepository; -use crate::tools::task::types::TaskSessionState; use axum::extract::State; use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; use axum::response::Response; @@ -790,6 +789,17 @@ async fn send_topic_history( tracing::info!(topic_id = %topic_id, message_count = messages.len(), "Sending topic history"); + // 收集已有 tool_result 的 tool_call_id 集合，用于判断任务是否已有结果 + let mut tool_call_ids_with_results: std::collections::HashSet = + std::collections::HashSet::new(); + for msg in &messages { + if msg.role == "tool" { + if let Some(ref tcid) = msg.tool_call_id { + tool_call_ids_with_results.insert(tcid.clone()); + } + } + } + // 将消息转换为 WsOutbound 并发送 for msg in messages { for outbound in chat_message_to_ws_outbound(&msg) { @@ -797,9 +807,9 @@ async fn send_topic_history( } } - // 查询该话题下所有运行中的子智能体任务，补发 TaskStarted 事件 + // 查询该话题下所有子智能体任务，补发 TaskStarted 事件 // 解决页面刷新后 navigateToTaskId 丢失的问题 - let running_tasks = match task_repository.list_tasks_for_topic(topic_id).await { + let tasks = match task_repository.list_tasks_for_topic(topic_id).await { Ok(tasks) => tasks, Err(e) => { tracing::warn!(error = %e, topic_id = %topic_id, "Failed to list tasks for topic"); @@ -807,28 +817,40 @@ async fn send_topic_history( } }; - for task in running_tasks { - if task.state == TaskSessionState::Running { - // 判断是否为孙智能体：parent_session_id 以 "sub:" 开头表示父会话是子智能体 - let parent_task_id = extract_parent_task_id(&task); - - tracing::info!( - task_id = %task.id, - description = %task.description, - parent_task_id = ?parent_task_id, - "Re-sending TaskStarted for running task after topic history load" - ); - let _ = sender - .send(WsOutbound::TaskStarted { - task_id: task.id.clone(), - description: task.description.clone(), - subagent_type: task.subagent_type.clone(), - topic_id: Some(topic_id.to_string()), - parent_task_id, - tool_call_id: None, - }) - .await; + for task in tasks { + // 判断是否需要补发 TaskStarted： + // - 如果该任务的 tool_call_id 已有对应的 tool_result，前端会显示结果，不需要补发 + // - 否则（Running 状态或已完成但结果未进入历史），补发 TaskStarted 以便前端显示"查看实时进度" + let has_tool_result = task + .tool_call_id + .as_ref() + .map(|tcid| tool_call_ids_with_results.contains(tcid)) + .unwrap_or(false); + if has_tool_result { + continue; } + + // 判断是否为孙智能体：parent_session_id 以 "sub:" 开头表示父会话是子智能体 + let parent_task_id = extract_parent_task_id(&task); + + tracing::info!( + task_id = %task.id, + description = %task.description, + parent_task_id = ?parent_task_id, + tool_call_id = ?task.tool_call_id, + state = ?task.state, + "Re-sending TaskStarted for task without tool_result after topic history load" + ); + let _ = sender + .send(WsOutbound::TaskStarted { + task_id: task.id.clone(), + description: task.description.clone(), + subagent_type: task.subagent_type.clone(), + topic_id: Some(topic_id.to_string()), + parent_task_id, + tool_call_id: task.tool_call_id.clone(), + }) + .await; } Ok(()) @@ -846,6 +868,17 @@ async fn send_task_messages( tracing::info!(session_id = %session_id, message_count = messages.len(), "Sending task messages"); + // 收集已有 tool_result 的 tool_call_id 集合，用于判断子任务是否已有结果 + let mut tool_call_ids_with_results: std::collections::HashSet = + std::collections::HashSet::new(); + for msg in &messages { + if msg.role == "tool" { + if let Some(ref tcid) = msg.tool_call_id { + tool_call_ids_with_results.insert(tcid.clone()); + } + } + } + for msg in messages { let mut outbounds = chat_message_to_ws_outbound(&msg); if let Some(ref task_id) = subagent_task_id { @@ -864,23 +897,33 @@ async fn send_task_messages( match repo.list_tasks_for_session(session_id).await { Ok(child_tasks) => { for child in child_tasks { - if child.state == TaskSessionState::Running { - tracing::info!( - child_task_id = %child.id, - parent_task_id = %parent_task_id, - "Re-sending TaskStarted for child task after sub-agent view re-enter" - ); - let _ = sender - .send(WsOutbound::TaskStarted { - task_id: child.id.clone(), - description: child.description.clone(), - subagent_type: child.subagent_type.clone(), - topic_id: child.parent_topic_id.clone(), - parent_task_id: Some(parent_task_id.clone()), - tool_call_id: None, - }) - .await; + // 如果该子任务的 tool_call_id 已有对应的 tool_result，前端会显示结果，不需要补发 + let has_tool_result = child + .tool_call_id + .as_ref() + .map(|tcid| tool_call_ids_with_results.contains(tcid)) + .unwrap_or(false); + if has_tool_result { + continue; } + + tracing::info!( + child_task_id = %child.id, + parent_task_id = %parent_task_id, + tool_call_id = ?child.tool_call_id, + state = ?child.state, + "Re-sending TaskStarted for child task without tool_result after sub-agent view re-enter" + ); + let _ = sender + .send(WsOutbound::TaskStarted { + task_id: child.id.clone(), + description: child.description.clone(), + subagent_type: child.subagent_type.clone(), + topic_id: child.parent_topic_id.clone(), + parent_task_id: Some(parent_task_id.clone()), + tool_call_id: child.tool_call_id.clone(), + }) + .await; } } Err(e) => { diff --git a/src/tools/task/runtime.rs b/src/tools/task/runtime.rs index 3f9c6e5..601f159 100644 --- a/src/tools/task/runtime.rs +++ b/src/tools/task/runtime.rs @@ -727,6 +727,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime { channel_name, task.description.clone(), task.subagent_type, + parent_context.tool_call_id.clone(), ); // 4. 在 sessions 表中创建子智能体会话（确保外键约束满足） diff --git a/src/tools/task/types.rs b/src/tools/task/types.rs index db2fd6b..72c7186 100644 --- a/src/tools/task/types.rs +++ b/src/tools/task/types.rs @@ -164,6 +164,11 @@ pub struct TaskSession { pub summary: Option, /// 错误信息 pub error: Option, + /// 触发本任务的 tool_call ID（由 agent_loop 注入）。 + /// 用于在 topic 历史补发 TaskStarted 时精确匹配前端的 tool_call 卡片， + /// 并判断该任务的 tool_result 是否已出现在历史消息中。 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_call_id: Option, } impl TaskSession { @@ -174,6 +179,7 @@ impl TaskSession { parent_channel_name: String, description: String, subagent_type: SubagentType, + tool_call_id: Option, ) -> Self { let id = format!("task:{}", uuid::Uuid::new_v4()); let session_id = format!("sub:{}:{}", parent_session_id, id); @@ -192,6 +198,7 @@ impl TaskSession { updated_at: now, summary: None, error: None, + tool_call_id, } }