ironflow_engine/executor/
shell.rs

1//! Shell step executor.
2
3use std::sync::Arc;
4use std::time::{Duration, Instant};
5
6use rust_decimal::Decimal;
7use serde_json::json;
8use tracing::info;
9
10use ironflow_core::operations::shell::Shell;
11use ironflow_core::provider::AgentProvider;
12
13use crate::config::ShellConfig;
14use crate::error::EngineError;
15
16use super::{StepExecutor, StepOutput};
17
18/// Executor for shell steps.
19///
20/// Runs a shell command and captures stdout, stderr, and exit code.
21pub struct ShellExecutor<'a> {
22    config: &'a ShellConfig,
23}
24
25impl<'a> ShellExecutor<'a> {
26    /// Create a new shell executor from a config reference.
27    pub fn new(config: &'a ShellConfig) -> Self {
28        Self { config }
29    }
30}
31
32impl StepExecutor for ShellExecutor<'_> {
33    async fn execute(&self, _provider: &Arc<dyn AgentProvider>) -> Result<StepOutput, EngineError> {
34        let start = Instant::now();
35
36        let mut shell = Shell::new(&self.config.command);
37        if let Some(secs) = self.config.timeout_secs {
38            shell = shell.timeout(Duration::from_secs(secs));
39        }
40        if let Some(ref dir) = self.config.dir {
41            shell = shell.dir(dir);
42        }
43        for (key, value) in &self.config.env {
44            shell = shell.env(key, value);
45        }
46        if self.config.clean_env {
47            shell = shell.clean_env();
48        }
49
50        let output = shell.run().await?;
51        let duration_ms = start.elapsed().as_millis() as u64;
52
53        info!(
54            step_kind = "shell",
55            command = %self.config.command,
56            exit_code = output.exit_code(),
57            duration_ms,
58            "shell step completed"
59        );
60
61        #[cfg(feature = "prometheus")]
62        {
63            use ironflow_core::metric_names::{
64                SHELL_DURATION_SECONDS, SHELL_TOTAL, STATUS_SUCCESS,
65            };
66            use metrics::{counter, histogram};
67            counter!(SHELL_TOTAL, "status" => STATUS_SUCCESS).increment(1);
68            histogram!(SHELL_DURATION_SECONDS).record(duration_ms as f64 / 1000.0);
69        }
70
71        Ok(StepOutput {
72            output: json!({
73                "stdout": output.stdout(),
74                "stderr": output.stderr(),
75                "exit_code": output.exit_code(),
76            }),
77            duration_ms,
78            cost_usd: Decimal::ZERO,
79            input_tokens: None,
80            output_tokens: None,
81        })
82    }
83}
84
85#[cfg(test)]
86mod tests {
87    use super::*;
88    use ironflow_core::providers::claude::ClaudeCodeProvider;
89    use ironflow_core::providers::record_replay::RecordReplayProvider;
90
91    fn create_test_provider() -> Arc<dyn AgentProvider> {
92        let inner = ClaudeCodeProvider::new();
93        Arc::new(RecordReplayProvider::replay(
94            inner,
95            "/tmp/ironflow-fixtures",
96        ))
97    }
98
99    #[tokio::test]
100    async fn shell_simple_command() {
101        let config = ShellConfig::new("echo hello");
102        let executor = ShellExecutor::new(&config);
103        let provider = create_test_provider();
104
105        let result = executor.execute(&provider).await;
106        assert!(result.is_ok());
107        let output = result.unwrap();
108        assert_eq!(output.output["exit_code"].as_i64().unwrap(), 0);
109        assert!(output.output["stdout"].as_str().unwrap().contains("hello"));
110    }
111
112    #[tokio::test]
113    async fn shell_nonzero_exit_returns_error() {
114        let config = ShellConfig::new("exit 1");
115        let executor = ShellExecutor::new(&config);
116        let provider = create_test_provider();
117
118        let result = executor.execute(&provider).await;
119        assert!(result.is_err());
120    }
121
122    #[tokio::test]
123    async fn shell_env_variables() {
124        let config = ShellConfig::new("echo $MY_VAR").env("MY_VAR", "test_value");
125        let executor = ShellExecutor::new(&config);
126        let provider = create_test_provider();
127
128        let result = executor.execute(&provider).await;
129        assert!(result.is_ok());
130        let output = result.unwrap();
131        assert!(
132            output.output["stdout"]
133                .as_str()
134                .unwrap()
135                .contains("test_value")
136        );
137    }
138
139    #[tokio::test]
140    async fn shell_step_output_has_structure() {
141        let config = ShellConfig::new("echo test");
142        let executor = ShellExecutor::new(&config);
143        let provider = create_test_provider();
144
145        let output = executor.execute(&provider).await.unwrap();
146        assert!(output.output.get("stdout").is_some());
147        assert!(output.output.get("stderr").is_some());
148        assert!(output.output.get("exit_code").is_some());
149        assert_eq!(output.cost_usd, Decimal::ZERO);
150        // duration_ms can be 0 when the command completes in under 1ms (CI)
151        assert!(output.duration_ms < 5000);
152    }
153
154    #[tokio::test]
155    async fn shell_command_with_pipe() {
156        let config = ShellConfig::new("echo hello | grep hello");
157        let executor = ShellExecutor::new(&config);
158        let provider = create_test_provider();
159
160        let result = executor.execute(&provider).await;
161        assert!(result.is_ok());
162        let output = result.unwrap();
163        assert_eq!(output.output["exit_code"].as_i64().unwrap(), 0);
164        assert!(output.output["stdout"].as_str().unwrap().contains("hello"));
165    }
166}
ironflow_engine/executor/shell.rs

ironflow_engine/executor/
shell.rs
