Struct Engine 

pub struct Engine { /* private fields */ }

The workflow orchestration engine.

Holds references to the store, agent provider, and a registry of WorkflowHandlers.

Examples

use std::sync::Arc;
use ironflow_engine::engine::Engine;
use ironflow_engine::config::ShellConfig;
use ironflow_engine::handler::{WorkflowHandler, HandlerFuture, WorkflowInfo};
use ironflow_engine::context::WorkflowContext;
use ironflow_store::memory::InMemoryStore;
use ironflow_store::models::TriggerKind;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use serde_json::json;

struct CiWorkflow;
impl WorkflowHandler for CiWorkflow {
    fn name(&self) -> &str { "ci" }
    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            ctx.shell("test", ShellConfig::new("cargo test")).await?;
            Ok(())
        })
    }
}

let store = Arc::new(InMemoryStore::new());
let provider = Arc::new(ClaudeCodeProvider::new());
let mut engine = Engine::new(store, provider);
engine.register(CiWorkflow)?;

let run = engine.run_handler("ci", TriggerKind::Manual, json!({})).await?;
tracing::info!(run_id = %run.id, status = ?run.status, "run completed");

Implementations

impl Engine

pub fn new(store: Arc<dyn Store>, provider: Arc<dyn AgentProvider>) -> Self

Create a new engine with the given store and agent provider.

Examples

use std::sync::Arc;
use ironflow_engine::engine::Engine;
use ironflow_store::memory::InMemoryStore;
use ironflow_core::providers::claude::ClaudeCodeProvider;

let engine = Engine::new(
    Arc::new(InMemoryStore::new()),
    Arc::new(ClaudeCodeProvider::new()),
);

pub fn with_budget_config(self, budget: BudgetConfig) -> Self

Apply cost guardrails to this engine.

Without this, both the per-run cap default and the monthly quota are disabled and the engine behaves exactly as before.

Examples

use std::sync::Arc;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use ironflow_engine::budget::BudgetConfig;
use ironflow_engine::engine::Engine;
use ironflow_store::memory::InMemoryStore;

let engine = Engine::new(
    Arc::new(InMemoryStore::new()),
    Arc::new(ClaudeCodeProvider::new()),
)
.with_budget_config(BudgetConfig::from_env());

pub fn budget_config(&self) -> &BudgetConfig

Returns the cost guardrails applied by this engine.

pub fn set_log_sender(&mut self, sender: LogSender)

Attach a log sender for real-time step output streaming.

When set, all workflow contexts created by this engine will forward step output (shell stdout/stderr, agent system messages) to the given sender.

pub fn set_artifact_sink(&mut self, sink: Arc<dyn ArtifactSink>)

Attach the backend that stores and serves artifact bytes.

Every context this engine builds inherits it. Without one, steps that declare artifacts fail explicitly and every other step is unaffected.

Examples

use std::sync::Arc;

use ironflow_engine::artifact::ArtifactSink;
use ironflow_engine::engine::Engine;

engine.set_artifact_sink(sink);

pub fn artifact_sink(&self) -> Option<&Arc<dyn ArtifactSink>>

The artifact backend attached to this engine, if any.

pub fn set_event_bus(&mut self, bus: WorkflowEventBus)

Attach a WorkflowEventBus for per-run real-time monitoring.

When set, all workflow contexts created by this engine will publish step-level events (StepStarted, StepCompleted, StepFailed) to the bus.

Examples

use ironflow_engine::engine::Engine;
use ironflow_engine::notify::WorkflowEventBus;

engine.set_event_bus(WorkflowEventBus::new());

pub fn event_bus(&self) -> Option<&WorkflowEventBus>

The workflow event bus attached to this engine, if any.

pub fn store(&self) -> &Arc<dyn Store>

Returns a reference to the backing store.

pub fn provider(&self) -> &Arc<dyn AgentProvider>

Returns a reference to the agent provider.

