diff --git a/crates/df-nodes/src/ai_node.rs b/crates/df-nodes/src/ai_node.rs index e9d0e8f..d139c59 100644 --- a/crates/df-nodes/src/ai_node.rs +++ b/crates/df-nodes/src/ai_node.rs @@ -6,8 +6,10 @@ //! 与 AI Chat（侧边栏交互对话）的区别：AI Node 由 DAG Executor 自动驱动， //! 适合嵌入自动化链路（如 想法 → AI分析 → 脚本落地 → 人工审批）。 //! -//! provider 解析/参数构造/自审 JSON 解析等纯逻辑见 `ai_node_helpers`(本文件只保留 -//! AiNode / AiSelfReviewNode 的 struct + impl,impl 块约束)。 +//! 拆分边界(纯结构,逻辑等价): +//! - provider 解析/参数构造/自审 JSON 解析等纯逻辑 → `ai_node_helpers` +//! - AiSelfReviewNode(自审闭环独立节点) → `ai_self_review_node` +//! - 本文件:仅 AiNode 的 struct + impl + tests。 use std::sync::Arc; @@ -17,12 +19,10 @@ use df_storage::crud::TaskRepo; use df_storage::db::Database; use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; -// 抽离的纯函数/类型(glob 引入,tests `use super::*` 间接可见) -#[allow(unused_imports)] -use crate::ai_node_helpers::{ - gate_should_block, parse_review_json, provider_from_params, resolve_and_parse, - truncate_for_summary, ResolvedProvider, REVIEW_SYSTEM_PROMPT, -}; +// AiNode::execute 用的纯函数(resolve_and_parse / provider_from_params); +// 其余 helpers(parse_review_json/gate_should_block/REVIEW_SYSTEM_PROMPT/ResolvedProvider/ +// truncate_for_summary)由 ai_self_review_node 与 tests 引入,不在此处 glob。 +use crate::ai_node_helpers::{provider_from_params, resolve_and_parse}; /// AI 节点 /// @@ -143,260 +143,13 @@ impl Node for AiNode { } } -// ============================================================ -// ai_self_review 节点 — AI 自审闭环(决策 a 步骤③) -// ============================================================ -// -// 复用 AiNode 的 provider 调用逻辑(parse_params + build_provider + complete),差异: -// 1. prompt 固定四维度审查模板(需求符合度/完整性/正确性/边界),不从 config/inputs 取 prompt -// 2. system_prompt 强约束「只输出 JSON」+ temperature=0 -// 3. LLM 输出按结构化 JSON 解析,失败兜底 verdict=unknown(防 LLM 不按要求输出) -// 4. 自审输入:task.description(需求,经 ctx.config["task_id"] 读 task) + task.output_json.text(产出) -// 5. 自审结果写回 task.output_json 加 review 子字段(不覆盖 text,保留对照) -// 6. review 摘要塞 NodeOutput.data,供下游 human_review 经 DAG inputs["ai_self_review"] 读 -// -// verdict 处理(首版保守):自审是辅助,最终决策权在人。verdict=fail 不主动 Err, -// 节点返 Ok 让 human_review 继续,人在审批卡看到 fail 结论再定(human reject 走 Err→工作流 -// failed→②-4 回调退回 in_review)。 - -/// AI 自审节点(决策 a 步骤③) -pub struct AiSelfReviewNode { - db: Arc, -} - -impl AiSelfReviewNode { - pub fn new(db: Arc) -> Self { - Self { db } - } - - /// 拼装自审 user prompt:任务需求 + 待审产出 + 四维度审查要求 + 输出格式。 - /// description / output_text 缺失时给占位(不报错,信任调用方注入合法 task_id)。 - fn build_review_prompt(description: &str, output_text: &str) -> String { - format!( - "\ -## 任务需求 -{description} - -## 待审产出 -{output_text} - -## 审查维度 -1. 需求符合度:产出是否覆盖需求描述的所有要点 -2. 产出完整性:是否有遗漏、未完成的部分 -3. 正确性:逻辑/事实/语法是否正确 -4. 边界处理:异常输入、空值、错误路径是否考虑 - -## 输出格式(严格 JSON) -{{\"verdict\":\"pass\"|\"fail\",\"dimensions\":{{\"requirement_fit\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"completeness\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"correctness\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"boundary\":{{\"score\":0-10,\"issues\":[\"...\"]}}}},\"summary\":\"一句话总结\",\"suggestions\":[\"改进建议\"]}} -verdict=fail 当且仅当任一维度 score<6 或有阻断性 issue。" - ) - } -} - -#[async_trait] -impl Node for AiSelfReviewNode { - async fn execute(&self, ctx: NodeContext) -> NodeResult { - tracing::info!("AiSelfReviewNode 执行: node_id={}", ctx.node_id); - - // FR-S1 注入链：provider 经 df_storage::secret 在 AiNode 内存解析，api_key 不进 config。 - let p = resolve_and_parse(&self.db, &ctx.config, &ctx.inputs).await?; - - // ── 读任务(需求 + 产出) ── - // task_id 必填(自审对象是具体任务的产出);缺失报错而非静默跳过。 - let task_id = ctx - .config - .get("task_id") - .and_then(|v| v.as_str()) - .ok_or_else(|| anyhow::anyhow!("AiSelfReviewNode 缺少必填参数: task_id"))?; - - let repo = TaskRepo::new(&self.db); - let task = repo - .get_by_id(task_id) - .await - .map_err(|e| anyhow::anyhow!("自审读任务失败: {}", e))? - .ok_or_else(|| anyhow::anyhow!("自审任务 {} 不存在", task_id))?; - - // output_json 可能无(ai_execute 未跑或失败) → 待审产出占位提示,不报错。 - let output_text = task - .output_json - .as_deref() - .and_then(|s| serde_json::from_str::(s).ok()) - .and_then(|v| v.get("text").and_then(|t| t.as_str()).map(String::from)) - .unwrap_or_else(|| "(无产出,仅依据需求审查)".to_string()); - - let prompt = Self::build_review_prompt(&task.description, &output_text); - - let provider: Box = provider_from_params(&p); - - let request = CompletionRequest { - model: p.model.clone(), - messages: vec![ - ChatMessage::system(REVIEW_SYSTEM_PROMPT), - ChatMessage::user(prompt), - ], - temperature: Some(0.0), // 强约束确定性输出 - max_tokens: p.max_tokens, - stream: false, - tools: None, - tool_choice: None, - reasoning_content: None, - }; - - tracing::info!( - "AiSelfReviewNode 调用 LLM: model={}, base_url={}", - p.provider.default_model, - p.provider.base_url - ); - let response = provider.complete(request).await?; - - // ── 解析自审 JSON(失败兜底 verdict=unknown) ── - let review = parse_review_json(&response.text); - let verdict = review - .get("verdict") - .and_then(|v| v.as_str()) - .unwrap_or("unknown") - .to_string(); - // summary / suggestions 在 move review 前先取走(供 NodeOutput.data 平铺) - let summary = review - .get("summary") - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - let suggestions = review - .get("suggestions") - .cloned() - .unwrap_or(serde_json::json!([])); - - tracing::info!( - "AiSelfReviewNode 完成: verdict={}, model={}", - verdict, - response.model - ); - - // ── 写回 task.output_json 加 review 子字段(不覆盖 text) ── - // 读现有 output_json(若有),合并 review 子字段后整体写回。 - let mut output_obj: serde_json::Map = task - .output_json - .as_deref() - .and_then(|s| serde_json::from_str::(s).ok()) - .and_then(|v| v.as_object().cloned()) - .unwrap_or_default(); - // review 子字段:自审结论 + reviewed_at + reviewer_model(便于前端展示与人对照) - let mut review_with_meta = match review { - serde_json::Value::Object(m) => m, - other => { - let mut m = serde_json::Map::new(); - m.insert("verdict".into(), other); - m - } - }; - review_with_meta.insert( - "reviewed_at".into(), - serde_json::Value::Number(df_types::now_millis().into()), - ); - review_with_meta.insert("reviewer_model".into(), serde_json::json!(response.model)); - output_obj.insert("review".into(), serde_json::Value::Object(review_with_meta)); - - let merged = serde_json::Value::Object(output_obj); - if let Ok(json_str) = serde_json::to_string(&merged) { - if let Err(e) = repo.update_field(task_id, "output_json", &json_str).await { - tracing::warn!( - task_id = task_id, - error = %e, - "AiSelfReviewNode 落 review 子字段失败(摘要已在 NodeOutput,节点继续)" - ); - } - } - - // ── NodeOutput.data 塞 review 摘要,供下游 human_review 经 inputs["ai_self_review"] 读 ── - // human_review(HumanNode)从 config/inputs 拼 description;此处把 verdict/summary/suggestions - // 平铺进 data,HumanNode 零改动即可透传(前端/模板层可据此拼审批卡描述)。 - let output = NodeOutput::from_value(serde_json::json!({ - "verdict": verdict, - "summary": summary, - "suggestions": suggestions, - "review": merged.get("review").cloned().unwrap_or(serde_json::Value::Null), - "model": response.model, - })); - - // ── 阶段3: 自审闸门(F-260616-07 决策 a 步骤③) ── - // config["gate"]==true 时,AiSelfReviewNode 从「自审辅助」升级为「DAG 节点闸门」: - // verdict="fail" → 返回 Err → executor first_err 中止后续层(下游 human_review 不跑) - // → 工作流 failed → ②-4 回调退回 in_review(对齐工作流失败语义) - // verdict="unknown" → 保持 Ok(LLM 输出不可靠时不阻断,人定权) - // verdict="pass" → Ok 放行下游 - // - // 落库在前(已 update_field output_json),故闸门 fail 时 task.output_json.review - // 已含结论,前端/二次工作流可读自审详情(不丢失审查痕迹)。 - // - // 设计选型(三方案对比,见 commit/设计文档): - // A) AiNode 内部门控(本方案):1 处 return Err,复用 executor first_err + ②-4 回调, - // 零 executor 核心改动,零依赖暂缓的条件引擎(T-260614-11)。最简。 - // B) DAG edges 条件 + ConditionEngine 求值:依赖 T-260614-11(暂缓),executor 当前 - // 完全不评估 edge.condition(topological_layers 无条件收录所有边),须先做条件 - // 引擎 Phase1+2 才能用 → 拆波。 - // C) executor 闸门检查钩子(节点 execute 后 executor 读 output.verdict):改 DagExecutor - // 核心循环,牵动所有节点,风险/范围不符「最简不破坏」。 - // 选 A:语义最贴近「自审结果作为闸门」(自审节点自行决定放行/阻断),且 gate 可按节点 - // config 开关(默认 false = 阶段2 行为不变,模板/前端零强制改动,向后兼容)。 - let gate_enabled = ctx - .config - .get("gate") - .and_then(|v| v.as_bool()) - .unwrap_or(false); - if gate_should_block(gate_enabled, &verdict) { - tracing::info!( - node_id = %ctx.node_id, - task_id, - "AiSelfReviewNode 闸门触发: verdict=fail, 阻断下游(工作流将 failed)" - ); - return Err(anyhow::anyhow!( - "AI 自审闸门未通过(verdict=fail): {}", - if summary.is_empty() { "自审结论 fail" } else { &summary } - )); - } - - Ok(output) - } - - fn schema(&self) -> NodeSchema { - NodeSchema { - params: serde_json::json!({ - "type": "object", - "properties": { - "task_id": { "type": "string", "description": "自审目标任务 ID(必填)" }, - "provider_id": { "type": "string", "description": "AI Provider ID(FR-S1:密钥经 secret 解析不进 config;留空走默认 provider)" }, - "model": { "type": "string", "description": "模型名(可选,留空用 record.default_model)" }, - "max_tokens": { "type": "integer" }, - "gate": { "type": "boolean", "description": "阶段3 闸门开关:false(默认)=自审辅助,verdict 仅透传展示;true=自审结果作 DAG 闸门,verdict=fail 返回 Err 阻断下游(工作流 failed → ②-4 退回),verdict=unknown/pass 放行" } - }, - "required": ["task_id", "provider_id"] - }), - output: serde_json::json!({ - "type": "object", - "properties": { - "verdict": { "type": "string", "enum": ["pass", "fail", "unknown"] }, - "summary": { "type": "string" }, - "suggestions": { "type": "array", "items": { "type": "string" } }, - "review": { "type": "object" }, - "model": { "type": "string" } - } - }), - } - } - - fn node_type(&self) -> &str { - "ai_self_review" - } -} - #[cfg(test)] mod tests { use super::*; - use crate::ai_node_helpers::{parse_params, resolve_provider}; - use df_storage::crud::{AiProviderRepo, ProjectRepo}; + use crate::ai_node_helpers::{parse_params, resolve_provider, ResolvedProvider}; + use df_storage::crud::AiProviderRepo; use df_storage::db::Database; - use df_storage::models::{AiProviderRecord, ProjectRecord, TaskRecord}; + use df_storage::models::AiProviderRecord; use serde_json::json; use std::collections::HashMap; @@ -664,173 +417,7 @@ mod tests { ); } - // ============================================================ - // 决策 a 自审闭环单测 — output_json 写回 + review JSON 解析兜底 - // ============================================================ - - /// 内存 DB 构造 helper:插父项目 + 指定 task,返回 (db, task_id)。 - async fn setup_task_db(task_output_json: Option<&str>) -> (Database, String) { - let db = Database::open_in_memory().await.expect("open_in_memory"); - ProjectRepo::new(&db) - .insert(ProjectRecord { - id: "p1".to_string(), - name: "proj".to_string(), - description: "".to_string(), - status: "planning".to_string(), - idea_id: None, - path: None, - stack: None, - created_at: "0".to_string(), - updated_at: "0".to_string(), - }) - .await - .expect("insert project"); - TaskRepo::new(&db) - .insert(TaskRecord { - id: "t1".to_string(), - project_id: "p1".to_string(), - title: "t1".to_string(), - description: "实现登录接口".to_string(), - status: "testing".to_string(), - priority: 2, - branch_name: None, - assignee: None, - workflow_def_id: None, - base_branch: None, - review_rounds: 0, - output_json: task_output_json.map(String::from), - idea_id: None, - created_at: "0".to_string(), - updated_at: "0".to_string(), - }) - .await - .expect("insert task"); - (db, "t1".to_string()) - } - - /// 决策 a 步骤②:AiNode.execute 落产出验证 — 有 task_id 时 output_json 写入。 - /// 用 mock provider 避免真调 LLM:自定义 Node 替换不便,改为直接验证 update_field 路径 - /// (execute 内部调 provider.complete 需真 LLM,此处降级验证 parse_review_json 兜底 + - /// 直接 repo.update_field 路径,与 execute 落库逻辑同一条 crud.update_field)。 - #[tokio::test] - async fn update_field_writes_output_json() { - // 验证 output_json 在 tasks 白名单(update_field 不报「不允许的字段名」)。 - // 这正是 AiNode.execute / AiSelfReviewNode.execute 落产出的底层路径。 - let (db, task_id) = setup_task_db(None).await; - let repo = TaskRepo::new(&db); - let payload = r#"{"text":"产出内容","model":"glm-4","usage":{}}"#; - let ok = repo - .update_field(&task_id, "output_json", payload) - .await - .expect("update_field output_json"); - assert!(ok, "update_field 应返回 affected>0=true"); - let after = repo.get_by_id(&task_id).await.unwrap().unwrap(); - assert_eq!( - after.output_json.as_deref(), - Some(payload), - "output_json 应被写入" - ); - // 原始 text 字段(产出对照)保留 - let parsed: serde_json::Value = serde_json::from_str(after.output_json.as_deref().unwrap()).unwrap(); - assert_eq!(parsed["text"], json!("产出内容")); - } - - /// 步骤③:parse_review_json 合规 JSON 直通(保留 verdict/dimensions/summary/suggestions)。 - #[test] - fn parse_review_json_valid_passes_through() { - let raw = r#"{"verdict":"pass","dimensions":{"requirement_fit":{"score":9,"issues":[]}},"summary":"符合需求","suggestions":["可加单测"]}"#; - let v = parse_review_json(raw); - assert_eq!(v["verdict"], json!("pass")); - assert_eq!(v["summary"], json!("符合需求")); - assert_eq!(v["suggestions"], json!(["可加单测"])); - assert_eq!(v["dimensions"]["requirement_fit"]["score"], json!(9)); - } - - /// 步骤③:parse_review_json 非 JSON / 缺 verdict → 兜底 verdict=unknown + summary 含原文。 - #[test] - fn parse_review_json_invalid_falls_back_unknown() { - // 非 JSON - let v = parse_review_json("这不是 JSON,LLM 没按要求输出"); - assert_eq!(v["verdict"], json!("unknown")); - assert!(v["summary"].as_str().unwrap().contains("不是 JSON")); - assert!(v["summary"].as_str().unwrap().contains("解析失败")); - - // JSON 但缺 verdict - let v = parse_review_json(r#"{"foo":"bar"}"#); - assert_eq!(v["verdict"], json!("unknown")); - - // JSON 非 object(数组) - let v = parse_review_json("[1,2,3]"); - assert_eq!(v["verdict"], json!("unknown")); - } - - /// 步骤③:LLM 偶尔包 ```json 代码块,parse_review_json 应剥离后解析成功。 - #[test] - fn parse_review_json_strips_markdown_code_fence() { - let raw = "```json\n{\"verdict\":\"fail\",\"summary\":\"缺边界处理\"}\n```"; - let v = parse_review_json(raw); - assert_eq!(v["verdict"], json!("fail")); - assert_eq!(v["summary"], json!("缺边界处理")); - } - - /// 步骤③:truncate_for_summary 长文截断。 - #[test] - fn truncate_for_summary_long_text() { - let short = "短文"; - assert_eq!(truncate_for_summary(short), "短文"); - let long: String = "字".repeat(500); - let t = truncate_for_summary(&long); - assert!(t.ends_with("…")); - assert!(t.chars().count() <= 302, "截断后含省略号应 ≈300 字"); - } - - /// 步骤③:AiSelfReviewNode.build_review_prompt 含需求 + 产出 + 四维度。 - #[test] - fn build_review_prompt_contains_inputs() { - let p = AiSelfReviewNode::build_review_prompt("登录接口", "已实现 /login"); - assert!(p.contains("登录接口"), "prompt 应含需求"); - assert!(p.contains("已实现 /login"), "prompt 应含产出"); - assert!(p.contains("需求符合度"), "prompt 应含四维度"); - assert!(p.contains("边界处理")); - assert!(p.contains("verdict"), "prompt 应含输出格式约束"); - } - - // ============================================================ - // 阶段3 自审闸门(gate_should_block)单测 - // ============================================================ - // - // gate 决策矩阵: - // gate=false(默认,阶段2 行为) → 任何 verdict 都放行(辅助模式) - // gate=true + verdict=pass → 放行 - // gate=true + verdict=unknown → 放行(LLM 不可靠时不阻断,人定权) - // gate=true + verdict=fail → 阻断(返回 Err,工作流 failed) - - #[test] - fn gate_disabled_never_blocks_any_verdict() { - // gate 默认 false(阶段2 兼容):无论 verdict 如何都不阻断,自审仅辅助展示 - assert!(!gate_should_block(false, "fail"), "gate 关闭时 fail 也不阻断"); - assert!(!gate_should_block(false, "pass"), "gate 关闭时 pass 放行"); - assert!(!gate_should_block(false, "unknown"), "gate 关闭时 unknown 放行"); - } - - #[test] - fn gate_enabled_blocks_only_on_fail() { - // gate 开启:仅 fail 阻断 - assert!(gate_should_block(true, "fail"), "gate 开 + fail 应阻断"); - assert!(!gate_should_block(true, "pass"), "gate 开 + pass 放行"); - assert!( - !gate_should_block(true, "unknown"), - "gate 开 + unknown 放行(LLM 不可靠时保人定权)" - ); - } - - #[test] - fn gate_unknown_verdict_does_not_block() { - // 防御:verdict 非标准值(如解析异常)不应触发阻断(对齐 unknown 保守不阻断语义) - assert!( - !gate_should_block(true, "weird"), - "非标准 verdict 不应阻断(对齐 unknown 保守)" - ); - assert!(!gate_should_block(true, ""), "空 verdict 不应阻断"); - } + // 自审闭环相关单测(parse_review_json / truncate_for_summary / gate_should_block / + // build_review_prompt / update_field_writes_output_json)已随 AiSelfReviewNode 迁移至 + // ai_self_review_node.rs(与被测代码同位,纯搬运)。 } diff --git a/crates/df-nodes/src/ai_self_review_node.rs b/crates/df-nodes/src/ai_self_review_node.rs new file mode 100644 index 0000000..3b2f200 --- /dev/null +++ b/crates/df-nodes/src/ai_self_review_node.rs @@ -0,0 +1,447 @@ +//! AI 自审节点(决策 a 步骤③)— AiSelfReviewNode +//! +//! 从 ai_node.rs 拆出的独立节点(纯结构搬运,逻辑等价)。复用 AiNode 的 provider +//! 调用链(resolve_and_parse + provider_from_params),差异在 prompt 模板/JSON 解析/ +//! verdict 闸门。provider 解析与自审 JSON 解析等纯逻辑见 `ai_node_helpers`。 +//! +//! 节点类型注册 key:`ai_self_review`(state.rs build_registry)。 +//! 模板对齐:task_workflow_templates.rs testing 模板 `ai_self_review` 节点 + gate:true。 + +use std::sync::Arc; + +use async_trait::async_trait; +use df_ai::provider::{ChatMessage, CompletionRequest, LlmProvider}; +use df_storage::crud::TaskRepo; +use df_storage::db::Database; +use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; + +// 抽离的纯函数/类型(与 AiNode 共用)。 +use crate::ai_node_helpers::{ + gate_should_block, parse_review_json, provider_from_params, resolve_and_parse, + REVIEW_SYSTEM_PROMPT, +}; + +// AiSelfReviewNode 节点 — AI 自审闭环(决策 a 步骤③) +// +// 复用 AiNode 的 provider 调用逻辑(parse_params + build_provider + complete),差异: +// 1. prompt 固定四维度审查模板(需求符合度/完整性/正确性/边界),不从 config/inputs 取 prompt +// 2. system_prompt 强约束「只输出 JSON」+ temperature=0 +// 3. LLM 输出按结构化 JSON 解析,失败兜底 verdict=unknown(防 LLM 不按要求输出) +// 4. 自审输入:task.description(需求,经 ctx.config["task_id"] 读 task) + task.output_json.text(产出) +// 5. 自审结果写回 task.output_json 加 review 子字段(不覆盖 text,保留对照) +// 6. review 摘要塞 NodeOutput.data,供下游 human_review 经 DAG inputs["ai_self_review"] 读 +// +// verdict 处理(首版保守):自审是辅助,最终决策权在人。verdict=fail 不主动 Err, +// 节点返 Ok 让 human_review 继续,人在审批卡看到 fail 结论再定(human reject 走 Err→工作流 +// failed→②-4 回调退回 in_review)。 + +/// AI 自审节点(决策 a 步骤③) +pub struct AiSelfReviewNode { + db: Arc, +} + +impl AiSelfReviewNode { + pub fn new(db: Arc) -> Self { + Self { db } + } + + /// 拼装自审 user prompt:任务需求 + 待审产出 + 四维度审查要求 + 输出格式。 + /// description / output_text 缺失时给占位(不报错,信任调用方注入合法 task_id)。 + fn build_review_prompt(description: &str, output_text: &str) -> String { + format!( + "\ +## 任务需求 +{description} + +## 待审产出 +{output_text} + +## 审查维度 +1. 需求符合度:产出是否覆盖需求描述的所有要点 +2. 产出完整性:是否有遗漏、未完成的部分 +3. 正确性:逻辑/事实/语法是否正确 +4. 边界处理:异常输入、空值、错误路径是否考虑 + +## 输出格式(严格 JSON) +{{\"verdict\":\"pass\"|\"fail\",\"dimensions\":{{\"requirement_fit\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"completeness\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"correctness\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"boundary\":{{\"score\":0-10,\"issues\":[\"...\"]}}}},\"summary\":\"一句话总结\",\"suggestions\":[\"改进建议\"]}} +verdict=fail 当且仅当任一维度 score<6 或有阻断性 issue。" + ) + } +} + +#[async_trait] +impl Node for AiSelfReviewNode { + async fn execute(&self, ctx: NodeContext) -> NodeResult { + tracing::info!("AiSelfReviewNode 执行: node_id={}", ctx.node_id); + + // FR-S1 注入链：provider 经 df_storage::secret 在 AiNode 内存解析，api_key 不进 config。 + let p = resolve_and_parse(&self.db, &ctx.config, &ctx.inputs).await?; + + // ── 读任务(需求 + 产出) ── + // task_id 必填(自审对象是具体任务的产出);缺失报错而非静默跳过。 + let task_id = ctx + .config + .get("task_id") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("AiSelfReviewNode 缺少必填参数: task_id"))?; + + let repo = TaskRepo::new(&self.db); + let task = repo + .get_by_id(task_id) + .await + .map_err(|e| anyhow::anyhow!("自审读任务失败: {}", e))? + .ok_or_else(|| anyhow::anyhow!("自审任务 {} 不存在", task_id))?; + + // output_json 可能无(ai_execute 未跑或失败) → 待审产出占位提示,不报错。 + let output_text = task + .output_json + .as_deref() + .and_then(|s| serde_json::from_str::(s).ok()) + .and_then(|v| v.get("text").and_then(|t| t.as_str()).map(String::from)) + .unwrap_or_else(|| "(无产出,仅依据需求审查)".to_string()); + + let prompt = Self::build_review_prompt(&task.description, &output_text); + + let provider: Box = provider_from_params(&p); + + let request = CompletionRequest { + model: p.model.clone(), + messages: vec![ + ChatMessage::system(REVIEW_SYSTEM_PROMPT), + ChatMessage::user(prompt), + ], + temperature: Some(0.0), // 强约束确定性输出 + max_tokens: p.max_tokens, + stream: false, + tools: None, + tool_choice: None, + reasoning_content: None, + }; + + tracing::info!( + "AiSelfReviewNode 调用 LLM: model={}, base_url={}", + p.provider.default_model, + p.provider.base_url + ); + let response = provider.complete(request).await?; + + // ── 解析自审 JSON(失败兜底 verdict=unknown) ── + let review = parse_review_json(&response.text); + let verdict = review + .get("verdict") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + // summary / suggestions 在 move review 前先取走(供 NodeOutput.data 平铺) + let summary = review + .get("summary") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let suggestions = review + .get("suggestions") + .cloned() + .unwrap_or(serde_json::json!([])); + + tracing::info!( + "AiSelfReviewNode 完成: verdict={}, model={}", + verdict, + response.model + ); + + // ── 写回 task.output_json 加 review 子字段(不覆盖 text) ── + // 读现有 output_json(若有),合并 review 子字段后整体写回。 + let mut output_obj: serde_json::Map = task + .output_json + .as_deref() + .and_then(|s| serde_json::from_str::(s).ok()) + .and_then(|v| v.as_object().cloned()) + .unwrap_or_default(); + // review 子字段:自审结论 + reviewed_at + reviewer_model(便于前端展示与人对照) + let mut review_with_meta = match review { + serde_json::Value::Object(m) => m, + other => { + let mut m = serde_json::Map::new(); + m.insert("verdict".into(), other); + m + } + }; + review_with_meta.insert( + "reviewed_at".into(), + serde_json::Value::Number(df_types::now_millis().into()), + ); + review_with_meta.insert("reviewer_model".into(), serde_json::json!(response.model)); + output_obj.insert("review".into(), serde_json::Value::Object(review_with_meta)); + + let merged = serde_json::Value::Object(output_obj); + if let Ok(json_str) = serde_json::to_string(&merged) { + if let Err(e) = repo.update_field(task_id, "output_json", &json_str).await { + tracing::warn!( + task_id = task_id, + error = %e, + "AiSelfReviewNode 落 review 子字段失败(摘要已在 NodeOutput,节点继续)" + ); + } + } + + // ── NodeOutput.data 塞 review 摘要,供下游 human_review 经 inputs["ai_self_review"] 读 ── + // human_review(HumanNode)从 config/inputs 拼 description;此处把 verdict/summary/suggestions + // 平铺进 data,HumanNode 零改动即可透传(前端/模板层可据此拼审批卡描述)。 + let output = NodeOutput::from_value(serde_json::json!({ + "verdict": verdict, + "summary": summary, + "suggestions": suggestions, + "review": merged.get("review").cloned().unwrap_or(serde_json::Value::Null), + "model": response.model, + })); + + // ── 阶段3: 自审闸门(F-260616-07 决策 a 步骤③) ── + // config["gate"]==true 时,AiSelfReviewNode 从「自审辅助」升级为「DAG 节点闸门」: + // verdict="fail" → 返回 Err → executor first_err 中止后续层(下游 human_review 不跑) + // → 工作流 failed → ②-4 回调退回 in_review(对齐工作流失败语义) + // verdict="unknown" → 保持 Ok(LLM 输出不可靠时不阻断,人定权) + // verdict="pass" → Ok 放行下游 + // + // 落库在前(已 update_field output_json),故闸门 fail 时 task.output_json.review + // 已含结论,前端/二次工作流可读自审详情(不丢失审查痕迹)。 + // + // 设计选型(三方案对比,见 commit/设计文档): + // A) AiNode 内部门控(本方案):1 处 return Err,复用 executor first_err + ②-4 回调, + // 零 executor 核心改动,零依赖暂缓的条件引擎(T-260614-11)。最简。 + // B) DAG edges 条件 + ConditionEngine 求值:依赖 T-260614-11(暂缓),executor 当前 + // 完全不评估 edge.condition(topological_layers 无条件收录所有边),须先做条件 + // 引擎 Phase1+2 才能用 → 拆波。 + // C) executor 闸门检查钩子(节点 execute 后 executor 读 output.verdict):改 DagExecutor + // 核心循环,牵动所有节点,风险/范围不符「最简不破坏」。 + // 选 A:语义最贴近「自审结果作为闸门」(自审节点自行决定放行/阻断),且 gate 可按节点 + // config 开关(默认 false = 阶段2 行为不变,模板/前端零强制改动,向后兼容)。 + let gate_enabled = ctx + .config + .get("gate") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + if gate_should_block(gate_enabled, &verdict) { + tracing::info!( + node_id = %ctx.node_id, + task_id, + "AiSelfReviewNode 闸门触发: verdict=fail, 阻断下游(工作流将 failed)" + ); + return Err(anyhow::anyhow!( + "AI 自审闸门未通过(verdict=fail): {}", + if summary.is_empty() { "自审结论 fail" } else { &summary } + )); + } + + Ok(output) + } + + fn schema(&self) -> NodeSchema { + NodeSchema { + params: serde_json::json!({ + "type": "object", + "properties": { + "task_id": { "type": "string", "description": "自审目标任务 ID(必填)" }, + "provider_id": { "type": "string", "description": "AI Provider ID(FR-S1:密钥经 secret 解析不进 config;留空走默认 provider)" }, + "model": { "type": "string", "description": "模型名(可选,留空用 record.default_model)" }, + "max_tokens": { "type": "integer" }, + "gate": { "type": "boolean", "description": "阶段3 闸门开关:false(默认)=自审辅助,verdict 仅透传展示;true=自审结果作 DAG 闸门,verdict=fail 返回 Err 阻断下游(工作流 failed → ②-4 退回),verdict=unknown/pass 放行" } + }, + "required": ["task_id", "provider_id"] + }), + output: serde_json::json!({ + "type": "object", + "properties": { + "verdict": { "type": "string", "enum": ["pass", "fail", "unknown"] }, + "summary": { "type": "string" }, + "suggestions": { "type": "array", "items": { "type": "string" } }, + "review": { "type": "object" }, + "model": { "type": "string" } + } + }), + } + } + + fn node_type(&self) -> &str { + "ai_self_review" + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ai_node_helpers::{gate_should_block, parse_review_json, truncate_for_summary}; + use df_storage::crud::{ProjectRepo, TaskRepo}; + use df_storage::db::Database; + use df_storage::models::{ProjectRecord, TaskRecord}; + use serde_json::json; + + // ============================================================ + // 决策 a 自审闭环单测 — output_json 写回 + review JSON 解析兜底 + // ============================================================ + + /// 内存 DB 构造 helper:插父项目 + 指定 task,返回 (db, task_id)。 + async fn setup_task_db(task_output_json: Option<&str>) -> (Database, String) { + let db = Database::open_in_memory().await.expect("open_in_memory"); + ProjectRepo::new(&db) + .insert(ProjectRecord { + id: "p1".to_string(), + name: "proj".to_string(), + description: "".to_string(), + status: "planning".to_string(), + idea_id: None, + path: None, + stack: None, + created_at: "0".to_string(), + updated_at: "0".to_string(), + }) + .await + .expect("insert project"); + TaskRepo::new(&db) + .insert(TaskRecord { + id: "t1".to_string(), + project_id: "p1".to_string(), + title: "t1".to_string(), + description: "实现登录接口".to_string(), + status: "testing".to_string(), + priority: 2, + branch_name: None, + assignee: None, + workflow_def_id: None, + base_branch: None, + review_rounds: 0, + output_json: task_output_json.map(String::from), + idea_id: None, + created_at: "0".to_string(), + updated_at: "0".to_string(), + }) + .await + .expect("insert task"); + (db, "t1".to_string()) + } + + /// 决策 a 步骤②:AiNode.execute 落产出验证 — 有 task_id 时 output_json 写入。 + /// 用 mock provider 避免真调 LLM:自定义 Node 替换不便,改为直接验证 update_field 路径 + /// (execute 内部调 provider.complete 需真 LLM,此处降级验证 parse_review_json 兜底 + + /// 直接 repo.update_field 路径,与 execute 落库逻辑同一条 crud.update_field)。 + #[tokio::test] + async fn update_field_writes_output_json() { + // 验证 output_json 在 tasks 白名单(update_field 不报「不允许的字段名」)。 + // 这正是 AiNode.execute / AiSelfReviewNode.execute 落产出的底层路径。 + let (db, task_id) = setup_task_db(None).await; + let repo = TaskRepo::new(&db); + let payload = r#"{"text":"产出内容","model":"glm-4","usage":{}}"#; + let ok = repo + .update_field(&task_id, "output_json", payload) + .await + .expect("update_field output_json"); + assert!(ok, "update_field 应返回 affected>0=true"); + let after = repo.get_by_id(&task_id).await.unwrap().unwrap(); + assert_eq!( + after.output_json.as_deref(), + Some(payload), + "output_json 应被写入" + ); + // 原始 text 字段(产出对照)保留 + let parsed: serde_json::Value = serde_json::from_str(after.output_json.as_deref().unwrap()).unwrap(); + assert_eq!(parsed["text"], json!("产出内容")); + } + + /// 步骤③:parse_review_json 合规 JSON 直通(保留 verdict/dimensions/summary/suggestions)。 + #[test] + fn parse_review_json_valid_passes_through() { + let raw = r#"{"verdict":"pass","dimensions":{"requirement_fit":{"score":9,"issues":[]}},"summary":"符合需求","suggestions":["可加单测"]}"#; + let v = parse_review_json(raw); + assert_eq!(v["verdict"], json!("pass")); + assert_eq!(v["summary"], json!("符合需求")); + assert_eq!(v["suggestions"], json!(["可加单测"])); + assert_eq!(v["dimensions"]["requirement_fit"]["score"], json!(9)); + } + + /// 步骤③:parse_review_json 非 JSON / 缺 verdict → 兜底 verdict=unknown + summary 含原文。 + #[test] + fn parse_review_json_invalid_falls_back_unknown() { + // 非 JSON + let v = parse_review_json("这不是 JSON,LLM 没按要求输出"); + assert_eq!(v["verdict"], json!("unknown")); + assert!(v["summary"].as_str().unwrap().contains("不是 JSON")); + assert!(v["summary"].as_str().unwrap().contains("解析失败")); + + // JSON 但缺 verdict + let v = parse_review_json(r#"{"foo":"bar"}"#); + assert_eq!(v["verdict"], json!("unknown")); + + // JSON 非 object(数组) + let v = parse_review_json("[1,2,3]"); + assert_eq!(v["verdict"], json!("unknown")); + } + + /// 步骤③:LLM 偶尔包 ```json 代码块,parse_review_json 应剥离后解析成功。 + #[test] + fn parse_review_json_strips_markdown_code_fence() { + let raw = "```json\n{\"verdict\":\"fail\",\"summary\":\"缺边界处理\"}\n```"; + let v = parse_review_json(raw); + assert_eq!(v["verdict"], json!("fail")); + assert_eq!(v["summary"], json!("缺边界处理")); + } + + /// 步骤③:truncate_for_summary 长文截断。 + #[test] + fn truncate_for_summary_long_text() { + let short = "短文"; + assert_eq!(truncate_for_summary(short), "短文"); + let long: String = "字".repeat(500); + let t = truncate_for_summary(&long); + assert!(t.ends_with("…")); + assert!(t.chars().count() <= 302, "截断后含省略号应 ≈300 字"); + } + + /// 步骤③:AiSelfReviewNode.build_review_prompt 含需求 + 产出 + 四维度。 + #[test] + fn build_review_prompt_contains_inputs() { + let p = AiSelfReviewNode::build_review_prompt("登录接口", "已实现 /login"); + assert!(p.contains("登录接口"), "prompt 应含需求"); + assert!(p.contains("已实现 /login"), "prompt 应含产出"); + assert!(p.contains("需求符合度"), "prompt 应含四维度"); + assert!(p.contains("边界处理")); + assert!(p.contains("verdict"), "prompt 应含输出格式约束"); + } + + // ============================================================ + // 阶段3 自审闸门(gate_should_block)单测 + // ============================================================ + // + // gate 决策矩阵: + // gate=false(默认,阶段2 行为) → 任何 verdict 都放行(辅助模式) + // gate=true + verdict=pass → 放行 + // gate=true + verdict=unknown → 放行(LLM 不可靠时不阻断,人定权) + // gate=true + verdict=fail → 阻断(返回 Err,工作流 failed) + + #[test] + fn gate_disabled_never_blocks_any_verdict() { + // gate 默认 false(阶段2 兼容):无论 verdict 如何都不阻断,自审仅辅助展示 + assert!(!gate_should_block(false, "fail"), "gate 关闭时 fail 也不阻断"); + assert!(!gate_should_block(false, "pass"), "gate 关闭时 pass 放行"); + assert!(!gate_should_block(false, "unknown"), "gate 关闭时 unknown 放行"); + } + + #[test] + fn gate_enabled_blocks_only_on_fail() { + // gate 开启:仅 fail 阻断 + assert!(gate_should_block(true, "fail"), "gate 开 + fail 应阻断"); + assert!(!gate_should_block(true, "pass"), "gate 开 + pass 放行"); + assert!( + !gate_should_block(true, "unknown"), + "gate 开 + unknown 放行(LLM 不可靠时保人定权)" + ); + } + + #[test] + fn gate_unknown_verdict_does_not_block() { + // 防御:verdict 非标准值(如解析异常)不应触发阻断(对齐 unknown 保守不阻断语义) + assert!( + !gate_should_block(true, "weird"), + "非标准 verdict 不应阻断(对齐 unknown 保守)" + ); + assert!(!gate_should_block(true, ""), "空 verdict 不应阻断"); + } +} diff --git a/crates/df-nodes/src/lib.rs b/crates/df-nodes/src/lib.rs index f4bb638..c5a687c 100644 --- a/crates/df-nodes/src/lib.rs +++ b/crates/df-nodes/src/lib.rs @@ -1,6 +1,7 @@ //! df-nodes: 内置节点集合 — AI、脚本、人工审批 pub mod ai_node; +pub mod ai_self_review_node; #[allow(dead_code)] mod ai_node_helpers; pub mod human_node; diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index 399c48d..0671a16 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -705,13 +705,13 @@ fn build_registry(db: Arc) -> NodeRegistry { registry.register("ai", move |_config| { Box::new(df_nodes::ai_node::AiNode::new(ai_db.clone())) }); - // AiSelfReviewNode(df_nodes::ai_node, impl Node trait): + // AiSelfReviewNode(df_nodes::ai_self_review_node, impl Node trait): // 决策 a 步骤③:四维度自审(prompt 固定+JSON 解析兜底),写回 output_json 加 review 子字段, // review 摘要塞 NodeOutput.data 供下游 human_review 经 DAG inputs 透传。 // testing 模板 ai_self_review 节点类型对齐此注册 key。 let review_db = db.clone(); registry.register("ai_self_review", move |_config| { - Box::new(df_nodes::ai_node::AiSelfReviewNode::new(review_db.clone())) + Box::new(df_nodes::ai_self_review_node::AiSelfReviewNode::new(review_db.clone())) }); // TaskAdvanceNode(df_nodes::task_advance_node, impl Node trait): // 推进链阶段 2 工作流联动入口 — DAG 内触发 advance_task。