ironflow_engine/executor/
mod.rs

1//! Step executor — reconstructs operations from configs and runs them.
2//!
3//! Each step type (shell, HTTP, agent) has its own executor implementing
4//! the [`StepExecutor`] trait. The [`execute_step_config`] function dispatches
5//! to the appropriate executor based on the [`StepConfig`] variant.
6
7mod agent;
8mod http;
9mod shell;
10
11use std::future::Future;
12use std::sync::Arc;
13
14use rust_decimal::Decimal;
15use serde_json::Value;
16use uuid::Uuid;
17
18use ironflow_core::provider::{AgentProvider, DebugMessage};
19
20use crate::config::StepConfig;
21use crate::error::EngineError;
22use crate::log_sender::StepLogSender;
23
24pub use agent::AgentExecutor;
25pub use http::HttpExecutor;
26pub use shell::ShellExecutor;
27
28/// Result of executing a single step.
29#[derive(Debug, Clone)]
30pub struct StepOutput {
31    /// Serialized output (stdout for shell, body for http, value for agent).
32    ///
33    /// For agent steps with a JSON schema, the value may not strictly conform
34    /// to the schema: Claude CLI can flatten wrapper objects with a single
35    /// array field, returning a bare array instead of `{"items": [...]}`.
36    /// Callers should handle both the expected wrapper and a bare value.
37    pub output: Value,
38    /// Wall-clock duration in milliseconds.
39    pub duration_ms: u64,
40    /// Cost in USD (agent steps only).
41    pub cost_usd: Decimal,
42    /// Input token count (agent steps only).
43    pub input_tokens: Option<u64>,
44    /// Output token count (agent steps only).
45    pub output_tokens: Option<u64>,
46    /// Model identifier used for agent steps (e.g. `"claude-sonnet-4-20250514"`).
47    pub model: Option<String>,
48    /// Conversation trace from verbose agent invocations.
49    pub debug_messages: Option<Vec<DebugMessage>>,
50}
51
52impl StepOutput {
53    /// Serialize debug messages to a JSON [`Value`] for store persistence.
54    ///
55    /// Returns `None` when verbose mode was off (no messages captured).
56    pub fn debug_messages_json(&self) -> Option<Value> {
57        self.debug_messages
58            .as_ref()
59            .and_then(|msgs| serde_json::to_value(msgs).ok())
60    }
61}
62
63/// Result of a single step within a [`parallel`](crate::context::WorkflowContext::parallel) batch.
64#[derive(Debug, Clone)]
65pub struct ParallelStepResult {
66    /// The step name (same as provided to `parallel()`).
67    pub name: String,
68    /// The step execution output.
69    pub output: StepOutput,
70    /// The step ID in the store (for dependency tracking).
71    pub step_id: Uuid,
72}
73
74/// Trait for step executors.
75///
76/// Each step type implements this trait to execute its specific operation
77/// and return a [`StepOutput`].
78pub trait StepExecutor: Send + Sync {
79    /// Execute the step and return structured output.
80    ///
81    /// # Errors
82    ///
83    /// Returns [`EngineError`] if the operation fails.
84    fn execute(
85        &self,
86        provider: &Arc<dyn AgentProvider>,
87    ) -> impl Future<Output = Result<StepOutput, EngineError>> + Send;
88}
89
90/// Execute a [`StepConfig`] and return structured output.
91///
92/// When a [`StepLogSender`] is provided, executors that support streaming
93/// will emit log lines in real time (e.g. shell stdout/stderr).
94///
95/// # Errors
96///
97/// Returns [`EngineError::Operation`] if the operation fails.
98///
99/// # Examples
100///
101/// ```no_run
102/// use ironflow_engine::config::{StepConfig, ShellConfig};
103/// use ironflow_engine::executor::execute_step_config;
104/// use ironflow_core::provider::AgentProvider;
105/// use ironflow_core::providers::claude::ClaudeCodeProvider;
106/// use std::sync::Arc;
107///
108/// # async fn example() -> Result<(), ironflow_engine::error::EngineError> {
109/// let provider: Arc<dyn AgentProvider> = Arc::new(ClaudeCodeProvider::new());
110/// let config = StepConfig::Shell(ShellConfig::new("echo hello"));
111/// let output = execute_step_config(&config, &provider, None).await?;
112/// # Ok(())
113/// # }
114/// ```
115#[tracing::instrument(name = "executor.execute_step", skip_all, fields(step.kind))]
116pub async fn execute_step_config(
117    config: &StepConfig,
118    provider: &Arc<dyn AgentProvider>,
119    log_sender: Option<StepLogSender>,
120) -> Result<StepOutput, EngineError> {
121    let kind = match config {
122        StepConfig::Shell(_) => "shell",
123        StepConfig::Http(_) => "http",
124        StepConfig::Agent(_) => "agent",
125        StepConfig::Workflow(_) => "workflow",
126        StepConfig::Approval(_) => "approval",
127    };
128    tracing::Span::current().record("step.kind", kind);
129
130    let result = match config {
131        StepConfig::Shell(cfg) => {
132            let mut executor = ShellExecutor::new(cfg);
133            if let Some(sender) = log_sender {
134                executor = executor.with_log_sender(sender);
135            }
136            executor.execute(provider).await
137        }
138        StepConfig::Http(cfg) => HttpExecutor::new(cfg).execute(provider).await,
139        StepConfig::Agent(cfg) => {
140            let mut executor = AgentExecutor::new(cfg);
141            if let Some(sender) = log_sender {
142                executor = executor.with_log_sender(sender);
143            }
144            executor.execute(provider).await
145        }
146        StepConfig::Workflow(_) => Err(EngineError::StepConfig(
147            "workflow steps are executed by WorkflowContext, not the executor".to_string(),
148        )),
149        StepConfig::Approval(_) => Err(EngineError::StepConfig(
150            "approval steps are executed by WorkflowContext, not the executor".to_string(),
151        )),
152    };
153
154    #[cfg(feature = "prometheus")]
155    {
156        use ironflow_core::metric_names::{
157            STATUS_ERROR, STATUS_SUCCESS, STEP_DURATION_SECONDS, STEPS_TOTAL,
158        };
159        use metrics::{counter, histogram};
160        let status = if result.is_ok() {
161            STATUS_SUCCESS
162        } else {
163            STATUS_ERROR
164        };
165        counter!(STEPS_TOTAL, "kind" => kind, "status" => status).increment(1);
166        if let Ok(ref output) = result {
167            histogram!(STEP_DURATION_SECONDS, "kind" => kind)
168                .record(output.duration_ms as f64 / 1000.0);
169        }
170    }
171
172    result
173}
174
175#[cfg(test)]
176mod tests {
177    use super::*;
178    use ironflow_core::provider::DebugMessage;
179    use serde_json::json;
180
181    #[test]
182    fn step_output_with_no_debug_messages_returns_none() {
183        let output = StepOutput {
184            output: json!({"result": "ok"}),
185            duration_ms: 100,
186            cost_usd: rust_decimal::Decimal::ZERO,
187            input_tokens: None,
188            output_tokens: None,
189            model: None,
190            debug_messages: None,
191        };
192
193        assert_eq!(output.debug_messages_json(), None);
194    }
195
196    #[test]
197    fn step_output_with_empty_debug_messages_returns_some_empty_array() {
198        let output = StepOutput {
199            output: json!({"result": "ok"}),
200            duration_ms: 100,
201            cost_usd: rust_decimal::Decimal::ZERO,
202            input_tokens: None,
203            output_tokens: None,
204            model: None,
205            debug_messages: Some(Vec::new()),
206        };
207
208        let json_val = output.debug_messages_json();
209        assert!(json_val.is_some());
210        let arr = json_val.unwrap();
211        assert!(arr.is_array());
212        assert_eq!(arr.as_array().unwrap().len(), 0);
213    }
214
215    #[test]
216    fn step_output_debug_messages_json_serializes_messages() {
217        let json_msgs = json!([
218            {
219                "text": "Hello",
220                "thinking": null,
221                "thinking_redacted": false,
222                "tool_calls": [],
223                "tool_results": [],
224                "stop_reason": "end_turn",
225                "input_tokens": 10,
226                "output_tokens": 20
227            },
228            {
229                "text": "Hi there",
230                "thinking": null,
231                "thinking_redacted": false,
232                "tool_calls": [],
233                "tool_results": [],
234                "stop_reason": "end_turn",
235                "input_tokens": 15,
236                "output_tokens": 25
237            }
238        ]);
239
240        let messages: Vec<DebugMessage> =
241            serde_json::from_value(json_msgs.clone()).expect("deserialize debug messages");
242
243        let output = StepOutput {
244            output: json!({"result": "ok"}),
245            duration_ms: 100,
246            cost_usd: rust_decimal::Decimal::ZERO,
247            input_tokens: None,
248            output_tokens: None,
249            model: None,
250            debug_messages: Some(messages),
251        };
252
253        let json_val = output.debug_messages_json();
254        assert!(json_val.is_some());
255
256        let arr = json_val.unwrap();
257        assert!(arr.is_array());
258        let messages_array = arr.as_array().unwrap();
259        assert_eq!(messages_array.len(), 2);
260        assert_eq!(messages_array[0]["text"], "Hello");
261        assert_eq!(messages_array[1]["text"], "Hi there");
262    }
263
264    #[test]
265    fn step_output_contains_all_metrics() {
266        let output = StepOutput {
267            output: json!({"data": "test"}),
268            duration_ms: 5000,
269            cost_usd: rust_decimal::Decimal::new(123, 2),
270            input_tokens: Some(100),
271            output_tokens: Some(200),
272            model: Some("claude-sonnet".to_string()),
273            debug_messages: None,
274        };
275
276        assert_eq!(output.duration_ms, 5000);
277        assert_eq!(output.cost_usd, rust_decimal::Decimal::new(123, 2));
278        assert_eq!(output.input_tokens, Some(100));
279        assert_eq!(output.output_tokens, Some(200));
280        assert_eq!(output.model, Some("claude-sonnet".to_string()));
281    }
282
283    #[test]
284    fn step_output_default_tokens_and_model_are_none() {
285        let output = StepOutput {
286            output: json!({}),
287            duration_ms: 0,
288            cost_usd: rust_decimal::Decimal::ZERO,
289            input_tokens: None,
290            output_tokens: None,
291            model: None,
292            debug_messages: None,
293        };
294
295        assert!(output.input_tokens.is_none());
296        assert!(output.output_tokens.is_none());
297        assert!(output.model.is_none());
298    }
299
300    #[test]
301    fn parallel_step_result_contains_step_metadata() {
302        let step_id = uuid::Uuid::now_v7();
303        let output = StepOutput {
304            output: json!({"done": true}),
305            duration_ms: 1000,
306            cost_usd: rust_decimal::Decimal::ZERO,
307            input_tokens: None,
308            output_tokens: None,
309            model: None,
310            debug_messages: None,
311        };
312
313        let result = ParallelStepResult {
314            name: "build".to_string(),
315            output,
316            step_id,
317        };
318
319        assert_eq!(result.name, "build");
320        assert_eq!(result.step_id, step_id);
321        assert_eq!(result.output.duration_ms, 1000);
322    }
323
324    #[test]
325    fn step_output_serializes_complex_json_output() {
326        let complex_output = json!({
327            "status": "success",
328            "data": {
329                "items": [1, 2, 3],
330                "nested": {
331                    "key": "value"
332                }
333            }
334        });
335
336        let output = StepOutput {
337            output: complex_output.clone(),
338            duration_ms: 100,
339            cost_usd: rust_decimal::Decimal::ZERO,
340            input_tokens: None,
341            output_tokens: None,
342            model: None,
343            debug_messages: None,
344        };
345
346        assert_eq!(output.output, complex_output);
347        assert_eq!(output.output["status"], "success");
348        assert_eq!(output.output["data"]["items"][0], 1);
349        assert_eq!(output.output["data"]["nested"]["key"], "value");
350    }
351}
ironflow_engine/executor/mod.rs

ironflow_engine/executor/
mod.rs