pub fn register( &mut self, handler: impl WorkflowHandler + 'static, ) -> Result<(), EngineError>

Register a WorkflowHandler for dynamic workflow execution.

The handler is looked up by WorkflowHandler::name when executing or enqueuing.

Errors

Returns EngineError::InvalidWorkflow if a handler with the same name is already registered.

Examples

use std::sync::Arc;
use ironflow_engine::engine::Engine;
use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::ShellConfig;
use ironflow_store::memory::InMemoryStore;
use ironflow_core::providers::claude::ClaudeCodeProvider;

struct MyWorkflow;
impl WorkflowHandler for MyWorkflow {
    fn name(&self) -> &str { "my-workflow" }
    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            ctx.shell("step1", ShellConfig::new("echo done")).await?;
            Ok(())
        })
    }
}

let mut engine = Engine::new(
    Arc::new(InMemoryStore::new()),
    Arc::new(ClaudeCodeProvider::new()),
);
engine.register(MyWorkflow)?;

pub fn register_boxed( &mut self, handler: Box<dyn WorkflowHandler>, ) -> Result<(), EngineError>

Register a pre-boxed workflow handler.

Errors

Returns EngineError::InvalidWorkflow if a handler with the same name is already registered or if its category is invalid.

pub fn get_handler(&self, name: &str) -> Option<&Arc<dyn WorkflowHandler>>

Get a registered handler by name.

pub fn handler_names(&self) -> Vec<&str>

List registered handler names.

pub fn handler_info(&self, name: &str) -> Option<WorkflowInfo>

Get detailed info about a registered workflow handler.

pub fn scheduled_handlers(&self) -> Vec<(&str, &CronSchedule)>

List handlers that have a cron schedule configured.

Returns pairs of (workflow_name, cron_expression) for all handlers where WorkflowHandler::schedule returns Some.

Use this to wire scheduled handlers into a cron scheduler (e.g. ironflow_runtime::Runtime::cron).

Examples

let engine = Engine::new(
    Arc::new(InMemoryStore::new()),
    Arc::new(ClaudeCodeProvider::new()),
);
for (name, schedule) in engine.scheduled_handlers() {
    tracing::info!("{name} runs on schedule: {schedule}");
}

pub fn subscribe( &mut self, subscriber: impl EventSubscriber + 'static, event_types: &[&'static str], )

Register an event subscriber for domain events.

The subscriber is called only for events whose type is in event_types. Pass Event::ALL to receive every event.

Examples

use ironflow_engine::engine::Engine;
use ironflow_engine::notify::{Event, WebhookSubscriber};
use ironflow_store::memory::InMemoryStore;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use std::sync::Arc;

let mut engine = Engine::new(
    Arc::new(InMemoryStore::new()),
    Arc::new(ClaudeCodeProvider::new()),
);

engine.subscribe(
    WebhookSubscriber::new("https://hooks.example.com/events"),
    &[Event::RUN_STATUS_CHANGED, Event::STEP_FAILED],
);

pub fn event_publisher(&self) -> &EventPublisher

Returns a reference to the event publisher.

Useful for publishing events from outside the engine (e.g. auth routes in the API layer).

pub async fn run_handler( &self, handler_name: &str, trigger: TriggerKind, payload: Value, ) -> Result<Run, EngineError>

Execute a registered handler inline.

Creates a run, builds a WorkflowContext, calls the handler’s execute, and finalizes the run.

Errors

Returns EngineError::InvalidWorkflow if no handler is registered with that name. Returns EngineError if execution fails.

Examples

use std::sync::Arc;
use ironflow_engine::engine::Engine;
use ironflow_store::memory::InMemoryStore;
use ironflow_store::models::TriggerKind;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use serde_json::json;

let run = engine.run_handler("deploy", TriggerKind::Manual, json!({})).await?;

pub async fn enqueue_handler( &self, handler_name: &str, trigger: TriggerKind, payload: Value, max_retries: u32, ) -> Result<Run, EngineError>

Enqueue a handler-based workflow for worker execution.

The workflow name is stored in the run. The worker looks up the handler by name when executing.

Errors

Returns EngineError::InvalidWorkflow if no handler is registered. Returns EngineError::MonthlyBudgetExceeded if the monthly cost quota is exhausted.

pub async fn enqueue_handler_with_options( &self, handler_name: &str, trigger: TriggerKind, payload: Value, options: EnqueueOptions, ) -> Result<RunCreation, EngineError>

