diff --git a/Cargo.lock b/Cargo.lock index de46f22..503fd9b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -881,7 +881,7 @@ dependencies = [ "df-types", "eventsource-stream", "futures", - "rand", + "rand 0.8.6", "reqwest 0.12.28", "serde", "serde_json", @@ -955,6 +955,7 @@ dependencies = [ "df-storage", "df-types", "df-workflow", + "reqwest 0.12.28", "serde", "serde_json", "tokio", @@ -1042,6 +1043,7 @@ dependencies = [ "futures", "serde", "serde_json", + "serde_yaml", "tokio", "tracing", ] @@ -1643,8 +1645,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -1654,9 +1658,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 5.3.0", "wasip2", + "wasm-bindgen", ] [[package]] @@ -1992,6 +1998,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", + "webpki-roots", ] [[package]] @@ -2493,6 +2500,12 @@ version = "0.4.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "markup5ever" version = "0.38.0" @@ -3347,6 +3360,61 @@ dependencies = [ "memchr", ] +[[package]] +name = "quinn" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror 2.0.18", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fcb935c5bec503c2f0e306bdd3e58bb9029dcb14fa8d9ac76e3a5256ac0763e" +dependencies = [ + "bytes", + "getrandom 0.3.4", + "lru-slab", + "rand 0.9.4", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror 2.0.18", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.60.2", +] + [[package]] name = "quote" version = "1.0.45" @@ -3375,8 +3443,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" dependencies = [ "libc", - "rand_chacha", - "rand_core", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", ] [[package]] @@ -3386,7 +3464,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", ] [[package]] @@ -3398,6 +3486,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "raw-window-handle" version = "0.6.2" @@ -3498,6 +3595,8 @@ dependencies = [ "native-tls", "percent-encoding", "pin-project-lite", + "quinn", + "rustls", "rustls-pki-types", "serde", "serde_json", @@ -3505,6 +3604,7 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-native-tls", + "tokio-rustls", "tokio-util", "tower", "tower-http", @@ -3514,6 +3614,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams 0.4.2", "web-sys", + "webpki-roots", ] [[package]] @@ -3637,6 +3738,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" dependencies = [ "once_cell", + "ring", "rustls-pki-types", "rustls-webpki", "subtle", @@ -3649,6 +3751,7 @@ version = "1.14.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" dependencies = [ + "web-time", "zeroize", ] @@ -3760,7 +3863,7 @@ dependencies = [ "generic-array", "num", "once_cell", - "rand", + "rand 0.8.6", "serde", "zbus 4.4.0", ] @@ -3981,6 +4084,19 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "serde_yaml" +version = "0.9.34+deprecated" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" +dependencies = [ + "indexmap 2.14.0", + "itoa", + "ryu", + "serde", + "unsafe-libyaml", +] + [[package]] name = "serialize-to-javascript" version = "0.1.2" @@ -5234,7 +5350,7 @@ dependencies = [ "httparse", "log", "native-tls", - "rand", + "rand 0.8.6", "sha1", "thiserror 1.0.69", "utf-8", @@ -5252,7 +5368,7 @@ dependencies = [ "http", "httparse", "log", - "rand", + "rand 0.8.6", "sha1", "thiserror 1.0.69", "utf-8", @@ -5340,6 +5456,12 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "unsafe-libyaml" +version = "0.2.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" + [[package]] name = "untrusted" version = "0.9.0" @@ -5607,6 +5729,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "web_atoms" version = "0.2.4" @@ -5663,6 +5795,15 @@ dependencies = [ "system-deps", ] +[[package]] +name = "webpki-roots" +version = "1.0.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf85cb06032201fa7c6f829d7db5a7e5aa45bcc0655327713065f6f0576731bf" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "webview2-com" version = "0.38.2" @@ -6400,7 +6541,7 @@ dependencies = [ "hex", "nix", "ordered-stream", - "rand", + "rand 0.8.6", "serde", "serde_repr", "sha1", diff --git a/crates/df-nodes/Cargo.toml b/crates/df-nodes/Cargo.toml index 94962cf..6518522 100644 --- a/crates/df-nodes/Cargo.toml +++ b/crates/df-nodes/Cargo.toml @@ -15,3 +15,4 @@ tokio = { workspace = true } async-trait = { workspace = true } anyhow = { workspace = true } tracing = { workspace = true } +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } diff --git a/crates/df-nodes/src/git_node.rs b/crates/df-nodes/src/git_node.rs new file mode 100644 index 0000000..83e6b10 --- /dev/null +++ b/crates/df-nodes/src/git_node.rs @@ -0,0 +1,426 @@ +//! Git 节点 — 执行 git CLI 命令(branch/checkout/commit/merge/push/status/log) +//! +//! 通过 df_execute::shell::execute 调用本地 git CLI,working_dir 指定仓库路径。 +//! 不内嵌 git2/libgit2:CLI 路径更通用、调试透明,与 ScriptNode 一致。 + +use async_trait::async_trait; +use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; + +/// Git 节点 +pub struct GitNode; + +/// 支持的 Git 动作 +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum GitAction { + Branch, + Checkout, + Commit, + Merge, + Push, + Status, + Log, +} + +impl GitAction { + /// 从 config.action 字符串解析动作,非法值返回 Err。 + pub fn parse(raw: &str) -> anyhow::Result { + match raw { + "branch" => Ok(GitAction::Branch), + "checkout" => Ok(GitAction::Checkout), + "commit" => Ok(GitAction::Commit), + "merge" => Ok(GitAction::Merge), + "push" => Ok(GitAction::Push), + "status" => Ok(GitAction::Status), + "log" => Ok(GitAction::Log), + other => anyhow::bail!( + "GitNode 非法 action: {}(合法值: branch|checkout|commit|merge|push|status|log)", + other + ), + } + } + + fn as_str(&self) -> &'static str { + match self { + GitAction::Branch => "branch", + GitAction::Checkout => "checkout", + GitAction::Commit => "commit", + GitAction::Merge => "merge", + GitAction::Push => "push", + GitAction::Status => "status", + GitAction::Log => "log", + } + } +} + +/// 从 NodeContext.config 解析出的 GitNode 参数。 +/// 抽离此结构便于单元测试 config 解析逻辑(无需构造完整 NodeContext / 起 shell)。 +#[derive(Debug, Clone)] +pub struct GitParams { + pub action: GitAction, + /// 目标分支(branch/checkout/merge) + pub branch_name: Option, + /// commit 信息(commit) + pub message: Option, + /// 仓库路径(None=当前目录) + pub working_dir: Option, +} + +/// 从 config JSON 解析 GitParams。action 必填;其余按需。 +pub fn parse_params(config: &serde_json::Value) -> anyhow::Result { + let action = config + .get("action") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("GitNode 缺少必填参数: action"))?; + let action = GitAction::parse(action)?; + + let branch_name = config + .get("branch_name") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + let message = config + .get("message") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + let working_dir = config + .get("working_dir") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + Ok(GitParams { + action, + branch_name, + message, + working_dir, + }) +} + +/// 根据 action + 参数构建 git CLI 命令字符串。working_dir 由 ShellRequest 处理,此处不含 cd。 +fn build_command(params: &GitParams) -> anyhow::Result { + match params.action { + GitAction::Branch => { + // 无 branch_name → 列出所有本地分支;有 → 创建新分支 + match ¶ms.branch_name { + Some(name) => Ok(format!("git branch {}", shell_quote(name))), + None => Ok("git branch".to_string()), + } + } + GitAction::Checkout => { + let name = params + .branch_name + .as_deref() + .ok_or_else(|| anyhow::anyhow!("GitNode action=checkout 缺少 branch_name"))?; + Ok(format!("git checkout {}", shell_quote(name))) + } + GitAction::Commit => { + let msg = params + .message + .as_deref() + .ok_or_else(|| anyhow::anyhow!("GitNode action=commit 缺少 message"))?; + Ok(format!("git commit -m {}", shell_quote(msg))) + } + GitAction::Merge => { + let name = params + .branch_name + .as_deref() + .ok_or_else(|| anyhow::anyhow!("GitNode action=merge 缺少 branch_name"))?; + Ok(format!("git merge {}", shell_quote(name))) + } + GitAction::Push => Ok("git push".to_string()), + GitAction::Status => Ok("git status".to_string()), + GitAction::Log => Ok("git log --oneline -20".to_string()), + } +} + +/// 简单 shell 引号包裹:含空格/特殊字符时用双引号包裹并转义内嵌双引号。 +/// 仅用于内部参数拼接(branch_name/message),命令名固定白名单不用户可控。 +fn shell_quote(s: &str) -> String { + if s.chars().any(|c| c.is_whitespace() || c == '"' || c == '$' || c == '`') { + format!("\"{}\"", s.replace('"', "\\\"")) + } else { + s.to_string() + } +} + +#[async_trait] +impl Node for GitNode { + async fn execute(&self, ctx: NodeContext) -> NodeResult { + tracing::info!("GitNode 执行: node_id={}", ctx.node_id); + + let params = parse_params(&ctx.config)?; + let command = build_command(¶ms)?; + + tracing::info!( + action = params.action.as_str(), + working_dir = ?params.working_dir, + "GitNode 构建命令: {}", + command + ); + + let request = df_execute::shell::ShellRequest { + command, + working_dir: params.working_dir.clone(), + env: std::collections::HashMap::new(), + timeout_secs: Some(60), + shell_type: Default::default(), + }; + + let result = df_execute::shell::execute(request).await?; + + let exit_code = result.exit_code.unwrap_or(-1); + if exit_code != 0 { + anyhow::bail!( + "GitNode 执行失败 (exit_code={}): {}", + exit_code, + result.stderr.trim() + ); + } + + // 按 action 提取结构化字段(branch 列表/commit hash 等),无法解析时回退原始 stdout。 + let parsed = parse_action_output(¶ms.action, &result.stdout); + + tracing::info!( + action = params.action.as_str(), + exit_code, + duration_ms = result.duration_ms, + "GitNode 完成" + ); + + Ok(NodeOutput::from_value(serde_json::json!({ + "action": params.action.as_str(), + "stdout": result.stdout, + "stderr": result.stderr, + "exit_code": exit_code, + "duration_ms": result.duration_ms, + "parsed": parsed, + }))) + } + + fn schema(&self) -> NodeSchema { + NodeSchema { + params: serde_json::json!({ + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["branch", "checkout", "commit", "merge", "push", "status", "log"] + }, + "branch_name": { "type": "string", "description": "目标分支(branch/checkout/merge)" }, + "message": { "type": "string", "description": "commit 信息(commit)" }, + "working_dir": { "type": "string", "description": "git 仓库路径(可选)" } + }, + "required": ["action"] + }), + output: serde_json::json!({ + "type": "object", + "properties": { + "action": { "type": "string" }, + "stdout": { "type": "string" }, + "stderr": { "type": "string" }, + "exit_code": { "type": "integer" }, + "duration_ms": { "type": "integer" }, + "parsed": { "type": "object" } + } + }), + } + } + + fn node_type(&self) -> &str { + "git" + } +} + +/// 按 action 解析 stdout 成结构化字段。 +/// - branch(无参列出):每行一个分支,`*` 标记当前分支 +/// - log:每行一条 `hash subject` +/// - commit:首段 hash +/// - 其余动作无强结构,返回空对象 +fn parse_action_output(action: &GitAction, stdout: &str) -> serde_json::Value { + match action { + GitAction::Branch => { + let branches: Vec<&str> = stdout.lines().map(|l| l.trim_start_matches("* ").trim()).collect(); + let current = stdout + .lines() + .find(|l| l.starts_with('*')) + .map(|l| l.trim_start_matches("* ").trim().to_string()); + serde_json::json!({ "branches": branches, "current": current }) + } + GitAction::Log => { + let entries: Vec = stdout + .lines() + .filter_map(|l| { + let l = l.trim(); + let mut parts = l.splitn(2, ' '); + let hash = parts.next()?.to_string(); + let subject = parts.next().unwrap_or("").to_string(); + Some(serde_json::json!({ "hash": hash, "subject": subject })) + }) + .collect(); + serde_json::json!({ "entries": entries }) + } + GitAction::Commit => { + // `git commit` 默认输出含 `[branch hash]` 形式;取首个 7+ 位十六进制段 + let hash = stdout + .split_whitespace() + .find(|t| t.len() >= 7 && t.chars().all(|c| c.is_ascii_hexdigit())) + .unwrap_or("") + .to_string(); + serde_json::json!({ "commit_hash": hash }) + } + _ => serde_json::json!({}), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + // ── GitAction::parse ── + + #[test] + fn parse_all_valid_actions() { + for raw in ["branch", "checkout", "commit", "merge", "push", "status", "log"] { + let a = GitAction::parse(raw).unwrap_or_else(|e| panic!("合法 action {} 应解析成功: {}", raw, e)); + assert_eq!(a.as_str(), raw); + } + } + + #[test] + fn parse_invalid_action_errors() { + let err = GitAction::parse("rebase").unwrap_err().to_string(); + assert!(err.contains("非法 action"), "实际: {}", err); + assert!(err.contains("rebase")); + } + + #[test] + fn parse_empty_action_errors() { + let err = GitAction::parse("").unwrap_err().to_string(); + assert!(err.contains("非法 action"), "实际: {}", err); + } + + // ── parse_params(config) ── + + #[test] + fn params_missing_action_errors() { + let err = parse_params(&json!({ "branch_name": "main" })) + .unwrap_err() + .to_string(); + assert!(err.contains("action"), "实际: {}", err); + } + + #[test] + fn params_minimal_valid() { + let p = parse_params(&json!({ "action": "status" })).unwrap(); + assert_eq!(p.action, GitAction::Status); + assert_eq!(p.branch_name, None); + assert_eq!(p.message, None); + assert_eq!(p.working_dir, None); + } + + #[test] + fn params_full_fields() { + let p = parse_params(&json!({ + "action": "commit", + "branch_name": "feature/x", + "message": "fix: 修复", + "working_dir": "/repo" + })) + .unwrap(); + assert_eq!(p.action, GitAction::Commit); + assert_eq!(p.branch_name.as_deref(), Some("feature/x")); + assert_eq!(p.message.as_deref(), Some("fix: 修复")); + assert_eq!(p.working_dir.as_deref(), Some("/repo")); + } + + // ── build_command ── + + #[test] + fn command_branch_list_when_no_name() { + let p = parse_params(&json!({ "action": "branch" })).unwrap(); + assert_eq!(build_command(&p).unwrap(), "git branch"); + } + + #[test] + fn command_branch_create_when_name() { + let p = parse_params(&json!({ "action": "branch", "branch_name": "feat" })).unwrap(); + assert_eq!(build_command(&p).unwrap(), "git branch feat"); + } + + #[test] + fn command_checkout_requires_name() { + let p = parse_params(&json!({ "action": "checkout" })).unwrap(); + let err = build_command(&p).unwrap_err().to_string(); + assert!(err.contains("branch_name"), "实际: {}", err); + } + + #[test] + fn command_commit_requires_message() { + let p = parse_params(&json!({ "action": "commit" })).unwrap(); + let err = build_command(&p).unwrap_err().to_string(); + assert!(err.contains("message"), "实际: {}", err); + } + + #[test] + fn command_commit_quotes_message_with_space() { + let p = parse_params(&json!({ "action": "commit", "message": "fix bug a" })).unwrap(); + let cmd = build_command(&p).unwrap(); + assert_eq!(cmd, "git commit -m \"fix bug a\""); + } + + #[test] + fn command_merge_requires_name() { + let p = parse_params(&json!({ "action": "merge" })).unwrap(); + assert!(build_command(&p).is_err()); + } + + #[test] + fn command_push_status_log_fixed() { + for (action, expected) in [ + ("push", "git push"), + ("status", "git status"), + ("log", "git log --oneline -20"), + ] { + let p = parse_params(&json!({ "action": action })).unwrap(); + assert_eq!(build_command(&p).unwrap(), expected); + } + } + + // ── shell_quote ── + + #[test] + fn shell_quote_plain_passthrough() { + assert_eq!(shell_quote("feat"), "feat"); + } + + #[test] + fn shell_quote_escapes_embedded_double_quote() { + let q = shell_quote("a\"b"); + assert_eq!(q, "\"a\\\"b\""); + } + + // ── parse_action_output ── + + #[test] + fn parse_branch_output() { + let out = parse_action_output( + &GitAction::Branch, + "* main\n develop\n feature/x\n", + ); + assert_eq!(out["current"], "main"); + assert_eq!(out["branches"].as_array().unwrap().len(), 3); + } + + #[test] + fn parse_log_output() { + let out = parse_action_output( + &GitAction::Log, + "abc1234 fix bug\n deadbeef add feature\n", + ); + let entries = out["entries"].as_array().unwrap(); + assert_eq!(entries.len(), 2); + assert_eq!(entries[0]["hash"], "abc1234"); + assert_eq!(entries[1]["subject"], "add feature"); + } +} diff --git a/crates/df-nodes/src/http_node.rs b/crates/df-nodes/src/http_node.rs new file mode 100644 index 0000000..af8758e --- /dev/null +++ b/crates/df-nodes/src/http_node.rs @@ -0,0 +1,319 @@ +//! HTTP 节点 — 发起 HTTP 请求(GET/POST/PUT/DELETE) +//! +//! 使用 reqwest(与 df-ai 同一版本 0.12)。config 提供 method/url/headers/body/timeout_secs。 +//! 输出 status_code + body + (可选)响应头子集。 + +use std::time::Duration; + +use async_trait::async_trait; +use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; + +/// HTTP 节点 +pub struct HttpNode; + +/// 支持的 HTTP 方法 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HttpMethod { + Get, + Post, + Put, + Delete, +} + +impl HttpMethod { + /// 从 config.method 字符串解析,大小写不敏感,非法值返回 Err。缺省返回 GET。 + pub fn parse(raw: Option<&str>) -> anyhow::Result { + match raw.map(|s| s.to_ascii_uppercase()).as_deref() { + None | Some("GET") => Ok(HttpMethod::Get), + Some("POST") => Ok(HttpMethod::Post), + Some("PUT") => Ok(HttpMethod::Put), + Some("DELETE") => Ok(HttpMethod::Delete), + Some(other) => anyhow::bail!( + "HttpNode 非法 method: {}(合法值: GET|POST|PUT|DELETE)", + other + ), + } + } + + fn as_str(&self) -> &'static str { + match self { + HttpMethod::Get => "GET", + HttpMethod::Post => "POST", + HttpMethod::Put => "PUT", + HttpMethod::Delete => "DELETE", + } + } +} + +/// 从 NodeContext.config 解析出的 HttpNode 参数。 +/// 抽离此结构便于单元测试 config 解析逻辑(无需发起真实网络请求)。 +#[derive(Debug, Clone)] +pub struct HttpParams { + pub method: HttpMethod, + pub url: String, + /// 请求头(JSON 对象 → Vec<(name, value)>) + pub headers: Vec<(String, String)>, + /// 请求体(可选) + pub body: Option, + /// 超时秒数(缺省 30) + pub timeout_secs: u64, +} + +/// 默认超时 +const DEFAULT_TIMEOUT_SECS: u64 = 30; + +/// 从 config JSON 解析 HttpParams。url 必填;method 缺省 GET;timeout 缺省 30。 +pub fn parse_params(config: &serde_json::Value) -> anyhow::Result { + let method = HttpMethod::parse(config.get("method").and_then(|v| v.as_str()))?; + + let url = config + .get("url") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("HttpNode 缺少必填参数: url"))?; + if url.trim().is_empty() { + anyhow::bail!("HttpNode url 不能为空"); + } + + // headers:JSON 对象 → 有序 (name, value) 列表。非对象/非字符串值忽略并 warn。 + let mut headers = Vec::new(); + if let Some(obj) = config.get("headers").and_then(|v| v.as_object()) { + for (k, v) in obj { + match v.as_str() { + Some(s) => headers.push((k.clone(), s.to_string())), + None => { + tracing::warn!( + header = %k, + "HttpNode headers 中 header 值非字符串,忽略" + ); + } + } + } + } + + let body = config + .get("body") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + let timeout_secs = config + .get("timeout_secs") + .and_then(|v| v.as_u64()) + .unwrap_or(DEFAULT_TIMEOUT_SECS); + + Ok(HttpParams { + method, + url: url.to_string(), + headers, + body, + timeout_secs, + }) +} + +#[async_trait] +impl Node for HttpNode { + async fn execute(&self, ctx: NodeContext) -> NodeResult { + tracing::info!("HttpNode 执行: node_id={}", ctx.node_id); + + let params = parse_params(&ctx.config)?; + + tracing::info!( + method = params.method.as_str(), + url = %params.url, + timeout_secs = params.timeout_secs, + "HttpNode 发起请求" + ); + + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(params.timeout_secs)) + .build()?; + + let mut req = match params.method { + HttpMethod::Get => client.get(¶ms.url), + HttpMethod::Post => client.post(¶ms.url), + HttpMethod::Put => client.put(¶ms.url), + HttpMethod::Delete => client.delete(¶ms.url), + }; + + for (name, value) in ¶ms.headers { + req = req.header(name, value); + } + + if let Some(body) = ¶ms.body { + req = req.body(body.clone()); + } + + let response = req.send().await?; + let status_code = response.status().as_u16(); + + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()); + + let body_text = response.text().await?; + + tracing::info!( + method = params.method.as_str(), + url = %params.url, + status_code, + body_len = body_text.len(), + "HttpNode 完成" + ); + + Ok(NodeOutput::from_value(serde_json::json!({ + "method": params.method.as_str(), + "url": params.url, + "status_code": status_code, + "body": body_text, + "content_type": content_type, + }))) + } + + fn schema(&self) -> NodeSchema { + NodeSchema { + params: serde_json::json!({ + "type": "object", + "properties": { + "method": { + "type": "string", + "enum": ["GET", "POST", "PUT", "DELETE"], + "default": "GET" + }, + "url": { "type": "string", "description": "请求 URL" }, + "headers": { + "type": "object", + "description": "请求头 JSON 对象", + "additionalProperties": { "type": "string" } + }, + "body": { "type": "string", "description": "请求体(可选)" }, + "timeout_secs": { "type": "integer", "default": 30 } + }, + "required": ["url"] + }), + output: serde_json::json!({ + "type": "object", + "properties": { + "method": { "type": "string" }, + "url": { "type": "string" }, + "status_code": { "type": "integer" }, + "body": { "type": "string" }, + "content_type": { "type": "string" } + } + }), + } + } + + fn node_type(&self) -> &str { + "http" + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + // ── HttpMethod::parse ── + + #[test] + fn method_defaults_to_get_when_absent() { + assert_eq!(HttpMethod::parse(None).unwrap(), HttpMethod::Get); + } + + #[test] + fn method_case_insensitive() { + assert_eq!(HttpMethod::parse(Some("get")).unwrap(), HttpMethod::Get); + assert_eq!(HttpMethod::parse(Some("Post")).unwrap(), HttpMethod::Post); + assert_eq!(HttpMethod::parse(Some("DELETE")).unwrap(), HttpMethod::Delete); + assert_eq!(HttpMethod::parse(Some("pUt")).unwrap(), HttpMethod::Put); + } + + #[test] + fn method_invalid_errors() { + let err = HttpMethod::parse(Some("PATCH")).unwrap_err().to_string(); + assert!(err.contains("非法 method"), "实际: {}", err); + assert!(err.contains("PATCH")); + } + + // ── parse_params(config) ── + + #[test] + fn params_missing_url_errors() { + let err = parse_params(&json!({ "method": "GET" })) + .unwrap_err() + .to_string(); + assert!(err.contains("url"), "实际: {}", err); + } + + #[test] + fn params_empty_url_errors() { + let err = parse_params(&json!({ "url": " " })).unwrap_err().to_string(); + assert!(err.contains("空"), "实际: {}", err); + } + + #[test] + fn params_defaults_method_and_timeout() { + let p = parse_params(&json!({ "url": "https://example.com" })).unwrap(); + assert_eq!(p.method, HttpMethod::Get); + assert_eq!(p.timeout_secs, DEFAULT_TIMEOUT_SECS); + assert!(p.headers.is_empty()); + assert_eq!(p.body, None); + } + + #[test] + fn params_custom_timeout() { + let p = parse_params(&json!({ "url": "https://x", "timeout_secs": 5 })).unwrap(); + assert_eq!(p.timeout_secs, 5); + } + + #[test] + fn params_parses_headers_object() { + let p = parse_params(&json!({ + "url": "https://x", + "headers": { + "Authorization": "Bearer abc", + "X-Trace-Id": "123" + } + })) + .unwrap(); + // headers 顺序由 serde_json BTreeMap 保证;只校验集合 + let map: std::collections::HashMap<&str, &str> = + p.headers.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); + assert_eq!(map.get("Authorization"), Some(&"Bearer abc")); + assert_eq!(map.get("X-Trace-Id"), Some(&"123")); + } + + #[test] + fn params_non_string_header_value_skipped() { + // 非字符串值不应炸,应被跳过 + let p = parse_params(&json!({ + "url": "https://x", + "headers": { "X-Num": 123, "X-Ok": "yes" } + })) + .unwrap(); + let names: Vec<&str> = p.headers.iter().map(|(k, _)| k.as_str()).collect(); + assert!(names.contains(&"X-Ok")); + assert!(!names.contains(&"X-Num"), "非字符串 header 应被跳过"); + } + + #[test] + fn params_parses_body() { + let p = parse_params(&json!({ + "url": "https://x", + "method": "POST", + "body": "{\"k\":1}" + })) + .unwrap(); + assert_eq!(p.method, HttpMethod::Post); + assert_eq!(p.body.as_deref(), Some("{\"k\":1}")); + } + + #[test] + fn params_invalid_method_propagates() { + let err = parse_params(&json!({ "url": "https://x", "method": "TRACE" })) + .unwrap_err() + .to_string(); + assert!(err.contains("非法 method"), "实际: {}", err); + } +} diff --git a/crates/df-nodes/src/lib.rs b/crates/df-nodes/src/lib.rs index e5e5af6..456fff3 100644 --- a/crates/df-nodes/src/lib.rs +++ b/crates/df-nodes/src/lib.rs @@ -1,10 +1,13 @@ -//! df-nodes: 内置节点集合 — AI、脚本、人工审批 +//! df-nodes: 内置节点集合 — AI、脚本、人工审批、Git、HTTP、通知 pub mod ai_node; pub mod ai_self_review_node; mod ai_node_helpers; +pub mod git_node; +pub mod http_node; pub mod human_node; mod human_node_helpers; +pub mod notify_node; pub mod script_node; pub mod task_advance_node; pub mod task_state_machine; diff --git a/crates/df-nodes/src/notify_node.rs b/crates/df-nodes/src/notify_node.rs new file mode 100644 index 0000000..8582005 --- /dev/null +++ b/crates/df-nodes/src/notify_node.rs @@ -0,0 +1,292 @@ +//! 通知节点 — desktop(本地日志,后续接 tauri-plugin-notification) / webhook +//! +//! - type=desktop:tracing::info! 输出(桌面通知集成延后,见 PROGRESS.md 后续 Sprint) +//! - type=webhook:POST JSON {title, message} 到 webhook_url +//! +//! 节点语义:尽力而为,通知失败不阻断工作流(webhook 发送失败时输出 success=false 但仍 Ok 返回)。 + +use std::time::Duration; + +use async_trait::async_trait; +use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; + +/// 通知节点 +pub struct NotifyNode; + +/// 通知类型 +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum NotifyType { + Desktop, + Webhook, +} + +impl NotifyType { + /// 从 config.type 字符串解析,非法值返回 Err。 + pub fn parse(raw: &str) -> anyhow::Result { + match raw { + "desktop" => Ok(NotifyType::Desktop), + "webhook" => Ok(NotifyType::Webhook), + other => anyhow::bail!( + "NotifyNode 非法 type: {}(合法值: desktop|webhook)", + other + ), + } + } + + fn as_str(&self) -> &'static str { + match self { + NotifyType::Desktop => "desktop", + NotifyType::Webhook => "webhook", + } + } +} + +/// 从 NodeContext.config 解析出的 NotifyNode 参数。 +/// 抽离此结构便于单元测试 config 解析逻辑(无需发起网络请求)。 +#[derive(Debug, Clone)] +pub struct NotifyParams { + pub notify_type: NotifyType, + pub title: String, + pub message: String, + /// webhook 类型必填;desktop 类型忽略 + pub webhook_url: Option, +} + +/// 从 config JSON 解析 NotifyParams。type 必填;webhook 类型要求 webhook_url。 +pub fn parse_params(config: &serde_json::Value) -> anyhow::Result { + let raw_type = config + .get("type") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("NotifyNode 缺少必填参数: type"))?; + let notify_type = NotifyType::parse(raw_type)?; + + let title = config + .get("title") + .and_then(|v| v.as_str()) + .unwrap_or("通知") + .to_string(); + + let message = config + .get("message") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + + let webhook_url = config + .get("webhook_url") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + // webhook 类型必须有 webhook_url + if notify_type == NotifyType::Webhook { + if webhook_url.as_deref().map(|s| s.trim().is_empty()).unwrap_or(true) { + anyhow::bail!("NotifyNode type=webhook 缺少 webhook_url"); + } + } + + Ok(NotifyParams { + notify_type, + title, + message, + webhook_url, + }) +} + +#[async_trait] +impl Node for NotifyNode { + async fn execute(&self, ctx: NodeContext) -> NodeResult { + tracing::info!("NotifyNode 执行: node_id={}", ctx.node_id); + + let params = parse_params(&ctx.config)?; + + match params.notify_type { + NotifyType::Desktop => { + // 桌面通知集成延后;当前仅日志,后续接 tauri-plugin-notification。 + tracing::info!( + title = %params.title, + message = %params.message, + "NotifyNode desktop 通知(日志占位,集成待后续 Sprint)" + ); + Ok(NodeOutput::from_value(serde_json::json!({ + "type": params.notify_type.as_str(), + "title": params.title, + "message": params.message, + "delivered": true, + }))) + } + NotifyType::Webhook => { + // webhook_url 在 parse_params 已校验非空 + let url = params.webhook_url.as_deref().unwrap(); + tracing::info!( + title = %params.title, + url = %url, + "NotifyNode webhook 推送" + ); + + let payload = serde_json::json!({ + "title": params.title, + "message": params.message, + }); + + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(15)) + .build()?; + + // 尽力而为:发送失败不阻断工作流,降级为 success=false + error 字段返回。 + let (status_code, success, error) = match client + .post(url) + .json(&payload) + .send() + .await + { + Ok(resp) => { + let code = resp.status().as_u16(); + (Some(code), code >= 200 && code < 300, None) + } + Err(e) => (None, false, Some(e.to_string())), + }; + + if !success { + tracing::warn!( + url = %url, + status_code = ?status_code, + error = ?error, + "NotifyNode webhook 发送失败(不阻断工作流)" + ); + } + + Ok(NodeOutput::from_value(serde_json::json!({ + "type": params.notify_type.as_str(), + "title": params.title, + "message": params.message, + "webhook_url": url, + "status_code": status_code, + "success": success, + "error": error, + }))) + } + } + } + + fn schema(&self) -> NodeSchema { + NodeSchema { + params: serde_json::json!({ + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["desktop", "webhook"] + }, + "title": { "type": "string", "description": "通知标题" }, + "message": { "type": "string", "description": "通知正文" }, + "webhook_url": { "type": "string", "description": "webhook URL(webhook 类型必填)" } + }, + "required": ["type"] + }), + output: serde_json::json!({ + "type": "object", + "properties": { + "type": { "type": "string" }, + "title": { "type": "string" }, + "message": { "type": "string" }, + "delivered": { "type": "boolean" }, + "webhook_url": { "type": "string" }, + "status_code": { "type": "integer" }, + "success": { "type": "boolean" }, + "error": { "type": "string" } + } + }), + } + } + + fn node_type(&self) -> &str { + "notify" + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + // ── NotifyType::parse ── + + #[test] + fn type_parses_desktop_and_webhook() { + assert_eq!(NotifyType::parse("desktop").unwrap(), NotifyType::Desktop); + assert_eq!(NotifyType::parse("webhook").unwrap(), NotifyType::Webhook); + } + + #[test] + fn type_invalid_errors() { + let err = NotifyType::parse("email").unwrap_err().to_string(); + assert!(err.contains("非法 type"), "实际: {}", err); + assert!(err.contains("email")); + } + + // ── parse_params(config) ── + + #[test] + fn params_missing_type_errors() { + let err = parse_params(&json!({ "title": "t" })).unwrap_err().to_string(); + assert!(err.contains("type"), "实际: {}", err); + } + + #[test] + fn params_invalid_type_errors() { + let err = parse_params(&json!({ "type": "sms" })).unwrap_err().to_string(); + assert!(err.contains("非法 type"), "实际: {}", err); + } + + #[test] + fn params_desktop_defaults_title_and_message() { + let p = parse_params(&json!({ "type": "desktop" })).unwrap(); + assert_eq!(p.notify_type, NotifyType::Desktop); + assert_eq!(p.title, "通知"); // 默认标题 + assert_eq!(p.message, ""); // 默认空正文 + assert_eq!(p.webhook_url, None); // desktop 不需要 + } + + #[test] + fn params_desktop_with_custom_fields() { + let p = parse_params(&json!({ + "type": "desktop", + "title": "构建完成", + "message": "已发布 v1.0" + })) + .unwrap(); + assert_eq!(p.title, "构建完成"); + assert_eq!(p.message, "已发布 v1.0"); + } + + #[test] + fn params_webhook_requires_url() { + let err = parse_params(&json!({ "type": "webhook" })) + .unwrap_err() + .to_string(); + assert!(err.contains("webhook_url"), "实际: {}", err); + } + + #[test] + fn params_webhook_empty_url_errors() { + let err = parse_params(&json!({ "type": "webhook", "webhook_url": " " })) + .unwrap_err() + .to_string(); + assert!(err.contains("webhook_url"), "实际: {}", err); + } + + #[test] + fn params_webhook_valid() { + let p = parse_params(&json!({ + "type": "webhook", + "title": "告警", + "message": "CPU > 90%", + "webhook_url": "https://hooks.example.com/x" + })) + .unwrap(); + assert_eq!(p.notify_type, NotifyType::Webhook); + assert_eq!(p.title, "告警"); + assert_eq!(p.message, "CPU > 90%"); + assert_eq!(p.webhook_url.as_deref(), Some("https://hooks.example.com/x")); + } +} diff --git a/crates/df-workflow/Cargo.toml b/crates/df-workflow/Cargo.toml index 5c55277..05ea0d0 100644 --- a/crates/df-workflow/Cargo.toml +++ b/crates/df-workflow/Cargo.toml @@ -23,3 +23,4 @@ async-trait = { workspace = true } futures = "0.3" anyhow = { workspace = true } tracing = { workspace = true } +serde_yaml = "0.9" diff --git a/crates/df-workflow/src/lib.rs b/crates/df-workflow/src/lib.rs index fc6a3f2..31edab2 100644 --- a/crates/df-workflow/src/lib.rs +++ b/crates/df-workflow/src/lib.rs @@ -8,3 +8,4 @@ pub mod executor; pub mod node; pub mod registry; pub mod state; +pub mod template_loader; diff --git a/crates/df-workflow/src/template_loader.rs b/crates/df-workflow/src/template_loader.rs new file mode 100644 index 0000000..1011610 --- /dev/null +++ b/crates/df-workflow/src/template_loader.rs @@ -0,0 +1,330 @@ +//! YAML 模板加载器 — 从 YAML 字符串加载 DagDef + 校验 +//! +//! 模板格式: +//! ```yaml +//! name: 代码审查 +//! description: AI 驱动的代码审查流程 +//! nodes: +//! review: +//! type: ai +//! config: +//! prompt: "审查以下代码变更..." +//! persona_id: reviewer +//! notify: +//! type: notify +//! config: +//! type: desktop +//! title: "审查完成" +//! edges: +//! - from: review +//! to: notify +//! ``` + +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +use crate::dag_def::{DagDef, EdgeDef, NodeDef}; + +/// YAML 模板顶层结构 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WorkflowTemplate { + pub name: String, + #[serde(default)] + pub description: String, + pub nodes: HashMap, + #[serde(default)] + pub edges: Vec, +} + +/// 模板节点(YAML 友好命名) +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TemplateNode { + /// 节点类型(ai/script/human/git/http/notify/subflow) + #[serde(rename = "type")] + pub node_type: String, + #[serde(default)] + pub config: serde_json::Value, + #[serde(default)] + pub label: Option, +} + +/// 模板边(YAML 友好命名) +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TemplateEdge { + pub from: String, + pub to: String, + #[serde(default)] + pub condition: Option, +} + +/// 模板加载错误 +#[derive(Debug)] +pub enum TemplateError { + /// YAML 解析失败 + YamlParse(String), + /// 校验失败(空节点/未知类型/依赖环) + Validation(String), +} + +impl std::fmt::Display for TemplateError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::YamlParse(msg) => write!(f, "YAML 解析失败: {}", msg), + Self::Validation(msg) => write!(f, "模板校验失败: {}", msg), + } + } +} + +impl std::error::Error for TemplateError {} + +/// 已知的节点类型白名单 +const KNOWN_NODE_TYPES: &[&str] = &[ + "ai", "script", "human", "git", "http", "notify", "subflow", +]; + +/// 从 YAML 字符串加载模板并转换为 DagDef +/// +/// 步骤: +/// 1. YAML → WorkflowTemplate(serde_yaml) +/// 2. 校验(节点非空/类型合法/边引用存在/无环) +/// 3. 转换为 DagDef +pub fn load_template(yaml: &str) -> Result { + // 1. YAML 解析 + let template: WorkflowTemplate = serde_yaml::from_str(yaml) + .map_err(|e| TemplateError::YamlParse(e.to_string()))?; + + // 2. 校验 + validate_template(&template)?; + + // 3. 转换为 DagDef + let mut dag = DagDef::new(); + for (id, node) in &template.nodes { + dag.nodes.insert( + id.clone(), + NodeDef { + id: id.clone(), + node_type: node.node_type.clone(), + config: node.config.clone(), + label: node.label.clone(), + }, + ); + } + for edge in &template.edges { + dag.edges.push(EdgeDef { + source: edge.from.clone(), + target: edge.to.clone(), + condition: edge.condition.clone(), + }); + } + + Ok(dag) +} + +/// 校验模板结构 +fn validate_template(t: &WorkflowTemplate) -> Result<(), TemplateError> { + // 节点非空 + if t.nodes.is_empty() { + return Err(TemplateError::Validation("模板节点不能为空".into())); + } + + // 节点类型合法 + for (id, node) in &t.nodes { + if !KNOWN_NODE_TYPES.contains(&node.node_type.as_str()) { + return Err(TemplateError::Validation(format!( + "节点 '{}' 使用了未知类型 '{}',已知类型: {:?}", + id, node.node_type, KNOWN_NODE_TYPES + ))); + } + } + + // 边引用的节点存在 + let node_ids: std::collections::HashSet<&String> = t.nodes.keys().collect(); + for edge in &t.edges { + if !node_ids.contains(&edge.from) { + return Err(TemplateError::Validation(format!( + "边的起点 '{}' 不存在", edge.from + ))); + } + if !node_ids.contains(&edge.to) { + return Err(TemplateError::Validation(format!( + "边的终点 '{}' 不存在", edge.to + ))); + } + } + + // 依赖无环(DFS 检测) + if has_cycle(&t.edges) { + return Err(TemplateError::Validation("依赖存在环".into())); + } + + Ok(()) +} + +/// DFS 检测边列表是否有环 +fn has_cycle(edges: &[TemplateEdge]) -> bool { + use std::collections::{HashMap, HashSet}; + + // 建邻接表 + let mut adj: HashMap<&str, Vec<&str>> = HashMap::new(); + for e in edges { + adj.entry(e.from.as_str()).or_default().push(e.to.as_str()); + } + + let mut visited: HashSet<&str> = HashSet::new(); + let mut in_stack: HashSet<&str> = HashSet::new(); + + fn dfs<'a>( + node: &'a str, + adj: &HashMap<&'a str, Vec<&'a str>>, + visited: &mut HashSet<&'a str>, + in_stack: &mut HashSet<&'a str>, + ) -> bool { + if in_stack.contains(node) { + return true; // 环 + } + if visited.contains(node) { + return false; // 已访问过,无环 + } + visited.insert(node); + in_stack.insert(node); + if let Some(neighbors) = adj.get(node) { + for next in neighbors { + if dfs(next, adj, visited, in_stack) { + return true; + } + } + } + in_stack.remove(node); + false + } + + for e in edges { + let start = e.from.as_str(); + if dfs(start, &adj, &mut visited, &mut in_stack) { + return true; + } + } + false +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tpl_01_valid_template_loads() { + let yaml = r#" +name: 代码审查 +description: AI 驱动的代码审查 +nodes: + review: + type: ai + config: + prompt: "审查代码" + notify: + type: notify + config: + type: desktop + title: "完成" +edges: + - from: review + to: notify +"#; + let dag = load_template(yaml).unwrap(); + assert_eq!(dag.nodes.len(), 2); + assert_eq!(dag.edges.len(), 1); + assert!(dag.nodes.contains_key("review")); + assert_eq!(dag.nodes.get("review").unwrap().node_type, "ai"); + } + + #[test] + fn tpl_02_empty_nodes_rejected() { + let yaml = "name: 空\nnodes: {}\n"; + let result = load_template(yaml); + assert!(matches!(result, Err(TemplateError::Validation(_)))); + } + + #[test] + fn tpl_03_unknown_node_type_rejected() { + let yaml = r#" +name: test +nodes: + bad: + type: unknown_type + config: {} +"#; + let result = load_template(yaml); + assert!(matches!(result, Err(TemplateError::Validation(_)))); + } + + #[test] + fn tpl_04_cycle_detected() { + let yaml = r#" +name: 环 +nodes: + a: + type: ai + config: {} + b: + type: ai + config: {} +edges: + - from: a + to: b + - from: b + to: a +"#; + let result = load_template(yaml); + assert!(matches!(result, Err(TemplateError::Validation(ref msg)) if msg.contains("环"))); + } + + #[test] + fn tpl_05_edge_to_nonexistent_rejected() { + let yaml = r#" +name: test +nodes: + a: + type: ai + config: {} +edges: + - from: a + to: ghost +"#; + let result = load_template(yaml); + assert!(matches!(result, Err(TemplateError::Validation(_)))); + } + + #[test] + fn tpl_06_no_edges_ok() { + let yaml = r#" +name: 单节点 +nodes: + solo: + type: script + config: + command: "echo hello" +"#; + let dag = load_template(yaml).unwrap(); + assert_eq!(dag.nodes.len(), 1); + assert!(dag.edges.is_empty()); + } + + #[test] + fn tpl_07_invalid_yaml_rejected() { + let yaml = "not: valid: yaml: {{{"; + let result = load_template(yaml); + assert!(matches!(result, Err(TemplateError::YamlParse(_)))); + } + + #[test] + fn tpl_08_all_known_node_types() { + for nt in KNOWN_NODE_TYPES { + let yaml = format!( + "name: test\nnodes:\n n:\n type: {}\n config: {{}}", + nt + ); + let result = load_template(&yaml); + assert!(result.is_ok(), "节点类型 {} 应合法", nt); + } + } +}