修复: 上下文记忆与工作流注入生产侧空壳接通

WorkingContext 写入链落地(用户轮目标钉扎+压缩后 JSON 卡片回填重锚定,含 6 单测);
工作流 DAG 摘要注入生产侧接通(AI 路径启动写入+终态回写,审批调用点透传会话)。
附知识库与记忆走查诊断(「笨/丢历史」根因:消息压缩视图丢失+知识检索四重收口),
todo 登记修复分档(含随批补登 09-28 交互走查索引条目)。
This commit is contained in:
lxy
2026-09-29 13:28:56 +08:00
parent f57e25a277
commit dfa38f6259
9 changed files with 333 additions and 27 deletions
@@ -272,10 +272,10 @@ pub(super) async fn maybe_auto_compress(
}
let conv = session.conv(&conv_id);
let _compressed = conv.messages.compress_old_messages(protect_start);
// T3: 插入 NL 摘要(向前兼容),JSON 卡片可供后续 WorkingContext(T5)使用
// NL 摘要插首位 + JSON 卡片回填 WorkingContext,决策随压缩重锚定
conv.messages.insert_at(0, ChatMessage::system(&summary.nl_summary));
conv.messages.set_compressing(false);
// TODO(T5): 如有 WorkingContext,调用 version.reset_all(summary.json_card, turn)
conv.working_context.sync_after_compress(&summary.json_card);
}
tracing::info!(
conv_id = %conv_id,
@@ -1,19 +1,14 @@
//! 工作流 DAG 上下文注入 — 当会话关联工作流时,将活跃路径注入 system prompt
//!
//! T4 实现。提供两个函数:
//! - `build_dag_summary`: 从 DagDef 构建可读的 DAG 摘要文本
//! - `extract_active_path`: 从完整 DAG 中提取当前活跃路径(简化)
//! T4 实现,完整链路(生产侧已接通):
//! - 生产:`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。
//!
//! 调用方(前端/workflow 引擎)在启动工作流时将摘要存入
//! `PerConvState.workflow_dag_summary`,agentic loop 自动注入 system prompt。
//!
//! 预留说明(dead_code):本模块的 3 个函数当前无生产调用方——
//! T4 消费侧已落地(`agentic/mod.rs` 在构建 system prompt 时读
//! `PerConvState.workflow_dag_summary` 并注入),但生产侧尚无人写入该字段
//! (`PerConvState::new` 初始化为 `None`,无代码调用 `build_dag_summary`)。
//! 即完整链路为「工作流启动器 → 调本模块产出摘要 → 写 workflow_dag_summary
//! → agentic loop 注入 system prompt」,中间「生产者」环节待接通。
//! 属为后续工作流上下文注入准备的预留,非真死代码,故 allow 抑制告警。
//! `build_dag_summary` 已有生产调用方;`extract_active_path` 仍无生产调用方
//! (仅单测引用),模块级 allow 为其保留——节点级中间进度刷新属后续扩展。
#![allow(dead_code)]
+6 -2
View File
@@ -647,6 +647,8 @@ pub async fn ai_chat_send(
} else {
conv.messages.push(ChatMessage::user(&user_content));
}
// 新用户轮把本请求钉扎为 WorkingContext 常驻目标(截断 200 字符,与 agentic G1 目标钉扎 GOAL_MAX_CHARS 同量级)
conv.working_context.begin_turn(&user_content, 200);
// F-260619-04 P1:push 后取末条 user 消息 id(本轮 user,原供知识注入 referenced 溯源)。
// DRY(B):知识注入已收敛至 inject_knowledge_into_prompt(helper 内部同消息取 text+id),
// 此处 user_msg_id 不再透传到注入逻辑,保留下划线占用(锁内 push 已发生,语义不变)。
@@ -891,7 +893,7 @@ pub async fn ai_approve(
// 回填 / emit_data_changed / save / try_continue)。workflow 联动任务推进由 run_workflow 内部
// 按 task_id+target_status 自动处理(workflow.rs:270-308),此处不重复推进。
let exec_result: anyhow::Result<serde_json::Value> = if approval.tool_name == "run_workflow" {
super::super::execute_run_workflow_for_tool(&app, &state, &args).await
super::super::execute_run_workflow_for_tool(&app, &state, &args, conv_id.as_deref()).await
} else {
// F-260620 对齐 ai_authorize_dir(:658):ai_approve 执行同样包 60s 超时。
// 防 list_directory 大目录/run_command 慢命令卡死 → 到不了 audit_finalize/emit/try_continue
@@ -1273,7 +1275,7 @@ pub async fn ai_authorize_dir(
// → 到不了 audit_finalize/emit/try_continue → IPC 永挂 → generating 永真。包 60s 超时(对齐
// run_command),超时返 Err → 走 failed 分支 finalize+emit+try_continue,IPC 必返回不卡。
let exec_result: anyhow::Result<serde_json::Value> = if approval.tool_name == "run_workflow" {
super::super::execute_run_workflow_for_tool(&app, &state, &args).await
super::super::execute_run_workflow_for_tool(&app, &state, &args, conv_id.as_deref()).await
} else {
match tokio::time::timeout(
std::time::Duration::from_secs(60),
@@ -1972,6 +1974,8 @@ pub async fn ai_chat_force_send(
} else {
conv.messages.push(ChatMessage::user(&user_content));
}
// 新用户轮把本请求钉扎为 WorkingContext 常驻目标(截断 200 字符,与 agentic G1 目标钉扎 GOAL_MAX_CHARS 同量级)
conv.working_context.begin_turn(&user_content, 200);
// F-260619-04 P1:push 后立即取末条 user 消息 id(供知识注入 referenced 溯源)。
let user_msg_id = conv.messages.last_user_message_id();
// G1 目标钉扎已迁移至 run_agentic_loop 工具调用推理(提取方式同 ai_chat_send)
+5
View File
@@ -1142,10 +1142,14 @@ pub(crate) struct ToolCallDraft {
///
/// 不在此推进任务状态 —— workflow.rs::run_workflow 内部已按 task_id+target_status 自动联动
/// (成功推进到 target / 失败按 regression_target 退回),此处不重复。
///
/// `conv_id`(T4 会话关联)末参透传 run_workflow_inner:调用方带发起会话 ID 时,
/// 启动写 workflow_id/workflow_dag_summary、终态回写摘要,供 agentic loop 注入 system prompt。
pub(crate) async fn execute_run_workflow_for_tool(
app: &tauri::AppHandle,
state: &crate::state::AppState,
args: &serde_json::Value,
conv_id: Option<&str>,
) -> anyhow::Result<serde_json::Value> {
let task_id = args
.get("task_id")
@@ -1178,6 +1182,7 @@ pub(crate) async fn execute_run_workflow_for_tool(
Some(task_id.clone()),
Some(target_status.clone()),
"ai",
conv_id,
)
.await
.map_err(|e| anyhow::anyhow!("工作流执行失败: {}", e))?;
+54 -7
View File
@@ -21,6 +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::state::AppState;
use super::{err_str, now_millis};
@@ -105,7 +106,7 @@ pub async fn run_workflow(
// 瘦转发到 run_workflow_inner抽取共用核心供 ai_approve 经
// execute_run_workflow_for_tool 直接调,无需构造 tauri::State)。
// 命令层(前端 invoke run_workflow)是人工手动触发 → triggered_by="manual"。
run_workflow_inner(&app, &state, name, dag, config, task_id, target_status, "manual").await
run_workflow_inner(&app, &state, name, dag, config, task_id, target_status, "manual", None).await
}
/// 工作流执行核心逻辑(B-260617-01 抽取,命令层瘦转发)。
@@ -118,6 +119,11 @@ pub async fn run_workflow(
/// - 命令层 run_workflow(前端 invoke,人工手动)传 "manual"
/// - execute_run_workflow_for_tool(AI 工具经 ai_approve 审批后执行)传 "ai"
/// 原硬编码 "manual" 致 AI 路径误标,改由调用方透传。
///
/// `conv_id` T4 会话关联:Some(AI 路径)时启动写入 workflow_id(存本次执行 ID,非模板 ID)与
/// workflow_dag_summary 进会话状态,供 agentic loop 每轮 build_runtime_prompt 注入;
/// 执行闭包终态(completed/failed)回写终态摘要。会话不存在则静默跳过。
/// 命令层手动触发无会话上下文,传 None 零关联。
pub async fn run_workflow_inner(
app: &AppHandle,
state: &AppState,
@@ -127,6 +133,7 @@ pub async fn run_workflow_inner(
task_id: Option<String>,
target_status: Option<String>,
triggered_by: &str,
conv_id: Option<&str>,
) -> Result<String, String> {
// 0. F-260616-06 ①-1: DagDef 来源选模板(方案A·最小)。
// 规则:
@@ -162,6 +169,10 @@ pub async fn run_workflow_inner(
// 3. 写入执行记录(status=running)
let execution_id: ExecutionId = new_id().into();
let dag_json = serde_json::to_string(&dag).map_err(err_str)?;
// T4: DAG 摘要启动态(无当前节点),启动写入与终态刷新共用同一基础文本,
// 闭包外算好 move 进闭包,闭包内只拼状态行不重算。
let dag_summary = build_dag_summary(&dag, None);
let conv_id = conv_id.map(|s| s.to_string());
let record = WorkflowRecord {
id: execution_id.to_string(),
name,
@@ -181,6 +192,18 @@ pub async fn run_workflow_inner(
.await
.map_err(err_str)?;
// T4 生产侧:AI 路径启动时把执行 ID 与 DAG 摘要写入会话状态,接通
// 「生产者 → workflow_dag_summary → build_runtime_prompt 注入」链路。
// 存的是本次执行 ID(非模板 ID);会话不存在(已删除/无效 conv_id)静默跳过,
// 与消费侧 conv_read 短路语义一致(不惰性创建空会话)。
if let Some(cid) = conv_id.as_deref() {
let mut session = state.ai_session.lock().await;
if let Some(conv) = session.per_conv.get_mut(cid) {
conv.workflow_id = Some(execution_id.to_string());
conv.workflow_dag_summary = Some(dag_summary.clone());
}
}
// 3. 订阅事件总线,把事件转发给前端(收到完成/失败事件后退出)
let mut rx = state.event_bus.subscribe();
let forward_app = app.clone();
@@ -281,6 +304,10 @@ pub async fn run_workflow_inner(
// F-260616-06 ②-2: move task_id / target_status 进闭包供完成/失败回调使用
let cb_task_id = task_id.clone();
let cb_target_status = target_status.clone();
// T4 终态刷新物料:conv_id + 基础摘要 + 会话锁句柄 move 进执行闭包
let cb_conv_id = conv_id.clone();
let cb_dag_summary = dag_summary.clone();
let ai_session = state.ai_session.clone();
// panic-guard:执行 spawn 闭包最外层包 catch_unwind(AssertUnwindSafe(..))。
// panic 时走已存在 failed 分支(update status=failed + emit WorkflowFailed + state_registry.remove),
// 防 panic 后终态不发/registry 不清永久卡。主逻辑/调度零改动,仅在 panic 时补发终态。
@@ -381,24 +408,44 @@ pub async fn run_workflow_inner(
// 执行结束(成功/失败)清理状态注册表,防内存泄漏
state_registry.lock().await.remove(&exec_id);
// 执行失败时补发 WorkflowFailed(执行器内部只发 NodeFailed)
// failed_node 从状态机取首个失败/取消节点(R9⑪: 原 String::new() 空值)
if let Some(error) = error {
let failed_node = executor
// failed_node 从状态机取首个失败/取消节点(R9⑪: 原 String::new() 空值),
// 失败补发事件与 T4 会话终态摘要共用,提到分支外计算。
let failed_node = if error.is_some() {
executor
.state_machine()
.snapshot()
.into_iter()
.find(|(_, s)| matches!(s, NodeStatus::Failed | NodeStatus::Cancelled))
.map(|(id, _)| id)
.unwrap_or_default();
.unwrap_or_default()
} else {
String::new()
};
// 执行失败时补发 WorkflowFailed(执行器内部只发 NodeFailed)
if let Some(error) = error {
event_bus
.send(WorkflowEvent::WorkflowFailed {
execution_id: exec_id.clone(),
error,
failed_node,
failed_node: failed_node.clone(),
})
.await;
}
// T4 终态刷新:工作流终态后把完成/失败结论写回会话摘要,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_dag_summary = Some(summary);
}
}
}) // 闭 catch_unwind 内 async move
.catch_unwind()
.await;