enact_core/workflow/

mod.rs

//! Workflow module - Opinionated workflow blueprints and execution
//!
//! Implements Enact v5 workflow patterns inspired by Antfarm:
//! - Workflow blueprints (feature-dev, bug-fix, security-audit)
//! - Step I/O contracts (STATUS: done|retry|blocked)
//! - Role-based capability profiles
//! - Verify/retry loops
//! - Story loop primitive
//! - Progress journal
//! - Health monitoring

pub mod contract;
pub mod health;
pub mod progress;
pub mod role;
pub mod story_loop;

use crate::graph::schema::{GraphDefinition, NodeDefinition};
use anyhow::{anyhow, Context, Result};
use contract::{FailureAction, StepContract};
use health::{HealthCheckConfig, RunHealthWatchdog};
use progress::ProgressJournalWriter;
use role::{RoleDefinition, RoleRegistry};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::Path;

/// Workflow definition (v5 format)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowDefinition {
    /// Workflow name
    pub name: String,
    /// Workflow version
    #[serde(default = "default_version")]
    pub version: String,
    /// Workflow description
    pub description: Option<String>,
    /// Optional default model for steps in this workflow
    #[serde(default)]
    pub model: Option<String>,
    /// Triggers
    #[serde(default)]
    pub triggers: Vec<String>,
    /// Input parameters
    #[serde(default)]
    pub inputs: HashMap<String, InputParameter>,
    /// Role definitions
    #[serde(default)]
    pub roles: Vec<RoleDefinition>,
    /// Workflow steps
    pub steps: Vec<WorkflowStep>,
    /// Progress journal configuration
    #[serde(default)]
    pub progress_journal: ProgressJournalConfig,
    /// Health check configuration
    #[serde(default)]
    pub health_checks: HealthCheckConfig,
}

fn default_version() -> String {
    "1.0.0".to_string()
}

/// Input parameter definition
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InputParameter {
    /// Parameter type
    #[serde(rename = "type")]
    pub param_type: String,
    /// Description
    pub description: Option<String>,
    /// Whether parameter is required
    #[serde(default)]
    pub required: bool,
    /// Default value
    #[serde(default)]
    pub default: Option<serde_json::Value>,
}

/// Workflow step definition
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowStep {
    /// Step ID
    pub id: String,
    /// Step name
    pub name: String,
    /// Role assigned to this step
    pub role: String,
    /// Step description
    pub description: Option<String>,
    /// Step type (standard, story_loop, etc.)
    #[serde(default = "default_step_type")]
    pub step_type: StepType,
    /// Loop configuration (if type is story_loop)
    #[serde(default)]
    pub loop_config: Option<story_loop::StoryLoopConfig>,
    /// Step input template (can use variable substitution)
    pub input: String,
    /// Optional model pin for this step
    #[serde(default)]
    pub model: Option<String>,
    /// Step contract for validation
    #[serde(default)]
    pub contract: Option<StepContract>,
}

fn default_step_type() -> StepType {
    StepType::Standard
}

/// Step types
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StepType {
    /// Standard step
    Standard,
    /// Story loop step
    StoryLoop,
    /// Verification step
    Verification,
    /// Test step
    Test,
}

/// Progress journal configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProgressJournalConfig {
    /// Whether journal is enabled
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Journal template
    #[serde(default)]
    pub template: Option<String>,
}

fn default_true() -> bool {
    true
}

impl Default for ProgressJournalConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            template: None,
        }
    }
}

/// Workflow loader
pub struct WorkflowLoader;

impl WorkflowLoader {
    /// Load workflow from YAML file
    pub async fn load_from_file(path: &Path) -> Result<WorkflowDefinition> {
        let content = tokio::fs::read_to_string(path)
            .await
            .context("Failed to read workflow file")?;

        Self::load_from_str(&content)
    }

    /// Load workflow from YAML string
    pub fn load_from_str(yaml: &str) -> Result<WorkflowDefinition> {
        serde_yaml::from_str(yaml).context("Failed to parse workflow YAML")
    }

    /// Load workflow from directory
    pub async fn load_from_directory(dir: &Path) -> Result<WorkflowDefinition> {
        let workflow_file = dir.join("workflow.yml");

        if !workflow_file.exists() {
            return Err(anyhow!(
                "Workflow file not found: {}",
                workflow_file.display()
            ));
        }

        Self::load_from_file(&workflow_file).await
    }
}

/// Workflow compiler - converts workflow to graph
pub struct WorkflowCompiler;

