ironflow_engine/
context.rs

1//! [`WorkflowContext`] — execution context for dynamic workflows.
2//!
3//! Provides step execution methods that automatically persist results to the
4//! store. Each call to [`shell`](WorkflowContext::shell),
5//! [`http`](WorkflowContext::http), [`agent`](WorkflowContext::agent), or
6//! [`workflow`](WorkflowContext::workflow) creates a step record, executes the
7//! operation, captures the output, and returns a
8//! [`StepOutput`](crate::executor::StepOutput) that the next step can
9//! reference.
10//!
11//! # Examples
12//!
13//! ```no_run
14//! use ironflow_engine::context::WorkflowContext;
15//! use ironflow_engine::config::{ShellConfig, AgentStepConfig};
16//! use ironflow_engine::error::EngineError;
17//!
18//! # async fn example(ctx: &mut WorkflowContext) -> Result<(), EngineError> {
19//! let build = ctx.shell("build", ShellConfig::new("cargo build")).await?;
20//! let review = ctx.agent("review", AgentStepConfig::new(
21//!     &format!("Build output:\n{}", build.stdout())
22//! )).await?;
23//! # Ok(())
24//! # }
25//! ```
26
27mod accessors;
28mod artifacts;
29mod error_handlers;
30mod failure;
31mod guard;
32mod lifecycle;
33mod steps;
34
35#[cfg(test)]
36mod tests;
37
38use std::collections::HashMap;
39use std::fmt;
40use std::sync::Arc;
41
42use rust_decimal::Decimal;
43use uuid::Uuid;
44
45use ironflow_core::decision::DecisionProvider;
46use ironflow_core::provider::AgentProvider;
47use ironflow_core::trace_context::WorkflowTraceContext;
48use ironflow_store::models::Step;
49use ironflow_store::store::Store;
50
51use crate::artifact::ArtifactSink;
52use crate::config::StepConfig;
53use crate::executor::{StepInterceptor, StepResult};
54use crate::guard::{SharedGuardState, WorkflowGuardConfig};
55use crate::handler::WorkflowHandler;
56use crate::log_sender::LogSender;
57use crate::notify::WorkflowEventBus;
58use crate::operation::OperationContext;
59use crate::plan::SharedPlanRecorder;
60
61/// Callback type for resolving workflow handlers by name.
62pub(crate) type HandlerResolver =
63    Arc<dyn Fn(&str) -> Option<Arc<dyn WorkflowHandler>> + Send + Sync>;
64
65/// Execution context for a single workflow run.
66///
67/// Tracks the current step position and provides convenience methods
68/// for executing operations with automatic persistence.
69///
70/// # Examples
71///
72/// ```no_run
73/// use ironflow_engine::context::WorkflowContext;
74/// use ironflow_engine::config::ShellConfig;
75/// use ironflow_engine::error::EngineError;
76///
77/// # async fn example(ctx: &mut WorkflowContext) -> Result<(), EngineError> {
78/// let result = ctx.shell("greet", ShellConfig::new("echo hello")).await?;
79/// assert!(result.stdout().contains("hello"));
80/// # Ok(())
81/// # }
82/// ```
83pub struct WorkflowContext {
84    run_id: Uuid,
85    workflow_name: String,
86    store: Arc<dyn Store>,
87    provider: Arc<dyn AgentProvider>,
88    /// Optional decision backend (System One / Jev) for `ctx.decision(...)`.
89    /// `None` when no decision provider was wired: a decision step then fails
90    /// explicitly instead of silently doing nothing.
91    decision_provider: Option<Arc<dyn DecisionProvider>>,
92    handler_resolver: Option<HandlerResolver>,
93    position: u32,
94    /// IDs of the last executed step(s) -- used to record DAG dependencies.
95    last_step_ids: Vec<Uuid>,
96    /// Accumulated cost across all steps in this run.
97    total_cost_usd: Decimal,
98    /// Accumulated duration across all steps.
99    total_duration_ms: u64,
100    /// Cumulative cost cap for this run, resolved at creation. `None` = no cap.
101    max_cost_usd: Option<Decimal>,
102    /// Cost already spent by ancestor runs when this context belongs to a
103    /// sub-workflow. Zero for a top-level run.
104    inherited_cost_usd: Decimal,
105    /// Steps from a previous execution of the *same* attempt, keyed by position.
106    /// Used when resuming after approval to replay completed steps.
107    replay_steps: HashMap<u32, Step>,
108    /// Approvals granted in an *earlier* attempt, keyed by position, holding the
109    /// attempt that granted them. An approval is carried by the run, not by the
110    /// attempt, so a retry never asks a human to approve the same gate twice.
111    granted_approvals: HashMap<u32, u32>,
112    /// Which run attempt this context is executing (1-based).
113    attempt: u32,
114    /// Wall-clock duration already recorded on the run by previous attempts.
115    /// Added to this attempt's duration when the run is finalized.
116    carried_duration_ms: u64,
117    /// Optional sender for real-time log streaming.
118    log_sender: Option<LogSender>,
119    /// Where artifact bytes are read and written. `None` when no artifact
120    /// storage is configured: steps that declare artifacts then fail explicitly
121    /// instead of silently dropping their files.
122    artifact_sink: Option<Arc<dyn ArtifactSink>>,
123    /// Set to `true` when at least one `allow_failure` step failed.
124    has_allowed_failure: bool,
125    /// Error handlers registered via [`on_error`](Self::on_error).
126    error_handlers: Vec<OnErrorHandler>,
127    /// Shared guard state for workflow execution limits.
128    guard_state: Option<SharedGuardState>,
129    /// Guard configuration for this workflow run.
130    guard_config: Option<WorkflowGuardConfig>,
131    /// Accumulated step results for post-execution inspection.
132    step_results: Vec<StepResult>,
133    /// Optional event bus for per-run real-time monitoring.
134    event_bus: Option<WorkflowEventBus>,
135    /// Optional hook that resolves steps without executing them. `None` in
136    /// production; set by [`crate::testing::TestEngine`].
137    interceptor: Option<Arc<dyn StepInterceptor>>,
138    /// W3C trace context for distributed tracing propagation.
139    trace_context: WorkflowTraceContext,
140    /// Shared operation context for custom operations.
141    operation_ctx: Option<OperationContext>,
142    /// Set when the context is recording an execution plan instead of running.
143    /// Every step method checks this first and records intent without executing.
144    plan: Option<SharedPlanRecorder>,
145}
146
147/// A registered error handler that fires when a subsequent step fails.
148struct OnErrorHandler {
149    name: String,
150    config: StepConfig,
151}
152
153impl fmt::Debug for WorkflowContext {
154    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
155        f.debug_struct("WorkflowContext")
156            .field("run_id", &self.run_id)
157            .field("position", &self.position)
158            .field("total_cost_usd", &self.total_cost_usd)
159            .field("inherited_cost_usd", &self.inherited_cost_usd)
160            .field("max_cost_usd", &self.max_cost_usd)
161            .field("planning", &self.plan.is_some())
162            .finish_non_exhaustive()
163    }
164}
ironflow_engine/context.rs

ironflow_engine/
context.rs
