use std::collections::{HashMap, HashSet};
use crate::{
ValueRef, WorkflowPlan, WorkflowRunDefinition, WorkflowSecretHandle, WorkflowStepDefinition,
WorkflowStepKind,
};
use thiserror::Error;
use super::schema::{validate_schema, validate_schema_shape as validate_shape};
#[derive(Debug, Error, Clone, PartialEq, Eq)]
pub enum WorkflowCompileError {
#[error("unsupported workflow_schema {0}")]
UnsupportedSchema(u32),
#[error("workflow id and revision are required")]
MissingIdentity,
#[error("invalid workflow budget: {0}")]
InvalidBudget(String),
#[error("duplicate or empty step id: {0}")]
DuplicateStep(String),
#[error("unknown step reference: {0}")]
UnknownStep(String),
#[error("invalid step {step}: {message}")]
InvalidStep { step: String, message: String },
#[error("cyclic step dependency: {0}")]
Cycle(String),
#[error("invalid schema: {0}")]
InvalidSchema(String),
}
#[derive(Debug, Clone)]
pub struct CompiledWorkflow {
pub definition: WorkflowRunDefinition,
pub steps: HashMap<String, WorkflowStepDefinition>,
}
impl CompiledWorkflow {
pub fn compile(definition: WorkflowRunDefinition) -> Result<Self, WorkflowCompileError> {
if definition.workflow_schema != 1 {
return Err(WorkflowCompileError::UnsupportedSchema(
definition.workflow_schema,
));
}
if definition.id.trim().is_empty() || definition.revision == 0 {
return Err(WorkflowCompileError::MissingIdentity);
}
validate_budget(&definition)?;
validate_schema_shape(&definition.input_schema)?;
if let Some(schema) = &definition.output_schema {
validate_schema_shape(schema)?;
}
let mut steps = HashMap::new();
for step in &definition.steps {
if step.id.trim().is_empty() || steps.insert(step.id.clone(), step.clone()).is_some() {
return Err(WorkflowCompileError::DuplicateStep(step.id.clone()));
}
validate_step(step)?;
}
if steps.is_empty() {
return Err(WorkflowCompileError::InvalidStep {
step: "<definition>".to_string(),
message: "at least one step is required".to_string(),
});
}
validate_plan(&definition.plan, &steps)?;
validate_dependencies(&definition.plan, &steps)?;
validate_execution_bindings(&definition, &steps)?;
Ok(Self { definition, steps })
}
pub fn validate_input(&self, value: &serde_json::Value) -> Result<(), String> {
validate_schema(&self.definition.input_schema, value)
}
}
fn validate_schema_shape(schema: &serde_json::Value) -> Result<(), WorkflowCompileError> {
validate_shape(schema, "$").map_err(WorkflowCompileError::InvalidSchema)
}
fn validate_budget(definition: &WorkflowRunDefinition) -> Result<(), WorkflowCompileError> {
let budget = &definition.budgets;
if budget.max_concurrency == 0 || budget.max_concurrency > 64 {
return Err(WorkflowCompileError::InvalidBudget(
"max_concurrency must be 1..=64".to_string(),
));
}
if budget.max_steps == 0
|| budget.max_agents > budget.max_steps
|| budget.max_nesting_depth == 0
|| budget.wall_time_ms == 0
{
return Err(WorkflowCompileError::InvalidBudget(
"step/agent/depth/wall budgets must be bounded and coherent".to_string(),
));
}
Ok(())
}
fn validate_step(step: &WorkflowStepDefinition) -> Result<(), WorkflowCompileError> {
let required = match &step.kind {
WorkflowStepKind::Tool { tool, .. } => tool,
WorkflowStepKind::Agent {
agent,
structured_output_attempts,
..
} => {
if *structured_output_attempts == 0 || *structured_output_attempts > 8 {
return Err(WorkflowCompileError::InvalidStep {
step: step.id.clone(),
message: "structured_output_attempts must be 1..=8".to_string(),
});
}
agent
}
WorkflowStepKind::Workflow {
workflow_id,
revision,
..
} => {
if *revision == 0 {
return Err(WorkflowCompileError::InvalidStep {
step: step.id.clone(),
message: "nested workflow revision must be fixed and non-zero".to_string(),
});
}
workflow_id
}
};
if required.trim().is_empty() {
return Err(WorkflowCompileError::InvalidStep {
step: step.id.clone(),
message: "target cannot be empty".to_string(),
});
}
if let Some(schema) = &step.output_schema {
validate_schema_shape(schema)?;
}
Ok(())
}
fn validate_plan(
plan: &WorkflowPlan,
steps: &HashMap<String, WorkflowStepDefinition>,
) -> Result<(), WorkflowCompileError> {
let mut seen = HashSet::new();
validate_plan_inner(plan, steps, &mut seen)?;
let unreferenced = steps
.keys()
.filter(|step| !seen.contains(*step))
.cloned()
.collect::<Vec<_>>();
if !unreferenced.is_empty() {
return Err(WorkflowCompileError::InvalidStep {
step: "<plan>".to_string(),
message: format!("unreferenced steps: {}", unreferenced.join(", ")),
});
}
Ok(())
}
fn validate_plan_inner(
plan: &WorkflowPlan,
steps: &HashMap<String, WorkflowStepDefinition>,
seen: &mut HashSet<String>,
) -> Result<(), WorkflowCompileError> {
match plan {
WorkflowPlan::Step { step } => {
if !steps.contains_key(step) {
return Err(WorkflowCompileError::UnknownStep(step.clone()));
}
if !seen.insert(step.clone()) {
return Err(WorkflowCompileError::InvalidStep {
step: step.clone(),
message: "step appears more than once in the execution plan".to_string(),
});
}
}
WorkflowPlan::Sequence { nodes } | WorkflowPlan::Parallel { nodes } => {
if nodes.is_empty() {
return Err(WorkflowCompileError::InvalidStep {
step: "<plan>".to_string(),
message: "sequence/parallel requires nodes".to_string(),
});
}
for node in nodes {
validate_plan_inner(node, steps, seen)?;
}
}
WorkflowPlan::Choice {
condition,
then_branch,
else_branch,
} => {
validate_ref(condition, steps)?;
validate_plan_inner(then_branch, steps, seen)?;
validate_plan_inner(else_branch, steps, seen)?;
}
WorkflowPlan::Map { source, item, body } => {
if item.trim().is_empty() {
return Err(WorkflowCompileError::InvalidStep {
step: "<map>".to_string(),
message: "map item name cannot be empty".to_string(),
});
}
validate_ref(source, steps)?;
validate_plan_inner(body, steps, seen)?;
}
WorkflowPlan::Retry {
node, max_attempts, ..
} => {
if *max_attempts == 0 {
return Err(WorkflowCompileError::InvalidStep {
step: "<retry>".to_string(),
message: "max_attempts must be greater than zero".to_string(),
});
}
validate_plan_inner(node, steps, seen)?;
}
}
Ok(())
}
fn validate_execution_bindings(
definition: &WorkflowRunDefinition,
steps: &HashMap<String, WorkflowStepDefinition>,
) -> Result<(), WorkflowCompileError> {
type ItemSchemas = HashMap<String, serde_json::Value>;
fn reference_schema(
owner: &str,
reference: &ValueRef,
definition: &WorkflowRunDefinition,
steps: &HashMap<String, WorkflowStepDefinition>,
available: &HashSet<String>,
items: &ItemSchemas,
) -> Result<serde_json::Value, WorkflowCompileError> {
let invalid = |message: String| WorkflowCompileError::InvalidStep {
step: owner.to_string(),
message,
};
match reference {
ValueRef::Args { pointer } => schema_pointer(&definition.input_schema, pointer)
.cloned()
.ok_or_else(|| invalid("args reference has an invalid schema pointer".to_string())),
ValueRef::Step { step, pointer } => {
if !available.contains(step) {
return Err(invalid(format!(
"step reference '{step}' is not available before this step"
)));
}
let schema = steps
.get(step)
.and_then(|step| step.output_schema.as_ref())
.ok_or_else(|| {
invalid(format!(
"step reference '{step}' requires a declared output_schema"
))
})?;
schema_pointer(schema, pointer).cloned().ok_or_else(|| {
invalid(format!(
"step reference '{step}' has an invalid schema pointer"
))
})
}
ValueRef::Item { name, pointer } => {
let schema = items.get(name).ok_or_else(|| {
invalid(format!(
"map item reference '{name}' is outside its map body"
))
})?;
schema_pointer(schema, pointer).cloned().ok_or_else(|| {
invalid(format!(
"map item reference '{name}' has an invalid schema pointer"
))
})
}
ValueRef::Literal { value } => Ok(schema_for_literal(value)),
}
}
fn walk_template(
owner: &str,
value: &serde_json::Value,
definition: &WorkflowRunDefinition,
steps: &HashMap<String, WorkflowStepDefinition>,
available: &HashSet<String>,
items: &ItemSchemas,
) -> Result<(), WorkflowCompileError> {
match value {
serde_json::Value::Object(object) if object.contains_key("$secret") => {
let handle: WorkflowSecretHandle =
serde_json::from_value(value.clone()).map_err(|error| {
WorkflowCompileError::InvalidStep {
step: owner.to_string(),
message: format!("malformed secret capability handle: {error}"),
}
})?;
if handle.capability.trim().is_empty() {
return Err(WorkflowCompileError::InvalidStep {
step: owner.to_string(),
message: "secret capability handle cannot be empty".to_string(),
});
}
Ok(())
}
serde_json::Value::Object(object) if object.contains_key("from") => {
let reference: ValueRef =
serde_json::from_value(value.clone()).map_err(|error| {
WorkflowCompileError::InvalidStep {
step: owner.to_string(),
message: format!("malformed value reference: {error}"),
}
})?;
reference_schema(owner, &reference, definition, steps, available, items)?;
Ok(())
}
serde_json::Value::Object(object) => object.values().try_for_each(|child| {
walk_template(owner, child, definition, steps, available, items)
}),
serde_json::Value::Array(array) => array.iter().try_for_each(|child| {
walk_template(owner, child, definition, steps, available, items)
}),
_ => Ok(()),
}
}
fn walk_plan(
plan: &WorkflowPlan,
definition: &WorkflowRunDefinition,
steps: &HashMap<String, WorkflowStepDefinition>,
available: &HashSet<String>,
items: &ItemSchemas,
) -> Result<HashSet<String>, WorkflowCompileError> {
match plan {
WorkflowPlan::Step { step } => {
let definition_step = &steps[step];
let template = match &definition_step.kind {
WorkflowStepKind::Tool { args, .. }
| WorkflowStepKind::Workflow { args, .. } => args,
WorkflowStepKind::Agent { prompt, .. } => prompt,
};
walk_template(step, template, definition, steps, available, items)?;
let mut result = available.clone();
result.insert(step.clone());
Ok(result)
}
WorkflowPlan::Sequence { nodes } => {
nodes.iter().try_fold(available.clone(), |available, node| {
walk_plan(node, definition, steps, &available, items)
})
}
WorkflowPlan::Parallel { nodes } => {
let mut result = available.clone();
for node in nodes {
result.extend(walk_plan(node, definition, steps, available, items)?);
}
Ok(result)
}
WorkflowPlan::Choice {
condition,
then_branch,
else_branch,
} => {
let schema =
reference_schema("<choice>", condition, definition, steps, available, items)?;
if schema.get("type").and_then(serde_json::Value::as_str) != Some("boolean") {
return Err(WorkflowCompileError::InvalidStep {
step: "<choice>".to_string(),
message: "choice condition schema must declare type boolean".to_string(),
});
}
let then_available = walk_plan(then_branch, definition, steps, available, items)?;
let else_available = walk_plan(else_branch, definition, steps, available, items)?;
Ok(then_available
.intersection(&else_available)
.cloned()
.collect())
}
WorkflowPlan::Map { source, item, body } => {
let item_schema =
reference_schema("<map>", source, definition, steps, available, items)?
.get("items")
.cloned()
.ok_or_else(|| WorkflowCompileError::InvalidStep {
step: "<map>".to_string(),
message: "map source schema must declare array items".to_string(),
})?;
let mut nested_items = items.clone();
nested_items.insert(item.clone(), item_schema);
walk_plan(body, definition, steps, available, &nested_items)?;
Ok(available.clone())
}
WorkflowPlan::Retry { node, .. } => {
walk_plan(node, definition, steps, available, items)
}
}
}
walk_plan(
&definition.plan,
definition,
steps,
&HashSet::new(),
&HashMap::new(),
)?;
Ok(())
}
fn schema_for_literal(value: &serde_json::Value) -> serde_json::Value {
let kind = match value {
serde_json::Value::Null => "null",
serde_json::Value::Bool(_) => "boolean",
serde_json::Value::Number(number) if number.is_i64() || number.is_u64() => "integer",
serde_json::Value::Number(_) => "number",
serde_json::Value::String(_) => "string",
serde_json::Value::Array(_) => "array",
serde_json::Value::Object(_) => "object",
};
serde_json::json!({"type": kind})
}
fn schema_pointer<'a>(
schema: &'a serde_json::Value,
pointer: &str,
) -> Option<&'a serde_json::Value> {
if pointer.is_empty() {
return Some(schema);
}
let mut current = schema;
for token in pointer.strip_prefix('/')?.split('/') {
let token = token.replace("~1", "/").replace("~0", "~");
current = if token.parse::<usize>().is_ok() {
current.get("items")?
} else {
current.get("properties")?.get(&token)?
};
}
Some(current)
}
fn validate_ref(
value_ref: &ValueRef,
steps: &HashMap<String, WorkflowStepDefinition>,
) -> Result<(), WorkflowCompileError> {
if let ValueRef::Step { step, .. } = value_ref {
if !steps.contains_key(step) {
return Err(WorkflowCompileError::UnknownStep(step.clone()));
}
}
Ok(())
}
fn validate_dependencies(
plan: &WorkflowPlan,
steps: &HashMap<String, WorkflowStepDefinition>,
) -> Result<(), WorkflowCompileError> {
let mut edges: HashMap<String, HashSet<String>> = HashMap::new();
collect_plan_edges(plan, HashSet::new(), &mut edges);
for (id, step) in steps {
collect_value_step_refs(step, |dependency| {
edges
.entry(id.clone())
.or_default()
.insert(dependency.to_string());
});
}
fn visit(
node: &str,
edges: &HashMap<String, HashSet<String>>,
visiting: &mut HashSet<String>,
visited: &mut HashSet<String>,
) -> Result<(), WorkflowCompileError> {
if visited.contains(node) {
return Ok(());
}
if !visiting.insert(node.to_string()) {
return Err(WorkflowCompileError::Cycle(node.to_string()));
}
if let Some(dependencies) = edges.get(node) {
for dependency in dependencies {
visit(dependency, edges, visiting, visited)?;
}
}
visiting.remove(node);
visited.insert(node.to_string());
Ok(())
}
let mut visited = HashSet::new();
for id in steps.keys() {
visit(id, &edges, &mut HashSet::new(), &mut visited)?;
}
Ok(())
}
fn collect_plan_edges(
plan: &WorkflowPlan,
previous: HashSet<String>,
edges: &mut HashMap<String, HashSet<String>>,
) -> HashSet<String> {
match plan {
WorkflowPlan::Step { step } => {
edges.entry(step.clone()).or_default().extend(previous);
HashSet::from([step.clone()])
}
WorkflowPlan::Sequence { nodes } => nodes.iter().fold(previous, |prior, node| {
collect_plan_edges(node, prior, edges)
}),
WorkflowPlan::Parallel { nodes } => {
let mut tails = HashSet::new();
for node in nodes {
tails.extend(collect_plan_edges(node, previous.clone(), edges));
}
tails
}
WorkflowPlan::Choice {
condition,
then_branch,
else_branch,
} => {
let mut incoming = previous;
if let ValueRef::Step { step, .. } = condition {
incoming.insert(step.clone());
}
let mut tails = collect_plan_edges(then_branch, incoming.clone(), edges);
tails.extend(collect_plan_edges(else_branch, incoming, edges));
tails
}
WorkflowPlan::Map { body, .. } | WorkflowPlan::Retry { node: body, .. } => {
collect_plan_edges(body, previous, edges)
}
}
}
fn collect_value_step_refs(step: &WorkflowStepDefinition, mut found: impl FnMut(&str)) {
let value = match &step.kind {
WorkflowStepKind::Tool { args, .. } | WorkflowStepKind::Workflow { args, .. } => args,
WorkflowStepKind::Agent { prompt, .. } => prompt,
};
fn walk(value: &serde_json::Value, found: &mut impl FnMut(&str)) {
match value {
serde_json::Value::Object(object) => {
if object.get("from").and_then(serde_json::Value::as_str) == Some("step") {
if let Some(step) = object.get("step").and_then(serde_json::Value::as_str) {
found(step);
}
}
for child in object.values() {
walk(child, found);
}
}
serde_json::Value::Array(array) => {
for child in array {
walk(child, found);
}
}
_ => {}
}
}
walk(value, &mut found);
}
#[cfg(test)]
mod choice_tests {
use super::*;
use serde_json::{json, Value};
fn step(id: &str, args: Value) -> WorkflowStepDefinition {
WorkflowStepDefinition {
id: id.to_string(),
kind: WorkflowStepKind::Tool {
tool: "echo".to_string(),
args,
capabilities: Vec::new(),
},
failure: Default::default(),
output_schema: Some(json!({"type":"boolean"})),
}
}
fn choice(condition: Value, then_branch: Value, else_branch: Value) -> Value {
json!({"type":"choice","condition":condition,"then_branch":then_branch,"else_branch":else_branch})
}
fn leaf(id: &str) -> Value {
json!({"type":"step","step":id})
}
fn definition(plan: Value, steps: Vec<WorkflowStepDefinition>) -> WorkflowRunDefinition {
WorkflowRunDefinition {
workflow_schema: 1,
id: "conditional".to_string(),
revision: 1,
input_schema: json!({"type":"object","properties":{"approved":{"type":"boolean"},"rows":{"type":"array","items":{"type":"boolean"}}},"additionalProperties":false}),
output_schema: None,
steps,
plan: serde_json::from_value(plan).expect("Choice plan loads"),
budgets: Default::default(),
}
}
#[test]
fn workflow_choice_loads_json_yaml_and_requires_both_branches() {
let plan = choice(
json!({"from":"literal","value":true}),
leaf("yes"),
leaf("no"),
);
let json_plan: WorkflowPlan = serde_json::from_value(plan.clone()).unwrap();
let yaml_plan: WorkflowPlan = serde_yaml::from_str(
"type: choice\ncondition: {from: literal, value: true}\nthen_branch: {type: step, step: yes}\nelse_branch: {type: step, step: no}\n",
).unwrap();
assert_eq!(json_plan, yaml_plan);
for missing in ["then_branch", "else_branch", "condition"] {
let mut incomplete = plan.clone();
incomplete.as_object_mut().unwrap().remove(missing);
assert!(serde_json::from_value::<WorkflowPlan>(incomplete).is_err());
}
}
#[test]
fn workflow_choice_accepts_boolean_args_step_item_and_literal() {
for condition in [
json!({"from":"args","pointer":"/approved"}),
json!({"from":"literal","value":false}),
] {
CompiledWorkflow::compile(definition(
choice(condition, leaf("yes"), leaf("no")),
vec![step("yes", json!(true)), step("no", json!(false))],
))
.unwrap();
}
let from_step = definition(
json!({"type":"sequence","nodes":[leaf("before"), choice(json!({"from":"step","step":"before"}), leaf("yes"), leaf("no"))]}),
vec![
step("before", json!(true)),
step("yes", json!(true)),
step("no", json!(false)),
],
);
CompiledWorkflow::compile(from_step).unwrap();
let from_item = definition(
json!({"type":"map","source":{"from":"args","pointer":"/rows"},"item":"row","body":choice(json!({"from":"item","name":"row"}), leaf("yes"), leaf("no"))}),
vec![step("yes", json!(true)), step("no", json!(false))],
);
CompiledWorkflow::compile(from_item).unwrap();
}
#[test]
fn workflow_choice_rejects_missing_and_non_boolean_reference_schemas() {
for condition in [
json!({"from":"literal","value":"false"}),
json!({"from":"literal","value":0}),
json!({"from":"literal","value":null}),
json!({"from":"args","pointer":"/missing"}),
json!({"from":"item","name":"outside"}),
json!({"from":"step","step":"unknown"}),
] {
assert!(CompiledWorkflow::compile(definition(
choice(condition, leaf("yes"), leaf("no")),
vec![step("yes", json!(true)), step("no", json!(false))],
))
.is_err());
}
for schema in [
json!({}),
json!({"type":"string"}),
json!({"type":"integer"}),
] {
let mut flow = definition(
choice(
json!({"from":"args","pointer":"/approved"}),
leaf("yes"),
leaf("no"),
),
vec![step("yes", json!(true)), step("no", json!(false))],
);
flow.input_schema["properties"]["approved"] = schema;
assert!(CompiledWorkflow::compile(flow).is_err());
}
let mut flow = definition(
json!({"type":"sequence","nodes":[leaf("before"), choice(json!({"from":"step","step":"before"}), leaf("yes"), leaf("no"))]}),
vec![
step("before", json!(true)),
step("yes", json!(true)),
step("no", json!(false)),
],
);
flow.steps[0].output_schema = None;
assert!(CompiledWorkflow::compile(flow).is_err());
}
#[test]
fn workflow_choice_rejects_forward_sibling_and_join_output_references() {
let condition = json!({"from":"literal","value":true});
let branch = choice(condition.clone(), leaf("yes"), leaf("no"));
for flow in [
definition(
json!({"type":"sequence","nodes":[choice(json!({"from":"step","step":"later"}),leaf("yes"),leaf("no")),leaf("later")]}),
vec![
step("yes", json!(true)),
step("no", json!(false)),
step("later", json!(true)),
],
),
definition(
branch.clone(),
vec![
step("yes", json!(true)),
step("no", json!({"from":"step","step":"yes"})),
],
),
definition(
json!({"type":"sequence","nodes":[branch,leaf("after")]}),
vec![
step("yes", json!(true)),
step("no", json!(false)),
step("after", json!({"from":"step","step":"yes"})),
],
),
] {
assert!(CompiledWorkflow::compile(flow).is_err());
}
}
#[test]
fn workflow_choice_preserves_incoming_and_branch_local_bindings() {
let from_before = json!({"from":"step","step":"before"});
let flow = definition(
json!({"type":"sequence","nodes":[leaf("before"),choice(from_before.clone(),json!({"type":"sequence","nodes":[leaf("yes"),leaf("yes-after")]}),leaf("no")),leaf("after")]}),
vec![
step("before", json!(true)),
step("yes", from_before.clone()),
step("yes-after", json!({"from":"step","step":"yes"})),
step("no", from_before.clone()),
step("after", from_before),
],
);
CompiledWorkflow::compile(flow).unwrap();
}
#[test]
fn workflow_choice_rejects_duplicate_steps_and_invalid_unselected_plan() {
let duplicate = definition(
choice(
json!({"from":"literal","value":true}),
leaf("same"),
leaf("same"),
),
vec![step("same", json!(true))],
);
assert!(CompiledWorkflow::compile(duplicate).is_err());
let empty_branch = definition(
choice(
json!({"from":"literal","value":true}),
leaf("yes"),
json!({"type":"sequence","nodes":[]}),
),
vec![step("yes", json!(true))],
);
assert!(CompiledWorkflow::compile(empty_branch).is_err());
}
}