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};
19use ironflow_store::entities::StepStatus;
20
21use crate::config::StepConfig;
22use crate::error::EngineError;
23use crate::log_sender::StepLogSender;
24
25pub use agent::AgentExecutor;
26pub use http::HttpExecutor;
27pub use shell::ShellExecutor;
28
29/// Result of executing a single step.
30#[derive(Debug, Clone)]
31pub struct StepOutput {
32    /// Serialized output (stdout for shell, body for http, value for agent).
33    ///
34    /// For agent steps with a JSON schema, the value may not strictly conform
35    /// to the schema: Claude CLI can flatten wrapper objects with a single
36    /// array field, returning a bare array instead of `{"items": [...]}`.
37    /// Callers should handle both the expected wrapper and a bare value.
38    pub output: Value,
39    /// Wall-clock duration in milliseconds.
40    pub duration_ms: u64,
41    /// Cost in USD (agent steps only).
42    pub cost_usd: Decimal,
43    /// Input token count (agent steps only).
44    pub input_tokens: Option<u64>,
45    /// Output token count (agent steps only).
46    pub output_tokens: Option<u64>,
47    /// Model identifier used for agent steps (e.g. `"claude-sonnet-4-20250514"`).
48    pub model: Option<String>,
49    /// Conversation trace from verbose agent invocations.
50    pub debug_messages: Option<Vec<DebugMessage>>,
51}
52
53impl StepOutput {
54    /// Serialize debug messages to a JSON [`Value`] for store persistence.
55    ///
56    /// Returns `None` when verbose mode was off (no messages captured).
57    pub fn debug_messages_json(&self) -> Option<Value> {
58        self.debug_messages
59            .as_ref()
60            .and_then(|msgs| serde_json::to_value(msgs).ok())
61    }
62}
63
64/// Result of a single step within a [`parallel`](crate::context::WorkflowContext::parallel) batch.
65#[derive(Debug, Clone)]
66pub struct ParallelStepResult {
67    /// The step name (same as provided to `parallel()`).
68    pub name: String,
69    /// The step execution output.
70    pub output: StepOutput,
71    /// The step ID in the store (for dependency tracking).
72    pub step_id: Uuid,
73}
74
75/// Enriched result of a completed step, for post-execution inspection.
76///
77/// Collects the step's trace ID, status, metrics, and a truncated output
78/// summary into a single struct that the [`WorkflowContext`](crate::context::WorkflowContext)
79/// accumulates over the run.
80///
81/// # Examples
82///
83/// ```
84/// use ironflow_engine::executor::StepResult;
85/// use ironflow_store::entities::StepStatus;
86/// use rust_decimal::Decimal;
87/// use uuid::Uuid;
88///
89/// let result = StepResult {
90///     trace_id: Uuid::nil(),
91///     name: "build".to_string(),
92///     status: StepStatus::Completed,
93///     duration_ms: 1200,
94///     cost_usd: Decimal::ZERO,
95///     input_tokens: None,
96///     output_tokens: None,
97///     error: None,
98///     output_summary: Some("ok".to_string()),
99/// };
100/// assert_eq!(result.status, StepStatus::Completed);
101/// ```
102#[derive(Debug, Clone, serde::Serialize)]
103pub struct StepResult {
104    /// Deterministic trace ID for log correlation.
105    pub trace_id: Uuid,
106    /// Step name.
107    pub name: String,
108    /// Terminal status.
109    pub status: StepStatus,
110    /// Wall-clock duration in milliseconds.
111    pub duration_ms: u64,
112    /// Cost in USD.
113    pub cost_usd: Decimal,
114    /// Input token count (agent steps only).
115    pub input_tokens: Option<u64>,
116    /// Output token count (agent steps only).
117    pub output_tokens: Option<u64>,
118    /// Error message if the step failed.
119    pub error: Option<String>,
120    /// First 500 characters of the serialized output.
121    pub output_summary: Option<String>,
122}
123
124/// Maximum length of [`StepResult::output_summary`].
125const OUTPUT_SUMMARY_MAX_LEN: usize = 500;
126
127impl StepResult {
128    /// Build from a completed step's output.
129    pub fn from_success(trace_id: Uuid, name: &str, output: &StepOutput) -> Self {
130        Self {
131            trace_id,
132            name: name.to_string(),
133            status: StepStatus::Completed,
134            duration_ms: output.duration_ms,
135            cost_usd: output.cost_usd,
136            input_tokens: output.input_tokens,
137            output_tokens: output.output_tokens,
138            error: None,
139            output_summary: summarize_output(&output.output),
140        }
141    }
142
143    /// Build from a failed step.
144    pub fn from_failure(
145        trace_id: Uuid,
146        name: &str,
147        error: &str,
148        duration_ms: u64,
149        cost_usd: Decimal,
150    ) -> Self {
151        Self {
152            trace_id,
153            name: name.to_string(),
154            status: StepStatus::Failed,
155            duration_ms,
156            cost_usd,
157            input_tokens: None,
158            output_tokens: None,
159            error: Some(error.to_string()),
160            output_summary: None,
161        }
162    }
163}
164
165fn summarize_output(value: &Value) -> Option<String> {
166    let raw = value.to_string();
167    match raw.char_indices().nth(OUTPUT_SUMMARY_MAX_LEN) {
168        None => Some(raw),
169        Some((byte_idx, _)) => Some(raw[..byte_idx].to_string()),
170    }
171}
172
173/// Trait for step executors.
174///
175/// Each step type implements this trait to execute its specific operation
176/// and return a [`StepOutput`].
177pub trait StepExecutor: Send + Sync {
178    /// Execute the step and return structured output.
179    ///
180    /// # Errors
181    ///
182    /// Returns [`EngineError`] if the operation fails.
183    fn execute(
184        &self,
185        provider: &Arc<dyn AgentProvider>,
186    ) -> impl Future<Output = Result<StepOutput, EngineError>> + Send;
187}
188
189/// Execute a [`StepConfig`] and return structured output.
190///
191/// When a [`StepLogSender`] is provided, executors that support streaming
192/// will emit log lines in real time (e.g. shell stdout/stderr).
193///
194/// # Errors
195///
196/// Returns [`EngineError::Operation`] if the operation fails.
197///
198/// # Examples
199///
200/// ```no_run
201/// use ironflow_engine::config::{StepConfig, ShellConfig};
202/// use ironflow_engine::executor::execute_step_config;
203/// use ironflow_core::provider::AgentProvider;
204/// use ironflow_core::providers::claude::ClaudeCodeProvider;
205/// use std::sync::Arc;
206///
207/// # async fn example() -> Result<(), ironflow_engine::error::EngineError> {
208/// let provider: Arc<dyn AgentProvider> = Arc::new(ClaudeCodeProvider::new());
209/// let config = StepConfig::Shell(ShellConfig::new("echo hello"));
210/// let output = execute_step_config(&config, &provider, None).await?;
211/// # Ok(())
212/// # }
213/// ```
214#[tracing::instrument(name = "executor.execute_step", skip_all, fields(step.kind))]
215pub async fn execute_step_config(
216    config: &StepConfig,
217    provider: &Arc<dyn AgentProvider>,
218    log_sender: Option<StepLogSender>,
219) -> Result<StepOutput, EngineError> {
220    let kind = match config {
221        StepConfig::Shell(_) => "shell",
222        StepConfig::Http(_) => "http",
223        StepConfig::Agent(_) => "agent",
224        StepConfig::Workflow(_) => "workflow",
225        StepConfig::Approval(_) => "approval",
226    };
227    tracing::Span::current().record("step.kind", kind);
228
229    let result = match config {
230        StepConfig::Shell(cfg) => {
231            let mut executor = ShellExecutor::new(cfg);
232            if let Some(sender) = log_sender {
233                executor = executor.with_log_sender(sender);
234            }
235            executor.execute(provider).await
236        }
237        StepConfig::Http(cfg) => HttpExecutor::new(cfg).execute(provider).await,
238        StepConfig::Agent(cfg) => {
239            let mut executor = AgentExecutor::new(cfg);
240            if let Some(sender) = log_sender {
241                executor = executor.with_log_sender(sender);
242            }
243            executor.execute(provider).await
244        }
245        StepConfig::Workflow(_) => Err(EngineError::StepConfig(
246            "workflow steps are executed by WorkflowContext, not the executor".to_string(),
247        )),
248        StepConfig::Approval(_) => Err(EngineError::StepConfig(
249            "approval steps are executed by WorkflowContext, not the executor".to_string(),
250        )),
251    };
252
253    #[cfg(feature = "prometheus")]
254    {
255        use ironflow_core::metric_names::{
256            STATUS_ERROR, STATUS_SUCCESS, STEP_DURATION_SECONDS, STEPS_TOTAL,
257        };
258        use metrics::{counter, histogram};
259        let status = if result.is_ok() {
260            STATUS_SUCCESS
261        } else {
262            STATUS_ERROR
263        };
264        counter!(STEPS_TOTAL, "kind" => kind, "status" => status).increment(1);
265        if let Ok(ref output) = result {
266            histogram!(STEP_DURATION_SECONDS, "kind" => kind)
267                .record(output.duration_ms as f64 / 1000.0);
268        }
269    }
270
271    result
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277    use ironflow_core::provider::DebugMessage;
278    use serde_json::json;
279
280    #[test]
281    fn step_output_with_no_debug_messages_returns_none() {
282        let output = StepOutput {
283            output: json!({"result": "ok"}),
284            duration_ms: 100,
285            cost_usd: rust_decimal::Decimal::ZERO,
286            input_tokens: None,
287            output_tokens: None,
288            model: None,
289            debug_messages: None,
290        };
291
292        assert_eq!(output.debug_messages_json(), None);
293    }
294
295    #[test]
296    fn step_output_with_empty_debug_messages_returns_some_empty_array() {
297        let output = StepOutput {
298            output: json!({"result": "ok"}),
299            duration_ms: 100,
300            cost_usd: rust_decimal::Decimal::ZERO,
301            input_tokens: None,
302            output_tokens: None,
303            model: None,
304            debug_messages: Some(Vec::new()),
305        };
306
307        let json_val = output.debug_messages_json();
308        assert!(json_val.is_some());
309        let arr = json_val.unwrap();
310        assert!(arr.is_array());
311        assert_eq!(arr.as_array().unwrap().len(), 0);
312    }
313
314    #[test]
315    fn step_output_debug_messages_json_serializes_messages() {
316        let json_msgs = json!([
317            {
318                "text": "Hello",
319                "thinking": null,
320                "thinking_redacted": false,
321                "tool_calls": [],
322                "tool_results": [],
323                "stop_reason": "end_turn",
324                "input_tokens": 10,
325                "output_tokens": 20
326            },
327            {
328                "text": "Hi there",
329                "thinking": null,
330                "thinking_redacted": false,
331                "tool_calls": [],
332                "tool_results": [],
333                "stop_reason": "end_turn",
334                "input_tokens": 15,
335                "output_tokens": 25
336            }
337        ]);
338
339        let messages: Vec<DebugMessage> =
340            serde_json::from_value(json_msgs.clone()).expect("deserialize debug messages");
341
342        let output = StepOutput {
343            output: json!({"result": "ok"}),
344            duration_ms: 100,
345            cost_usd: rust_decimal::Decimal::ZERO,
346            input_tokens: None,
347            output_tokens: None,
348            model: None,
349            debug_messages: Some(messages),
350        };
351
352        let json_val = output.debug_messages_json();
353        assert!(json_val.is_some());
354
355        let arr = json_val.unwrap();
356        assert!(arr.is_array());
357        let messages_array = arr.as_array().unwrap();
358        assert_eq!(messages_array.len(), 2);
359        assert_eq!(messages_array[0]["text"], "Hello");
360        assert_eq!(messages_array[1]["text"], "Hi there");
361    }
362
363    #[test]
364    fn step_output_contains_all_metrics() {
365        let output = StepOutput {
366            output: json!({"data": "test"}),
367            duration_ms: 5000,
368            cost_usd: rust_decimal::Decimal::new(123, 2),
369            input_tokens: Some(100),
370            output_tokens: Some(200),
371            model: Some("claude-sonnet".to_string()),
372            debug_messages: None,
373        };
374
375        assert_eq!(output.duration_ms, 5000);
376        assert_eq!(output.cost_usd, rust_decimal::Decimal::new(123, 2));
377        assert_eq!(output.input_tokens, Some(100));
378        assert_eq!(output.output_tokens, Some(200));
379        assert_eq!(output.model, Some("claude-sonnet".to_string()));
380    }
381
382    #[test]
383    fn step_output_default_tokens_and_model_are_none() {
384        let output = StepOutput {
385            output: json!({}),
386            duration_ms: 0,
387            cost_usd: rust_decimal::Decimal::ZERO,
388            input_tokens: None,
389            output_tokens: None,
390            model: None,
391            debug_messages: None,
392        };
393
394        assert!(output.input_tokens.is_none());
395        assert!(output.output_tokens.is_none());
396        assert!(output.model.is_none());
397    }
398
399    #[test]
400    fn parallel_step_result_contains_step_metadata() {
401        let step_id = uuid::Uuid::now_v7();
402        let output = StepOutput {
403            output: json!({"done": true}),
404            duration_ms: 1000,
405            cost_usd: rust_decimal::Decimal::ZERO,
406            input_tokens: None,
407            output_tokens: None,
408            model: None,
409            debug_messages: None,
410        };
411
412        let result = ParallelStepResult {
413            name: "build".to_string(),
414            output,
415            step_id,
416        };
417
418        assert_eq!(result.name, "build");
419        assert_eq!(result.step_id, step_id);
420        assert_eq!(result.output.duration_ms, 1000);
421    }
422
423    #[test]
424    fn step_output_serializes_complex_json_output() {
425        let complex_output = json!({
426            "status": "success",
427            "data": {
428                "items": [1, 2, 3],
429                "nested": {
430                    "key": "value"
431                }
432            }
433        });
434
435        let output = StepOutput {
436            output: complex_output.clone(),
437            duration_ms: 100,
438            cost_usd: rust_decimal::Decimal::ZERO,
439            input_tokens: None,
440            output_tokens: None,
441            model: None,
442            debug_messages: None,
443        };
444
445        assert_eq!(output.output, complex_output);
446        assert_eq!(output.output["status"], "success");
447        assert_eq!(output.output["data"]["items"][0], 1);
448        assert_eq!(output.output["data"]["nested"]["key"], "value");
449    }
450
451    #[test]
452    fn step_result_from_success_captures_all_fields() {
453        let trace_id = Uuid::nil();
454        let output = StepOutput {
455            output: json!({"stdout": "ok"}),
456            duration_ms: 1500,
457            cost_usd: Decimal::new(42, 2),
458            input_tokens: Some(100),
459            output_tokens: Some(200),
460            model: Some("claude-sonnet".to_string()),
461            debug_messages: None,
462        };
463
464        let result = StepResult::from_success(trace_id, "build", &output);
465
466        assert_eq!(result.trace_id, trace_id);
467        assert_eq!(result.name, "build");
468        assert_eq!(result.status, StepStatus::Completed);
469        assert_eq!(result.duration_ms, 1500);
470        assert_eq!(result.cost_usd, Decimal::new(42, 2));
471        assert_eq!(result.input_tokens, Some(100));
472        assert_eq!(result.output_tokens, Some(200));
473        assert!(result.error.is_none());
474        assert!(result.output_summary.is_some());
475        assert!(result.output_summary.unwrap().contains("stdout"));
476    }
477
478    #[test]
479    fn step_result_from_failure_captures_error() {
480        let trace_id = Uuid::nil();
481        let result =
482            StepResult::from_failure(trace_id, "deploy", "connection refused", 500, Decimal::ZERO);
483
484        assert_eq!(result.trace_id, trace_id);
485        assert_eq!(result.name, "deploy");
486        assert_eq!(result.status, StepStatus::Failed);
487        assert_eq!(result.duration_ms, 500);
488        assert_eq!(result.error, Some("connection refused".to_string()));
489        assert!(result.output_summary.is_none());
490    }
491
492    #[test]
493    fn step_result_output_summary_truncates_long_output() {
494        let long_value = json!({"data": "x".repeat(1000)});
495        let output = StepOutput {
496            output: long_value,
497            duration_ms: 0,
498            cost_usd: Decimal::ZERO,
499            input_tokens: None,
500            output_tokens: None,
501            model: None,
502            debug_messages: None,
503        };
504
505        let result = StepResult::from_success(Uuid::nil(), "test", &output);
506        let summary = result.output_summary.unwrap();
507        assert_eq!(summary.len(), 500);
508    }
509}
ironflow_engine/executor/mod.rs

ironflow_engine/executor/
mod.rs
