bamboo_domain/workflow/
compiler.rs

1use std::collections::{HashMap, HashSet};
2
3use crate::{
4    ValueRef, WorkflowPlan, WorkflowRunDefinition, WorkflowSecretHandle, WorkflowStepDefinition,
5    WorkflowStepKind,
6};
7use thiserror::Error;
8
9use super::schema::{validate_schema, validate_schema_shape as validate_shape};
10
11#[derive(Debug, Error, Clone, PartialEq, Eq)]
12pub enum WorkflowCompileError {
13    #[error("unsupported workflow_schema {0}")]
14    UnsupportedSchema(u32),
15    #[error("workflow id and revision are required")]
16    MissingIdentity,
17    #[error("invalid workflow budget: {0}")]
18    InvalidBudget(String),
19    #[error("duplicate or empty step id: {0}")]
20    DuplicateStep(String),
21    #[error("unknown step reference: {0}")]
22    UnknownStep(String),
23    #[error("invalid step {step}: {message}")]
24    InvalidStep { step: String, message: String },
25    #[error("cyclic step dependency: {0}")]
26    Cycle(String),
27    #[error("invalid schema: {0}")]
28    InvalidSchema(String),
29}
30
31#[derive(Debug, Clone)]
32pub struct CompiledWorkflow {
33    pub definition: WorkflowRunDefinition,
34    pub steps: HashMap<String, WorkflowStepDefinition>,
35}
36
37impl CompiledWorkflow {
38    pub fn compile(definition: WorkflowRunDefinition) -> Result<Self, WorkflowCompileError> {
39        if definition.workflow_schema != 1 {
40            return Err(WorkflowCompileError::UnsupportedSchema(
41                definition.workflow_schema,
42            ));
43        }
44        if definition.id.trim().is_empty() || definition.revision == 0 {
45            return Err(WorkflowCompileError::MissingIdentity);
46        }
47        validate_budget(&definition)?;
48        validate_schema_shape(&definition.input_schema)?;
49        if let Some(schema) = &definition.output_schema {
50            validate_schema_shape(schema)?;
51        }
52        let mut steps = HashMap::new();
53        for step in &definition.steps {
54            if step.id.trim().is_empty() || steps.insert(step.id.clone(), step.clone()).is_some() {
55                return Err(WorkflowCompileError::DuplicateStep(step.id.clone()));
56            }
57            validate_step(step)?;
58        }
59        if steps.is_empty() {
60            return Err(WorkflowCompileError::InvalidStep {
61                step: "<definition>".to_string(),
62                message: "at least one step is required".to_string(),
63            });
64        }
65        validate_plan(&definition.plan, &steps)?;
66        validate_dependencies(&definition.plan, &steps)?;
67        validate_execution_bindings(&definition, &steps)?;
68        Ok(Self { definition, steps })
69    }
70
71    pub fn validate_input(&self, value: &serde_json::Value) -> Result<(), String> {
72        validate_schema(&self.definition.input_schema, value)
73    }
74}
75
76fn validate_schema_shape(schema: &serde_json::Value) -> Result<(), WorkflowCompileError> {
77    validate_shape(schema, "$").map_err(WorkflowCompileError::InvalidSchema)
78}
79
80fn validate_budget(definition: &WorkflowRunDefinition) -> Result<(), WorkflowCompileError> {
81    let budget = &definition.budgets;
82    if budget.max_concurrency == 0 || budget.max_concurrency > 64 {
83        return Err(WorkflowCompileError::InvalidBudget(
84            "max_concurrency must be 1..=64".to_string(),
85        ));
86    }
87    if budget.max_steps == 0
88        || budget.max_agents > budget.max_steps
89        || budget.max_nesting_depth == 0
90        || budget.wall_time_ms == 0
91    {
92        return Err(WorkflowCompileError::InvalidBudget(
93            "step/agent/depth/wall budgets must be bounded and coherent".to_string(),
94        ));
95    }
96    Ok(())
97}
98
99fn validate_step(step: &WorkflowStepDefinition) -> Result<(), WorkflowCompileError> {
100    let required = match &step.kind {
101        WorkflowStepKind::Tool { tool, .. } => tool,
102        WorkflowStepKind::Agent {
103            agent,
104            structured_output_attempts,
105            ..
106        } => {
107            if *structured_output_attempts == 0 || *structured_output_attempts > 8 {
108                return Err(WorkflowCompileError::InvalidStep {
109                    step: step.id.clone(),
110                    message: "structured_output_attempts must be 1..=8".to_string(),
111                });
112            }
113            agent
114        }
115        WorkflowStepKind::Workflow {
116            workflow_id,
117            revision,
118            ..
119        } => {
120            if *revision == 0 {
121                return Err(WorkflowCompileError::InvalidStep {
122                    step: step.id.clone(),
123                    message: "nested workflow revision must be fixed and non-zero".to_string(),
124                });
125            }
126            workflow_id
127        }
128    };
129    if required.trim().is_empty() {
130        return Err(WorkflowCompileError::InvalidStep {
131            step: step.id.clone(),
132            message: "target cannot be empty".to_string(),
133        });
134    }
135    if let Some(schema) = &step.output_schema {
136        validate_schema_shape(schema)?;
137    }
138    Ok(())
139}
140
141fn validate_plan(
142    plan: &WorkflowPlan,
143    steps: &HashMap<String, WorkflowStepDefinition>,
144) -> Result<(), WorkflowCompileError> {
145    let mut seen = HashSet::new();
146    validate_plan_inner(plan, steps, &mut seen)?;
147    let unreferenced = steps
148        .keys()
149        .filter(|step| !seen.contains(*step))
150        .cloned()
151        .collect::<Vec<_>>();
152    if !unreferenced.is_empty() {
153        return Err(WorkflowCompileError::InvalidStep {
154            step: "<plan>".to_string(),
155            message: format!("unreferenced steps: {}", unreferenced.join(", ")),
156        });
157    }
158    Ok(())
159}
160
161fn validate_plan_inner(
162    plan: &WorkflowPlan,
163    steps: &HashMap<String, WorkflowStepDefinition>,
164    seen: &mut HashSet<String>,
165) -> Result<(), WorkflowCompileError> {
166    match plan {
167        WorkflowPlan::Step { step } => {
168            if !steps.contains_key(step) {
169                return Err(WorkflowCompileError::UnknownStep(step.clone()));
170            }
171            if !seen.insert(step.clone()) {
172                return Err(WorkflowCompileError::InvalidStep {
173                    step: step.clone(),
174                    message: "step appears more than once in the execution plan".to_string(),
175                });
176            }
177        }
178        WorkflowPlan::Sequence { nodes } | WorkflowPlan::Parallel { nodes } => {
179            if nodes.is_empty() {
180                return Err(WorkflowCompileError::InvalidStep {
181                    step: "<plan>".to_string(),
182                    message: "sequence/parallel requires nodes".to_string(),
183                });
184            }
185            for node in nodes {
186                validate_plan_inner(node, steps, seen)?;
187            }
188        }
189        WorkflowPlan::Choice {
190            condition,
191            then_branch,
192            else_branch,
193        } => {
194            validate_ref(condition, steps)?;
195            validate_plan_inner(then_branch, steps, seen)?;
196            validate_plan_inner(else_branch, steps, seen)?;
197        }
198        WorkflowPlan::Map { source, item, body } => {
199            if item.trim().is_empty() {
200                return Err(WorkflowCompileError::InvalidStep {
201                    step: "<map>".to_string(),
202                    message: "map item name cannot be empty".to_string(),
203                });
204            }
205            validate_ref(source, steps)?;
206            validate_plan_inner(body, steps, seen)?;
207        }
208        WorkflowPlan::Retry {
209            node, max_attempts, ..
210        } => {
211            if *max_attempts == 0 {
212                return Err(WorkflowCompileError::InvalidStep {
213                    step: "<retry>".to_string(),
214                    message: "max_attempts must be greater than zero".to_string(),
215                });
216            }
217            validate_plan_inner(node, steps, seen)?;
218        }
219    }
220    Ok(())
221}
222
223fn validate_execution_bindings(
224    definition: &WorkflowRunDefinition,
225    steps: &HashMap<String, WorkflowStepDefinition>,
226) -> Result<(), WorkflowCompileError> {
227    type ItemSchemas = HashMap<String, serde_json::Value>;
228
229    fn reference_schema(
230        owner: &str,
231        reference: &ValueRef,
232        definition: &WorkflowRunDefinition,
233        steps: &HashMap<String, WorkflowStepDefinition>,
234        available: &HashSet<String>,
235        items: &ItemSchemas,
236    ) -> Result<serde_json::Value, WorkflowCompileError> {
237        let invalid = |message: String| WorkflowCompileError::InvalidStep {
238            step: owner.to_string(),
239            message,
240        };
241        match reference {
242            ValueRef::Args { pointer } => schema_pointer(&definition.input_schema, pointer)
243                .cloned()
244                .ok_or_else(|| invalid("args reference has an invalid schema pointer".to_string())),
245            ValueRef::Step { step, pointer } => {
246                if !available.contains(step) {
247                    return Err(invalid(format!(
248                        "step reference '{step}' is not available before this step"
249                    )));
250                }
251                let schema = steps
252                    .get(step)
253                    .and_then(|step| step.output_schema.as_ref())
254                    .ok_or_else(|| {
255                        invalid(format!(
256                            "step reference '{step}' requires a declared output_schema"
257                        ))
258                    })?;
259                schema_pointer(schema, pointer).cloned().ok_or_else(|| {
260                    invalid(format!(
261                        "step reference '{step}' has an invalid schema pointer"
262                    ))
263                })
264            }
265            ValueRef::Item { name, pointer } => {
266                let schema = items.get(name).ok_or_else(|| {
267                    invalid(format!(
268                        "map item reference '{name}' is outside its map body"
269                    ))
270                })?;
271                schema_pointer(schema, pointer).cloned().ok_or_else(|| {
272                    invalid(format!(
273                        "map item reference '{name}' has an invalid schema pointer"
274                    ))
275                })
276            }
277            ValueRef::Literal { value } => Ok(schema_for_literal(value)),
278        }
279    }
280
281    fn walk_template(
282        owner: &str,
283        value: &serde_json::Value,
284        definition: &WorkflowRunDefinition,
285        steps: &HashMap<String, WorkflowStepDefinition>,
286        available: &HashSet<String>,
287        items: &ItemSchemas,
288    ) -> Result<(), WorkflowCompileError> {
289        match value {
290            serde_json::Value::Object(object) if object.contains_key("$secret") => {
291                let handle: WorkflowSecretHandle =
292                    serde_json::from_value(value.clone()).map_err(|error| {
293                        WorkflowCompileError::InvalidStep {
294                            step: owner.to_string(),
295                            message: format!("malformed secret capability handle: {error}"),
296                        }
297                    })?;
298                if handle.capability.trim().is_empty() {
299                    return Err(WorkflowCompileError::InvalidStep {
300                        step: owner.to_string(),
301                        message: "secret capability handle cannot be empty".to_string(),
302                    });
303                }
304                Ok(())
305            }
306            serde_json::Value::Object(object) if object.contains_key("from") => {
307                let reference: ValueRef =
308                    serde_json::from_value(value.clone()).map_err(|error| {
309                        WorkflowCompileError::InvalidStep {
310                            step: owner.to_string(),
311                            message: format!("malformed value reference: {error}"),
312                        }
313                    })?;
314                reference_schema(owner, &reference, definition, steps, available, items)?;
315                Ok(())
316            }
317            serde_json::Value::Object(object) => object.values().try_for_each(|child| {
318                walk_template(owner, child, definition, steps, available, items)
319            }),
320            serde_json::Value::Array(array) => array.iter().try_for_each(|child| {
321                walk_template(owner, child, definition, steps, available, items)
322            }),
323            _ => Ok(()),
324        }
325    }
326
327    fn walk_plan(
328        plan: &WorkflowPlan,
329        definition: &WorkflowRunDefinition,
330        steps: &HashMap<String, WorkflowStepDefinition>,
331        available: &HashSet<String>,
332        items: &ItemSchemas,
333    ) -> Result<HashSet<String>, WorkflowCompileError> {
334        match plan {
335            WorkflowPlan::Step { step } => {
336                let definition_step = &steps[step];
337                let template = match &definition_step.kind {
338                    WorkflowStepKind::Tool { args, .. }
339                    | WorkflowStepKind::Workflow { args, .. } => args,
340                    WorkflowStepKind::Agent { prompt, .. } => prompt,
341                };
342                walk_template(step, template, definition, steps, available, items)?;
343                let mut result = available.clone();
344                result.insert(step.clone());
345                Ok(result)
346            }
347            WorkflowPlan::Sequence { nodes } => {
348                nodes.iter().try_fold(available.clone(), |available, node| {
349                    walk_plan(node, definition, steps, &available, items)
350                })
351            }
352            WorkflowPlan::Parallel { nodes } => {
353                let mut result = available.clone();
354                for node in nodes {
355                    // Every sibling receives the same pre-parallel availability;
356                    // no sibling may consume another sibling's output.
357                    result.extend(walk_plan(node, definition, steps, available, items)?);
358                }
359                Ok(result)
360            }
361            WorkflowPlan::Choice {
362                condition,
363                then_branch,
364                else_branch,
365            } => {
366                let schema =
367                    reference_schema("<choice>", condition, definition, steps, available, items)?;
368                if schema.get("type").and_then(serde_json::Value::as_str) != Some("boolean") {
369                    return Err(WorkflowCompileError::InvalidStep {
370                        step: "<choice>".to_string(),
371                        message: "choice condition schema must declare type boolean".to_string(),
372                    });
373                }
374                // Both paths receive the same incoming bindings; expose only
375                // values available on both paths after the choice.
376                let then_available = walk_plan(then_branch, definition, steps, available, items)?;
377                let else_available = walk_plan(else_branch, definition, steps, available, items)?;
378                Ok(then_available
379                    .intersection(&else_available)
380                    .cloned()
381                    .collect())
382            }
383            WorkflowPlan::Map { source, item, body } => {
384                let item_schema =
385                    reference_schema("<map>", source, definition, steps, available, items)?
386                        .get("items")
387                        .cloned()
388                        .ok_or_else(|| WorkflowCompileError::InvalidStep {
389                            step: "<map>".to_string(),
390                            message: "map source schema must declare array items".to_string(),
391                        })?;
392                let mut nested_items = items.clone();
393                nested_items.insert(item.clone(), item_schema);
394                // Map body step outputs are materialized under per-item scoped
395                // runtime ids (`step@map[index]`), not as root-level step ids.
396                // Validate the body, but do not leak its availability to nodes
397                // that execute after the map.
398                walk_plan(body, definition, steps, available, &nested_items)?;
399                Ok(available.clone())
400            }
401            WorkflowPlan::Retry { node, .. } => {
402                walk_plan(node, definition, steps, available, items)
403            }
404        }
405    }
406
407    walk_plan(
408        &definition.plan,
409        definition,
410        steps,
411        &HashSet::new(),
412        &HashMap::new(),
413    )?;
414    Ok(())
415}
416
417fn schema_for_literal(value: &serde_json::Value) -> serde_json::Value {
418    let kind = match value {
419        serde_json::Value::Null => "null",
420        serde_json::Value::Bool(_) => "boolean",
421        serde_json::Value::Number(number) if number.is_i64() || number.is_u64() => "integer",
422        serde_json::Value::Number(_) => "number",
423        serde_json::Value::String(_) => "string",
424        serde_json::Value::Array(_) => "array",
425        serde_json::Value::Object(_) => "object",
426    };
427    serde_json::json!({"type": kind})
428}
429
430fn schema_pointer<'a>(
431    schema: &'a serde_json::Value,
432    pointer: &str,
433) -> Option<&'a serde_json::Value> {
434    if pointer.is_empty() {
435        return Some(schema);
436    }
437    let mut current = schema;
438    for token in pointer.strip_prefix('/')?.split('/') {
439        let token = token.replace("~1", "/").replace("~0", "~");
440        current = if token.parse::<usize>().is_ok() {
441            current.get("items")?
442        } else {
443            current.get("properties")?.get(&token)?
444        };
445    }
446    Some(current)
447}
448
449fn validate_ref(
450    value_ref: &ValueRef,
451    steps: &HashMap<String, WorkflowStepDefinition>,
452) -> Result<(), WorkflowCompileError> {
453    if let ValueRef::Step { step, .. } = value_ref {
454        if !steps.contains_key(step) {
455            return Err(WorkflowCompileError::UnknownStep(step.clone()));
456        }
457    }
458    Ok(())
459}
460
461fn validate_dependencies(
462    plan: &WorkflowPlan,
463    steps: &HashMap<String, WorkflowStepDefinition>,
464) -> Result<(), WorkflowCompileError> {
465    let mut edges: HashMap<String, HashSet<String>> = HashMap::new();
466    collect_plan_edges(plan, HashSet::new(), &mut edges);
467    for (id, step) in steps {
468        collect_value_step_refs(step, |dependency| {
469            edges
470                .entry(id.clone())
471                .or_default()
472                .insert(dependency.to_string());
473        });
474    }
475    fn visit(
476        node: &str,
477        edges: &HashMap<String, HashSet<String>>,
478        visiting: &mut HashSet<String>,
479        visited: &mut HashSet<String>,
480    ) -> Result<(), WorkflowCompileError> {
481        if visited.contains(node) {
482            return Ok(());
483        }
484        if !visiting.insert(node.to_string()) {
485            return Err(WorkflowCompileError::Cycle(node.to_string()));
486        }
487        if let Some(dependencies) = edges.get(node) {
488            for dependency in dependencies {
489                visit(dependency, edges, visiting, visited)?;
490            }
491        }
492        visiting.remove(node);
493        visited.insert(node.to_string());
494        Ok(())
495    }
496    let mut visited = HashSet::new();
497    for id in steps.keys() {
498        visit(id, &edges, &mut HashSet::new(), &mut visited)?;
499    }
500    Ok(())
501}
502
503fn collect_plan_edges(
504    plan: &WorkflowPlan,
505    previous: HashSet<String>,
506    edges: &mut HashMap<String, HashSet<String>>,
507) -> HashSet<String> {
508    match plan {
509        WorkflowPlan::Step { step } => {
510            edges.entry(step.clone()).or_default().extend(previous);
511            HashSet::from([step.clone()])
512        }
513        WorkflowPlan::Sequence { nodes } => nodes.iter().fold(previous, |prior, node| {
514            collect_plan_edges(node, prior, edges)
515        }),
516        WorkflowPlan::Parallel { nodes } => {
517            let mut tails = HashSet::new();
518            for node in nodes {
519                tails.extend(collect_plan_edges(node, previous.clone(), edges));
520            }
521            tails
522        }
523        WorkflowPlan::Choice {
524            condition,
525            then_branch,
526            else_branch,
527        } => {
528            let mut incoming = previous;
529            if let ValueRef::Step { step, .. } = condition {
530                incoming.insert(step.clone());
531            }
532            let mut tails = collect_plan_edges(then_branch, incoming.clone(), edges);
533            tails.extend(collect_plan_edges(else_branch, incoming, edges));
534            tails
535        }
536        WorkflowPlan::Map { body, .. } | WorkflowPlan::Retry { node: body, .. } => {
537            collect_plan_edges(body, previous, edges)
538        }
539    }
540}
541
542fn collect_value_step_refs(step: &WorkflowStepDefinition, mut found: impl FnMut(&str)) {
543    let value = match &step.kind {
544        WorkflowStepKind::Tool { args, .. } | WorkflowStepKind::Workflow { args, .. } => args,
545        WorkflowStepKind::Agent { prompt, .. } => prompt,
546    };
547    fn walk(value: &serde_json::Value, found: &mut impl FnMut(&str)) {
548        match value {
549            serde_json::Value::Object(object) => {
550                if object.get("from").and_then(serde_json::Value::as_str) == Some("step") {
551                    if let Some(step) = object.get("step").and_then(serde_json::Value::as_str) {
552                        found(step);
553                    }
554                }
555                for child in object.values() {
556                    walk(child, found);
557                }
558            }
559            serde_json::Value::Array(array) => {
560                for child in array {
561                    walk(child, found);
562                }
563            }
564            _ => {}
565        }
566    }
567    walk(value, &mut found);
568}
569
570#[cfg(test)]
571mod choice_tests {
572    use super::*;
573    use serde_json::{json, Value};
574
575    fn step(id: &str, args: Value) -> WorkflowStepDefinition {
576        WorkflowStepDefinition {
577            id: id.to_string(),
578            kind: WorkflowStepKind::Tool {
579                tool: "echo".to_string(),
580                args,
581                capabilities: Vec::new(),
582            },
583            failure: Default::default(),
584            output_schema: Some(json!({"type":"boolean"})),
585        }
586    }
587
588    fn choice(condition: Value, then_branch: Value, else_branch: Value) -> Value {
589        json!({"type":"choice","condition":condition,"then_branch":then_branch,"else_branch":else_branch})
590    }
591
592    fn leaf(id: &str) -> Value {
593        json!({"type":"step","step":id})
594    }
595
596    fn definition(plan: Value, steps: Vec<WorkflowStepDefinition>) -> WorkflowRunDefinition {
597        WorkflowRunDefinition {
598            workflow_schema: 1,
599            id: "conditional".to_string(),
600            revision: 1,
601            input_schema: json!({"type":"object","properties":{"approved":{"type":"boolean"},"rows":{"type":"array","items":{"type":"boolean"}}},"additionalProperties":false}),
602            output_schema: None,
603            steps,
604            plan: serde_json::from_value(plan).expect("Choice plan loads"),
605            budgets: Default::default(),
606        }
607    }
608
609    #[test]
610    fn workflow_choice_loads_json_yaml_and_requires_both_branches() {
611        let plan = choice(
612            json!({"from":"literal","value":true}),
613            leaf("yes"),
614            leaf("no"),
615        );
616        let json_plan: WorkflowPlan = serde_json::from_value(plan.clone()).unwrap();
617        let yaml_plan: WorkflowPlan = serde_yaml::from_str(
618            "type: choice\ncondition: {from: literal, value: true}\nthen_branch: {type: step, step: yes}\nelse_branch: {type: step, step: no}\n",
619        ).unwrap();
620        assert_eq!(json_plan, yaml_plan);
621        for missing in ["then_branch", "else_branch", "condition"] {
622            let mut incomplete = plan.clone();
623            incomplete.as_object_mut().unwrap().remove(missing);
624            assert!(serde_json::from_value::<WorkflowPlan>(incomplete).is_err());
625        }
626    }
627
628    #[test]
629    fn workflow_choice_accepts_boolean_args_step_item_and_literal() {
630        for condition in [
631            json!({"from":"args","pointer":"/approved"}),
632            json!({"from":"literal","value":false}),
633        ] {
634            CompiledWorkflow::compile(definition(
635                choice(condition, leaf("yes"), leaf("no")),
636                vec![step("yes", json!(true)), step("no", json!(false))],
637            ))
638            .unwrap();
639        }
640        let from_step = definition(
641            json!({"type":"sequence","nodes":[leaf("before"), choice(json!({"from":"step","step":"before"}), leaf("yes"), leaf("no"))]}),
642            vec![
643                step("before", json!(true)),
644                step("yes", json!(true)),
645                step("no", json!(false)),
646            ],
647        );
648        CompiledWorkflow::compile(from_step).unwrap();
649        let from_item = definition(
650            json!({"type":"map","source":{"from":"args","pointer":"/rows"},"item":"row","body":choice(json!({"from":"item","name":"row"}), leaf("yes"), leaf("no"))}),
651            vec![step("yes", json!(true)), step("no", json!(false))],
652        );
653        CompiledWorkflow::compile(from_item).unwrap();
654    }
655
656    #[test]
657    fn workflow_choice_rejects_missing_and_non_boolean_reference_schemas() {
658        for condition in [
659            json!({"from":"literal","value":"false"}),
660            json!({"from":"literal","value":0}),
661            json!({"from":"literal","value":null}),
662            json!({"from":"args","pointer":"/missing"}),
663            json!({"from":"item","name":"outside"}),
664            json!({"from":"step","step":"unknown"}),
665        ] {
666            assert!(CompiledWorkflow::compile(definition(
667                choice(condition, leaf("yes"), leaf("no")),
668                vec![step("yes", json!(true)), step("no", json!(false))],
669            ))
670            .is_err());
671        }
672        for schema in [
673            json!({}),
674            json!({"type":"string"}),
675            json!({"type":"integer"}),
676        ] {
677            let mut flow = definition(
678                choice(
679                    json!({"from":"args","pointer":"/approved"}),
680                    leaf("yes"),
681                    leaf("no"),
682                ),
683                vec![step("yes", json!(true)), step("no", json!(false))],
684            );
685            flow.input_schema["properties"]["approved"] = schema;
686            assert!(CompiledWorkflow::compile(flow).is_err());
687        }
688        let mut flow = definition(
689            json!({"type":"sequence","nodes":[leaf("before"), choice(json!({"from":"step","step":"before"}), leaf("yes"), leaf("no"))]}),
690            vec![
691                step("before", json!(true)),
692                step("yes", json!(true)),
693                step("no", json!(false)),
694            ],
695        );
696        flow.steps[0].output_schema = None;
697        assert!(CompiledWorkflow::compile(flow).is_err());
698    }
699
700    #[test]
701    fn workflow_choice_rejects_forward_sibling_and_join_output_references() {
702        let condition = json!({"from":"literal","value":true});
703        let branch = choice(condition.clone(), leaf("yes"), leaf("no"));
704        for flow in [
705            definition(
706                json!({"type":"sequence","nodes":[choice(json!({"from":"step","step":"later"}),leaf("yes"),leaf("no")),leaf("later")]}),
707                vec![
708                    step("yes", json!(true)),
709                    step("no", json!(false)),
710                    step("later", json!(true)),
711                ],
712            ),
713            definition(
714                branch.clone(),
715                vec![
716                    step("yes", json!(true)),
717                    step("no", json!({"from":"step","step":"yes"})),
718                ],
719            ),
720            definition(
721                json!({"type":"sequence","nodes":[branch,leaf("after")]}),
722                vec![
723                    step("yes", json!(true)),
724                    step("no", json!(false)),
725                    step("after", json!({"from":"step","step":"yes"})),
726                ],
727            ),
728        ] {
729            assert!(CompiledWorkflow::compile(flow).is_err());
730        }
731    }
732
733    #[test]
734    fn workflow_choice_preserves_incoming_and_branch_local_bindings() {
735        let from_before = json!({"from":"step","step":"before"});
736        let flow = definition(
737            json!({"type":"sequence","nodes":[leaf("before"),choice(from_before.clone(),json!({"type":"sequence","nodes":[leaf("yes"),leaf("yes-after")]}),leaf("no")),leaf("after")]}),
738            vec![
739                step("before", json!(true)),
740                step("yes", from_before.clone()),
741                step("yes-after", json!({"from":"step","step":"yes"})),
742                step("no", from_before.clone()),
743                step("after", from_before),
744            ],
745        );
746        CompiledWorkflow::compile(flow).unwrap();
747    }
748
749    #[test]
750    fn workflow_choice_rejects_duplicate_steps_and_invalid_unselected_plan() {
751        let duplicate = definition(
752            choice(
753                json!({"from":"literal","value":true}),
754                leaf("same"),
755                leaf("same"),
756            ),
757            vec![step("same", json!(true))],
758        );
759        assert!(CompiledWorkflow::compile(duplicate).is_err());
760        let empty_branch = definition(
761            choice(
762                json!({"from":"literal","value":true}),
763                leaf("yes"),
764                json!({"type":"sequence","nodes":[]}),
765            ),
766            vec![step("yes", json!(true))],
767        );
768        assert!(CompiledWorkflow::compile(empty_branch).is_err());
769    }
770}
bamboo_domain/workflow/compiler.rs

bamboo_domain/workflow/
compiler.rs
