Struct WorkflowContext 

pub struct WorkflowContext { /* private fields */ }

Execution context for a single workflow run.

Tracks the current step position and provides convenience methods for executing operations with automatic persistence.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::ShellConfig;
use ironflow_engine::error::EngineError;

let result = ctx.shell("greet", ShellConfig::new("echo hello")).await?;
assert!(result.output["stdout"].as_str().unwrap().contains("hello"));

Implementations

impl WorkflowContext

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

Create a new context for a run.

Not typically called directly — the Engine creates this when executing a WorkflowHandler.

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

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

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

Attach the backend that stores and serves artifact bytes.

Without one, any step that declares an output or calls put_artifact fails with EngineError::ArtifactsUnavailable. Every other step is unaffected, so an existing deployment keeps working until artifacts are configured.

Examples

use std::sync::Arc;

use ironflow_engine::artifact::ArtifactSink;
use ironflow_engine::context::WorkflowContext;

ctx.set_artifact_sink(sink);

pub fn trace_context(&self) -> &WorkflowTraceContext

Return the W3C trace context for this workflow run.

The trace context is derived from the run ID and can be used to correlate spans across distributed services. Each step automatically receives a child context.

pub fn set_guard( &mut self, config: WorkflowGuardConfig, state: SharedGuardState, )

Attach a workflow guard configuration and shared state.

When set, the guard is checked before every sub-workflow invocation. The shared state is propagated to child workflows so that limits apply globally across the entire run tree.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::guard::{WorkflowGuardConfig, new_shared_guard_state};

ctx.set_guard(WorkflowGuardConfig::default(), new_shared_guard_state());

pub fn guard_config(&self) -> Option<&WorkflowGuardConfig>

The current guard configuration, if any.

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

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

When set, step transitions automatically publish WorkflowEvents to the bus.

pub async fn put_artifact( &self, step_id: Uuid, name: &str, content_type: Option<&str>, content: Vec<u8>, ) -> Result<Artifact, EngineError>

Store an in-memory payload as an artifact of the given step.

The declarative ShellConfig::output covers shell steps; this covers custom operations and agent steps, which have no working directory to collect from.

The MIME type is guessed from name unless content_type is set.

Errors

Returns EngineError::ArtifactsUnavailable when no backend is attached, EngineError::Artifact when the name is invalid or storage fails, and EngineError::Store when the step already owns that name.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::error::EngineError;
use uuid::Uuid;

let artifact = ctx
    .put_artifact(step_id, "summary.json", None, br#"{"ok":true}"#.to_vec())
    .await?;
assert_eq!(artifact.content_type, "application/json");

pub async fn get_artifact( &self, step: &str, name: &str, ) -> Result<Vec<u8>, EngineError>

Read back an artifact produced earlier in this run.

Resolution follows the same rule as a declared input: same run and attempt, steps positioned strictly before the current one, closest producer wins.

Errors

Returns EngineError::ArtifactNotFound when nothing matches, EngineError::ArtifactsUnavailable when no backend is attached, and EngineError::Artifact when the bytes cannot be read.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::error::EngineError;

let bytes = ctx.get_artifact("build", "report.html").await?;
println!("{} bytes", bytes.len());

pub fn attempt(&self) -> u32

The run attempt this context is executing (1-based).

pub fn set_max_cost_usd(&mut self, cap: Option<Decimal>)

Set the cumulative cost cap enforced before every agent step.

Called by the Engine with the run’s persisted max_cost_usd. None disables the check.

Examples

use ironflow_engine::context::WorkflowContext;
use rust_decimal::Decimal;

ctx.set_max_cost_usd(Some(Decimal::new(200, 2))); // $2.00

pub fn max_cost_usd(&self) -> Option<Decimal>

The cumulative cost cap of this run, if any.

pub fn charged_cost_usd(&self) -> Decimal

Total cost charged against the cap: this run plus every ancestor run.

For a top-level run this equals total_cost_usd. For a sub-workflow it also includes what the parent chain already spent.

pub fn run_id(&self) -> Uuid

The run ID this context is executing for.

pub fn workflow_name(&self) -> &str

The workflow name this run belongs to.

pub fn total_cost_usd(&self) -> Decimal

Accumulated cost across all executed steps so far.

pub fn has_allowed_failure(&self) -> bool

Whether at least one allow_failure step failed during this run.

pub fn total_duration_ms(&self) -> u64

Accumulated duration across all executed steps so far.

pub fn step_results(&self) -> &[StepResult]

Enriched results of all completed steps in execution order.

pub async fn parallel( &mut self, steps: Vec<(&str, StepConfig)>, fail_fast: bool, ) -> Result<Vec<ParallelStepResult>, EngineError>

Execute multiple steps concurrently (wait-all model).

All steps in the batch execute in parallel via tokio::JoinSet. Each step is recorded with the same position (execution wave). Dependencies on previous steps are recorded automatically.

When fail_fast is true, remaining steps are aborted on the first failure. When false, all steps run to completion and the first error is returned.

Errors

Returns EngineError if any step fails.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::{StepConfig, ShellConfig};
use ironflow_engine::error::EngineError;

let results = ctx.parallel(
    vec![
        ("test-unit", StepConfig::Shell(ShellConfig::new("cargo test --lib"))),
        ("lint", StepConfig::Shell(ShellConfig::new("cargo clippy"))),
    ],
    true,
).await?;

for r in &results {
    println!("{}: {:?}", r.name, r.output.output);
}

pub async fn shell( &mut self, name: &str, config: ShellConfig, ) -> Result<StepOutput, EngineError>

Execute a shell step.

Creates the step record, runs the command, persists the result, and returns the output for use in subsequent steps.

Errors

Returns EngineError if the command fails or the store errors.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::ShellConfig;
use ironflow_engine::error::EngineError;

let files = ctx.shell("list", ShellConfig::new("ls -la")).await?;
println!("stdout: {}", files.output["stdout"]);

pub async fn http( &mut self, name: &str, config: HttpConfig, ) -> Result<StepOutput, EngineError>

Execute an HTTP step.

Errors

Returns EngineError if the request fails or the store errors.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::HttpConfig;
use ironflow_engine::error::EngineError;

let resp = ctx.http("health", HttpConfig::get("https://api.example.com/health")).await?;
println!("status: {}", resp.output["status"]);

pub async fn agent( &mut self, name: &str, config: impl Into<AgentStepConfig>, ) -> Result<StepOutput, EngineError>

Execute an agent step.

Errors

Returns EngineError if the agent invocation fails or the store errors.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::AgentStepConfig;
use ironflow_engine::error::EngineError;

let review = ctx.agent("review", AgentStepConfig::new("Review the code")).await?;
println!("review: {}", review.output);

pub async fn approval( &mut self, name: &str, config: ApprovalConfig, ) -> Result<(), EngineError>

Create a human approval gate.

On first execution, records an approval step and returns EngineError::ApprovalRequired to suspend the run. The engine transitions the run to AwaitingApproval.

On resume (after a human approved via the API), the approval step is replayed: it is marked as Completed and execution continues past it. Multiple approval gates in the same handler work – each one pauses and resumes independently.

Errors

Returns EngineError::ApprovalRequired to pause the run on first execution. Returns other EngineError variants on store failures.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::ApprovalConfig;
use ironflow_engine::error::EngineError;

ctx.approval("deploy-gate", ApprovalConfig::new("Approve deployment?")).await?;
// Execution continues here after approval

pub async fn skip( &mut self, name: &str, reason: &str, ) -> Result<(), EngineError>

Record a step as explicitly skipped.

Use this inside an if/else branch when a step should not execute but must still appear in the DAG and timeline with its reason.

The step is created directly in StepStatus::Skipped state and the reason is stored in the output as {"reason": "..."}.

Errors

Returns EngineError if the store fails.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::error::EngineError;

let tests_passed = false;
if tests_passed {
    // ctx.shell("deploy", ...).await?;
} else {
    ctx.skip("deploy", "tests failed").await?;
}

pub async fn operation( &mut self, name: &str, op: &dyn Operation, ) -> Result<StepOutput, EngineError>

Execute a custom operation step.

Runs a user-defined Operation with full step lifecycle management: creates the step record, transitions to Running, executes the operation, persists the output and duration, and marks the step Completed or Failed.

The operation’s kind() is stored as StepKind::Custom.

Errors

Returns EngineError if the operation fails or the store errors.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::operation::Operation;
use ironflow_engine::error::EngineError;
use serde_json::{Value, json};
use std::pin::Pin;
use std::future::Future;

struct MyOp;
impl Operation for MyOp {
    fn kind(&self) -> &str { "my-service" }
    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<Value, EngineError>> + Send + '_>> {
        Box::pin(async { Ok(json!({"ok": true})) })
    }
}

