attune/docs/guides/workflow-quickstart.md

Workflow Orchestration Quick-Start Guide

This guide helps developers get started implementing the workflow orchestration feature in Attune.

Overview

Workflows are composable YAML-based action graphs that enable complex automation. This implementation adds workflow capabilities to Attune over 5 phases spanning 9 weeks.

Before You Start

Required Reading

1. docs/workflow-orchestration.md - Full technical design (1,063 lines)
2. docs/workflow-implementation-plan.md - Implementation roadmap (562 lines)
3. docs/workflow-summary.md - Quick reference (400 lines)

Required Knowledge

• Rust async programming (tokio)
• PostgreSQL and SQLx
• RabbitMQ message patterns
• Graph algorithms (basic traversal)
• Template engines (Jinja2-style syntax)

Development Environment

# Ensure you have:
- Rust 1.70+
- PostgreSQL 14+
- RabbitMQ 3.12+
- Docker (for testing)

# Clone and setup
cd attune
cargo build

Implementation Phases

Phase 1: Foundation (Weeks 1-2)

Goal: Database schema, models, parser, template engine

Step 1.1: Database Migration

# Create migration file
cd migrations
touch 020_workflow_orchestration.sql

Copy content from docs/examples/workflow-migration.sql:

• 3 new tables: workflow_definition, workflow_execution, workflow_task_execution
• Modify action table with is_workflow and workflow_def columns
• Add indexes, triggers, views

Run migration:

sqlx migrate run

Step 1.2: Data Models

Add to crates/common/src/models.rs:

pub mod workflow {
    use super::*;
    
    #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
    pub struct WorkflowDefinition {
        pub id: Id,
        pub r#ref: String,
        pub pack: Id,
        pub pack_ref: String,
        pub label: String,
        pub description: Option<String>,
        pub version: String,
        pub param_schema: Option<JsonSchema>,
        pub out_schema: Option<JsonSchema>,
        pub definition: JsonValue,  // Full workflow YAML as JSON
        pub tags: Vec<String>,
        pub enabled: bool,
        pub created: DateTime<Utc>,
        pub updated: DateTime<Utc>,
    }
    
    #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
    pub struct WorkflowExecution {
        pub id: Id,
        pub execution: Id,
        pub workflow_def: Id,
        pub current_tasks: Vec<String>,
        pub completed_tasks: Vec<String>,
        pub failed_tasks: Vec<String>,
        pub skipped_tasks: Vec<String>,
        pub variables: JsonValue,
        pub task_graph: JsonValue,
        pub status: ExecutionStatus,
        pub error_message: Option<String>,
        pub paused: bool,
        pub pause_reason: Option<String>,
        pub created: DateTime<Utc>,
        pub updated: DateTime<Utc>,
    }
    
    #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
    pub struct WorkflowTaskExecution {
        pub id: Id,
        pub workflow_execution: Id,
        pub execution: Id,
        pub task_name: String,
        pub task_index: Option<i32>,
        pub task_batch: Option<i32>,
        pub status: ExecutionStatus,
        pub started_at: Option<DateTime<Utc>>,
        pub completed_at: Option<DateTime<Utc>>,
        pub duration_ms: Option<i64>,
        pub result: Option<JsonValue>,
        pub error: Option<JsonValue>,
        pub retry_count: i32,
        pub max_retries: i32,
        pub next_retry_at: Option<DateTime<Utc>>,
        pub timeout_seconds: Option<i32>,
        pub timed_out: bool,
        pub created: DateTime<Utc>,
        pub updated: DateTime<Utc>,
    }
}

Step 1.3: Repositories

Create crates/common/src/repositories/workflow_definition.rs:

use sqlx::PgPool;
use crate::models::workflow::WorkflowDefinition;
use crate::error::Result;

pub struct WorkflowDefinitionRepository;

impl WorkflowDefinitionRepository {
    pub async fn create(pool: &PgPool, def: &WorkflowDefinition) -> Result<WorkflowDefinition> {
        // INSERT implementation
    }
    
    pub async fn find_by_ref(pool: &PgPool, ref_: &str) -> Result<Option<WorkflowDefinition>> {
        // SELECT WHERE ref = ?
    }
    
    pub async fn list_by_pack(pool: &PgPool, pack_id: i64) -> Result<Vec<WorkflowDefinition>> {
        // SELECT WHERE pack = ?
    }
    
    // ... other CRUD methods
}

Create similar repositories for workflow_execution and workflow_task_execution.

Step 1.4: YAML Parser

Create crates/executor/src/workflow/parser.rs:

use serde::{Deserialize, Serialize};
use std::collections::HashMap;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowSpec {
    pub r#ref: String,
    pub label: String,
    pub description: Option<String>,
    pub version: String,
    pub parameters: Option<serde_json::Value>,
    pub output: Option<serde_json::Value>,
    pub vars: HashMap<String, serde_json::Value>,
    pub tasks: Vec<TaskSpec>,
    pub output_map: Option<HashMap<String, String>>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskSpec {
    pub name: String,
    #[serde(rename = "type")]
    pub task_type: Option<TaskType>,
    pub action: Option<String>,
    pub input: HashMap<String, String>,
    pub publish: Option<Vec<String>>,
    pub on_success: Option<String>,
    pub on_failure: Option<String>,
    pub on_complete: Option<String>,
    pub on_timeout: Option<String>,
    pub decision: Option<Vec<DecisionBranch>>,
    pub when: Option<String>,
    pub with_items: Option<String>,
    pub batch_size: Option<usize>,
    pub retry: Option<RetryPolicy>,
    pub timeout: Option<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum TaskType {
    Action,
    Parallel,
    Workflow,
}

pub fn parse_workflow_yaml(yaml: &str) -> Result<WorkflowSpec> {
    serde_yaml::from_str(yaml)
        .map_err(|e| Error::InvalidWorkflowDefinition(e.to_string()))
}

Step 1.5: Template Engine

Add to crates/executor/Cargo.toml:

[dependencies]
tera = "1.19"

Create crates/executor/src/workflow/context.rs:

use tera::{Tera, Context};
use std::collections::HashMap;
use serde_json::Value as JsonValue;

pub struct WorkflowContext {
    pub execution_id: i64,
    pub parameters: JsonValue,
    pub vars: HashMap<String, JsonValue>,
    pub task_results: HashMap<String, TaskResult>,
    pub pack_config: JsonValue,
    pub system: SystemContext,
}

impl WorkflowContext {
    pub fn new(execution_id: i64, parameters: JsonValue) -> Self {
        Self {
            execution_id,
            parameters,
            vars: HashMap::new(),
            task_results: HashMap::new(),
            pack_config: JsonValue::Null,
            system: SystemContext::default(),
        }
    }
    
    pub fn render_template(&self, template: &str) -> Result<String> {
        let mut tera = Tera::default();
        let context = self.to_tera_context();
        tera.render_str(template, &context)
            .map_err(|e| Error::TemplateError(e.to_string()))
    }
    
    fn to_tera_context(&self) -> Context {
        let mut ctx = Context::new();
        ctx.insert("parameters", &self.parameters);
        ctx.insert("vars", &self.vars);
        ctx.insert("task", &self.task_results);
        ctx.insert("system", &self.system);
        ctx.insert("pack", &json!({"config": self.pack_config}));
        ctx
    }
}

Phase 1 Testing:

cargo test -p attune-common workflow
cargo test -p attune-executor workflow::parser

---

Phase 2: Execution Engine (Weeks 3-4)

Goal: Graph builder, workflow executor, message handlers

Step 2.1: Task Graph

Create crates/executor/src/workflow/graph.rs:

use std::collections::HashMap;

pub struct TaskGraph {
    nodes: HashMap<String, TaskNode>,
    edges: HashMap<String, Vec<Edge>>,
}

impl TaskGraph {
    pub fn from_workflow_spec(spec: &WorkflowSpec) -> Result<Self> {
        // Build graph from task definitions
        // Create nodes for each task
        // Create edges based on transitions (on_success, on_failure, etc.)
    }
    
    pub fn get_entry_tasks(&self) -> Vec<&TaskNode> {
        // Return tasks with no incoming edges
    }
    
    pub fn get_next_tasks(&self, completed_task: &str, result: &TaskResult) -> Vec<&TaskNode> {
        // Follow edges based on result (success/failure)
    }
}

Step 2.2: Workflow Executor

Create crates/executor/src/workflow/executor.rs:

pub struct WorkflowExecutor {
    pool: PgPool,
    publisher: MessagePublisher,
}

impl WorkflowExecutor {
    pub async fn execute_workflow(
        &self,
        execution_id: i64,
        workflow_ref: &str,
        parameters: JsonValue,
    ) -> Result<()> {
        // 1. Load workflow definition
        // 2. Create workflow_execution record
        // 3. Initialize context
        // 4. Build task graph
        // 5. Schedule initial tasks
    }
    
    pub async fn handle_task_completion(
        &self,
        workflow_execution_id: i64,
        task_name: String,
        result: TaskResult,
    ) -> Result<()> {
        // 1. Update workflow_task_execution
        // 2. Publish variables
        // 3. Determine next tasks
        // 4. Schedule next tasks
        // 5. Check if workflow complete
    }
}

Step 2.3: Message Handlers

Integrate with existing executor message loops:

// In executor/src/main.rs
async fn start_workflow_message_handlers(
    pool: PgPool,
    publisher: MessagePublisher,
) -> Result<()> {
    let executor = WorkflowExecutor::new(pool.clone(), publisher.clone());
    
    // Listen for execution.completed on workflow tasks
    let consumer = create_consumer("workflow.task.completions").await?;
    
    consumer.consume_with_handler(move |envelope| {
        let executor = executor.clone();
        async move {
            executor.handle_task_completion(
                envelope.payload.workflow_execution_id,
                envelope.payload.task_name,
                envelope.payload.result,
            ).await
        }
    }).await?;
    
    Ok(())
}

Phase 2 Testing:

cargo test -p attune-executor workflow::graph
cargo test -p attune-executor workflow::executor

---

Phase 3: Advanced Features (Weeks 5-6)

Goal: Iteration, parallelism, retry, conditionals

Step 3.1: Iteration

Create crates/executor/src/workflow/iterator.rs:

pub struct TaskIterator {
    items: Vec<JsonValue>,
    batch_size: Option<usize>,
}

impl TaskIterator {
    pub fn from_template(
        template: &str,
        context: &WorkflowContext,
        batch_size: Option<usize>,
    ) -> Result<Self> {
        let rendered = context.render_template(template)?;
        let items: Vec<JsonValue> = serde_json::from_str(&rendered)?;
        Ok(Self { items, batch_size })
    }
    
    pub fn batches(&self) -> Vec<Vec<&JsonValue>> {
        if let Some(size) = self.batch_size {
            self.items.chunks(size).map(|c| c.iter().collect()).collect()
        } else {
            vec![self.items.iter().collect()]
        }
    }
}

Step 3.2: Parallel Execution

Create crates/executor/src/workflow/parallel.rs:

pub struct ParallelExecutor {
    // Execute multiple tasks simultaneously
    // Wait for all to complete
    // Aggregate results
}

Step 3.3: Retry Logic

Create crates/executor/src/workflow/retry.rs:

pub struct RetryHandler {
    // Exponential/linear/constant backoff
    // Max retries
    // Condition evaluation
}

---

Phase 4: API & Tools (Weeks 7-8)

Goal: REST endpoints, validation, pack integration

Step 4.1: API Routes

Create crates/api/src/routes/workflows.rs:

pub fn workflow_routes() -> Router {
    Router::new()
        .route("/packs/:pack_ref/workflows", post(create_workflow))
        .route("/packs/:pack_ref/workflows", get(list_workflows))
        .route("/workflows/:workflow_ref", get(get_workflow))
        .route("/workflows/:workflow_ref", put(update_workflow))
        .route("/workflows/:workflow_ref", delete(delete_workflow))
        .route("/workflows/:workflow_ref/execute", post(execute_workflow))
}

Step 4.2: Pack Integration

Update pack registration to scan workflows/ directory:

// In pack registration logic
async fn register_workflows_in_pack(pool: &PgPool, pack_id: i64, pack_path: &Path) -> Result<()> {
    let workflows_dir = pack_path.join("workflows");
    if !workflows_dir.exists() {
        return Ok(());
    }
    
    for entry in std::fs::read_dir(workflows_dir)? {
        let path = entry?.path();
        if path.extension() == Some("yaml".as_ref()) {
            let yaml = std::fs::read_to_string(&path)?;
            let spec = parse_workflow_yaml(&yaml)?;
            
            // Create workflow_definition
            // Create synthetic action with is_workflow=true
        }
    }
    
    Ok(())
}

---

Phase 5: Testing & Documentation (Week 9)

Goal: Comprehensive tests and documentation

Integration Tests

Create crates/executor/tests/workflow_integration.rs:

#[tokio::test]
async fn test_simple_sequential_workflow() {
    // Test basic workflow execution
}

#[tokio::test]
async fn test_parallel_execution() {
    // Test parallel tasks
}

#[tokio::test]
async fn test_conditional_branching() {
    // Test decision trees
}

#[tokio::test]
async fn test_iteration_with_batching() {
    // Test with-items
}

---

Development Tips

Debugging Workflows

# Enable debug logging
RUST_LOG=attune_executor::workflow=debug cargo run

# Watch workflow execution
psql -d attune -c "SELECT * FROM attune.workflow_execution_summary;"

# Check task status
psql -d attune -c "SELECT * FROM attune.workflow_task_detail WHERE workflow_execution = ?;"

Testing YAML Parsing

# Validate workflow YAML
cargo run --bin attune-cli -- workflow validate path/to/workflow.yaml

Common Pitfalls

1. Circular Dependencies: Validate graph for cycles
2. Template Errors: Always handle template rendering failures
3. Variable Scope: Test all 6 scopes independently
4. Message Ordering: Ensure task completions processed in order
5. Resource Limits: Enforce max tasks/depth/iterations

---

Resources

• Design Docs: docs/workflow-*.md
• Examples: docs/examples/simple-workflow.yaml, complete-workflow.yaml
• Migration: docs/examples/workflow-migration.sql
• TODO Tasks: work-summary/TODO.md Phase 8.1

---

Getting Help

• Review full design: docs/workflow-orchestration.md
• Check implementation plan: docs/workflow-implementation-plan.md
• See examples: docs/examples/
• Ask questions in project discussions

---

Ready to start? Begin with Phase 1, Step 1.1: Database Migration