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 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 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_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.