let result = ctx.operation("call-service", &MyOp).await?;
println!("output: {}", result.output);

pub async fn workflow( &mut self, handler: &dyn WorkflowHandler, payload: Value, ) -> Result<StepOutput, EngineError>

Execute a sub-workflow step.

Creates a child run for the named workflow handler, executes it with its own steps and lifecycle, and returns a StepOutput containing the child run ID and aggregated metrics.

Requires the context to be created with with_handler_resolver.

Errors

Returns EngineError::InvalidWorkflow if no handler is registered with the given name, or if no handler resolver is available.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::error::EngineError;
use serde_json::json;

// let result = ctx.workflow(&MySubWorkflow, json!({})).await?;

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

Access the store directly (advanced usage).

pub async fn payload(&self) -> Result<Value, EngineError>

Access the payload that triggered this run.

Fetches the run from the store and returns its payload.

Errors

Returns EngineError::Store if the run is not found.

pub async fn input<T: DeserializeOwned>(&self) -> Result<T, EngineError>

Deserialize the run payload into a typed input struct.

Shorthand for serde_json::from_value(ctx.payload().await?).

Errors

Returns EngineError::Store if the run is not found, or EngineError::Serialization if the payload does not match T.

Examples

use serde::Deserialize;

#[derive(Deserialize)]
struct DeployInput {
    environment: String,
    dry_run: Option<bool>,
}

let input: DeployInput = ctx.input().await?;

pub fn on_error(&mut self, name: &str, config: impl Into<StepConfig>)

Register an error handler that fires when any subsequent step fails.

The handler is consumed after firing (fire-once). Multiple handlers can be registered; they fire in registration order.

Error handler execution is best-effort: if a handler fails, the error is logged but the original step error is preserved. Error handler steps appear in the run timeline with Step::is_error_handler set to true.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::ShellConfig;
use ironflow_engine::error::EngineError;

ctx.on_error("cleanup", ShellConfig::new("rm -rf /tmp/build"));
ctx.shell("build", ShellConfig::new("cargo build")).await?;

pub fn clear_error_handlers(&mut self)

Remove all registered error handlers.

Examples

use ironflow_engine::context::WorkflowContext;
use ironflow_engine::config::ShellConfig;
use ironflow_engine::error::EngineError;

ctx.on_error("cleanup", ShellConfig::new("rm -rf /tmp/build"));
ctx.shell("build", ShellConfig::new("cargo build")).await?;
ctx.clear_error_handlers();
// cleanup will NOT fire if deploy fails
ctx.shell("deploy", ShellConfig::new("./deploy.sh")).await?;

Trait Implementations

impl Debug for WorkflowContext

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

Formats the value using the given formatter.