fix(gateway): 修复已完成子智能体任务卡在"子智能体正在执行..."

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,卡在"子智能体正在执行..."无法变为"查看实时进度"。
This commit is contained in:
oudecheng 2026-08-05 17:34:03 +08:00
parent bc26c66169
commit c790ee1609
4 changed files with 92 additions and 40 deletions

View File

@ -188,6 +188,7 @@ fn reconstruct_task_from_db(
updated_at: now, updated_at: now,
summary: None, summary: None,
error: None, error: None,
tool_call_id: None,
})) }))
} }

View File

@ -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::protocol::{MediaSummary, WsInbound, WsOutbound, parse_inbound, serialize_outbound};
use crate::storage::persistent_session_id; use crate::storage::persistent_session_id;
use crate::tools::task::repository::TaskRepository; use crate::tools::task::repository::TaskRepository;
use crate::tools::task::types::TaskSessionState;
use axum::extract::State; use axum::extract::State;
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
use axum::response::Response; 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"); 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<String> =
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 并发送 // 将消息转换为 WsOutbound 并发送
for msg in messages { for msg in messages {
for outbound in chat_message_to_ws_outbound(&msg) { for outbound in chat_message_to_ws_outbound(&msg) {
@ -797,9 +807,9 @@ async fn send_topic_history(
} }
} }
// 查询该话题下所有运行中的子智能体任务,补发 TaskStarted 事件 // 查询该话题下所有子智能体任务,补发 TaskStarted 事件
// 解决页面刷新后 navigateToTaskId 丢失的问题 // 解决页面刷新后 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, Ok(tasks) => tasks,
Err(e) => { Err(e) => {
tracing::warn!(error = %e, topic_id = %topic_id, "Failed to list tasks for topic"); 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 { for task in tasks {
if task.state == TaskSessionState::Running { // 判断是否需要补发 TaskStarted
// 判断是否为孙智能体parent_session_id 以 "sub:" 开头表示父会话是子智能体 // - 如果该任务的 tool_call_id 已有对应的 tool_result前端会显示结果不需要补发
let parent_task_id = extract_parent_task_id(&task); // - 否则Running 状态或已完成但结果未进入历史),补发 TaskStarted 以便前端显示"查看实时进度"
let has_tool_result = task
tracing::info!( .tool_call_id
task_id = %task.id, .as_ref()
description = %task.description, .map(|tcid| tool_call_ids_with_results.contains(tcid))
parent_task_id = ?parent_task_id, .unwrap_or(false);
"Re-sending TaskStarted for running task after topic history load" if has_tool_result {
); continue;
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;
} }
// 判断是否为孙智能体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(()) Ok(())
@ -846,6 +868,17 @@ async fn send_task_messages(
tracing::info!(session_id = %session_id, message_count = messages.len(), "Sending 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<String> =
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 { for msg in messages {
let mut outbounds = chat_message_to_ws_outbound(&msg); let mut outbounds = chat_message_to_ws_outbound(&msg);
if let Some(ref task_id) = subagent_task_id { 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 { match repo.list_tasks_for_session(session_id).await {
Ok(child_tasks) => { Ok(child_tasks) => {
for child in child_tasks { for child in child_tasks {
if child.state == TaskSessionState::Running { // 如果该子任务的 tool_call_id 已有对应的 tool_result前端会显示结果不需要补发
tracing::info!( let has_tool_result = child
child_task_id = %child.id, .tool_call_id
parent_task_id = %parent_task_id, .as_ref()
"Re-sending TaskStarted for child task after sub-agent view re-enter" .map(|tcid| tool_call_ids_with_results.contains(tcid))
); .unwrap_or(false);
let _ = sender if has_tool_result {
.send(WsOutbound::TaskStarted { continue;
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;
} }
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) => { Err(e) => {

View File

@ -727,6 +727,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
channel_name, channel_name,
task.description.clone(), task.description.clone(),
task.subagent_type, task.subagent_type,
parent_context.tool_call_id.clone(),
); );
// 4. 在 sessions 表中创建子智能体会话(确保外键约束满足) // 4. 在 sessions 表中创建子智能体会话(确保外键约束满足)

View File

@ -164,6 +164,11 @@ pub struct TaskSession {
pub summary: Option<String>, pub summary: Option<String>,
/// 错误信息 /// 错误信息
pub error: Option<String>, pub error: Option<String>,
/// 触发本任务的 tool_call ID由 agent_loop 注入)。
/// 用于在 topic 历史补发 TaskStarted 时精确匹配前端的 tool_call 卡片,
/// 并判断该任务的 tool_result 是否已出现在历史消息中。
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tool_call_id: Option<String>,
} }
impl TaskSession { impl TaskSession {
@ -174,6 +179,7 @@ impl TaskSession {
parent_channel_name: String, parent_channel_name: String,
description: String, description: String,
subagent_type: SubagentType, subagent_type: SubagentType,
tool_call_id: Option<String>,
) -> Self { ) -> Self {
let id = format!("task:{}", uuid::Uuid::new_v4()); let id = format!("task:{}", uuid::Uuid::new_v4());
let session_id = format!("sub:{}:{}", parent_session_id, id); let session_id = format!("sub:{}:{}", parent_session_id, id);
@ -192,6 +198,7 @@ impl TaskSession {
updated_at: now, updated_at: now,
summary: None, summary: None,
error: None, error: None,
tool_call_id,
} }
} }