Enqueue a handler-based workflow with labels, deferred scheduling, an optional cost cap, an optional author, and an optional idempotency key.

See EnqueueOptions for the individual settings.

When EnqueueOptions::idempotency_key is set and already bound to a run created within IDEMPOTENCY_WINDOW, nothing is enqueued and the original run is returned as RunCreation::Existing.

Errors

Returns EngineError::InvalidWorkflow if no handler is registered. Returns EngineError::MonthlyBudgetExceeded if the monthly cost quota is exhausted. Returns EngineError::Store if the run cannot be persisted.

Examples

use ironflow_engine::engine::{Engine, EnqueueOptions};
use ironflow_store::models::TriggerKind;
use serde_json::json;

let creation = engine
    .enqueue_handler_with_options(
        "deploy",
        TriggerKind::Api,
        json!({"env": "prod"}),
        EnqueueOptions {
            max_retries: 3,
            idempotency_key: Some("github:abc-123".to_string()),
            ..Default::default()
        },
    )
    .await?;

if creation.is_created() {
    println!("enqueued {}", creation.run().id);
}

pub async fn execute_handler_run( &self, run_id: Uuid, ) -> Result<Run, EngineError>

Execute a handler-based run (used by the worker after pick_next_pending).

Looks up the handler by the run’s workflow_name and executes it with a fresh WorkflowContext.

Errors

Returns EngineError::InvalidWorkflow if no handler matches.

pub async fn execute_run(&self, run_id: Uuid) -> Result<Run, EngineError>

Execute a run by its ID (used by the worker after pick_next_pending).

Delegates to execute_handler_run.

Errors

Returns EngineError if the run is not found or execution fails.

pub async fn resume_run(&self, run_id: Uuid) -> Result<Run, EngineError>

Resume a run after human approval.

Re-executes the handler with step replay: completed steps return cached output, approved approval steps are skipped, and execution continues from the first unexecuted step.

Supports multiple approval gates – each resume replays all prior steps and stops at the next approval (or completes the run).

Errors

Returns EngineError::InvalidWorkflow if no handler matches. Returns EngineError if execution fails or hits another approval.

pub async fn fail_or_schedule_retry( &self, run_id: Uuid, error: &str, retryable: bool, cost_usd: Option<Decimal>, duration_ms: Option<u64>, ) -> Result<RunStatus, EngineError>

Record a run failure, replaying the run later when retries remain.

This is the single place where a failed run’s fate is decided. When retryable is true and the run has not exhausted max_retries, the run moves to RunStatus::Retrying with scheduled_at set to now + backoff – pick_next_pending picks it up again once that time has passed. Otherwise the run moves to RunStatus::Failed.

Either way, steps left non-terminal by the failed attempt are closed via fail_orphaned_steps so they are never confused with the next attempt’s steps.

Callers pass retryable explicitly rather than an error value, because the worker classifies failures it observes from the outside (a timeout, a panicked task) that never produce an EngineError. Use is_run_retryable to classify an EngineError.

cost_usd and duration_ms are None when the caller does not know the totals (a timeout or a panic observed from outside the handler); the values already stored on the run are then left untouched.

Returns the status the run was moved to.

Errors

Returns EngineError::Store if the run does not exist or the update cannot be persisted.

Examples

use ironflow_engine::engine::Engine;
use ironflow_engine::error::EngineError;
use uuid::Uuid;

let status = engine
    .fail_or_schedule_retry(run_id, "run timed out after 300s", true, None, None)
    .await?;

pub async fn fail_orphaned_steps( &self, run_id: Uuid, error_message: &str, ) -> Result<(), EngineError>

Fail all non-terminal steps for a run.

Called after a run is marked as failed (timeout, error, panic) to clean up orphaned steps that are still in Running, Pending, or AwaitingApproval.

• Running / AwaitingApproval steps are marked Failed.
• Pending steps are marked Skipped (FSM does not allow Pending -> Failed).

Errors from individual step updates are logged but do not abort the cleanup.

Errors

Returns EngineError if listing steps fails.

Trait Implementations

impl Debug for Engine

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.