impl WorkflowCompiler {
    /// Compile workflow to graph definition
    pub fn compile(workflow: &WorkflowDefinition) -> Result<GraphDefinition> {
        let mut nodes = HashMap::new();

        // Compile each step to a node
        for step in workflow.steps.iter() {
            let node_def = Self::compile_step(step)?;
            nodes.insert(step.id.clone(), node_def);
        }

        // Connect steps in sequence (for now)
        // TODO: Support conditional routing based on contracts
        for i in 0..workflow.steps.len() - 1 {
            let current_id = &workflow.steps[i].id;
            let next_id = &workflow.steps[i + 1].id;

            if let Some(node) = nodes.get_mut(current_id) {
                let mut edges = node.edges().clone();
                edges.insert("_default".to_string(), next_id.clone());
                *node = Self::update_node_edges(node.clone(), edges)?;
            }
        }

        // Last step connects to END
        if let Some(last_step) = workflow.steps.last() {
            if let Some(node) = nodes.get_mut(&last_step.id) {
                let mut edges = node.edges().clone();
                edges.insert("_default".to_string(), "END".to_string());
                *node = Self::update_node_edges(node.clone(), edges)?;
            }
        }

        Ok(GraphDefinition {
            name: workflow.name.clone(),
            version: workflow.version.clone(),
            description: workflow.description.clone(),
            model: workflow.model.clone(),
            triggers: workflow.triggers.clone(),
            inputs: workflow
                .inputs
                .iter()
                .map(|(k, v)| (k.clone(), v.description.clone().unwrap_or_default()))
                .collect(),
            nodes,
        })
    }

    /// Compile a single step to node definition
    fn compile_step(step: &WorkflowStep) -> Result<NodeDefinition> {
        match step.step_type {
            StepType::Standard | StepType::Verification | StepType::Test => {
                Ok(NodeDefinition::Llm {
                    model: step.model.clone(),
                    system_prompt: format!(
                        "You are the {} agent. {}",
                        step.name,
                        step.description.clone().unwrap_or_default()
                    ),
                    tools: vec![],
                    edges: HashMap::new(),
                })
            }
            StepType::StoryLoop => {
                // Story loop compiles to a subgraph
                Ok(NodeDefinition::Graph {
                    graph_name: format!("{}_loop", step.id),
                    edges: HashMap::new(),
                })
            }
        }
    }

    /// Update node edges
    fn update_node_edges(
        node: NodeDefinition,
        edges: HashMap<String, String>,
    ) -> Result<NodeDefinition> {
        match node {
            NodeDefinition::Llm {
                model,
                system_prompt,
                tools,
                ..
            } => Ok(NodeDefinition::Llm {
                model,
                system_prompt,
                tools,
                edges,
            }),
            NodeDefinition::Function { action, .. } => {
                Ok(NodeDefinition::Function { action, edges })
            }
            NodeDefinition::Condition { expr, .. } => Ok(NodeDefinition::Condition { expr, edges }),
            NodeDefinition::Graph { graph_name, .. } => {
                Ok(NodeDefinition::Graph { graph_name, edges })
            }
        }
    }
}

/// Workflow validator
pub struct WorkflowValidator;

impl WorkflowValidator {
    /// Validate workflow definition
    pub fn validate(workflow: &WorkflowDefinition) -> Result<Vec<ValidationIssue>> {
        let mut issues = vec![];

        // Check for duplicate step IDs
        let mut seen_ids = std::collections::HashSet::new();
        for step in &workflow.steps {
            if !seen_ids.insert(step.id.clone()) {
                issues.push(ValidationIssue {
                    severity: ValidationSeverity::Error,
                    message: format!("Duplicate step ID: {}", step.id),
                    location: format!("steps.{}", step.id),
                });
            }
        }

        // Check that all role references exist
        let role_ids: std::collections::HashSet<_> = workflow.roles.iter().map(|r| &r.id).collect();

        for step in &workflow.steps {
            if !role_ids.contains(&step.role) {
                issues.push(ValidationIssue {
                    severity: ValidationSeverity::Error,
                    message: format!(
                        "Step '{}' references undefined role '{}'",
                        step.id, step.role
                    ),
                    location: format!("steps.{}.role", step.id),
                });
            }
        }

        // Validate step contracts
        for step in &workflow.steps {
            if let Some(contract) = &step.contract {
                if let Err(e) = Self::validate_contract(contract) {
                    issues.push(ValidationIssue {
                        severity: ValidationSeverity::Warning,
                        message: format!("Invalid contract in step '{}': {}", step.id, e),
                        location: format!("steps.{}.contract", step.id),
                    });
                }
            }
        }

        Ok(issues)
    }

    /// Validate a contract definition
    fn validate_contract(contract: &StepContract) -> Result<()> {
        // Validate that on_failure references valid retry targets
        if let Some(FailureAction::Retry {
            retry_target: Some(target),
            ..
        }) = &contract.on_failure
        {
            // Target step must exist - this will be checked during workflow validation
            // Here we just ensure the format is valid
            if target.is_empty() {
                return Err(anyhow!("Empty retry target"));
            }
        }

        Ok(())
    }
}

/// Validation issue
#[derive(Debug, Clone)]
pub struct ValidationIssue {
    /// Issue severity
    pub severity: ValidationSeverity,
    /// Error message
    pub message: String,
    /// Location in workflow
    pub location: String,
}

