lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增: advance_task auto_path跳态自动推进

Browse Source 
- 跳态目标按 shortest_path 自动逐跳推进(advance_task_auto_path),每跳独立校验+review_rounds 回归计数照旧
- 直达单跳与严格路径行为一致;workflow.rs 回调等机器生成单跳场景仍直用 advance_task_atomic 保持严格语义
- update_task 工具描述与 status 拦截错误文案同步 auto_path 语义
- 治 LLM 反复请求跳态推进被状态机拒绝的空耗往返
This commit is contained in:
lxy
2026-10-05 00:35:03 +08:00
parent c10f0e46f0
commit c83ea0c315
2 changed files with 189 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/df-nodes/src/task_advance_node.rs
+187 -3
View File
@@ -8,9 +8,14 @@
//!
//! 两条调用路径复用同一推进逻辑:
//! - DAG 工作流:TaskAdvanceNode 实现 Node trait,从 NodeContext.config 读 task_id /
//! target_status,持有 Arc<Database> 构造 TaskRepo。
//! target_status,持有 Arc<Database> 构造 TaskRepo(严格单跳,advance_task_atomic)。
//! - IPC 直驱:advance_task_atomic 公开 async fn,src-tauri commands::task::advance_task
//! thin 入口直接调用(无需构造 NodeContext,避免 EventBus/StateMachine 等重依赖)。
//!
//! auto_path 层(默认开启):advance_task_with_parent(IPC/AI 工具/MCP 同源入口)走
//! advance_task_auto_path,跳态目标按 shortest_path 自动逐跳推进(每跳独立校验 +
//! review_rounds 回归计数照旧);直达单跳与严格路径行为一致,workflow.rs 回调等
//! 机器生成单跳场景仍直用 advance_task_atomic 保持严格语义。
use std::sync::Arc;
@@ -113,6 +118,69 @@ pub async fn advance_task_atomic(
})
}
// ============================================================
// auto_path — 跳态目标自动逐跳推进(治 LLM 直取跳态被状态机拒绝的空耗往返)
// ============================================================
/// advance_task 自动路径推进(auto_path):目标不可直达时按最短路径逐跳推进。
///
/// 背景:实测 LLM 反复请求跳态推进(todo→done / in_progress→done),错误文本虽已附
/// shortest_path 提示,LLM 仍不走多跳,空耗往返。根修:跳态目标自动按
/// [`crate::task_state_machine::shortest_path`] 逐跳推进,每跳独立走
/// [`advance_task_atomic`](状态机校验 + 原子 CAS + review_rounds 回归计数规则照旧,
/// 如路径途经 in_review→in_progress 退回边照样 +1)。
///
/// 分派:
/// - 直达/委托(路径长 0 或 2):交 [`advance_task_atomic`] 单跳完成,同态/非法
/// target/终态不可达的错误文案零变更。
/// - 多跳(路径长 > 2):逐跳推进,任一跳失败即止(任务停在中间态,错误注明第几跳;
/// 不回滚已完成跳——回滚本身也要过状态机)。并发改动致某跳 CAS 失败时,调用方
/// 重读任务现状再决策即可。
///
/// 返回 (最终 TaskRecord, 实际路径):路径含首尾([from, ..., to]),直达时长 2、
/// 多跳时长 > 2,供调用方在 result 透出实际推进路径。
pub async fn advance_task_auto_path(
repo: &TaskRepo,
id: &str,
target_status: &str,
) -> df_types::error::Result<(TaskRecord, Vec<&'static str>)> {
let current = repo
.get_by_id(id)
.await?
.ok_or_else(|| df_types::error::Error::NotFound(format!("任务 {} 不存在", id)))?;
let from = current.status.as_str();
let path = crate::task_state_machine::shortest_path(from, target_status);
let record = if path.len() > 2 {
let mut record = current;
for (idx, &to) in path.iter().enumerate().skip(1) {
record = advance_task_atomic(repo, id, to).await.map_err(|e| match e {
// 任务被删(NotFound)原样透传;其余(状态机/CAS 冲突)包上跳次上下文
nf @ df_types::error::Error::NotFound(_) => nf,
other => df_types::error::Error::InvalidState {
current: format!(
"自动路径第 {}/{} 跳({} → {})失败: {other};任务停在中间态,请重读任务现状后再决策",
idx,
path.len() - 1,
path[idx - 1],
to
),
expected: to.to_string(),
},
})?;
}
record
} else {
// 直达(含同态/非法 target/终态不可达):atomic 原样校验与报错,语义零变更
advance_task_atomic(repo, id, target_status).await?
};
// 路径承载:shortest_path 产物全为 ALL_STATES 常量(&'static);空路径(同态/
// 非法/不可达)不会走到成功返回,此处兜底空列表。
let full_path = if path.len() >= 2 { path } else { Vec::new() };
Ok((record, full_path))
}
// ============================================================
// 父任务聚合 — 父任务 status 重算 + 推进联动(容器模型,设计 §2.1)
// ============================================================
@@ -200,7 +268,10 @@ pub async fn recompute_parent_status(
/// 推进任务 + 若为子任务则触发父聚合(父聚合失败仅 warn 不阻断,宽容语义)。
///
/// 推进链唯一 status 写入路径的两段式(设计 D3 统一,IPC/AI 工具/MCP 同源):
/// 1. 推进自身:调 `advance_task_atomic`(状态机校验 + 原子 CAS + review_rounds 累加)。
/// 1. 推进自身:调 `advance_task_auto_path`(auto_path 默认开启:跳态目标按
/// shortest_path 自动逐跳推进,治实测 LLM todo→done / in_progress→done 直取被
/// 状态机拒绝后不走多跳的空耗往返;直达单跳行为与原 advance_task_atomic 一致)。
/// 实际路径结构化返回值在 auto_path,多跳时此处 tracing 留痕。
/// 2. 父聚合:推进成功后若 `updated.parent_id` 有值,触发 `recompute_parent_status` 重算
/// 父 status(父任务=容器模型,聚合规则见 recompute_parent_status)。父聚合失败仅
/// tracing::warn 不阻断——子任务推进成功是主结果,父 status 漂移可后续修正。
@@ -211,7 +282,16 @@ pub async fn advance_task_with_parent(
id: &str,
target_status: &str,
) -> df_types::error::Result<TaskRecord> {
let updated = advance_task_atomic(repo, id, target_status).await?;
let (updated, path) = advance_task_auto_path(repo, id, target_status).await?;
if path.len() > 2 {
tracing::info!(
task_id = %id,
target = %target_status,
path = %crate::task_state_machine::format_path(&path),
review_rounds = updated.review_rounds,
"[advance_task] 跳态目标已按最短路径自动逐跳推进"
);
}
if let Some(pid) = &updated.parent_id {
if let Err(e) = recompute_parent_status(repo, pid).await {
tracing::warn!(
@@ -576,6 +656,110 @@ mod tests {
assert_eq!(r.review_rounds, 0, "解除 blocked 不累加");
}
// ============================================================
// auto_path — 跳态目标自动逐跳推进(治实测 todo→done / in_progress→done 直取失败)
// ============================================================
#[tokio::test]
async fn auto_path_todo_to_done_walks_full_gates() {
// 实测失败形态 1:todo 直取 done,auto_path 走满闸门链直达目标
let repo = setup().await;
repo.insert(rec("t1", TaskStatus::Todo)).await.unwrap();
let (r, path) = advance_task_auto_path(&repo, "t1", "done").await.unwrap();
assert_eq!(r.status.as_str(), "done");
assert_eq!(r.review_rounds, 0, "纯前向路径不累加 review_rounds");
assert_eq!(path, vec!["todo", "in_progress", "in_review", "testing", "done"]);
}
#[tokio::test]
async fn auto_path_in_progress_to_done() {
// 实测失败形态 2:in_progress 直取 done
let repo = setup().await;
repo.insert(rec("t1", TaskStatus::InProgress)).await.unwrap();
let (r, path) = advance_task_auto_path(&repo, "t1", "done").await.unwrap();
assert_eq!(r.status.as_str(), "done");
assert_eq!(path, vec!["in_progress", "in_review", "testing", "done"]);
}
#[tokio::test]
async fn auto_path_regression_hop_bumps_rounds() {
// 目标不可直达且路径途经退回边:testing→in_review→in_progress 两跳均为退回,
// 每跳独立按 is_regression 累加 → review_rounds = 2(规则照旧,逐跳生效)
let repo = setup().await;
repo.insert(rec("t1", TaskStatus::Testing)).await.unwrap();
let (r, path) = advance_task_auto_path(&repo, "t1", "in_progress").await.unwrap();
assert_eq!(r.status.as_str(), "in_progress");
assert_eq!(path, vec!["testing", "in_review", "in_progress"]);
assert_eq!(r.review_rounds, 2, "两跳退回各 +1,应累加到 2");
}
#[tokio::test]
async fn auto_path_multihop_forward_no_bump() {
// deferred → blocked:3 跳(deferred→todo→in_progress→blocked),全前向不累加
let repo = setup().await;
repo.insert(rec("t1", TaskStatus::Deferred)).await.unwrap();
let (r, path) = advance_task_auto_path(&repo, "t1", "blocked").await.unwrap();
assert_eq!(r.status.as_str(), "blocked");
assert_eq!(path, vec!["deferred", "todo", "in_progress", "blocked"]);
assert_eq!(r.review_rounds, 0);
}
#[tokio::test]
async fn auto_path_direct_transition_unchanged() {
// 直达单跳:行为与旧 advance_task_atomic 完全一致(路径长 2,rounds 不动)
let repo = setup().await;
repo.insert(rec("t1", TaskStatus::Todo)).await.unwrap();
let (r, path) = advance_task_auto_path(&repo, "t1", "in_progress").await.unwrap();
assert_eq!(r.status.as_str(), "in_progress");
assert_eq!(r.review_rounds, 0);
assert_eq!(path, vec!["todo", "in_progress"]);
}
#[tokio::test]
async fn auto_path_with_parent_multihop_lands_done() {
// LLM 入口(advance_task_with_parent)默认开启 auto_path:todo 直取 done 成功
let repo = setup().await;
repo.insert(rec("t1", TaskStatus::Todo)).await.unwrap();
let r = advance_task_with_parent(&repo, "t1", "done").await.unwrap();
assert_eq!(r.status.as_str(), "done");
assert_eq!(r.review_rounds, 0);
}
#[tokio::test]
async fn auto_path_terminal_and_same_state_still_rejected() {
// auto_path 不越状态机:终态不可达 / 同态仍按原错误类型拒绝(委托 atomic 原样报错)
let repo = setup().await;
repo.insert(rec("t1", TaskStatus::Done)).await.unwrap();
let err = advance_task_with_parent(&repo, "t1", "todo").await.unwrap_err();
assert!(
matches!(err, df_types::error::Error::InvalidState { .. }),
"终态不可达应是 InvalidState,实际: {err:?}"
);
repo.insert(rec("t2", TaskStatus::InProgress)).await.unwrap();
let err = advance_task_with_parent(&repo, "t2", "in_progress").await.unwrap_err();
match err {
df_types::error::Error::Validation(msg) => {
assert!(msg.contains("相同状态"), "同态消息应含「相同状态」,实际: {msg}");
}
other => panic!("同态应是 Validation,实际: {other:?}"),
}
}
#[tokio::test]
async fn auto_path_invalid_target_still_rejected() {
let repo = setup().await;
repo.insert(rec("t1", TaskStatus::Todo)).await.unwrap();
let err = advance_task_with_parent(&repo, "t1", "merged").await.unwrap_err();
assert!(matches!(err, df_types::error::Error::Validation(_)));
}
#[tokio::test]
async fn auto_path_not_found_still_rejected() {
let repo = setup().await;
let err = advance_task_with_parent(&repo, "nope", "done").await.unwrap_err();
assert!(matches!(err, df_types::error::Error::NotFound(_)));
}
// ============================================================
// CR-06 观察项 ② 测试补强 — ②-3(completed)/②-4(failed) 回调推进覆盖
//
src-tauri/src/commands/ai/tools/task.rs
+2 -2
View File
@@ -134,7 +134,7 @@ pub fn register(registry: &mut AiToolRegistry, db: &Arc<Database>) {
registry,
db: Arc<Database>,
"update_task",
"更新任务的指定字段（title/description/priority/assignee 等），需要提供任务 ID、字段名和新值。注意：status 改动须走 advance_task 工具（状态机推进，8 态 todo/in_progress/in_review/testing/blocked/done/cancelled/deferred，按 target_status 推进不可跳跃），本工具不接受 status 字段（防绕过状态机）",
"更新任务的指定字段（title/description/priority/assignee 等），需要提供任务 ID、字段名和新值。注意：status 改动须走 advance_task 工具（状态机推进，8 态 todo/in_progress/in_review/testing/blocked/done/cancelled/deferred，按 target_status 推进,目标不可直达时 auto_path 沿合法路径自动多跳），本工具不接受 status 字段（防绕过状态机）",
RiskLevel::Medium,
meta: ToolMeta {
entity_action: Some(("task", "update")),
@@ -151,7 +151,7 @@ pub fn register(registry: &mut AiToolRegistry, db: &Arc<Database>) {
// 对齐 Agent B df-storage tasks 白名单移 status——双重防御(schema 拒绝 + 白名单拒绝)。
if field == "status" {
anyhow::bail!(
"status 改动须走 advance_task 工具(状态机推进,8 态不可跳跃),本工具不接受 status 字段"
"status 改动须走 advance_task 工具(状态机推进,8 态,目标不可直达时沿合法路径自动多跳),本工具不接受 status 字段"
);
}
// 复用 df-storage CRUD 白名单(按表隔离),与 update_field 校验同源