beam_core/
workflow_definition.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use anyhow::{Context, Result};
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6
7#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
8#[serde(rename_all = "camelCase")]
9pub struct ParamDef {
10    #[serde(rename = "type")]
11    pub param_type: String,
12    #[serde(default, skip_serializing_if = "Option::is_none")]
13    pub format: Option<String>,
14    #[serde(default, skip_serializing_if = "Option::is_none")]
15    pub required: Option<bool>,
16    #[serde(default, skip_serializing_if = "Option::is_none")]
17    pub default: Option<Value>,
18    #[serde(default, skip_serializing_if = "Option::is_none")]
19    pub description: Option<String>,
20}
21
22#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
23#[serde(rename_all = "camelCase")]
24pub struct RetryPolicy {
25    pub max_attempts: u64,
26    pub backoff: String,
27    pub base_ms: u64,
28    #[serde(default, skip_serializing_if = "Option::is_none")]
29    pub factor: Option<f64>,
30    #[serde(default, skip_serializing_if = "Option::is_none")]
31    pub jitter: Option<bool>,
32}
33
34#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
35#[serde(rename_all = "camelCase")]
36pub struct HumanGate {
37    pub stage: String,
38    pub prompt: Value,
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    pub approvers: Option<Vec<String>>,
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub deadline_ms: Option<u64>,
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    pub on_timeout: Option<String>,
45}
46
47#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
48#[serde(rename_all = "camelCase")]
49pub struct LoopTerminate {
50    pub node: String,
51    pub via: String,
52}
53
54#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
55#[serde(rename_all = "camelCase")]
56pub struct LoopOutputProjection {
57    pub from: String,
58}
59
60#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
61#[serde(rename_all = "camelCase")]
62pub struct NodeBase {
63    #[serde(default, skip_serializing_if = "Option::is_none")]
64    pub description: Option<String>,
65    #[serde(default, skip_serializing_if = "Option::is_none")]
66    pub depends: Option<Vec<String>>,
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    pub human_gate: Option<HumanGate>,
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub retry_policy: Option<RetryPolicy>,
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    pub timeout_ms: Option<u64>,
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    pub max_output_bytes: Option<u64>,
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub output_schema: Option<Value>,
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub unsafe_allow_ungated: Option<bool>,
79}
80
81#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
82#[serde(rename_all = "camelCase")]
83pub struct SubagentNode {
84    #[serde(flatten)]
85    pub base: NodeBase,
86    pub bot: String,
87    pub prompt: Value,
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub working_dir: Option<String>,
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub model_overrides: Option<Value>,
92    #[serde(default, skip_serializing_if = "Option::is_none")]
93    pub tool_policy: Option<Value>,
94}
95
96#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
97#[serde(rename_all = "camelCase")]
98pub struct HostExecutorNode {
99    #[serde(flatten)]
100    pub base: NodeBase,
101    pub executor: String,
102    pub input: Value,
103}
104
105#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
106#[serde(rename_all = "camelCase")]
107pub struct LoopNode {
108    #[serde(flatten)]
109    pub base: NodeBase,
110    pub max_iterations: u64,
111    pub body: Vec<String>,
112    pub terminate: LoopTerminate,
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub output: Option<LoopOutputProjection>,
115}
116
117#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
118#[serde(rename_all = "camelCase")]
119pub struct DecisionNode {
120    #[serde(flatten)]
121    pub base: NodeBase,
122}
123
124#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
125#[serde(tag = "type", rename_all = "camelCase")]
126pub enum WorkflowNode {
127    Subagent(SubagentNode),
128    HostExecutor(HostExecutorNode),
129    Loop(LoopNode),
130    Decision(DecisionNode),
131}
132
133#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
134#[serde(rename_all = "camelCase")]
135pub struct WorkflowDefaults {
136    #[serde(default, skip_serializing_if = "Option::is_none")]
137    pub retry_policy: Option<RetryPolicy>,
138    #[serde(default, skip_serializing_if = "Option::is_none")]
139    pub timeout_ms: Option<u64>,
140    #[serde(default, skip_serializing_if = "Option::is_none")]
141    pub max_output_bytes: Option<u64>,
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    pub max_concurrency: Option<u64>,
144}
145
146#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
147#[serde(rename_all = "camelCase")]
148pub struct WorkflowDefinition {
149    pub workflow_id: String,
150    pub version: u64,
151    #[serde(default, skip_serializing_if = "Option::is_none")]
152    pub params: Option<BTreeMap<String, ParamDef>>,
153    #[serde(default, skip_serializing_if = "Option::is_none")]
154    pub defaults: Option<WorkflowDefaults>,
155    pub nodes: BTreeMap<String, WorkflowNode>,
156}
157
158pub fn parse_workflow_definition(raw: &str) -> Result<WorkflowDefinition> {
159    let def: WorkflowDefinition =
160        serde_json::from_str(raw).context("failed to parse workflow json")?;
161    validate_workflow_definition(&def)?;
162    Ok(def)
163}
164
165pub fn validate_workflow_definition(def: &WorkflowDefinition) -> Result<()> {
166    if def.workflow_id.trim().is_empty() {
167        anyhow::bail!("workflowId 缺失");
168    }
169    if def.version == 0 {
170        anyhow::bail!("version must be positive");
171    }
172    if def.nodes.is_empty() {
173        anyhow::bail!("Workflow must declare at least one node");
174    }
175    for node_id in def.nodes.keys() {
176        if node_id == "." || node_id == ".." || node_id.contains("..") {
177            anyhow::bail!(
178                "nodeId '{}' rejected: path-traversal style ids are not allowed",
179                node_id
180            );
181        }
182    }
183    for (node_id, node) in &def.nodes {
184        for dep in node_depends(node) {
185            if !def.nodes.contains_key(dep) {
186                anyhow::bail!("Node '{}' depends on unknown node '{}'", node_id, dep);
187            }
188            if dep == node_id {
189                anyhow::bail!("Node '{}' depends on itself", node_id);
190            }
191        }
192    }
193    detect_cycles(def)?;
194    if !def.nodes.values().any(|node| node_depends(node).is_empty()) {
195        anyhow::bail!(
196            "Workflow has no scheduler-visible root node (every non-loop-body node has dependencies)"
197        );
198    }
199    Ok(())
200}
201
202fn node_depends(node: &WorkflowNode) -> &[String] {
203    match node {
204        WorkflowNode::Subagent(n) => n.base.depends.as_deref().unwrap_or(&[]),
205        WorkflowNode::HostExecutor(n) => n.base.depends.as_deref().unwrap_or(&[]),
206        WorkflowNode::Loop(n) => n.base.depends.as_deref().unwrap_or(&[]),
207        WorkflowNode::Decision(n) => n.base.depends.as_deref().unwrap_or(&[]),
208    }
209}
210
211fn detect_cycles(def: &WorkflowDefinition) -> Result<()> {
212    #[derive(Clone, Copy, PartialEq, Eq)]
213    enum Mark {
214        Temp,
215        Perm,
216    }
217    fn visit(
218        node_id: &str,
219        def: &WorkflowDefinition,
220        marks: &mut BTreeMap<String, Mark>,
221        stack: &mut BTreeSet<String>,
222    ) -> Result<()> {
223        if matches!(marks.get(node_id), Some(Mark::Perm)) {
224            return Ok(());
225        }
226        if matches!(marks.get(node_id), Some(Mark::Temp)) {
227            anyhow::bail!("Workflow graph contains a cycle at '{}'", node_id);
228        }
229        marks.insert(node_id.to_string(), Mark::Temp);
230        stack.insert(node_id.to_string());
231        for dep in node_depends(def.nodes.get(node_id).expect("node must exist")) {
232            visit(dep, def, marks, stack)?;
233        }
234        stack.remove(node_id);
235        marks.insert(node_id.to_string(), Mark::Perm);
236        Ok(())
237    }
238
239    let mut marks = BTreeMap::new();
240    let mut stack = BTreeSet::new();
241    for node_id in def.nodes.keys() {
242        visit(node_id, def, &mut marks, &mut stack)?;
243    }
244    Ok(())
245}
246
247#[cfg(test)]
248mod tests {
249    use super::*;
250
251    #[test]
252    fn parse_workflow_definition_accepts_basic_graph() {
253        let def = parse_workflow_definition(
254            r#"{
255                "workflowId":"flow-a",
256                "version":1,
257                "nodes":{
258                    "a":{"type":"subagent","bot":"bot-a","prompt":"hi"},
259                    "b":{"type":"hostExecutor","executor":"feishu-send","input":1,"depends":["a"],"humanGate":{"stage":"before","prompt":"approve?"}}
260                }
261            }"#,
262        )
263        .expect("definition");
264        assert_eq!(def.workflow_id, "flow-a");
265        assert_eq!(def.nodes.len(), 2);
266    }
267}
beam_core/workflow_definition.rs

beam_core/
workflow_definition.rs
