新增: 任务工作流 AiAgentNode 直连会话编排

- ai 节点 mode agent 开关派发,默认旧路径零回归 - 完成通知桥注册触发,收敛出口单点投递报告 - 节点超时取消兜底,结果落任务产出 - 会话编排走独立新会话,不劫持活跃面板
2026-10-01 19:35:22 +08:00
parent 7dcbdb1feb
commit bf0db10ef7
6 changed files with 513 additions and 11 deletions
@@ -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<Database> 同款。
+
+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<String>);
+
+/// 会话启动 future(Err = 启动失败,如 provider 缺失/应用句柄未就绪)。
+pub type SpawnFuture = Pin<Box<dyn Future<Output = Result<SpawnedSession, String>> + Send>>;
+
+/// 会话启动通道:入参任务文本,由 src-tauri 节点工厂注入实现(见 chat.rs
+/// orchestration_session_spawner)。
+pub type SpawnFn = Arc<dyn Fn(String) -> SpawnFuture + Send + Sync>;
+
+/// 默认编排超时(秒)。1800s = 30min,与 HumanNode 默认审批窗口同量级;
+/// agentic 会话含多轮工具调用,给宽值,到点 fail 退回任务状态。
+const DEFAULT_TIMEOUT_SECS: u64 = 1800;
+
+/// AI 编排节点
+///
+/// 持 Arc<Database> 落 task.output_json(有 task_id 时,与 AiNode 同模式),
+/// 持 SpawnFn 启动独立编排会话。
+pub struct AiAgentNode {
+    db: Arc<Database>,
+    spawn: SpawnFn,
+}
+
+impl AiAgentNode {
+    /// 构造节点(注册表工厂调用,注入数据库句柄与会话启动通道)
+    pub fn new(db: Arc<Database>, 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<Database> {
+        Arc::new(Database::open_in_memory().await.expect("open_in_memory"))
+    }
+
+    /// 内存 DB helper:插父项目 + 指定 task,返回 (db, task_id)。
+    async fn setup_task_db() -> (Arc<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: 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::<String>();
+                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::<String>();
+                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::<String>();
+                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::<String>();
+                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)");
+    }
+}