/// Validation severity
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ValidationSeverity {
    Info,
    Warning,
    Error,
}

/// Workflow execution context
pub struct WorkflowContext {
    /// Workflow definition
    pub workflow: WorkflowDefinition,
    /// Role registry
    pub role_registry: RoleRegistry,
    /// Progress journal writer
    pub progress_writer: Option<ProgressJournalWriter>,
    /// Health watchdog
    pub health_watchdog: Option<RunHealthWatchdog>,
    /// Input values
    pub inputs: HashMap<String, serde_json::Value>,
    /// Step outputs
    pub step_outputs: HashMap<String, contract::ParsedOutput>,
}

impl WorkflowContext {
    /// Create new workflow context
    pub fn new(workflow: WorkflowDefinition, inputs: HashMap<String, serde_json::Value>) -> Self {
        let mut role_registry = RoleRegistry::new();
        role_registry.load_from_workflow(workflow.roles.clone());

        Self {
            workflow,
            role_registry,
            progress_writer: None,
            health_watchdog: None,
            inputs,
            step_outputs: HashMap::new(),
        }
    }

    /// Enable progress journal
    pub fn with_progress_journal(mut self, run_id: String, task: String) -> Self {
        self.progress_writer = Some(ProgressJournalWriter::new(
            run_id,
            self.workflow.name.clone(),
            task,
        ));
        self
    }

    /// Enable health monitoring
    pub fn with_health_watchdog(mut self, config: HealthCheckConfig) -> Self {
        self.health_watchdog = Some(RunHealthWatchdog::new(config));
        self
    }

    /// Get step output
    pub fn get_step_output(&self, step_id: &str) -> Option<&contract::ParsedOutput> {
        self.step_outputs.get(step_id)
    }

    /// Set step output
    pub fn set_step_output(&mut self, step_id: String, output: contract::ParsedOutput) {
        self.step_outputs.insert(step_id, output);
    }

    /// Substitute variables in template
    pub fn substitute_variables(&self, template: &str) -> String {
        let mut result = template.to_string();

        // Substitute inputs
        for (key, value) in &self.inputs {
            let placeholder = format!("{{{{{}}}}}", key);
            let value_str = match value {
                serde_json::Value::String(s) => s.clone(),
                _ => value.to_string(),
            };
            result = result.replace(&placeholder, &value_str);
        }

        // Substitute step outputs (dot notation: step_id.FIELD)
        for (step_id, output) in &self.step_outputs {
            for (field_name, field_value) in &output.fields {
                let placeholder = format!("{{{{{}.{}}}}}", step_id, field_name);
                let value_str = match field_value {
                    serde_json::Value::String(s) => s.clone(),
                    _ => field_value.to_string(),
                };
                result = result.replace(&placeholder, &value_str);
            }
        }

        result
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use contract::StepStatus;

    const TEST_WORKFLOW_YAML: &str = r#"
name: test-workflow
version: "1.0.0"
description: A test workflow

inputs:
  task:
    type: string
    required: true

roles:
  - id: planner
    name: Planner
    profile: analysis

steps:
  - id: plan
    name: Plan
    role: planner
    input: "Task: {{task}}"
    contract:
      expects:
        status: done
      on_failure:
        action: retry
        max_retries: 2
"#;

    #[test]
    fn test_load_workflow() {
        let workflow = WorkflowLoader::load_from_str(TEST_WORKFLOW_YAML).unwrap();
        assert_eq!(workflow.name, "test-workflow");
        assert_eq!(workflow.steps.len(), 1);
    }

    #[test]
    fn test_validate_workflow() {
        let workflow = WorkflowLoader::load_from_str(TEST_WORKFLOW_YAML).unwrap();
        let issues = WorkflowValidator::validate(&workflow).unwrap();
        assert!(issues.is_empty());
    }

    #[test]
    fn test_workflow_context_substitution() {
        let workflow = WorkflowLoader::load_from_str(TEST_WORKFLOW_YAML).unwrap();
        let mut context = WorkflowContext::new(
            workflow,
            [(
                "task".to_string(),
                serde_json::Value::String("Implement feature".to_string()),
            )]
            .into_iter()
            .collect(),
        );

        // Set a step output
        context.set_step_output(
            "plan".to_string(),
            contract::ParsedOutput {
                status: StepStatus::Done,
                fields: [(
                    "REPO".to_string(),
                    serde_json::Value::String("/path/to/repo".to_string()),
                )]
                .into_iter()
                .collect(),
                raw_output: "STATUS: done\nREPO: /path/to/repo".to_string(),
            },
        );

        let template = "Task: {{task}}, Repo: {{plan.REPO}}";
        let result = context.substitute_variables(template);

        assert!(result.contains("Implement feature"));
        assert!(result.contains("/path/to/repo"));
    }
}