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