attune-system/attune
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP

Browse Source
This commit is contained in:
David Culbreth
2026-03-02 19:27:52 -06:00
parent 42a9f1d31a
commit 5da940639a
40 changed files with 3931 additions and 2785 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/executor/src/lib.rs
View File

@@ -28,7 +28,3 @@ pub use queue_manager::{ExecutionQueueManager, QueueConfig, QueueStats};
pub use retry_manager::{RetryAnalysis, RetryConfig, RetryManager, RetryReason};
pub use timeout_monitor::{ExecutionTimeoutMonitor, TimeoutMonitorConfig};
pub use worker_health::{HealthMetrics, HealthProbeConfig, HealthStatus, WorkerHealthProbe};
pub use workflow::{
parse_workflow_yaml, BackoffStrategy, ParseError, TemplateEngine, VariableContext,
WorkflowDefinition, WorkflowValidator,
};

62
crates/executor/src/workflow/context.rs
View File

@@ -61,9 +61,6 @@ pub type ContextResult<T> = Result<T, ContextError>;
/// Errors that can occur during context operations
#[derive(Debug, Error)]
pub enum ContextError {
#[error("Template rendering error: {0}")]
TemplateError(String),
#[error("Variable not found: {0}")]
VariableNotFound(String),
@@ -200,16 +197,19 @@ impl WorkflowContext {
}
/// Get a workflow-scoped variable by name.
#[allow(dead_code)] // Part of complete context API; used in tests
pub fn get_var(&self, name: &str) -> Option<JsonValue> {
self.variables.get(name).map(|entry| entry.value().clone())
}
/// Store a completed task's result (accessible as `task.<name>.*`).
#[allow(dead_code)] // Part of complete context API; used in tests
pub fn set_task_result(&mut self, task_name: &str, result: JsonValue) {
self.task_results.insert(task_name.to_string(), result);
}
/// Get a task result by task name.
#[allow(dead_code)] // Part of complete context API; used in tests
pub fn get_task_result(&self, task_name: &str) -> Option<JsonValue> {
self.task_results
.get(task_name)
@@ -217,11 +217,13 @@ impl WorkflowContext {
}
/// Set the pack configuration (accessible as `config.<key>`).
#[allow(dead_code)] // Part of complete context API; used in tests
pub fn set_pack_config(&mut self, config: JsonValue) {
self.pack_config = Arc::new(config);
}
/// Set the keystore secrets (accessible as `keystore.<key>`).
#[allow(dead_code)] // Part of complete context API; used in tests
pub fn set_keystore(&mut self, secrets: JsonValue) {
self.keystore = Arc::new(secrets);
}
@@ -233,6 +235,7 @@ impl WorkflowContext {
}
/// Clear current item
#[allow(dead_code)] // Part of complete context API; symmetric with set_current_item
pub fn clear_current_item(&mut self) {
self.current_item = None;
self.current_index = None;
@@ -440,6 +443,7 @@ impl WorkflowContext {
}
/// Export context for storage
#[allow(dead_code)] // Part of complete context API; used in tests
pub fn export(&self) -> JsonValue {
let variables: HashMap<String, JsonValue> = self
.variables
@@ -470,6 +474,7 @@ impl WorkflowContext {
}
/// Import context from stored data
#[allow(dead_code)] // Part of complete context API; used in tests
pub fn import(data: JsonValue) -> ContextResult<Self> {
let variables = DashMap::new();
if let Some(obj) = data["variables"].as_object() {
@@ -677,7 +682,9 @@ mod tests {
ctx.set_var("greeting", json!("Hello"));
// Canonical: workflow.<name>
let result = ctx.render_template("{{ workflow.greeting }} World").unwrap();
let result = ctx
.render_template("{{ workflow.greeting }} World")
.unwrap();
assert_eq!(result, "Hello World");
}
@@ -699,7 +706,9 @@ mod tests {
let ctx = WorkflowContext::new(json!({}), vars);
// Backward-compat alias: variables.<name>
let result = ctx.render_template("{{ variables.greeting }} World").unwrap();
let result = ctx
.render_template("{{ variables.greeting }} World")
.unwrap();
assert_eq!(result, "Hello World");
}
@@ -735,7 +744,9 @@ mod tests {
let mut ctx = WorkflowContext::new(json!({}), HashMap::new());
ctx.set_task_result("fetch", json!({"result": {"data": {"id": 42}}}));
let val = ctx.evaluate_expression("task.fetch.result.data.id").unwrap();
let val = ctx
.evaluate_expression("task.fetch.result.data.id")
.unwrap();
assert_eq!(val, json!(42));
}
@@ -744,7 +755,9 @@ mod tests {
let mut ctx = WorkflowContext::new(json!({}), HashMap::new());
ctx.set_task_result("run_cmd", json!({"result": {"stdout": "hello world"}}));
let val = ctx.evaluate_expression("task.run_cmd.result.stdout").unwrap();
let val = ctx
.evaluate_expression("task.run_cmd.result.stdout")
.unwrap();
assert_eq!(val, json!("hello world"));
}
@@ -755,14 +768,14 @@ mod tests {
#[test]
fn test_config_namespace() {
let mut ctx = WorkflowContext::new(json!({}), HashMap::new());
ctx.set_pack_config(json!({"api_token": "tok_abc123", "base_url": "https://api.example.com"}));
ctx.set_pack_config(
json!({"api_token": "tok_abc123", "base_url": "https://api.example.com"}),
);
let val = ctx.evaluate_expression("config.api_token").unwrap();
assert_eq!(val, json!("tok_abc123"));
let result = ctx
.render_template("URL: {{ config.base_url }}")
.unwrap();
let result = ctx.render_template("URL: {{ config.base_url }}").unwrap();
assert_eq!(result, "URL: https://api.example.com");
}
@@ -796,7 +809,9 @@ mod tests {
let mut ctx = WorkflowContext::new(json!({}), HashMap::new());
ctx.set_keystore(json!({"My Secret Key": "value123"}));
let val = ctx.evaluate_expression("keystore[\"My Secret Key\"]").unwrap();
let val = ctx
.evaluate_expression("keystore[\"My Secret Key\"]")
.unwrap();
assert_eq!(val, json!("value123"));
}
@@ -850,9 +865,7 @@ mod tests {
assert!(ctx
.evaluate_condition("parameters.x > 50 or parameters.y > 15")
.unwrap());
assert!(ctx
.evaluate_condition("not parameters.x > 50")
.unwrap());
assert!(ctx.evaluate_condition("not parameters.x > 50").unwrap());
}
#[test]
@@ -863,16 +876,15 @@ mod tests {
assert!(ctx.evaluate_condition("\"admin\" in roles").unwrap());
assert!(!ctx.evaluate_condition("\"root\" in roles").unwrap());
// Via canonical workflow namespace
assert!(ctx.evaluate_condition("\"admin\" in workflow.roles").unwrap());
assert!(ctx
.evaluate_condition("\"admin\" in workflow.roles")
.unwrap());
}
#[test]
fn test_condition_with_function_calls() {
let mut ctx = WorkflowContext::new(json!({}), HashMap::new());
ctx.set_last_task_outcome(
json!({"status": "ok", "code": 200}),
TaskOutcome::Succeeded,
);
ctx.set_last_task_outcome(json!({"status": "ok", "code": 200}), TaskOutcome::Succeeded);
assert!(ctx.evaluate_condition("succeeded()").unwrap());
assert!(!ctx.evaluate_condition("failed()").unwrap());
assert!(ctx
@@ -889,9 +901,7 @@ mod tests {
ctx.set_var("items", json!([1, 2, 3, 4, 5]));
assert!(ctx.evaluate_condition("length(items) > 3").unwrap());
assert!(!ctx.evaluate_condition("length(items) > 10").unwrap());
assert!(ctx
.evaluate_condition("length(items) == 5")
.unwrap());
assert!(ctx.evaluate_condition("length(items) == 5").unwrap());
}
#[test]
@@ -916,10 +926,8 @@ mod tests {
#[test]
fn test_expression_string_concat() {
let ctx = WorkflowContext::new(
json!({"first": "Hello", "second": "World"}),
HashMap::new(),
);
let ctx =
WorkflowContext::new(json!({"first": "Hello", "second": "World"}), HashMap::new());
let input = json!({"msg": "{{ parameters.first + \" \" + parameters.second }}"});
let result = ctx.render_json(&input).unwrap();
assert_eq!(result["msg"], json!("Hello World"));

776
crates/executor/src/workflow/coordinator.rs
View File

@@ -1,776 +0,0 @@
//! Workflow Execution Coordinator
//!
//! This module orchestrates workflow execution, managing task dependencies,
//! parallel execution, state transitions, and error handling.
use crate::workflow::context::WorkflowContext;
use crate::workflow::graph::{TaskGraph, TaskNode};
use crate::workflow::task_executor::{TaskExecutionResult, TaskExecutionStatus, TaskExecutor};
use attune_common::error::{Error, Result};
use attune_common::models::{
execution::{Execution, WorkflowTaskMetadata},
ExecutionStatus, Id, WorkflowExecution,
};
use attune_common::mq::MessageQueue;
use attune_common::workflow::WorkflowDefinition;
use chrono::Utc;
use serde_json::{json, Value as JsonValue};
use sqlx::PgPool;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::{debug, error, info, warn};
/// Workflow execution coordinator
pub struct WorkflowCoordinator {
db_pool: PgPool,
mq: MessageQueue,
task_executor: TaskExecutor,
}
impl WorkflowCoordinator {
/// Create a new workflow coordinator
pub fn new(db_pool: PgPool, mq: MessageQueue) -> Self {
let task_executor = TaskExecutor::new(db_pool.clone(), mq.clone());
Self {
db_pool,
mq,
task_executor,
}
}
/// Start a new workflow execution
pub async fn start_workflow(
&self,
workflow_ref: &str,
parameters: JsonValue,
parent_execution_id: Option<Id>,
) -> Result<WorkflowExecutionHandle> {
info!(
"Starting workflow: {} with params: {:?}",
workflow_ref, parameters
);
// Load workflow definition
let workflow_def = sqlx::query_as::<_, attune_common::models::WorkflowDefinition>(
"SELECT * FROM attune.workflow_definition WHERE ref = $1",
)
.bind(workflow_ref)
.fetch_optional(&self.db_pool)
.await?
.ok_or_else(|| Error::not_found("workflow_definition", "ref", workflow_ref))?;
if !workflow_def.enabled {
return Err(Error::validation("Workflow is disabled"));
}
// Parse workflow definition
let definition: WorkflowDefinition = serde_json::from_value(workflow_def.definition)
.map_err(|e| Error::validation(format!("Invalid workflow definition: {}", e)))?;
// Build task graph
let graph = TaskGraph::from_workflow(&definition)
.map_err(|e| Error::validation(format!("Failed to build task graph: {}", e)))?;
// Create parent execution record
// TODO: Implement proper execution creation
let _parent_execution_id_temp = parent_execution_id.unwrap_or(1); // Placeholder
let parent_execution = sqlx::query_as::<_, attune_common::models::Execution>(
r#"
INSERT INTO attune.execution (action_ref, pack, input, parent, status)
VALUES ($1, $2, $3, $4, $5)
RETURNING *
"#,
)
.bind(workflow_ref)
.bind(workflow_def.pack)
.bind(&parameters)
.bind(parent_execution_id)
.bind(ExecutionStatus::Running)
.fetch_one(&self.db_pool)
.await?;
// Initialize workflow context
let initial_vars: HashMap<String, JsonValue> = definition
.vars
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let context = WorkflowContext::new(parameters, initial_vars);
// Create workflow execution record
let workflow_execution = self
.create_workflow_execution_record(
parent_execution.id,
workflow_def.id,
&graph,
&context,
)
.await?;
info!(
"Created workflow execution {} for workflow {}",
workflow_execution.id, workflow_ref
);
// Create execution handle
let handle = WorkflowExecutionHandle {
coordinator: Arc::new(self.clone_ref()),
execution_id: workflow_execution.id,
parent_execution_id: parent_execution.id,
workflow_def_id: workflow_def.id,
graph,
state: Arc::new(Mutex::new(WorkflowExecutionState {
context,
status: ExecutionStatus::Running,
completed_tasks: HashSet::new(),
failed_tasks: HashSet::new(),
skipped_tasks: HashSet::new(),
executing_tasks: HashSet::new(),
scheduled_tasks: HashSet::new(),
join_state: HashMap::new(),
task_executions: HashMap::new(),
paused: false,
pause_reason: None,
error_message: None,
})),
};
// Update execution status to running
self.update_workflow_execution_status(workflow_execution.id, ExecutionStatus::Running)
.await?;
Ok(handle)
}
/// Create workflow execution record in database
async fn create_workflow_execution_record(
&self,
execution_id: Id,
workflow_def_id: Id,
graph: &TaskGraph,
context: &WorkflowContext,
) -> Result<WorkflowExecution> {
let task_graph_json = serde_json::to_value(graph)
.map_err(|e| Error::internal(format!("Failed to serialize task graph: {}", e)))?;
let variables = context.export();
sqlx::query_as::<_, WorkflowExecution>(
r#"
INSERT INTO attune.workflow_execution (
execution, workflow_def, current_tasks, completed_tasks,
failed_tasks, skipped_tasks, variables, task_graph,
status, paused
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
RETURNING *
"#,
)
.bind(execution_id)
.bind(workflow_def_id)
.bind(&[] as &[String])
.bind(&[] as &[String])
.bind(&[] as &[String])
.bind(&[] as &[String])
.bind(variables)
.bind(task_graph_json)
.bind(ExecutionStatus::Running)
.bind(false)
.fetch_one(&self.db_pool)
.await
.map_err(Into::into)
}
/// Update workflow execution status
async fn update_workflow_execution_status(
&self,
workflow_execution_id: Id,
status: ExecutionStatus,
) -> Result<()> {
sqlx::query(
r#"
UPDATE attune.workflow_execution
SET status = $1, updated = NOW()
WHERE id = $2
"#,
)
.bind(status)
.bind(workflow_execution_id)
.execute(&self.db_pool)
.await?;
Ok(())
}
/// Update workflow execution state
async fn update_workflow_execution_state(
&self,
workflow_execution_id: Id,
state: &WorkflowExecutionState,
) -> Result<()> {
let current_tasks: Vec<String> = state.executing_tasks.iter().cloned().collect();
let completed_tasks: Vec<String> = state.completed_tasks.iter().cloned().collect();
let failed_tasks: Vec<String> = state.failed_tasks.iter().cloned().collect();
let skipped_tasks: Vec<String> = state.skipped_tasks.iter().cloned().collect();
sqlx::query(
r#"
UPDATE attune.workflow_execution
SET
current_tasks = $1,
completed_tasks = $2,
failed_tasks = $3,
skipped_tasks = $4,
variables = $5,
status = $6,
paused = $7,
pause_reason = $8,
error_message = $9,
updated = NOW()
WHERE id = $10
"#,
)
.bind(&current_tasks)
.bind(&completed_tasks)
.bind(&failed_tasks)
.bind(&skipped_tasks)
.bind(state.context.export())
.bind(state.status)
.bind(state.paused)
.bind(&state.pause_reason)
.bind(&state.error_message)
.bind(workflow_execution_id)
.execute(&self.db_pool)
.await?;
Ok(())
}
/// Create a task execution record
async fn create_task_execution_record(
&self,
workflow_execution_id: Id,
parent_execution_id: Id,
task: &TaskNode,
task_index: Option<i32>,
task_batch: Option<i32>,
) -> Result<Execution> {
let max_retries = task.retry.as_ref().map(|r| r.count as i32).unwrap_or(0);
let timeout = task.timeout.map(|t| t as i32);
// Create workflow task metadata
let workflow_task = WorkflowTaskMetadata {
workflow_execution: workflow_execution_id,
task_name: task.name.clone(),
task_index,
task_batch,
retry_count: 0,
max_retries,
next_retry_at: None,
timeout_seconds: timeout,
timed_out: false,
duration_ms: None,
started_at: Some(Utc::now()),
completed_at: None,
};
sqlx::query_as::<_, Execution>(
r#"
INSERT INTO attune.execution (
action_ref, parent, status, workflow_task
)
VALUES ($1, $2, $3, $4)
RETURNING *
"#,
)
.bind(&task.name)
.bind(parent_execution_id)
.bind(ExecutionStatus::Running)
.bind(sqlx::types::Json(&workflow_task))
.fetch_one(&self.db_pool)
.await
.map_err(Into::into)
}
/// Update task execution record
async fn update_task_execution_record(
&self,
task_execution_id: Id,
result: &TaskExecutionResult,
) -> Result<()> {
let status = match result.status {
TaskExecutionStatus::Success => ExecutionStatus::Completed,
TaskExecutionStatus::Failed => ExecutionStatus::Failed,
TaskExecutionStatus::Timeout => ExecutionStatus::Timeout,
TaskExecutionStatus::Skipped => ExecutionStatus::Cancelled,
};
// Fetch current execution to get workflow_task metadata
let execution =
sqlx::query_as::<_, Execution>("SELECT * FROM attune.execution WHERE id = $1")
.bind(task_execution_id)
.fetch_one(&self.db_pool)
.await?;
// Update workflow_task metadata
if let Some(mut workflow_task) = execution.workflow_task {
workflow_task.completed_at = if result.status == TaskExecutionStatus::Success {
Some(Utc::now())
} else {
None
};
workflow_task.duration_ms = Some(result.duration_ms);
workflow_task.retry_count = result.retry_count;
workflow_task.next_retry_at = result.next_retry_at;
workflow_task.timed_out = result.status == TaskExecutionStatus::Timeout;
let _error_json = result.error.as_ref().map(|e| {
json!({
"message": e.message,
"type": e.error_type,
"details": e.details
})
});
sqlx::query(
r#"
UPDATE attune.execution
SET
status = $1,
result = $2,
workflow_task = $3,
updated = NOW()
WHERE id = $4
"#,
)
.bind(status)
.bind(&result.output)
.bind(sqlx::types::Json(&workflow_task))
.bind(task_execution_id)
.execute(&self.db_pool)
.await?;
}
Ok(())
}
/// Clone reference for Arc sharing
fn clone_ref(&self) -> Self {
Self {
db_pool: self.db_pool.clone(),
mq: self.mq.clone(),
task_executor: TaskExecutor::new(self.db_pool.clone(), self.mq.clone()),
}
}
}
/// Workflow execution state
#[derive(Debug, Clone)]
pub struct WorkflowExecutionState {
pub context: WorkflowContext,
pub status: ExecutionStatus,
pub completed_tasks: HashSet<String>,
pub failed_tasks: HashSet<String>,
pub skipped_tasks: HashSet<String>,
/// Tasks currently executing
pub executing_tasks: HashSet<String>,
/// Tasks scheduled but not yet executing
pub scheduled_tasks: HashSet<String>,
/// Join state tracking: task_name -> set of completed predecessor tasks
pub join_state: HashMap<String, HashSet<String>>,
pub task_executions: HashMap<String, Vec<Id>>,
pub paused: bool,
pub pause_reason: Option<String>,
pub error_message: Option<String>,
}
/// Handle for managing a workflow execution
pub struct WorkflowExecutionHandle {
coordinator: Arc<WorkflowCoordinator>,
execution_id: Id,
parent_execution_id: Id,
#[allow(dead_code)]
workflow_def_id: Id,
graph: TaskGraph,
state: Arc<Mutex<WorkflowExecutionState>>,
}
impl WorkflowExecutionHandle {
/// Execute the workflow to completion
pub async fn execute(&self) -> Result<WorkflowExecutionResult> {
info!("Executing workflow {}", self.execution_id);
// Start with entry point tasks
{
let mut state = self.state.lock().await;
for task_name in &self.graph.entry_points {
info!("Scheduling entry point task: {}", task_name);
state.scheduled_tasks.insert(task_name.clone());
}
}
// Wait for all tasks to complete
loop {
// Check for and spawn scheduled tasks
let tasks_to_spawn = {
let mut state = self.state.lock().await;
let mut to_spawn = Vec::new();
for task_name in state.scheduled_tasks.iter() {
to_spawn.push(task_name.clone());
}
// Clear scheduled tasks as we're about to spawn them
state.scheduled_tasks.clear();
to_spawn
};
// Spawn scheduled tasks
for task_name in tasks_to_spawn {
self.spawn_task_execution(task_name).await;
}
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
let state = self.state.lock().await;
// Check if workflow is paused
if state.paused {
info!("Workflow {} is paused", self.execution_id);
break;
}
// Check if workflow is complete (nothing executing and nothing scheduled)
if state.executing_tasks.is_empty() && state.scheduled_tasks.is_empty() {
info!("Workflow {} completed", self.execution_id);
drop(state);
let mut state = self.state.lock().await;
if state.failed_tasks.is_empty() {
state.status = ExecutionStatus::Completed;
} else {
state.status = ExecutionStatus::Failed;
state.error_message = Some(format!(
"Workflow failed: {} tasks failed",
state.failed_tasks.len()
));
}
self.coordinator
.update_workflow_execution_state(self.execution_id, &state)
.await?;
break;
}
}
let state = self.state.lock().await;
Ok(WorkflowExecutionResult {
status: state.status,
output: state.context.export(),
completed_tasks: state.completed_tasks.len(),
failed_tasks: state.failed_tasks.len(),
skipped_tasks: state.skipped_tasks.len(),
error_message: state.error_message.clone(),
})
}
/// Spawn a task execution in a new tokio task
async fn spawn_task_execution(&self, task_name: String) {
let coordinator = self.coordinator.clone();
let state_arc = self.state.clone();
let workflow_execution_id = self.execution_id;
let parent_execution_id = self.parent_execution_id;
let graph = self.graph.clone();
tokio::spawn(async move {
if let Err(e) = Self::execute_task_async(
coordinator,
state_arc,
workflow_execution_id,
parent_execution_id,
graph,
task_name,
)
.await
{
error!("Task execution failed: {}", e);
}
});
}
/// Execute a single task asynchronously
async fn execute_task_async(
coordinator: Arc<WorkflowCoordinator>,
state: Arc<Mutex<WorkflowExecutionState>>,
workflow_execution_id: Id,
parent_execution_id: Id,
graph: TaskGraph,
task_name: String,
) -> Result<()> {
// Move task from scheduled to executing
let task = {
let mut state = state.lock().await;
state.scheduled_tasks.remove(&task_name);
state.executing_tasks.insert(task_name.clone());
// Get the task node
match graph.get_task(&task_name) {
Some(task) => task.clone(),
None => {
error!("Task {} not found in graph", task_name);
return Ok(());
}
}
};
info!("Executing task: {}", task.name);
// Create task execution record
let task_execution = coordinator
.create_task_execution_record(
workflow_execution_id,
parent_execution_id,
&task,
None,
None,
)
.await?;
// Get context for execution
let mut context = {
let state = state.lock().await;
state.context.clone()
};
// Execute task
let result = coordinator
.task_executor
.execute_task(
&task,
&mut context,
workflow_execution_id,
parent_execution_id,
)
.await?;
// Update task execution record
coordinator
.update_task_execution_record(task_execution.id, &result)
.await?;
// Update workflow state based on result
let success = matches!(result.status, TaskExecutionStatus::Success);
{
let mut state = state.lock().await;
state.executing_tasks.remove(&task.name);
match result.status {
TaskExecutionStatus::Success => {
state.completed_tasks.insert(task.name.clone());
// Update context with task result
if let Some(output) = result.output {
state.context.set_task_result(&task.name, output);
}
}
TaskExecutionStatus::Failed => {
if result.should_retry {
// Task will be retried, keep it in scheduled
info!("Task {} will be retried", task.name);
state.scheduled_tasks.insert(task.name.clone());
// TODO: Schedule retry with delay
} else {
state.failed_tasks.insert(task.name.clone());
if let Some(ref error) = result.error {
warn!("Task {} failed: {}", task.name, error.message);
}
}
}
TaskExecutionStatus::Timeout => {
state.failed_tasks.insert(task.name.clone());
warn!("Task {} timed out", task.name);
}
TaskExecutionStatus::Skipped => {
state.skipped_tasks.insert(task.name.clone());
debug!("Task {} skipped", task.name);
}
}
// Persist state
coordinator
.update_workflow_execution_state(workflow_execution_id, &state)
.await?;
}
// Evaluate transitions and schedule next tasks
Self::on_task_completion(state.clone(), graph.clone(), task.name.clone(), success).await?;
Ok(())
}
/// Handle task completion by evaluating transitions and scheduling next tasks
async fn on_task_completion(
state: Arc<Mutex<WorkflowExecutionState>>,
graph: TaskGraph,
completed_task: String,
success: bool,
) -> Result<()> {
// Get next tasks based on transitions
let next_tasks = graph.next_tasks(&completed_task, success);
info!(
"Task {} completed (success={}), next tasks: {:?}",
completed_task, success, next_tasks
);
// Collect tasks to schedule
let mut tasks_to_schedule = Vec::new();
for next_task_name in next_tasks {
let mut state = state.lock().await;
// Check if task already scheduled or executing
if state.scheduled_tasks.contains(&next_task_name)
|| state.executing_tasks.contains(&next_task_name)
{
continue;
}
if let Some(task_node) = graph.get_task(&next_task_name) {
// Check join conditions
if let Some(join_count) = task_node.join {
// Update join state
let join_completions = state
.join_state
.entry(next_task_name.clone())
.or_insert_with(HashSet::new);
join_completions.insert(completed_task.clone());
// Check if join is satisfied
if join_completions.len() >= join_count {
info!(
"Join condition satisfied for task {}: {}/{} completed",
next_task_name,
join_completions.len(),
join_count
);
state.scheduled_tasks.insert(next_task_name.clone());
tasks_to_schedule.push(next_task_name);
} else {
info!(
"Join condition not yet satisfied for task {}: {}/{} completed",
next_task_name,
join_completions.len(),
join_count
);
}
} else {
// No join, schedule immediately
state.scheduled_tasks.insert(next_task_name.clone());
tasks_to_schedule.push(next_task_name);
}
} else {
error!("Next task {} not found in graph", next_task_name);
}
}
Ok(())
}
/// Pause workflow execution
pub async fn pause(&self, reason: Option<String>) -> Result<()> {
let mut state = self.state.lock().await;
state.paused = true;
state.pause_reason = reason;
self.coordinator
.update_workflow_execution_state(self.execution_id, &state)
.await?;
info!("Workflow {} paused", self.execution_id);
Ok(())
}
/// Resume workflow execution
pub async fn resume(&self) -> Result<()> {
let mut state = self.state.lock().await;
state.paused = false;
state.pause_reason = None;
self.coordinator
.update_workflow_execution_state(self.execution_id, &state)
.await?;
info!("Workflow {} resumed", self.execution_id);
Ok(())
}
/// Cancel workflow execution
pub async fn cancel(&self) -> Result<()> {
let mut state = self.state.lock().await;
state.status = ExecutionStatus::Cancelled;
self.coordinator
.update_workflow_execution_state(self.execution_id, &state)
.await?;
info!("Workflow {} cancelled", self.execution_id);
Ok(())
}
/// Get current execution status
pub async fn status(&self) -> WorkflowExecutionStatus {
let state = self.state.lock().await;
WorkflowExecutionStatus {
execution_id: self.execution_id,
status: state.status,
completed_tasks: state.completed_tasks.len(),
failed_tasks: state.failed_tasks.len(),
skipped_tasks: state.skipped_tasks.len(),
executing_tasks: state.executing_tasks.iter().cloned().collect(),
scheduled_tasks: state.scheduled_tasks.iter().cloned().collect(),
total_tasks: self.graph.nodes.len(),
paused: state.paused,
}
}
}
/// Result of workflow execution
#[derive(Debug, Clone)]
pub struct WorkflowExecutionResult {
pub status: ExecutionStatus,
pub output: JsonValue,
pub completed_tasks: usize,
pub failed_tasks: usize,
pub skipped_tasks: usize,
pub error_message: Option<String>,
}
/// Current status of workflow execution
#[derive(Debug, Clone)]
pub struct WorkflowExecutionStatus {
pub execution_id: Id,
pub status: ExecutionStatus,
pub completed_tasks: usize,
pub failed_tasks: usize,
pub skipped_tasks: usize,
pub executing_tasks: Vec<String>,
pub scheduled_tasks: Vec<String>,
pub total_tasks: usize,
pub paused: bool,
}
#[cfg(test)]
mod tests {
// Note: These tests require a database connection and are integration tests
// They should be run with `cargo test --features integration-tests`
#[tokio::test]
#[ignore] // Requires database
async fn test_workflow_coordinator_creation() {
// This is a placeholder test
// Actual tests would require database setup
assert!(true);
}
}

11
crates/executor/src/workflow/graph.rs
View File

@@ -21,9 +21,6 @@ pub type GraphResult<T> = Result<T, GraphError>;
pub enum GraphError {
#[error("Invalid task reference: {0}")]
InvalidTaskReference(String),
#[error("Graph building error: {0}")]
BuildError(String),
}
/// Executable task graph
@@ -197,6 +194,7 @@ impl TaskGraph {
}
/// Get all tasks that can transition into the given task (inbound edges)
#[allow(dead_code)] // Part of complete graph API; used in tests
pub fn get_inbound_tasks(&self, task_name: &str) -> Vec<String> {
self.inbound_edges
.get(task_name)
@@ -221,7 +219,8 @@ impl TaskGraph {
/// * `success` - Whether the task succeeded
///
/// # Returns
/// A vector of (task_name, publish_vars) tuples to schedule next
/// A vector of task names to schedule next
#[allow(dead_code)] // Part of complete graph API; used in tests
pub fn next_tasks(&self, task_name: &str, success: bool) -> Vec<String> {
let mut next = Vec::new();
@@ -251,7 +250,8 @@ impl TaskGraph {
/// Get the next tasks with full transition information.
///
/// Returns matching transitions with their publish directives and targets,
/// giving the coordinator full context for variable publishing.
/// giving the caller full context for variable publishing.
#[allow(dead_code)] // Part of complete graph API; used in tests
pub fn matching_transitions(&self, task_name: &str, success: bool) -> Vec<&GraphTransition> {
let mut matching = Vec::new();
@@ -275,6 +275,7 @@ impl TaskGraph {
}
/// Collect all unique target task names from all transitions of a given task.
#[allow(dead_code)] // Part of complete graph API; used in tests
pub fn all_transition_targets(&self, task_name: &str) -> HashSet<String> {
let mut targets = HashSet::new();
if let Some(node) = self.nodes.get(task_name) {

54
crates/executor/src/workflow/mod.rs
View File

@@ -1,60 +1,12 @@
//! Workflow orchestration module
//!
//! This module provides workflow execution, orchestration, parsing, validation,
//! and template rendering capabilities for the Attune workflow orchestration system.
//! This module provides workflow execution context, graph building, and
//! orchestration capabilities for the Attune workflow engine.
//!
//! # Modules
//!
//! - `parser`: Parse YAML workflow definitions into structured types
//! - `graph`: Build executable task graphs from workflow definitions
//! - `context`: Manage workflow execution context and variables
//! - `task_executor`: Execute individual workflow tasks
//! - `coordinator`: Orchestrate workflow execution with state management
//! - `template`: Template engine for variable interpolation (Jinja2-like syntax)
//!
//! # Example
//!
//! ```no_run
//! use attune_executor::workflow::{parse_workflow_yaml, WorkflowCoordinator};
//!
//! // Parse a workflow YAML file
//! let yaml = r#"
//! ref: my_pack.my_workflow
//! label: My Workflow
//! version: 1.0.0
//! tasks:
//! - name: hello
//! action: core.echo
//! input:
//! message: "{{ parameters.name }}"
//! "#;
//!
//! let workflow = parse_workflow_yaml(yaml).expect("Failed to parse workflow");
//! ```
//! - `graph`: Build executable task graphs from workflow definitions
// Phase 2: Workflow Execution Engine
pub mod context;
pub mod coordinator;
pub mod graph;
pub mod task_executor;
pub mod template;
// Re-export workflow utilities from common crate
pub use attune_common::workflow::{
parse_workflow_file, parse_workflow_yaml, workflow_to_json, BackoffStrategy, DecisionBranch,
LoadedWorkflow, LoaderConfig, ParseError, ParseResult, PublishDirective, RegistrationOptions,
RegistrationResult, RetryConfig, Task, TaskType, ValidationError, ValidationResult,
WorkflowDefinition, WorkflowFile, WorkflowLoader, WorkflowRegistrar, WorkflowValidator,
};
// Re-export Phase 2 components
pub use context::{ContextError, ContextResult, WorkflowContext};
pub use coordinator::{
WorkflowCoordinator, WorkflowExecutionHandle, WorkflowExecutionResult, WorkflowExecutionState,
WorkflowExecutionStatus,
};
pub use graph::{GraphError, GraphResult, GraphTransition, TaskGraph, TaskNode};
pub use task_executor::{
TaskExecutionError, TaskExecutionResult, TaskExecutionStatus, TaskExecutor,
};
pub use template::{TemplateEngine, TemplateError, TemplateResult, VariableContext, VariableScope};

871
crates/executor/src/workflow/task_executor.rs
View File

@@ -1,871 +0,0 @@
//! Task Executor
//!
//! This module handles the execution of individual workflow tasks,
//! including action invocation, retries, timeouts, and with-items iteration.
use crate::workflow::context::WorkflowContext;
use crate::workflow::graph::{BackoffStrategy, RetryConfig, TaskNode};
use attune_common::error::{Error, Result};
use attune_common::models::Id;
use attune_common::mq::MessageQueue;
use chrono::{DateTime, Utc};
use serde_json::{json, Value as JsonValue};
use sqlx::PgPool;
use std::time::Duration;
use tokio::time::timeout;
use tracing::{debug, error, info, warn};
/// Task execution result
#[derive(Debug, Clone)]
pub struct TaskExecutionResult {
/// Execution status
pub status: TaskExecutionStatus,
/// Task output/result
pub output: Option<JsonValue>,
/// Error information
pub error: Option<TaskExecutionError>,
/// Execution duration in milliseconds
pub duration_ms: i64,
/// Whether the task should be retried
pub should_retry: bool,
/// Next retry time (if applicable)
pub next_retry_at: Option<DateTime<Utc>>,
/// Number of retries performed
pub retry_count: i32,
}
/// Task execution status
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TaskExecutionStatus {
Success,
Failed,
Timeout,
Skipped,
}
/// Task execution error
#[derive(Debug, Clone)]
pub struct TaskExecutionError {
pub message: String,
pub error_type: String,
pub details: Option<JsonValue>,
}
/// Task executor
pub struct TaskExecutor {
db_pool: PgPool,
mq: MessageQueue,
}
impl TaskExecutor {
/// Create a new task executor
pub fn new(db_pool: PgPool, mq: MessageQueue) -> Self {
Self { db_pool, mq }
}
/// Execute a task
pub async fn execute_task(
&self,
task: &TaskNode,
context: &mut WorkflowContext,
workflow_execution_id: Id,
parent_execution_id: Id,
) -> Result<TaskExecutionResult> {
info!("Executing task: {}", task.name);
let start_time = Utc::now();
// Check if task should be skipped (when condition)
if let Some(ref condition) = task.when {
match context.evaluate_condition(condition) {
Ok(should_run) => {
if !should_run {
info!("Task {} skipped due to when condition", task.name);
return Ok(TaskExecutionResult {
status: TaskExecutionStatus::Skipped,
output: None,
error: None,
duration_ms: 0,
should_retry: false,
next_retry_at: None,
retry_count: 0,
});
}
}
Err(e) => {
warn!(
"Failed to evaluate when condition for task {}: {}",
task.name, e
);
// Continue execution if condition evaluation fails
}
}
}
// Check if this is a with-items task
if let Some(ref with_items_expr) = task.with_items {
return self
.execute_with_items(
task,
context,
workflow_execution_id,
parent_execution_id,
with_items_expr,
)
.await;
}
// Execute single task
let result = self
.execute_single_task(task, context, workflow_execution_id, parent_execution_id, 0)
.await?;
let duration_ms = (Utc::now() - start_time).num_milliseconds();
// Store task result in context
if let Some(ref output) = result.output {
context.set_task_result(&task.name, output.clone());
// Publish variables from matching transitions
let success = matches!(result.status, TaskExecutionStatus::Success);
for transition in &task.transitions {
let should_fire = match transition.kind() {
super::graph::TransitionKind::Succeeded => success,
super::graph::TransitionKind::Failed => !success,
super::graph::TransitionKind::TimedOut => !success,
super::graph::TransitionKind::Always => true,
super::graph::TransitionKind::Custom => true,
};
if should_fire && !transition.publish.is_empty() {
let var_names: Vec<String> =
transition.publish.iter().map(|p| p.name.clone()).collect();
if let Err(e) = context.publish_from_result(output, &var_names, None) {
warn!("Failed to publish variables for task {}: {}", task.name, e);
}
}
}
}
Ok(TaskExecutionResult {
duration_ms,
..result
})
}
/// Execute a single task (without with-items iteration)
async fn execute_single_task(
&self,
task: &TaskNode,
context: &WorkflowContext,
workflow_execution_id: Id,
parent_execution_id: Id,
retry_count: i32,
) -> Result<TaskExecutionResult> {
let start_time = Utc::now();
// Render task input
let input = match context.render_json(&task.input) {
Ok(rendered) => rendered,
Err(e) => {
error!("Failed to render task input for {}: {}", task.name, e);
return Ok(TaskExecutionResult {
status: TaskExecutionStatus::Failed,
output: None,
error: Some(TaskExecutionError {
message: format!("Failed to render task input: {}", e),
error_type: "template_error".to_string(),
details: None,
}),
duration_ms: 0,
should_retry: false,
next_retry_at: None,
retry_count,
});
}
};
// Execute based on task type
let result = match task.task_type {
attune_common::workflow::TaskType::Action => {
self.execute_action(task, input, workflow_execution_id, parent_execution_id)
.await
}
attune_common::workflow::TaskType::Parallel => {
self.execute_parallel(task, context, workflow_execution_id, parent_execution_id)
.await
}
attune_common::workflow::TaskType::Workflow => {
self.execute_workflow(task, input, workflow_execution_id, parent_execution_id)
.await
}
};
let duration_ms = (Utc::now() - start_time).num_milliseconds();
// Apply timeout if specified
let result = if let Some(timeout_secs) = task.timeout {
self.apply_timeout(result, timeout_secs).await
} else {
result
};
// Handle retries
let mut result = result?;
result.retry_count = retry_count;
if result.status == TaskExecutionStatus::Failed {
if let Some(ref retry_config) = task.retry {
if retry_count < retry_config.count as i32 {
// Check if we should retry based on error condition
let should_retry = if let Some(ref _on_error) = retry_config.on_error {
// TODO: Evaluate error condition
true
} else {
true
};
if should_retry {
result.should_retry = true;
result.next_retry_at =
Some(calculate_retry_time(retry_config, retry_count));
info!(
"Task {} failed, will retry (attempt {}/{})",
task.name,
retry_count + 1,
retry_config.count
);
}
}
}
}
result.duration_ms = duration_ms;
Ok(result)
}
/// Execute an action task
async fn execute_action(
&self,
task: &TaskNode,
input: JsonValue,
_workflow_execution_id: Id,
parent_execution_id: Id,
) -> Result<TaskExecutionResult> {
let action_ref = match &task.action {
Some(action) => action,
None => {
return Ok(TaskExecutionResult {
status: TaskExecutionStatus::Failed,
output: None,
error: Some(TaskExecutionError {
message: "Action task missing action reference".to_string(),
error_type: "configuration_error".to_string(),
details: None,
}),
duration_ms: 0,
should_retry: false,
next_retry_at: None,
retry_count: 0,
});
}
};
debug!("Executing action: {} with input: {:?}", action_ref, input);
// Create execution record in database
let execution = sqlx::query_as::<_, attune_common::models::Execution>(
r#"
INSERT INTO attune.execution (action_ref, input, parent, status)
VALUES ($1, $2, $3, $4)
RETURNING *
"#,
)
.bind(action_ref)
.bind(&input)
.bind(parent_execution_id)
.bind(attune_common::models::ExecutionStatus::Scheduled)
.fetch_one(&self.db_pool)
.await?;
// Queue action for execution by worker
// TODO: Implement proper message queue publishing
info!(
"Created action execution {} for task {} (queuing not yet implemented)",
execution.id, task.name
);
// For now, return pending status
// In a real implementation, we would wait for completion via message queue
Ok(TaskExecutionResult {
status: TaskExecutionStatus::Success,
output: Some(json!({
"execution_id": execution.id,
"status": "queued"
})),
error: None,
duration_ms: 0,
should_retry: false,
next_retry_at: None,
retry_count: 0,
})
}
/// Execute parallel tasks
async fn execute_parallel(
&self,
task: &TaskNode,
context: &WorkflowContext,
workflow_execution_id: Id,
parent_execution_id: Id,
) -> Result<TaskExecutionResult> {
let sub_tasks = match &task.sub_tasks {
Some(tasks) => tasks,
None => {
return Ok(TaskExecutionResult {
status: TaskExecutionStatus::Failed,
output: None,
error: Some(TaskExecutionError {
message: "Parallel task missing sub-tasks".to_string(),
error_type: "configuration_error".to_string(),
details: None,
}),
duration_ms: 0,
should_retry: false,
next_retry_at: None,
retry_count: 0,
});
}
};
info!("Executing {} parallel tasks", sub_tasks.len());
// Execute all sub-tasks in parallel
let mut futures = Vec::new();
for subtask in sub_tasks {
let subtask_clone = subtask.clone();
let subtask_name = subtask.name.clone();
let context = context.clone();
let db_pool = self.db_pool.clone();
let mq = self.mq.clone();
let future = async move {
let executor = TaskExecutor::new(db_pool, mq);
let result = executor
.execute_single_task(
&subtask_clone,
&context,
workflow_execution_id,
parent_execution_id,
0,
)
.await;
(subtask_name, result)
};
futures.push(future);
}
// Wait for all tasks to complete
let task_results = futures::future::join_all(futures).await;
let mut results = Vec::new();
let mut all_succeeded = true;
let mut errors = Vec::new();
for (task_name, result) in task_results {
match result {
Ok(result) => {
if result.status != TaskExecutionStatus::Success {
all_succeeded = false;
if let Some(error) = &result.error {
errors.push(json!({
"task": task_name,
"error": error.message
}));
}
}
results.push(json!({
"task": task_name,
"status": format!("{:?}", result.status),
"output": result.output
}));
}
Err(e) => {
all_succeeded = false;
errors.push(json!({
"task": task_name,
"error": e.to_string()
}));
}
}
}
let status = if all_succeeded {
TaskExecutionStatus::Success
} else {
TaskExecutionStatus::Failed
};
Ok(TaskExecutionResult {
status,
output: Some(json!({
"results": results
})),
error: if errors.is_empty() {
None
} else {
Some(TaskExecutionError {
message: format!("{} parallel tasks failed", errors.len()),
error_type: "parallel_execution_error".to_string(),
details: Some(json!({"errors": errors})),
})
},
duration_ms: 0,
should_retry: false,
next_retry_at: None,
retry_count: 0,
})
}
/// Execute a workflow task (nested workflow)
async fn execute_workflow(
&self,
_task: &TaskNode,
_input: JsonValue,
_workflow_execution_id: Id,
_parent_execution_id: Id,
) -> Result<TaskExecutionResult> {
// TODO: Implement nested workflow execution
// For now, return not implemented
warn!("Workflow task execution not yet implemented");
Ok(TaskExecutionResult {
status: TaskExecutionStatus::Failed,
output: None,
error: Some(TaskExecutionError {
message: "Nested workflow execution not yet implemented".to_string(),
error_type: "not_implemented".to_string(),
details: None,
}),
duration_ms: 0,
should_retry: false,
next_retry_at: None,
retry_count: 0,
})
}
/// Execute task with with-items iteration
async fn execute_with_items(
&self,
task: &TaskNode,
context: &mut WorkflowContext,
workflow_execution_id: Id,
parent_execution_id: Id,
items_expr: &str,
) -> Result<TaskExecutionResult> {
// Render items expression
let items_str = context.render_template(items_expr).map_err(|e| {
Error::validation(format!("Failed to render with-items expression: {}", e))
})?;
// Parse items (should be a JSON array)
let items: Vec<JsonValue> = serde_json::from_str(&items_str).map_err(|e| {
Error::validation(format!(
"with-items expression did not produce valid JSON array: {}",
e
))
})?;
info!("Executing task {} with {} items", task.name, items.len());
let items_len = items.len(); // Store length before consuming items
let concurrency = task.concurrency.unwrap_or(10);
let mut all_results = Vec::new();
let mut all_succeeded = true;
let mut errors = Vec::new();
// Check if batch processing is enabled
if let Some(batch_size) = task.batch_size {
// Batch mode: split items into batches and pass as arrays
debug!(
"Processing {} items in batches of {} (batch mode)",
items.len(),
batch_size
);
let batches: Vec<Vec<JsonValue>> = items
.chunks(batch_size)
.map(|chunk| chunk.to_vec())
.collect();
debug!("Created {} batches", batches.len());
// Execute batches with concurrency limit
let mut handles = Vec::new();
let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(concurrency));
for (batch_idx, batch) in batches.into_iter().enumerate() {
let permit = semaphore.clone().acquire_owned().await.unwrap();
let executor = TaskExecutor::new(self.db_pool.clone(), self.mq.clone());
let task = task.clone();
let mut batch_context = context.clone();
// Set current_item to the batch array
batch_context.set_current_item(json!(batch), batch_idx);
let handle = tokio::spawn(async move {
let result = executor
.execute_single_task(
&task,
&batch_context,
workflow_execution_id,
parent_execution_id,
0,
)
.await;
drop(permit);
(batch_idx, result)
});
handles.push(handle);
}
// Wait for all batches to complete
for handle in handles {
match handle.await {
Ok((batch_idx, Ok(result))) => {
if result.status != TaskExecutionStatus::Success {
all_succeeded = false;
if let Some(error) = &result.error {
errors.push(json!({
"batch": batch_idx,
"error": error.message
}));
}
}
all_results.push(json!({
"batch": batch_idx,
"status": format!("{:?}", result.status),
"output": result.output
}));
}
Ok((batch_idx, Err(e))) => {
all_succeeded = false;
errors.push(json!({
"batch": batch_idx,
"error": e.to_string()
}));
}
Err(e) => {
all_succeeded = false;
errors.push(json!({
"error": format!("Task panicked: {}", e)
}));
}
}
}
} else {
// Individual mode: process each item separately
debug!(
"Processing {} items individually (no batch_size specified)",
items.len()
);
// Execute items with concurrency limit
let mut handles = Vec::new();
let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(concurrency));
for (item_idx, item) in items.into_iter().enumerate() {
let permit = semaphore.clone().acquire_owned().await.unwrap();
let executor = TaskExecutor::new(self.db_pool.clone(), self.mq.clone());
let task = task.clone();
let mut item_context = context.clone();
// Set current_item to the individual item
item_context.set_current_item(item, item_idx);
let handle = tokio::spawn(async move {
let result = executor
.execute_single_task(
&task,
&item_context,
workflow_execution_id,
parent_execution_id,
0,
)
.await;
drop(permit);
(item_idx, result)
});
handles.push(handle);
}
// Wait for all items to complete
for handle in handles {
match handle.await {
Ok((idx, Ok(result))) => {
if result.status != TaskExecutionStatus::Success {
all_succeeded = false;
if let Some(error) = &result.error {
errors.push(json!({
"index": idx,
"error": error.message
}));
}
}
all_results.push(json!({
"index": idx,
"status": format!("{:?}", result.status),
"output": result.output
}));
}
Ok((idx, Err(e))) => {
all_succeeded = false;
errors.push(json!({
"index": idx,
"error": e.to_string()
}));
}
Err(e) => {
all_succeeded = false;
errors.push(json!({
"error": format!("Task panicked: {}", e)
}));
}
}
}
}
context.clear_current_item();
let status = if all_succeeded {
TaskExecutionStatus::Success
} else {
TaskExecutionStatus::Failed
};
Ok(TaskExecutionResult {
status,
output: Some(json!({
"results": all_results,
"total": items_len
})),
error: if errors.is_empty() {
None
} else {
Some(TaskExecutionError {
message: format!("{} items failed", errors.len()),
error_type: "with_items_error".to_string(),
details: Some(json!({"errors": errors})),
})
},
duration_ms: 0,
should_retry: false,
next_retry_at: None,
retry_count: 0,
})
}
/// Apply timeout to task execution
async fn apply_timeout(
&self,
result_future: Result<TaskExecutionResult>,
timeout_secs: u32,
) -> Result<TaskExecutionResult> {
match timeout(Duration::from_secs(timeout_secs as u64), async {
result_future
})
.await
{
Ok(result) => result,
Err(_) => {
warn!("Task execution timed out after {} seconds", timeout_secs);
Ok(TaskExecutionResult {
status: TaskExecutionStatus::Timeout,
output: None,
error: Some(TaskExecutionError {
message: format!("Task timed out after {} seconds", timeout_secs),
error_type: "timeout".to_string(),
details: None,
}),
duration_ms: (timeout_secs * 1000) as i64,
should_retry: false,
next_retry_at: None,
retry_count: 0,
})
}
}
}
}
/// Calculate next retry time based on retry configuration
fn calculate_retry_time(config: &RetryConfig, retry_count: i32) -> DateTime<Utc> {
let delay_secs = match config.backoff {
BackoffStrategy::Constant => config.delay,
BackoffStrategy::Linear => config.delay * (retry_count as u32 + 1),
BackoffStrategy::Exponential => {
let exp_delay = config.delay * 2_u32.pow(retry_count as u32);
if let Some(max_delay) = config.max_delay {
exp_delay.min(max_delay)
} else {
exp_delay
}
}
};
Utc::now() + chrono::Duration::seconds(delay_secs as i64)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_calculate_retry_time_constant() {
let config = RetryConfig {
count: 3,
delay: 10,
backoff: BackoffStrategy::Constant,
max_delay: None,
on_error: None,
};
let now = Utc::now();
let retry_time = calculate_retry_time(&config, 0);
let diff = (retry_time - now).num_seconds();
assert!(diff >= 9 && diff <= 11); // Allow 1 second tolerance
}
#[test]
fn test_calculate_retry_time_exponential() {
let config = RetryConfig {
count: 3,
delay: 10,
backoff: BackoffStrategy::Exponential,
max_delay: Some(100),
on_error: None,
};
let now = Utc::now();
// First retry: 10 * 2^0 = 10
let retry1 = calculate_retry_time(&config, 0);
assert!((retry1 - now).num_seconds() >= 9 && (retry1 - now).num_seconds() <= 11);
// Second retry: 10 * 2^1 = 20
let retry2 = calculate_retry_time(&config, 1);
assert!((retry2 - now).num_seconds() >= 19 && (retry2 - now).num_seconds() <= 21);
// Third retry: 10 * 2^2 = 40
let retry3 = calculate_retry_time(&config, 2);
assert!((retry3 - now).num_seconds() >= 39 && (retry3 - now).num_seconds() <= 41);
}
#[test]
fn test_calculate_retry_time_exponential_with_max() {
let config = RetryConfig {
count: 10,
delay: 10,
backoff: BackoffStrategy::Exponential,
max_delay: Some(100),
on_error: None,
};
let now = Utc::now();
// Retry with high count should be capped at max_delay
let retry = calculate_retry_time(&config, 10);
assert!((retry - now).num_seconds() >= 99 && (retry - now).num_seconds() <= 101);
}
#[test]
fn test_with_items_batch_creation() {
use serde_json::json;
// Test batch_size=3 with 7 items
let items = vec![
json!({"id": 1}),
json!({"id": 2}),
json!({"id": 3}),
json!({"id": 4}),
json!({"id": 5}),
json!({"id": 6}),
json!({"id": 7}),
];
let batch_size = 3;
let batches: Vec<Vec<JsonValue>> = items
.chunks(batch_size)
.map(|chunk| chunk.to_vec())
.collect();
// Should create 3 batches: [1,2,3], [4,5,6], [7]
assert_eq!(batches.len(), 3);
assert_eq!(batches[0].len(), 3);
assert_eq!(batches[1].len(), 3);
assert_eq!(batches[2].len(), 1); // Last batch can be smaller
// Verify content - batches are arrays
assert_eq!(batches[0][0], json!({"id": 1}));
assert_eq!(batches[2][0], json!({"id": 7}));
}
#[test]
fn test_with_items_no_batch_size_individual_processing() {
use serde_json::json;
// Without batch_size, items are processed individually
let items = vec![json!({"id": 1}), json!({"id": 2}), json!({"id": 3})];
// Each item should be processed separately (not as batches)
assert_eq!(items.len(), 3);
// Verify individual items
assert_eq!(items[0], json!({"id": 1}));
assert_eq!(items[1], json!({"id": 2}));
assert_eq!(items[2], json!({"id": 3}));
}
#[test]
fn test_with_items_batch_vs_individual() {
use serde_json::json;
let items = vec![json!({"id": 1}), json!({"id": 2}), json!({"id": 3})];
// With batch_size: items are grouped into batches (arrays)
let batch_size = Some(2);
if let Some(bs) = batch_size {
let batches: Vec<Vec<JsonValue>> = items
.clone()
.chunks(bs)
.map(|chunk| chunk.to_vec())
.collect();
// 2 batches: [1,2], [3]
assert_eq!(batches.len(), 2);
assert_eq!(batches[0], vec![json!({"id": 1}), json!({"id": 2})]);
assert_eq!(batches[1], vec![json!({"id": 3})]);
}
// Without batch_size: items processed individually
let batch_size: Option<usize> = None;
if batch_size.is_none() {
// Each item is a single value, not wrapped in array
for (idx, item) in items.iter().enumerate() {
assert_eq!(item["id"], idx + 1);
}
}
}
}

360
crates/executor/src/workflow/template.rs
View File

@@ -1,360 +0,0 @@
//! Template engine for workflow variable interpolation
//!
//! This module provides template rendering using Tera (Jinja2-like syntax)
//! with support for multi-scope variable contexts.
use serde_json::Value as JsonValue;
use std::collections::HashMap;
use tera::{Context, Tera};
/// Result type for template operations
pub type TemplateResult<T> = Result<T, TemplateError>;
/// Errors that can occur during template rendering
#[derive(Debug, thiserror::Error)]
pub enum TemplateError {
#[error("Template rendering error: {0}")]
RenderError(#[from] tera::Error),
#[error("Invalid template syntax: {0}")]
SyntaxError(String),
#[error("Variable not found: {0}")]
VariableNotFound(String),
#[error("JSON serialization error: {0}")]
JsonError(#[from] serde_json::Error),
#[error("Invalid scope: {0}")]
InvalidScope(String),
}
/// Variable scope priority (higher number = higher priority)
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum VariableScope {
/// System-level variables (lowest priority)
System = 1,
/// Key-value store variables
KeyValue = 2,
/// Pack configuration
PackConfig = 3,
/// Workflow parameters (input)
Parameters = 4,
/// Workflow vars (defined in workflow)
Vars = 5,
/// Task-specific variables (highest priority)
Task = 6,
}
/// Template engine with multi-scope variable context
pub struct TemplateEngine {
// Note: We can't use custom filters with Tera::one_off, so we need to keep tera instance
// But Tera doesn't expose a way to register templates without files in the new() constructor
// So we'll just use one_off for now and skip custom filters in basic rendering
}
impl Default for TemplateEngine {
fn default() -> Self {
Self::new()
}
}
impl TemplateEngine {
/// Create a new template engine
pub fn new() -> Self {
Self {}
}
/// Render a template string with the given context
pub fn render(&self, template: &str, context: &VariableContext) -> TemplateResult<String> {
let tera_context = context.to_tera_context()?;
// Use one-off template rendering
// Note: Custom filters are not supported with one_off rendering
Tera::one_off(template, &tera_context, true).map_err(TemplateError::from)
}
/// Render a template and parse result as JSON
pub fn render_json(
&self,
template: &str,
context: &VariableContext,
) -> TemplateResult<JsonValue> {
let rendered = self.render(template, context)?;
serde_json::from_str(&rendered).map_err(TemplateError::from)
}
/// Check if a template string contains valid syntax
pub fn validate_template(&self, template: &str) -> TemplateResult<()> {
Tera::one_off(template, &Context::new(), true)
.map(|_| ())
.map_err(TemplateError::from)
}
}
/// Multi-scope variable context for template rendering
#[derive(Debug, Clone)]
pub struct VariableContext {
/// System-level variables
system: HashMap<String, JsonValue>,
/// Key-value store variables
kv: HashMap<String, JsonValue>,
/// Pack configuration
pack_config: HashMap<String, JsonValue>,
/// Workflow parameters (input)
parameters: HashMap<String, JsonValue>,
/// Workflow vars
vars: HashMap<String, JsonValue>,
/// Task results and metadata
task: HashMap<String, JsonValue>,
}
impl Default for VariableContext {
fn default() -> Self {
Self::new()
}
}
impl VariableContext {
/// Create a new empty variable context
pub fn new() -> Self {
Self {
system: HashMap::new(),
kv: HashMap::new(),
pack_config: HashMap::new(),
parameters: HashMap::new(),
vars: HashMap::new(),
task: HashMap::new(),
}
}
/// Set system variables
pub fn with_system(mut self, vars: HashMap<String, JsonValue>) -> Self {
self.system = vars;
self
}
/// Set key-value store variables
pub fn with_kv(mut self, vars: HashMap<String, JsonValue>) -> Self {
self.kv = vars;
self
}
/// Set pack configuration
pub fn with_pack_config(mut self, config: HashMap<String, JsonValue>) -> Self {
self.pack_config = config;
self
}
/// Set workflow parameters
pub fn with_parameters(mut self, params: HashMap<String, JsonValue>) -> Self {
self.parameters = params;
self
}
/// Set workflow vars
pub fn with_vars(mut self, vars: HashMap<String, JsonValue>) -> Self {
self.vars = vars;
self
}
/// Set task variables
pub fn with_task(mut self, task_vars: HashMap<String, JsonValue>) -> Self {
self.task = task_vars;
self
}
/// Add a single variable to a scope
pub fn set(&mut self, scope: VariableScope, key: String, value: JsonValue) {
match scope {
VariableScope::System => self.system.insert(key, value),
VariableScope::KeyValue => self.kv.insert(key, value),
VariableScope::PackConfig => self.pack_config.insert(key, value),
VariableScope::Parameters => self.parameters.insert(key, value),
VariableScope::Vars => self.vars.insert(key, value),
VariableScope::Task => self.task.insert(key, value),
};
}
/// Get a variable from any scope (respects priority)
pub fn get(&self, key: &str) -> Option<&JsonValue> {
// Check scopes in priority order (highest to lowest)
self.task
.get(key)
.or_else(|| self.vars.get(key))
.or_else(|| self.parameters.get(key))
.or_else(|| self.pack_config.get(key))
.or_else(|| self.kv.get(key))
.or_else(|| self.system.get(key))
}
/// Convert to Tera context for rendering
pub fn to_tera_context(&self) -> TemplateResult<Context> {
let mut context = Context::new();
// Insert scopes as nested objects
context.insert("system", &self.system);
context.insert("kv", &self.kv);
context.insert("pack", &serde_json::json!({ "config": self.pack_config }));
context.insert("parameters", &self.parameters);
context.insert("vars", &self.vars);
context.insert("task", &self.task);
Ok(context)
}
/// Merge another context into this one (preserves priority)
pub fn merge(&mut self, other: &VariableContext) {
self.system.extend(other.system.clone());
self.kv.extend(other.kv.clone());
self.pack_config.extend(other.pack_config.clone());
self.parameters.extend(other.parameters.clone());
self.vars.extend(other.vars.clone());
self.task.extend(other.task.clone());
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn test_basic_template_rendering() {
let engine = TemplateEngine::new();
let mut context = VariableContext::new();
context.set(
VariableScope::Parameters,
"name".to_string(),
json!("World"),
);
let result = engine.render("Hello {{ parameters.name }}!", &context);
assert!(result.is_ok());
assert_eq!(result.unwrap(), "Hello World!");
}
#[test]
fn test_scope_priority() {
let engine = TemplateEngine::new();
let mut context = VariableContext::new();
// Set same variable in multiple scopes
context.set(VariableScope::System, "value".to_string(), json!("system"));
context.set(VariableScope::Vars, "value".to_string(), json!("vars"));
context.set(VariableScope::Task, "value".to_string(), json!("task"));
// Task scope should win (highest priority)
let result = engine.render("{{ task.value }}", &context);
assert_eq!(result.unwrap(), "task");
}
#[test]
fn test_nested_variables() {
let engine = TemplateEngine::new();
let mut context = VariableContext::new();
context.set(
VariableScope::Parameters,
"config".to_string(),
json!({"database": {"host": "localhost", "port": 5432}}),
);
let result = engine.render(
"postgres://{{ parameters.config.database.host }}:{{ parameters.config.database.port }}",
&context,
);
assert_eq!(result.unwrap(), "postgres://localhost:5432");
}
// Note: Custom filter tests are disabled since we're using Tera::one_off
// which doesn't support custom filters. In production, we would need to
// use a pre-configured Tera instance with templates registered.
#[test]
fn test_json_operations() {
let engine = TemplateEngine::new();
let mut context = VariableContext::new();
context.set(
VariableScope::Parameters,
"data".to_string(),
json!({"key": "value"}),
);
// Test accessing JSON properties
let result = engine.render("{{ parameters.data.key }}", &context);
assert_eq!(result.unwrap(), "value");
}
#[test]
fn test_conditional_rendering() {
let engine = TemplateEngine::new();
let mut context = VariableContext::new();
context.set(
VariableScope::Parameters,
"env".to_string(),
json!("production"),
);
let result = engine.render(
"{% if parameters.env == 'production' %}prod{% else %}dev{% endif %}",
&context,
);
assert_eq!(result.unwrap(), "prod");
}
#[test]
fn test_loop_rendering() {
let engine = TemplateEngine::new();
let mut context = VariableContext::new();
context.set(
VariableScope::Parameters,
"items".to_string(),
json!(["a", "b", "c"]),
);
let result = engine.render(
"{% for item in parameters.items %}{{ item }}{% endfor %}",
&context,
);
assert_eq!(result.unwrap(), "abc");
}
#[test]
fn test_context_merge() {
let mut ctx1 = VariableContext::new();
ctx1.set(VariableScope::Vars, "a".to_string(), json!(1));
ctx1.set(VariableScope::Vars, "b".to_string(), json!(2));
let mut ctx2 = VariableContext::new();
ctx2.set(VariableScope::Vars, "b".to_string(), json!(3));
ctx2.set(VariableScope::Vars, "c".to_string(), json!(4));
ctx1.merge(&ctx2);
assert_eq!(ctx1.get("a"), Some(&json!(1)));
assert_eq!(ctx1.get("b"), Some(&json!(3))); // ctx2 overwrites
assert_eq!(ctx1.get("c"), Some(&json!(4)));
}
#[test]
fn test_all_scopes() {
let engine = TemplateEngine::new();
let context = VariableContext::new()
.with_system(HashMap::from([("sys_var".to_string(), json!("system"))]))
.with_kv(HashMap::from([("kv_var".to_string(), json!("keyvalue"))]))
.with_pack_config(HashMap::from([("setting".to_string(), json!("config"))]))
.with_parameters(HashMap::from([("param".to_string(), json!("parameter"))]))
.with_vars(HashMap::from([("var".to_string(), json!("variable"))]))
.with_task(HashMap::from([(
"result".to_string(),
json!("task_result"),
)]));
let template = "{{ system.sys_var }}-{{ kv.kv_var }}-{{ pack.config.setting }}-{{ parameters.param }}-{{ vars.var }}-{{ task.result }}";
let result = engine.render(template, &context);
assert_eq!(
result.unwrap(),
"system-keyvalue-config-parameter-variable-task_result"
);
}
}