diff --git a/crates/df-nodes/src/ai_agent_node.rs b/crates/df-nodes/src/ai_agent_node.rs new file mode 100644 index 0000000..1cfea77 --- /dev/null +++ b/crates/df-nodes/src/ai_agent_node.rs @@ -0,0 +1,409 @@ +//! AI 编排节点 — 直连 agentic 会话执行工作流任务 +//! +//! 与 AiNode(单次 LLM 补全)的区别:AiAgentNode 在节点 execute 时以编程方式启动一个 +//! 独立 agentic 会话(新 conv_id,与用户活跃会话隔离,防 epoch 互踩),由 run_agentic_loop +//! 完整驱动(多轮迭代 + 工具调用),会话收敛后取最终 assistant 报告作为节点输出。 +//! +//! 阻塞语义与 HumanNode 同构:tokio::select! 三路等待 +//! - 会话完成通知(AppState.agent_completion_bridge 的 oneshot,由 agentic loop 收敛 +//! 统一出口 emit_ai_completed_once 投递,epoch 守卫保证仅 owner loop 触发) +//! - deadline 超时(config.timeout_secs,默认 1800s)→ Err(工作流 failed → 推进链退回) +//! - 500ms tick 轮询 is_cancelled +//! +//! spawn 通道经 [`SpawnFn`] 注入(src-tauri 节点工厂提供,内部取 AppState 起会话), +//! 本 crate 不依赖 src-tauri 类型,依赖方向与 AiNode 持 Arc 同款。 + +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use df_storage::crud::TaskRepo; +use df_storage::db::Database; +use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; + +// 任务文本兜底 prompt(与 AiNode 同款,经 ai_node_helpers 复用) +use crate::ai_node_helpers::{default_fallback_prompt, resolve_fallback_prompt}; + +/// 会话启动结果:(conv_id, 完成通知接收端)。接收端值 = 最终 assistant 报告文本。 +pub type SpawnedSession = (String, tokio::sync::oneshot::Receiver); + +/// 会话启动 future(Err = 启动失败,如 provider 缺失/应用句柄未就绪)。 +pub type SpawnFuture = Pin> + Send>>; + +/// 会话启动通道:入参任务文本,由 src-tauri 节点工厂注入实现(见 chat.rs +/// orchestration_session_spawner)。 +pub type SpawnFn = Arc SpawnFuture + Send + Sync>; + +/// 默认编排超时(秒)。1800s = 30min,与 HumanNode 默认审批窗口同量级; +/// agentic 会话含多轮工具调用,给宽值,到点 fail 退回任务状态。 +const DEFAULT_TIMEOUT_SECS: u64 = 1800; + +/// AI 编排节点 +/// +/// 持 Arc 落 task.output_json(有 task_id 时,与 AiNode 同模式), +/// 持 SpawnFn 启动独立编排会话。 +pub struct AiAgentNode { + db: Arc, + spawn: SpawnFn, +} + +impl AiAgentNode { + /// 构造节点(注册表工厂调用,注入数据库句柄与会话启动通道) + pub fn new(db: Arc, spawn: SpawnFn) -> Self { + Self { db, spawn } + } +} + +#[async_trait] +impl Node for AiAgentNode { + async fn execute(&self, ctx: NodeContext) -> NodeResult { + tracing::info!("AiAgentNode 执行: node_id={}", ctx.node_id); + + // 任务文本取值优先级与 AiNode 一致:上游 prompt 输出 > config.prompt > + // task_id 兜底(读 TaskRecord 生成) > 通用默认。推进链模板 config 无 prompt, + // 兜底链保证单节点 DAG 也能拿到任务书。 + let task = match ctx + .inputs + .get("prompt") + .and_then(|o| o.data.as_str()) + .map(String::from) + .or_else(|| { + ctx.config + .get("prompt") + .and_then(|v| v.as_str()) + .map(String::from) + }) { + Some(t) if !t.trim().is_empty() => t, + _ => { + resolve_fallback_prompt(&self.db, &ctx.config) + .await + .unwrap_or_else(default_fallback_prompt) + } + }; + + let timeout_secs = ctx + .config + .get("timeout_secs") + .and_then(|v| v.as_u64()) + .unwrap_or(DEFAULT_TIMEOUT_SECS); + let task_id = ctx + .config + .get("task_id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + + // 启动独立编排会话(内部生成新 conv_id 并注册完成通知桥,注册先于 spawn) + let (conv_id, mut done_rx) = (self.spawn)(task) + .await + .map_err(|e| anyhow::anyhow!("AI 编排会话启动失败: {}", e))?; + + // select! 骨架与 HumanNode 同构:完成通知 / deadline 超时 / 500ms 取消轮询。 + // oneshot 接收端在 loop 里重 poll,须借 &mut(移动进 select! 会破坏循环重入)。 + let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs); + let mut cancel_tick = tokio::time::interval(Duration::from_millis(500)); + cancel_tick.tick().await; // 丢弃首个立即触发的 tick + + let report = loop { + tokio::select! { + done = &mut done_rx => match done { + Ok(text) => break text, + Err(_) => { + // 发送端被 drop 且未投递:会话条目被覆盖注册或异常清除,无报告可取 + return Err(anyhow::anyhow!( + "AI 编排会话完成信号丢失(conv_id={})", + conv_id + )); + } + }, + _ = tokio::time::sleep_until(deadline) => { + // 超时 fail:工作流 failed → 推进链按 regression_target 退回任务状态 + return Err(anyhow::anyhow!( + "AI 编排会话超时({}s), conv_id={}", + timeout_secs, + conv_id + )); + } + _ = cancel_tick.tick() => { + if ctx.node_status.is_cancelled(&ctx.node_id) { + return Err(anyhow::anyhow!("AI 编排会话被取消")); + } + } + } + }; + + // 空报告判失败:stop 在任何输出前 / 会话被清理的场景,不作为有效产出推进任务 + if report.trim().is_empty() { + return Err(anyhow::anyhow!( + "AI 编排会话无输出(会话被停止或未产出报告), conv_id={}", + conv_id + )); + } + + let output = serde_json::json!({ + "text": report, + "conv_id": conv_id, + "task_id": task_id, + }); + + // 有 task_id 落 task.output_json(与 AiNode 同模式:通用 update_field 白名单字段, + // 落库失败不阻断节点返回,产出已在内存工作流可继续)。 + if !task_id.is_empty() { + if let Ok(json_str) = serde_json::to_string(&output) { + let repo = TaskRepo::new(&self.db); + if let Err(e) = repo.update_field(&task_id, "output_json", &json_str).await { + tracing::warn!( + task_id = task_id, + error = %e, + "AiAgentNode 落 output_json 失败(产出已在内存,节点继续)" + ); + } + } + } + + Ok(NodeOutput::from_value(output)) + } + + fn schema(&self) -> NodeSchema { + NodeSchema { + params: serde_json::json!({ + "type": "object", + "properties": { + "prompt": { "type": "string", "description": "任务书(缺省依次取上游 prompt 输出 / 按 task_id 读任务生成 / 通用默认)" }, + "timeout_secs": { "type": "integer", "description": "编排超时秒数(默认 1800,到点节点 fail 触发工作流退回)" }, + "task_id": { "type": "string", "description": "关联任务 ID(run_workflow 注入;有值时产出落 task.output_json)" } + }, + "required": [] + }), + output: serde_json::json!({ + "type": "object", + "properties": { + "text": { "type": "string", "description": "最终 assistant 报告文本" }, + "conv_id": { "type": "string", "description": "编排会话 ID(独立于用户活跃会话)" }, + "task_id": { "type": "string" } + } + }), + } + } + + fn is_blocking(&self) -> bool { + true + } + + fn node_type(&self) -> &str { + "ai_agent" + } +} + +#[cfg(test)] +mod tests { + use super::*; + use df_storage::crud::ProjectRepo; + use df_storage::models::{ProjectRecord, TaskRecord}; + use df_types::types::{ProjectStatus, TaskStatus}; + use df_workflow::eventbus::EventBus; + use df_workflow::state::StateMachine; + use serde_json::json; + use std::collections::HashMap; + + /// 构造 NodeContext:自定义 config,共享指定状态机(取消轮询测试用)。 + fn make_ctx_with(config: serde_json::Value, sm: StateMachine) -> NodeContext { + NodeContext { + node_id: "node-a".to_string(), + inputs: HashMap::new(), + config, + execution_id: "exec-1".into(), + event_bus: EventBus::new(), + node_status: sm, + } + } + + fn make_ctx(config: serde_json::Value) -> NodeContext { + make_ctx_with(config, StateMachine::new()) + } + + /// 内存 DB(无表数据;task_id 为空时节点不触库)。 + async fn empty_db() -> Arc { + Arc::new(Database::open_in_memory().await.expect("open_in_memory")) + } + + /// 内存 DB helper:插父项目 + 指定 task,返回 (db, task_id)。 + async fn setup_task_db() -> (Arc, 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: ProjectStatus::Planning, + 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: "实现登录接口".to_string(), + description: "含前端表单与后端校验".to_string(), + status: TaskStatus::Todo, + priority: 2, + branch_name: None, + assignee: None, + workflow_def_id: None, + base_branch: None, + review_rounds: 0, + output_json: None, + idea_id: None, + module_id: None, + queue: "todo".to_string(), + parent_id: None, + content_json: None, + created_at: "0".to_string(), + updated_at: "0".to_string(), + }) + .await + .expect("insert task"); + (Arc::new(db), "t1".to_string()) + } + + /// spawn 桩:启动后 50ms 投递指定报告文本。 + fn spawn_delivering(report: &'static str) -> SpawnFn { + Arc::new(move |_task: String| { + Box::pin(async move { + let (tx, rx) = tokio::sync::oneshot::channel::(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(50)).await; + let _ = tx.send(report.to_string()); + }); + Ok(("conv-1".to_string(), rx)) + }) + }) + } + + /// spawn 桩:发送端保活不投递(模拟会话长时间未收敛)。 + fn spawn_silent() -> SpawnFn { + Arc::new(|_task: String| { + Box::pin(async move { + let (tx, rx) = tokio::sync::oneshot::channel::(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_secs(5)).await; + let _ = tx.send("迟到的报告".to_string()); + }); + Ok(("conv-silent".to_string(), rx)) + }) + }) + } + + /// 完成通知桥触发:spawn 后收到投递 → Ok 输出 {text, conv_id, task_id}。 + #[tokio::test] + async fn completion_notify_returns_report() { + let node = AiAgentNode::new(empty_db().await, spawn_delivering("任务完成报告")); + let ctx = make_ctx(json!({ "task_id": "t1" })); + let out = node.execute(ctx).await.unwrap(); + assert_eq!(out.data["text"], json!("任务完成报告")); + assert_eq!(out.data["conv_id"], json!("conv-1")); + assert_eq!(out.data["task_id"], json!("t1")); + } + + /// deadline 超时:发送端保活不投递 + 短超时 → Err 超时(工作流 failed 退回路径)。 + #[tokio::test] + async fn timeout_fails_node() { + let node = AiAgentNode::new(empty_db().await, spawn_silent()); + let ctx = make_ctx(json!({ "timeout_secs": 1 })); + let err = node.execute(ctx).await.unwrap_err().to_string(); + assert!(err.contains("超时"), "应报超时, 实际: {}", err); + assert!(err.contains("conv-silent"), "超时信息应带 conv_id, 实际: {}", err); + } + + /// 取消轮询:500ms tick 命中 is_cancelled → Err 取消(状态保持 Cancelled 由 executor 保证)。 + #[tokio::test] + async fn cancelled_poll_fails_node() { + let sm = StateMachine::new(); + let sm_external = sm.clone(); + let node = AiAgentNode::new(empty_db().await, spawn_silent()); + let handle = tokio::spawn(async move { node.execute(make_ctx_with(json!({}), sm)).await }); + // 等 execute 进入 select! 轮询后再取消(过早 set 会因 tick 未启动而漏检一轮,仍会被下轮命中) + tokio::time::sleep(Duration::from_millis(50)).await; + sm_external.set_cancelled("node-a".to_string()); + let err = handle.await.unwrap().unwrap_err().to_string(); + assert!(err.contains("取消"), "应报取消, 实际: {}", err); + } + + /// 启动失败:spawn 返 Err(provider 缺失等)→ 节点 Err 带原因。 + #[tokio::test] + async fn spawn_failure_fails_node() { + let spawn: SpawnFn = Arc::new(|_task: String| { + Box::pin(async { Err("未配置 AI Provider".to_string()) }) + }); + let node = AiAgentNode::new(empty_db().await, spawn); + let err = node.execute(make_ctx(json!({}))).await.unwrap_err().to_string(); + assert!(err.contains("启动失败"), "应报启动失败, 实际: {}", err); + assert!(err.contains("未配置 AI Provider"), "应携带原因, 实际: {}", err); + } + + /// 信号丢失:发送端被 drop 且未投递 → Err 信号丢失(不误判为成功)。 + #[tokio::test] + async fn dropped_sender_fails_node() { + let spawn: SpawnFn = Arc::new(|_task: String| { + Box::pin(async move { + let (tx, rx) = tokio::sync::oneshot::channel::(); + drop(tx); + Ok(("conv-drop".to_string(), rx)) + }) + }); + let node = AiAgentNode::new(empty_db().await, spawn); + let err = node.execute(make_ctx(json!({}))).await.unwrap_err().to_string(); + assert!(err.contains("信号丢失"), "应报信号丢失, 实际: {}", err); + } + + /// 空报告:投递了纯空白文本 → Err 无输出(不作为有效产出推进任务)。 + #[tokio::test] + async fn blank_report_fails_node() { + let spawn: SpawnFn = Arc::new(|_task: String| { + Box::pin(async move { + let (tx, rx) = tokio::sync::oneshot::channel::(); + let _ = tx.send(" ".to_string()); + Ok(("conv-blank".to_string(), rx)) + }) + }); + let node = AiAgentNode::new(empty_db().await, spawn); + let err = node.execute(make_ctx(json!({}))).await.unwrap_err().to_string(); + assert!(err.contains("无输出"), "应报无输出, 实际: {}", err); + } + + /// 有 task_id 落 task.output_json(与 AiNode 同一条 update_field 白名单路径), + /// 且任务书兜底链生效:config 无 prompt 时按 task_id 读任务生成。 + #[tokio::test] + async fn task_output_json_written_and_fallback_prompt_used() { + let (db, task_id) = setup_task_db().await; + let node = AiAgentNode::new(db.clone(), spawn_delivering("登录接口已完成")); + // config 只有 task_id(推进链模板形态),prompt 走兜底生成 + let ctx = make_ctx(json!({ "task_id": task_id })); + let out = node.execute(ctx).await.unwrap(); + assert_eq!(out.data["text"], json!("登录接口已完成")); + + let after = TaskRepo::new(&db) + .get_by_id("t1") + .await + .unwrap() + .expect("task 存在"); + let saved = after.output_json.expect("output_json 应被写入"); + let parsed: serde_json::Value = serde_json::from_str(&saved).unwrap(); + assert_eq!(parsed["text"], json!("登录接口已完成")); + assert_eq!(parsed["task_id"], json!("t1")); + assert!(parsed["conv_id"].as_str().is_some(), "conv_id 应落 output_json"); + } + + /// 默认编排超时 1800s(30min),与 HumanNode 默认审批窗口同量级。 + #[test] + fn default_timeout_is_1800_secs() { + assert_eq!(DEFAULT_TIMEOUT_SECS, 1800, "默认编排超时应为 1800s(30min)"); + } +} diff --git a/crates/df-nodes/src/ai_node_helpers.rs b/crates/df-nodes/src/ai_node_helpers.rs index c184043..0436df6 100644 --- a/crates/df-nodes/src/ai_node_helpers.rs +++ b/crates/df-nodes/src/ai_node_helpers.rs @@ -108,7 +108,8 @@ pub(crate) async fn resolve_and_parse( /// /// 无 task_id / 查询失败 / task 不存在 → None,调用方降级到 `default_fallback_prompt`。 /// description 空只取 title;非空拼「标题(描述)」。truncate 防 token 滥用。 -async fn resolve_fallback_prompt(db: &Arc, config: &serde_json::Value) -> Option { +/// pub(crate):AiAgentNode 复用同款任务书兜底(与 AiNode 取值语义一致)。 +pub(crate) async fn resolve_fallback_prompt(db: &Arc, config: &serde_json::Value) -> Option { let task_id = config.get("task_id").and_then(|v| v.as_str())?; if task_id.is_empty() { return None; @@ -130,7 +131,8 @@ async fn resolve_fallback_prompt(db: &Arc, config: &serde_json::Value) } /// P0-B 通用兜底 prompt:无 task_id 时用(独立 AiNode 调用 / config 无 task 关联)。 -fn default_fallback_prompt() -> String { +/// pub(crate):AiAgentNode 复用同款通用兜底。 +pub(crate) fn default_fallback_prompt() -> String { "请根据上下文完成当前任务,产出可执行结果并自检。若需更多信息请说明。".to_string() } diff --git a/crates/df-nodes/src/lib.rs b/crates/df-nodes/src/lib.rs index 50511b5..cd442e9 100644 --- a/crates/df-nodes/src/lib.rs +++ b/crates/df-nodes/src/lib.rs @@ -1,5 +1,6 @@ //! df-nodes: 内置节点集合 — AI、脚本、人工审批、Git、Docker、HTTP、通知 +pub mod ai_agent_node; pub mod ai_node; pub mod ai_self_review_node; mod ai_node_helpers; diff --git a/crates/df-nodes/src/task_workflow_templates.rs b/crates/df-nodes/src/task_workflow_templates.rs index 3bc510e..4530e53 100644 --- a/crates/df-nodes/src/task_workflow_templates.rs +++ b/crates/df-nodes/src/task_workflow_templates.rs @@ -37,14 +37,17 @@ pub fn template_for(target_status: &str) -> Option { /// todo → in_progress:单 AiNode 执行任务。 /// /// 拓扑:1 个 ai 节点。完成回调(②-3)推进 status 到 in_progress。 +/// +/// mode:"agent" → ai 节点工厂构造 AiAgentNode(直连 agentic 会话编排执行,多轮工具 +/// 调用后取最终报告);缺省/"ask" 走旧 AiNode(单次补全,旧调用方零回归)。 +/// 其余运行时参数(prompt/provider/task_id)仍由 run_workflow 全局 config 注入 +/// (④-1 deep_merge 合并),模板只描述拓扑 + 执行模式。 fn in_progress_template() -> DagDef { let mut dag = DagDef::new(); - // config 留空:prompt / provider / task_id 均由 run_workflow 全局 config 注入 - // (④-1 deep_merge 合并),模板只描述拓扑。 dag.add_node( "ai_execute", "ai", - serde_json::json!({}), + serde_json::json!({ "mode": "agent" }), ); dag } @@ -120,10 +123,24 @@ mod tests { assert!(dag.edges.is_empty(), "in_progress: 无边"); let node = dag.nodes.get("ai_execute").expect("ai_execute 节点存在"); assert_eq!(node.node_type, "ai"); - // config 留空(运行时注入) + // mode:"agent" → 工作流编排执行(AiAgentNode 直连 agentic 会话); + // 其余运行时参数仍留空(由 run_workflow 全局 config 注入)。 + assert_eq!( + node.config.get("mode").and_then(|v| v.as_str()), + Some("agent"), + "ai_execute 应以 mode=agent 走编排节点" + ); + let non_mode_keys: Vec<&String> = node + .config + .as_object() + .expect("config 应为对象") + .keys() + .filter(|k| *k != "mode") + .collect(); assert!( - node.config.as_object().map(|o| o.is_empty()).unwrap_or(true), - "ai_execute config 应为空对象" + non_mode_keys.is_empty(), + "ai_execute 除 mode 外 config 应留空(运行时注入),实际: {:?}", + non_mode_keys ); } diff --git a/src-tauri/src/commands/ai/tools/spawn_agent.rs b/src-tauri/src/commands/ai/tools/spawn_agent.rs index b1e5cfb..ef44493 100644 --- a/src-tauri/src/commands/ai/tools/spawn_agent.rs +++ b/src-tauri/src/commands/ai/tools/spawn_agent.rs @@ -79,7 +79,8 @@ pub(crate) fn init_app_handle(app: tauri::AppHandle) { } /// handler 读取 AppHandle(未初始化返 None,handler 报明确错误)。 -fn app_handle() -> Option { +/// pub(crate):编排会话启动通道(chat.rs orchestration_session_spawner)同款复用。 +pub(crate) fn app_handle() -> Option { GLOBAL_APP.get().cloned() } diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index eb7d3d1..993b107 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -201,6 +201,14 @@ pub struct AppState { /// guard/reset/ai_is_generating 全部零锁操作。PerConvState.conv_state 字段已删。 /// 治卡死连环(AiCompleted 延迟 / 工具后中断 / 第二条进队列同源根因)。 pub conv_states: Arc, + /// 编排会话完成通知桥(conv_id → 完成信号 oneshot 发送端,值=最终 assistant 报告文本)。 + /// + /// 工作流 AiAgentNode 经编排会话启动通道(chat.rs orchestration_session_spawner) + /// 在 spawn 前注册等待;agentic loop 收敛统一出口(agentic/mod.rs emit_ai_completed_once, + /// epoch 守卫已由调用方保证,仅 owner loop 到达)投递并移除条目。 + /// std Mutex:临界区为纯内存 insert/remove,无 await,不持锁跨等待点。 + pub agent_completion_bridge: + Arc>>>, // ── 知识库 ── /// 知识库 Repo pub knowledge: KnowledgeRepo, @@ -396,6 +404,8 @@ impl AppState { // B-Phase1: ConvStateStore 无锁存储(空构造;双轨期无写点,B-Phase2 写侧切入)。 // 复用上方提前构建的 Arc(spawn_agent ctx 注入同一实例)。 conv_states, + // 编排会话完成通知桥(空表起步,AiAgentNode 会话启动时注册,loop 收敛出口投递)。 + agent_completion_bridge: Arc::new(std::sync::Mutex::new(HashMap::new())), // Phase3 Layer1:tunnel 客户端未连接实例,lib.rs setup 内 connect tunnel: std::sync::Arc::new(df_tunnel::WsTunnelClient::new()), knowledge: KnowledgeRepo::new(&db), @@ -685,6 +695,46 @@ impl AppState { pub async fn clear_once_allowed_dirs(&self) { self.allowed_dirs.write().await.once.clear(); } + + /// 注册编排会话完成等待(工作流 AiAgentNode 会话 spawn 前调用)。 + /// + /// 同 conv 重复注册覆盖旧条目(旧 oneshot 发送端随值 drop,等待方收 Err 判定信号丢失)。 + /// 锁中毒降级 into_inner 取数据继续(桥内只有内存条目,无一致性风险)。 + pub fn register_agent_completion(&self, conv_id: String, tx: tokio::sync::oneshot::Sender) { + let mut map = self + .agent_completion_bridge + .lock() + .unwrap_or_else(|p| p.into_inner()); + map.insert(conv_id, tx); + } + + /// 是否存在挂起的编排会话等待方(收敛出口短路判据,普通用户会话零后续开销)。 + pub fn has_agent_completion_waiter(&self, conv_id: &str) -> bool { + let map = self + .agent_completion_bridge + .lock() + .unwrap_or_else(|p| p.into_inner()); + map.contains_key(conv_id) + } + + /// 投递编排会话完成通知并移除条目(无等待方 no-op)。 + /// + /// 条目无论投递成败都移除:接收端已 drop(节点超时/取消)时 send 失败静默,不残留。 + /// 返回是否找到等待方(供调用方日志)。 + pub fn notify_agent_completion(&self, conv_id: &str, report: String) -> bool { + let taken = self + .agent_completion_bridge + .lock() + .unwrap_or_else(|p| p.into_inner()) + .remove(conv_id); + match taken { + Some(tx) => { + let _ = tx.send(report); + true + } + None => false, + } + } } /// 构建节点注册表 — 注册内置节点 @@ -706,9 +756,31 @@ fn build_registry(db: Arc) -> NodeRegistry { // AiNode(df_nodes::ai_node, impl Node trait): // 决策 a(AiNode 自审闭环)步骤②:AiNode 持 Arc,execute 完成后若有 task_id // 则把产出落 task.output_json。工厂闭包 move 捕获 db 句柄(Arc clone 廉价)。 + // mode 开关(工作流 AiAgentNode 直连 agentic 会话):config.mode == "agent" → 构造 + // AiAgentNode(编排会话),缺省/"ask" 走旧 AiNode(单次补全,旧调用方零回归)。 + // 工厂期 AppHandle 经进程级 OnceLock 读取(spawn_agent 同款,lib.rs setup 已注入, + // 工厂调用点在 run_workflow → build_dag,时序晚于 setup,句柄必已就绪)。 let ai_db = db.clone(); - registry.register("ai", move |_config| { - Box::new(df_nodes::ai_node::AiNode::new(ai_db.clone())) + registry.register("ai", move |config| { + if config.get("mode").and_then(|v| v.as_str()) == Some("agent") { + Box::new(df_nodes::ai_agent_node::AiAgentNode::new( + ai_db.clone(), + crate::commands::ai::commands::chat::orchestration_session_spawner(), + )) + } else { + Box::new(df_nodes::ai_node::AiNode::new(ai_db.clone())) + } + }); + // AiAgentNode(df_nodes::ai_agent_node, impl Node trait): + // 独立注册类型:节点 execute 时以编程方式起独立 agentic 会话跑任务,收敛后取最终 + // assistant 报告回写。spawn 通道经 orchestration_session_spawner 注入(内部经 + // AppHandle OnceLock 取 AppState),db 句柄供 task_id 落 output_json(ai_db 同款)。 + let agent_db = db.clone(); + registry.register("ai_agent", move |_config| { + Box::new(df_nodes::ai_agent_node::AiAgentNode::new( + agent_db.clone(), + crate::commands::ai::commands::chat::orchestration_session_spawner(), + )) }); // AiSelfReviewNode(df_nodes::ai_self_review_node, impl Node trait): // 决策 a 步骤③:四维度自审(prompt 固定+JSON 解析兜底),写回 output_json 加 review 子字段,