修复: 工作流终态回写归属校验与状态文案
终态摘要仅 workflow_id 归属一致才回写,防同会话并发互踩 完成态替换待上报行,消除已完成却待上报的自相矛盾 步骤链改用节点 label,无 label 退回 ID
This commit is contained in:
@@ -4,7 +4,7 @@
|
||||
//! - 生产:`run_workflow_inner`(workflow.rs)AI 路径(conv_id Some)启动时调
|
||||
//! `build_dag_summary` 产出摘要写入 `PerConvState.workflow_dag_summary`
|
||||
//! (`workflow_id` 存本次执行 ID,非模板 ID);执行闭包终态(completed/failed)
|
||||
//! 回写摘要并追加「状态: 已完成/已失败」行。
|
||||
//! 回写摘要并把「完成: 待执行器上报」行替换为「完成: 已完成/已失败」。
|
||||
//! - 消费:`agentic/mod.rs` 的 `build_runtime_prompt` 每轮读该字段注入 system prompt。
|
||||
//!
|
||||
//! `build_dag_summary` 已有生产调用方;`extract_active_path` 仍无生产调用方
|
||||
@@ -14,6 +14,10 @@
|
||||
|
||||
use df_workflow::dag_def::DagDef;
|
||||
|
||||
/// 启动态摘要末行(执行器尚未上报结果的占位)。终态回写(workflow.rs)按此 strip 后
|
||||
/// 替换为真实结论行,两处共用同一常量防文案漂移。
|
||||
pub(crate) const PENDING_RESULT_LINE: &str = "完成: 待执行器上报";
|
||||
|
||||
/// 从 DagDef 构建可读的 DAG 摘要文本
|
||||
///
|
||||
/// 输出格式(5 节点 DAG 为例):
|
||||
@@ -31,9 +35,13 @@ pub fn build_dag_summary(dag: &DagDef, current_node_id: Option<&str>) -> String
|
||||
// 拓扑排序(简化:按 edges 计算入度,分层输出)
|
||||
let layers = topological_layers(dag);
|
||||
|
||||
// 构建步骤链
|
||||
let step_chain: Vec<&str> = layers.iter()
|
||||
.flat_map(|layer| layer.iter().map(|id| id.as_str()))
|
||||
// 构建步骤链(label 优先,节点无 label 退回节点 ID;与 extract_active_path 同款映射)
|
||||
let step_chain: Vec<String> = layers.iter()
|
||||
.flat_map(|layer| layer.iter().map(|id| {
|
||||
dag.nodes.get(id)
|
||||
.and_then(|n| n.label.clone())
|
||||
.unwrap_or_else(|| id.clone())
|
||||
}))
|
||||
.collect();
|
||||
let chain_text = step_chain.join(" → ");
|
||||
|
||||
@@ -47,7 +55,7 @@ pub fn build_dag_summary(dag: &DagDef, current_node_id: Option<&str>) -> String
|
||||
if !current_label.is_empty() {
|
||||
result.push_str(&format!("当前: {}\n", current_label));
|
||||
}
|
||||
result.push_str(&format!("完成: 待执行器上报"));
|
||||
result.push_str(PENDING_RESULT_LINE);
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
@@ -21,7 +21,7 @@ use df_storage::models::WorkflowRecord;
|
||||
use df_workflow::dag_def::DagDef;
|
||||
use df_workflow::executor::DagExecutor;
|
||||
|
||||
use crate::commands::ai::agentic::workflow_context::build_dag_summary;
|
||||
use crate::commands::ai::agentic::workflow_context::{build_dag_summary, PENDING_RESULT_LINE};
|
||||
use crate::state::AppState;
|
||||
|
||||
use super::{err_str, now_millis};
|
||||
@@ -433,19 +433,27 @@ pub async fn run_workflow_inner(
|
||||
}
|
||||
|
||||
// 终态刷新:工作流终态后把完成/失败结论写回会话摘要,agentic loop
|
||||
// 下一轮注入终态而非启动时的「完成: 待执行器上报」。
|
||||
// 下一轮注入终态而非启动时的「待执行器上报」占位行。
|
||||
// 会话不存在(已删除)静默跳过;手动路径 conv_id None 无刷新。
|
||||
if let Some(cid) = cb_conv_id {
|
||||
let summary = if status == "completed" {
|
||||
format!("{}\n状态: 已完成", cb_dag_summary)
|
||||
} else {
|
||||
format!("{}\n状态: 已失败(节点: {})", cb_dag_summary, failed_node)
|
||||
};
|
||||
let mut session = ai_session.lock().await;
|
||||
if let Some(conv) = session.per_conv.get_mut(&cid) {
|
||||
// 归属校验:同会话先后启动多个工作流时,conv.workflow_id 指向最新一次执行,
|
||||
// 仅当会话仍持有本次执行 ID 才回写,防慢终态把 A 的完成态覆写到 B 的摘要上
|
||||
if conv.workflow_id.as_deref() == Some(&*exec_id) {
|
||||
// 终态替换启动态末行占位(追加会让「待执行器上报」与「已完成」并存自相矛盾)
|
||||
let base = cb_dag_summary
|
||||
.strip_suffix(PENDING_RESULT_LINE)
|
||||
.unwrap_or(&cb_dag_summary);
|
||||
let summary = if status == "completed" {
|
||||
format!("{base}完成: 已完成")
|
||||
} else {
|
||||
format!("{base}完成: 已失败(节点: {})", failed_node)
|
||||
};
|
||||
conv.workflow_dag_summary = Some(summary);
|
||||
}
|
||||
}
|
||||
}
|
||||
}) // 闭 catch_unwind 内 async move
|
||||
.catch_unwind()
|
||||
.await;
|
||||
|
||||
Reference in New Issue
Block a user