use std::collections::{BTreeMap, BTreeSet};
use anyhow::{bail, Result};
use serde::{Deserialize, Serialize};
pub const WORKFLOW_SCHEMA_VERSION: u32 = 1;
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct WorkflowDefinition {
pub version: u32,
pub id: String,
pub title: String,
#[serde(default)]
pub inputs: BTreeMap<String, String>,
#[serde(default)]
pub nodes: Vec<WorkflowNode>,
#[serde(default)]
pub edges: Vec<WorkflowEdge>,
#[serde(default)]
pub max_iterations: Option<u32>,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct WorkflowNode {
pub id: String,
pub action: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stage: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub skill: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub command: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct WorkflowEdge {
pub from: String,
pub to: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub condition: Option<String>,
}
#[derive(Clone, Debug, Default, Deserialize, Serialize)]
pub struct WorkflowRunReport {
pub workflow_id: String,
pub status: String,
pub iterations: u32,
#[serde(default)]
pub events: Vec<WorkflowRunEvent>,
#[serde(default)]
pub produced: Vec<(String, String)>,
#[serde(default)]
pub paused: Vec<String>,
#[serde(default)]
pub ready: Vec<String>,
#[serde(default)]
pub errors: Vec<String>,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct WorkflowRunEvent {
pub node: String,
pub action: String,
pub status: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum WorkflowActionOutcome {
Continue,
Paused(String),
Failed(String),
}
pub trait WorkflowActionExecutor {
fn execute(
&mut self,
workflow: &WorkflowDefinition,
node: &WorkflowNode,
) -> WorkflowActionOutcome;
}
pub fn parse_workflow(text: &str) -> Result<WorkflowDefinition> {
let workflow: WorkflowDefinition = serde_yaml::from_str(text)?;
validate_workflow(&workflow)?;
Ok(workflow)
}
pub fn validate_workflow(workflow: &WorkflowDefinition) -> Result<()> {
if workflow.version != WORKFLOW_SCHEMA_VERSION {
bail!(
"workflow {} usa version {} (esperado {})",
workflow.id,
workflow.version,
WORKFLOW_SCHEMA_VERSION
);
}
validate_id("workflow", &workflow.id)?;
if workflow.title.trim().is_empty() {
bail!("workflow {} sem title", workflow.id);
}
if workflow.nodes.is_empty() {
bail!("workflow {} precisa declarar nodes", workflow.id);
}
let allowed_actions: BTreeSet<&str> = [
"stage",
"skill",
"eval",
"checkpoint",
"parallel",
"command",
"notify",
]
.into_iter()
.collect();
let mut node_ids = BTreeSet::new();
let mut order = BTreeMap::new();
for (index, node) in workflow.nodes.iter().enumerate() {
validate_id("node", &node.id)?;
if !node_ids.insert(node.id.as_str()) {
bail!("workflow {} tem node duplicado: {}", workflow.id, node.id);
}
if !allowed_actions.contains(node.action.as_str()) {
bail!(
"workflow {} node {} usa action desconhecida: {}",
workflow.id,
node.id,
node.action
);
}
match node.action.as_str() {
"stage" if node.stage.as_deref().unwrap_or_default().trim().is_empty() => {
bail!("workflow {} node {} stage sem stage", workflow.id, node.id);
}
"skill" if node.skill.as_deref().unwrap_or_default().trim().is_empty() => {
bail!("workflow {} node {} skill sem skill", workflow.id, node.id);
}
"command"
if node
.command
.as_deref()
.unwrap_or_default()
.trim()
.is_empty() =>
{
bail!(
"workflow {} node {} command sem command",
workflow.id,
node.id
);
}
"checkpoint" => validate_checkpoint_node(workflow, node)?,
_ => {}
}
order.insert(node.id.as_str(), index);
}
let mut outgoing = BTreeSet::new();
for edge in &workflow.edges {
if !node_ids.contains(edge.from.as_str()) {
bail!(
"workflow {} edge referencia from inexistente: {}",
workflow.id,
edge.from
);
}
if !node_ids.contains(edge.to.as_str()) {
bail!(
"workflow {} edge referencia to inexistente: {}",
workflow.id,
edge.to
);
}
if let Some(condition) = edge.condition.as_deref() {
validate_condition(workflow, edge, condition)?;
}
let from = order[edge.from.as_str()];
let to = order[edge.to.as_str()];
if to <= from && workflow.max_iterations.unwrap_or(0) == 0 {
bail!(
"workflow {} tem loop {} -> {} sem max_iterations",
workflow.id,
edge.from,
edge.to
);
}
outgoing.insert(edge.from.as_str());
}
for node in workflow
.nodes
.iter()
.take(workflow.nodes.len().saturating_sub(1))
{
if !outgoing.contains(node.id.as_str()) {
bail!("workflow {} node {} sem destino", workflow.id, node.id);
}
}
Ok(())
}
fn validate_condition(
workflow: &WorkflowDefinition,
edge: &WorkflowEdge,
condition: &str,
) -> Result<()> {
let normalized = condition.trim().to_ascii_lowercase();
if matches!(
normalized.as_str(),
"ok" | "success" | "passed" | "failed" | "error" | "retry" | "always"
) {
return Ok(());
}
bail!(
"workflow {} edge {} -> {} usa condition desconhecida: {}",
workflow.id,
edge.from,
edge.to,
condition
)
}
fn validate_checkpoint_node(workflow: &WorkflowDefinition, node: &WorkflowNode) -> Result<()> {
let stage = node.stage.as_deref().unwrap_or_default();
if !matches!(
stage,
"prd" | "techspec" | "refinement" | "review" | "deploy"
) {
bail!(
"workflow {} checkpoint {} precisa stage prd|techspec|refinement|review|deploy",
workflow.id,
node.id
);
}
if node
.command
.as_deref()
.unwrap_or_default()
.contains("approve")
{
bail!(
"workflow {} checkpoint {} não pode autoaprovar gate humano",
workflow.id,
node.id
);
}
Ok(())
}
fn validate_id(label: &str, value: &str) -> Result<()> {
if value.trim().is_empty() {
bail!("{label} id vazio");
}
if !value
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
{
bail!("{label} id inválido: {value}");
}
Ok(())
}
pub fn run_workflow(
workflow: &WorkflowDefinition,
max_iterations_override: Option<u32>,
executor: &mut dyn WorkflowActionExecutor,
) -> Result<WorkflowRunReport> {
validate_workflow(workflow)?;
let loop_limit = max_iterations_override
.or(workflow.max_iterations)
.map(|value| value.max(1));
let step_limit = loop_limit
.map(|limit| {
(workflow.nodes.len() as u32)
.saturating_mul(limit.saturating_add(1))
.saturating_add(1)
})
.unwrap_or_else(|| workflow.nodes.len().saturating_add(1) as u32);
let by_id = workflow
.nodes
.iter()
.map(|node| (node.id.as_str(), node))
.collect::<BTreeMap<_, _>>();
let order = workflow
.nodes
.iter()
.enumerate()
.map(|(index, node)| (node.id.as_str(), index))
.collect::<BTreeMap<_, _>>();
let mut current = workflow.nodes[0].id.clone();
let mut report = WorkflowRunReport {
workflow_id: workflow.id.clone(),
status: "running".to_string(),
..Default::default()
};
for _step in 0..step_limit {
let Some(node) = by_id.get(current.as_str()) else {
bail!("workflow {} cursor inválido: {current}", workflow.id);
};
let outcome = executor.execute(workflow, node);
let outcome_status = match outcome {
WorkflowActionOutcome::Continue => {
report.events.push(WorkflowRunEvent {
node: node.id.clone(),
action: node.action.clone(),
status: "ok".to_string(),
message: None,
});
WorkflowStepStatus::Ok
}
WorkflowActionOutcome::Paused(message) => {
report.events.push(WorkflowRunEvent {
node: node.id.clone(),
action: node.action.clone(),
status: "paused".to_string(),
message: Some(message),
});
report.status = "awaiting_checkpoint".to_string();
return Ok(report);
}
WorkflowActionOutcome::Failed(message) => {
report.events.push(WorkflowRunEvent {
node: node.id.clone(),
action: node.action.clone(),
status: "error".to_string(),
message: Some(message.clone()),
});
report.errors.push(message);
WorkflowStepStatus::Failed
}
};
let outgoing = select_outgoing_edge(workflow, ¤t, outcome_status);
let Some(edge) = outgoing else {
report.status = match outcome_status {
WorkflowStepStatus::Ok => "completed".to_string(),
WorkflowStepStatus::Failed => "error".to_string(),
};
return Ok(report);
};
if order[edge.to.as_str()] <= order[current.as_str()] {
if let Some(limit) = loop_limit {
if report.iterations >= limit {
report.status = "iteration_limit".to_string();
return Ok(report);
}
}
report.iterations = report.iterations.saturating_add(1);
}
current = edge.to.clone();
}
report.status = "iteration_limit".to_string();
Ok(report)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum WorkflowStepStatus {
Ok,
Failed,
}
fn select_outgoing_edge<'a>(
workflow: &'a WorkflowDefinition,
current: &str,
status: WorkflowStepStatus,
) -> Option<&'a WorkflowEdge> {
workflow
.edges
.iter()
.find(|edge| {
edge.from == current
&& edge
.condition
.as_deref()
.is_some_and(|condition| condition_matches(condition, status))
})
.or_else(|| {
workflow.edges.iter().find(|edge| {
edge.from == current && edge.condition.is_none() && status == WorkflowStepStatus::Ok
})
})
}
fn condition_matches(condition: &str, status: WorkflowStepStatus) -> bool {
match condition.trim().to_ascii_lowercase().as_str() {
"always" => true,
"ok" | "success" | "passed" => status == WorkflowStepStatus::Ok,
"failed" | "error" => status == WorkflowStepStatus::Failed,
"retry" => status == WorkflowStepStatus::Ok,
_ => false,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Default)]
struct StaticExecutor {
outcomes: BTreeMap<String, WorkflowActionOutcome>,
visited: Vec<String>,
}
impl WorkflowActionExecutor for StaticExecutor {
fn execute(
&mut self,
_workflow: &WorkflowDefinition,
node: &WorkflowNode,
) -> WorkflowActionOutcome {
self.visited.push(node.id.clone());
self.outcomes
.get(&node.id)
.cloned()
.unwrap_or(WorkflowActionOutcome::Continue)
}
}
fn workflow(text: &str) -> WorkflowDefinition {
parse_workflow(text).unwrap()
}
#[test]
fn failed_condition_edge_is_not_taken_after_ok_outcome() {
let definition = workflow(
r#"
version: 1
id: conditional-ok
title: Conditional OK
nodes:
- id: execution
action: stage
stage: execution
- id: review
action: stage
stage: review
- id: fix
action: skill
skill: execution-discipline
edges:
- from: execution
to: review
- from: review
to: fix
condition: failed
"#,
);
let mut executor = StaticExecutor::default();
let report = run_workflow(&definition, None, &mut executor).unwrap();
assert_eq!(report.status, "completed");
assert_eq!(executor.visited, vec!["execution", "review"]);
}
#[test]
fn failed_condition_edge_is_taken_after_failure() {
let definition = workflow(
r#"
version: 1
id: conditional-failed
title: Conditional Failed
max_iterations: 1
nodes:
- id: execution
action: stage
stage: execution
- id: review
action: stage
stage: review
- id: fix
action: skill
skill: execution-discipline
edges:
- from: execution
to: review
- from: review
to: fix
condition: failed
"#,
);
let mut executor = StaticExecutor::default();
executor.outcomes.insert(
"review".to_string(),
WorkflowActionOutcome::Failed("review failed".to_string()),
);
let report = run_workflow(&definition, None, &mut executor).unwrap();
assert_eq!(report.status, "completed");
assert_eq!(executor.visited, vec!["execution", "review", "fix"]);
assert_eq!(report.errors, vec!["review failed".to_string()]);
}
#[test]
fn unknown_condition_is_rejected() {
let err = parse_workflow(
r#"
version: 1
id: bad-condition
title: Bad Condition
nodes:
- id: a
action: command
command: echo a
- id: b
action: command
command: echo b
edges:
- from: a
to: b
condition: maybe
"#,
)
.unwrap_err();
assert!(format!("{err:#}").contains("condition desconhecida"));
}
}