use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use anyhow::{bail, Context, 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)
}
#[allow(dead_code)]
fn workflows_dir(root: &Path) -> PathBuf {
root.join(".sdd").join("workflows")
}
#[allow(dead_code)]
pub fn load_run_report(root: &Path, id: &str) -> Result<Option<WorkflowRunReport>> {
if id.trim().is_empty()
|| !id
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
{
bail!("workflow run id inválido: {id}");
}
let path = workflows_dir(root).join(format!("{id}.json"));
if !path.exists() {
return Ok(None);
}
let text = std::fs::read_to_string(&path)
.with_context(|| format!("lendo workflow run report {}", path.display()))?;
let report: WorkflowRunReport = serde_json::from_str(&text)
.with_context(|| format!("desserializando workflow run report {}", path.display()))?;
Ok(Some(report))
}
#[allow(dead_code)]
pub fn list_run_reports(root: &Path) -> Result<Vec<WorkflowRunReport>> {
let dir = workflows_dir(root);
let mut out = Vec::new();
let Ok(entries) = std::fs::read_dir(&dir) else {
return Ok(out);
};
let mut paths: Vec<PathBuf> = entries
.flatten()
.map(|e| e.path())
.filter(|p| p.extension().and_then(|x| x.to_str()) == Some("json"))
.collect();
paths.sort();
for path in paths {
let text = std::fs::read_to_string(&path)
.with_context(|| format!("lendo workflow run report {}", path.display()))?;
if let Ok(report) = serde_json::from_str::<WorkflowRunReport>(&text) {
out.push(report);
}
}
Ok(out)
}
#[allow(dead_code)]
pub fn list_run_report_ids(root: &Path) -> Vec<String> {
let dir = workflows_dir(root);
let Ok(entries) = std::fs::read_dir(&dir) else {
return Vec::new();
};
let mut ids: Vec<String> = entries
.flatten()
.filter_map(|e| {
let path = e.path();
if path.extension().and_then(|x| x.to_str()) == Some("json") {
path.file_stem()
.and_then(|s| s.to_str())
.map(|s| s.to_string())
} else {
None
}
})
.collect();
ids.sort();
ids
}
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"));
}
#[test]
fn load_run_report_returns_none_when_missing() {
let dir = tempfile::tempdir().unwrap();
assert!(load_run_report(dir.path(), "nope").unwrap().is_none());
}
#[test]
fn load_run_report_round_trips() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let wdir = root.join(".sdd/workflows");
std::fs::create_dir_all(&wdir).unwrap();
let report = WorkflowRunReport {
workflow_id: "test-wf".to_string(),
status: "completed".to_string(),
iterations: 2,
..Default::default()
};
std::fs::write(
wdir.join("test-wf.json"),
serde_json::to_string_pretty(&report).unwrap(),
)
.unwrap();
let loaded = load_run_report(root, "test-wf").unwrap().unwrap();
assert_eq!(loaded.workflow_id, "test-wf");
assert_eq!(loaded.status, "completed");
assert_eq!(loaded.iterations, 2);
}
#[test]
fn load_run_report_rejects_invalid_id() {
let dir = tempfile::tempdir().unwrap();
assert!(load_run_report(dir.path(), "../escape").is_err());
assert!(load_run_report(dir.path(), "").is_err());
}
#[test]
fn list_run_reports_returns_all() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let wdir = root.join(".sdd/workflows");
std::fs::create_dir_all(&wdir).unwrap();
for id in ["alpha", "beta"] {
let report = WorkflowRunReport {
workflow_id: id.to_string(),
status: "completed".to_string(),
..Default::default()
};
std::fs::write(
wdir.join(format!("{id}.json")),
serde_json::to_string(&report).unwrap(),
)
.unwrap();
}
let reports = list_run_reports(root).unwrap();
assert_eq!(reports.len(), 2);
}
#[test]
fn list_run_reports_empty_when_no_dir() {
let dir = tempfile::tempdir().unwrap();
assert!(list_run_reports(dir.path()).unwrap().is_empty());
}
#[test]
fn list_run_report_ids_returns_sorted() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let wdir = root.join(".sdd/workflows");
std::fs::create_dir_all(&wdir).unwrap();
std::fs::write(wdir.join("beta.json"), "{}").unwrap();
std::fs::write(wdir.join("alpha.json"), "{}").unwrap();
let ids = list_run_report_ids(root);
assert_eq!(ids, vec!["alpha".to_string(), "beta".to_string()]);
}
}