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
38pub(crate) use lifecycle::interrupt_running_steps;
39
40use std::collections::HashMap;
41use std::fmt;
42use std::sync::Arc;
43
44use chrono::{DateTime, Utc};
45use rust_decimal::Decimal;
46use serde_json::Value;
47use uuid::Uuid;
48
49use ironflow_core::decision::DecisionProvider;
50use ironflow_core::provider::AgentProvider;
51use ironflow_core::trace_context::WorkflowTraceContext;
52use ironflow_store::models::Step;
53use ironflow_store::store::Store;
54
55use crate::artifact::ArtifactSink;
56use crate::config::StepConfig;
57use crate::executor::{StepInterceptor, StepResult};
58use crate::guard::{SharedGuardState, WorkflowGuardConfig};
59use crate::handler::WorkflowHandler;
60use crate::log_sender::LogSender;
61use crate::notify::WorkflowEventBus;
62use crate::operation::OperationContext;
63use crate::plan::SharedPlanRecorder;
64
65/// Label set on every child run of a sub-workflow step, holding the id of the
66/// run that started it.
67///
68/// The root of the chain is recorded under
69/// [`LABEL_ROOT_RUN_ID`](ironflow_core::provider::LABEL_ROOT_RUN_ID). Both are
70/// set on the child run when it is created, so a suspended child can be
71/// listed by label and resumed through its root.
72///
73/// # Examples
74///
75/// ```no_run
76/// use std::collections::HashMap;
77/// use ironflow_engine::context::PARENT_RUN_ID_LABEL;
78/// use ironflow_store::models::RunFilter;
79/// use uuid::Uuid;
80///
81/// # fn example(parent: Uuid) {
82/// let children = RunFilter {
83/// labels: Some(HashMap::from([(PARENT_RUN_ID_LABEL.to_string(), parent.to_string())])),
84/// ..RunFilter::default()
85/// };
86/// # }
87/// ```
88pub const PARENT_RUN_ID_LABEL: &str = "ironflow.io/parent-run-id";
89
90/// Callback type for resolving workflow handlers by name.
91pub(crate) type HandlerResolver =
92 Arc<dyn Fn(&str) -> Option<Arc<dyn WorkflowHandler>> + Send + Sync>;
93
94/// Execution context for a single workflow run.
95///
96/// Tracks the current step position and provides convenience methods
97/// for executing operations with automatic persistence.
98///
99/// # Examples
100///
101/// ```no_run
102/// use ironflow_engine::context::WorkflowContext;
103/// use ironflow_engine::config::ShellConfig;
104/// use ironflow_engine::error::EngineError;
105///
106/// # async fn example(ctx: &mut WorkflowContext) -> Result<(), EngineError> {
107/// let result = ctx.shell("greet", ShellConfig::new("echo hello")).await?;
108/// assert!(result.stdout().contains("hello"));
109/// # Ok(())
110/// # }
111/// ```
112pub struct WorkflowContext {
113 run_id: Uuid,
114 /// The top-level run: `run_id` itself, or the parent's root for a
115 /// sub-workflow. Stamped on agent pods so a retry releases the children.
116 root_run_id: Uuid,
117 workflow_name: String,
118 store: Arc<dyn Store>,
119 provider: Arc<dyn AgentProvider>,
120 /// Optional decision backend (System One / Jev) for `ctx.decision(...)`.
121 /// `None` when no decision provider was wired: a decision step then fails
122 /// explicitly instead of silently doing nothing.
123 decision_provider: Option<Arc<dyn DecisionProvider>>,
124 handler_resolver: Option<HandlerResolver>,
125 position: u32,
126 /// IDs of the last executed step(s) -- used to record DAG dependencies.
127 last_step_ids: Vec<Uuid>,
128 /// Accumulated cost across all steps in this run.
129 total_cost_usd: Decimal,
130 /// Accumulated duration across all steps.
131 total_duration_ms: u64,
132 /// Cumulative cost cap for this run, resolved at creation. `None` = no cap.
133 max_cost_usd: Option<Decimal>,
134 /// Cost already spent by ancestor runs when this context belongs to a
135 /// sub-workflow. Zero for a top-level run.
136 inherited_cost_usd: Decimal,
137 /// Steps from a previous execution of the *same* attempt, keyed by position.
138 /// Used when resuming after approval to replay completed steps.
139 replay_steps: HashMap<u32, Step>,
140 /// All steps of a previous execution of the *same* attempt, keyed by
141 /// `(position, step name)`. A `parallel` wave shares one position across
142 /// several steps, which `replay_steps` cannot represent -- this index lets
143 /// `parallel()` check that every step of a wave already completed before
144 /// replaying the whole wave from the store, without re-running any item.
145 replay_wave_steps: HashMap<(u32, String), Step>,
146 /// Approvals granted in an *earlier* attempt, keyed by position, holding the
147 /// attempt that granted them. An approval is carried by the run, not by the
148 /// attempt, so a retry never asks a human to approve the same gate twice.
149 granted_approvals: HashMap<u32, u32>,
150 /// Human inputs answered in an *earlier* attempt, keyed by position:
151 /// (attempt, answer). Like an approval, an answer is carried by the run, so
152 /// a retry never asks a human to answer the same input twice.
153 answered_inputs: HashMap<u32, (u32, Value)>,
154 /// `Workflow` steps of the current attempt interrupted by a lost worker
155 /// lease, keyed by position. The step executed again at that position
156 /// re-enters the child run they recorded instead of starting a new one.
157 interrupted_children: HashMap<u32, Step>,
158 /// Which run attempt this context is executing (1-based).
159 attempt: u32,
160 /// Wall-clock duration already recorded on the run by previous attempts.
161 /// Added to this attempt's duration when the run is finalized.
162 carried_duration_ms: u64,
163 /// Optional sender for real-time log streaming.
164 log_sender: Option<LogSender>,
165 /// Where artifact bytes are read and written. `None` when no artifact
166 /// storage is configured: steps that declare artifacts then fail explicitly
167 /// instead of silently dropping their files.
168 artifact_sink: Option<Arc<dyn ArtifactSink>>,
169 /// Set to `true` when at least one `allow_failure` step failed.
170 has_allowed_failure: bool,
171 /// Error handlers registered via [`on_error`](Self::on_error).
172 error_handlers: Vec<OnErrorHandler>,
173 /// Shared guard state for workflow execution limits.
174 guard_state: Option<SharedGuardState>,
175 /// Guard configuration for this workflow run.
176 guard_config: Option<WorkflowGuardConfig>,
177 /// Accumulated step results for post-execution inspection.
178 step_results: Vec<StepResult>,
179 /// Optional event bus for per-run real-time monitoring.
180 event_bus: Option<WorkflowEventBus>,
181 /// Optional hook that resolves steps without executing them. `None` in
182 /// production; set by [`crate::testing::TestEngine`].
183 interceptor: Option<Arc<dyn StepInterceptor>>,
184 /// W3C trace context for distributed tracing propagation.
185 trace_context: WorkflowTraceContext,
186 /// Shared operation context for custom operations.
187 operation_ctx: Option<OperationContext>,
188 /// When the run was created, set by the engine. Bounds the signals a wait
189 /// step accepts: a signal received before the run existed is not for it.
190 /// `None` falls back to reading the run from the store.
191 run_created_at: Option<DateTime<Utc>>,
192 /// Set when the context is recording an execution plan instead of running.
193 /// Every step method checks this first and records intent without executing.
194 plan: Option<SharedPlanRecorder>,
195 /// Output set by the handler with [`set_output`](Self::set_output),
196 /// persisted on the run when the execution ends.
197 output: Option<Value>,
198}
199
200/// A registered error handler that fires when a subsequent step fails.
201struct OnErrorHandler {
202 name: String,
203 config: StepConfig,
204}
205
206impl fmt::Debug for WorkflowContext {
207 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
208 f.debug_struct("WorkflowContext")
209 .field("run_id", &self.run_id)
210 .field("position", &self.position)
211 .field("total_cost_usd", &self.total_cost_usd)
212 .field("inherited_cost_usd", &self.inherited_cost_usd)
213 .field("max_cost_usd", &self.max_cost_usd)
214 .field("planning", &self.plan.is_some())
215 .finish_non_exhaustive()
216 }
217}