ironflow_engine/executor/
shell.rs

1//! Shell step executor.
2
3use std::process::Stdio;
4use std::sync::Arc;
5use std::time::{Duration, Instant};
6
7use rust_decimal::Decimal;
8use serde_json::json;
9use tokio::io::{AsyncBufReadExt, BufReader};
10use tokio::process::Command;
11use tokio::spawn;
12use tracing::info;
13
14use ironflow_core::error::OperationError;
15use ironflow_core::operations::shell::Shell;
16use ironflow_core::provider::AgentProvider;
17use ironflow_core::utils::truncate_output;
18use ironflow_store::entities::StepKind;
19
20use crate::config::ShellConfig;
21use crate::error::EngineError;
22use crate::log_sender::StepLogSender;
23use crate::notify::LogStream;
24
25use super::{StepArtifacts, StepExecutor, StepOutput};
26
27const DEFAULT_SHELL_TIMEOUT: Duration = Duration::from_secs(300);
28
29/// Read lines from an async reader, emit each line to the sender, and
30/// accumulate the full output as a single `String`.
31async fn read_and_stream<R: tokio::io::AsyncRead + Unpin>(
32    reader: R,
33    sender: StepLogSender,
34    stream: LogStream,
35) -> String {
36    let mut lines = BufReader::new(reader).lines();
37    let mut collected = String::new();
38    while let Ok(Some(line)) = lines.next_line().await {
39        sender.emit(stream, &line);
40        if !collected.is_empty() {
41            collected.push('\n');
42        }
43        collected.push_str(&line);
44    }
45    collected
46}
47
48/// Executor for shell steps.
49///
50/// Runs a shell command and captures stdout, stderr, and exit code.
51/// When a [`StepLogSender`] is attached, stdout and stderr are streamed
52/// line-by-line in real time.
53pub struct ShellExecutor<'a> {
54    config: &'a ShellConfig,
55    log_sender: Option<StepLogSender>,
56}
57
58impl<'a> ShellExecutor<'a> {
59    /// Create a new shell executor from a config reference.
60    pub fn new(config: &'a ShellConfig) -> Self {
61        Self {
62            config,
63            log_sender: None,
64        }
65    }
66
67    /// Attach a log sender for real-time line streaming.
68    pub fn with_log_sender(mut self, sender: StepLogSender) -> Self {
69        self.log_sender = Some(sender);
70        self
71    }
72}
73
74impl StepExecutor for ShellExecutor<'_> {
75    fn kind(&self) -> StepKind {
76        StepKind::Shell
77    }
78
79    async fn execute(&self, _provider: &Arc<dyn AgentProvider>) -> Result<StepOutput, EngineError> {
80        match self.log_sender {
81            Some(ref sender) => self.execute_streaming(sender.clone()).await,
82            None => self.execute_buffered().await,
83        }
84    }
85}
86
87impl ShellExecutor<'_> {
88    /// Non-streaming execution via [`Shell::run()`].
89    async fn execute_buffered(&self) -> Result<StepOutput, EngineError> {
90        let start = Instant::now();
91
92        let mut shell = Shell::new(&self.config.command);
93        if let Some(secs) = self.config.timeout_secs {
94            shell = shell.timeout(Duration::from_secs(secs));
95        }
96        if let Some(ref dir) = self.config.dir {
97            shell = shell.dir(dir);
98        }
99        for (key, value) in &self.config.env {
100            shell = shell.env(key, value);
101        }
102        if self.config.clean_env {
103            shell = shell.clean_env();
104        }
105
106        let output = shell.run().await?;
107        let duration_ms = start.elapsed().as_millis() as u64;
108
109        info!(
110            step_kind = "shell",
111            command = %self.config.command,
112            exit_code = output.exit_code(),
113            duration_ms,
114            "shell step completed"
115        );
116
117        self.record_metrics(duration_ms);
118
119        Ok(StepOutput {
120            output: json!({
121                "stdout": output.stdout(),
122                "stderr": output.stderr(),
123                "exit_code": output.exit_code(),
124            }),
125            duration_ms,
126            cost_usd: Decimal::ZERO,
127            input_tokens: None,
128            cache_read_input_tokens: None,
129            cache_creation_input_tokens: None,
130            output_tokens: None,
131            model: None,
132            debug_messages: None,
133            artifacts: StepArtifacts::default(),
134        })
135    }
136
137    /// Streaming execution: reads stdout/stderr line-by-line and forwards
138    /// each line to the [`StepLogSender`] in real time.
139    async fn execute_streaming(&self, sender: StepLogSender) -> Result<StepOutput, EngineError> {
140        let start = Instant::now();
141
142        let mut cmd = Command::new("sh");
143        cmd.arg("-c").arg(&self.config.command);
144        cmd.stdout(Stdio::piped())
145            .stderr(Stdio::piped())
146            .kill_on_drop(true);
147
148        if self.config.clean_env {
149            cmd.env_clear();
150        }
151        if let Some(ref dir) = self.config.dir {
152            cmd.current_dir(dir);
153        }
154        for (key, value) in &self.config.env {
155            cmd.env(key, value);
156        }
157
158        let mut child = cmd.spawn().map_err(|e| {
159            EngineError::Operation(OperationError::Shell {
160                exit_code: -1,
161                stderr: format!("failed to spawn shell: {e}"),
162            })
163        })?;
164
165        let stdout_pipe = child.stdout.take().expect("stdout piped");
166        let stderr_pipe = child.stderr.take().expect("stderr piped");
167
168        let stdout_task = spawn(read_and_stream(
169            stdout_pipe,
170            sender.clone(),
171            LogStream::Stdout,
172        ));
173        let stderr_task = spawn(read_and_stream(stderr_pipe, sender, LogStream::Stderr));
174
175        let timeout_dur = self
176            .config
177            .timeout_secs
178            .map(Duration::from_secs)
179            .unwrap_or(DEFAULT_SHELL_TIMEOUT);
180
181        let status = match tokio::time::timeout(timeout_dur, child.wait()).await {
182            Ok(Ok(status)) => status,
183            Ok(Err(e)) => {
184                return Err(EngineError::Operation(OperationError::Shell {
185                    exit_code: -1,
186                    stderr: format!("failed to wait for shell: {e}"),
187                }));
188            }
189            Err(_) => {
190                child.kill().await.ok();
191                return Err(EngineError::Operation(OperationError::Timeout {
192                    step: self.config.command.clone(),
193                    limit: timeout_dur,
194                }));
195            }
196        };
197
198        let raw_stdout = stdout_task.await.unwrap_or_default();
199        let raw_stderr = stderr_task.await.unwrap_or_default();
200
201        let stdout = truncate_output(raw_stdout.as_bytes(), "shell stdout");
202        let stderr = truncate_output(raw_stderr.as_bytes(), "shell stderr");
203
204        let exit_code = status.code().unwrap_or(-1);
205        let duration_ms = start.elapsed().as_millis() as u64;
206
207        info!(
208            step_kind = "shell",
209            command = %self.config.command,
210            exit_code,
211            duration_ms,
212            streaming = true,
213            "shell step completed"
214        );
215
216        self.record_metrics(duration_ms);
217
218        if exit_code != 0 {
219            return Err(EngineError::Operation(OperationError::Shell {
220                exit_code,
221                stderr: stderr.clone(),
222            }));
223        }
224
225        Ok(StepOutput {
226            output: json!({
227                "stdout": stdout,
228                "stderr": stderr,
229                "exit_code": exit_code,
230            }),
231            duration_ms,
232            cost_usd: Decimal::ZERO,
233            input_tokens: None,
234            cache_read_input_tokens: None,
235            cache_creation_input_tokens: None,
236            output_tokens: None,
237            model: None,
238            debug_messages: None,
239            artifacts: StepArtifacts::default(),
240        })
241    }
242
243    #[allow(unused_variables)]
244    fn record_metrics(&self, duration_ms: u64) {
245        #[cfg(feature = "prometheus")]
246        {
247            use ironflow_core::metric_names::{
248                SHELL_DURATION_SECONDS, SHELL_TOTAL, STATUS_SUCCESS,
249            };
250            use metrics::{counter, histogram};
251            counter!(SHELL_TOTAL, "status" => STATUS_SUCCESS).increment(1);
252            histogram!(SHELL_DURATION_SECONDS).record(duration_ms as f64 / 1000.0);
253        }
254    }
255}
256
257#[cfg(test)]
258mod tests {
259    use super::*;
260    use ironflow_core::providers::claude::ClaudeCodeProvider;
261    use ironflow_core::providers::record_replay::RecordReplayProvider;
262
263    fn create_test_provider() -> Arc<dyn AgentProvider> {
264        let inner = ClaudeCodeProvider::new();
265        Arc::new(RecordReplayProvider::replay(
266            inner,
267            "/tmp/ironflow-fixtures",
268        ))
269    }
270
271    #[tokio::test]
272    async fn shell_simple_command() {
273        let config = ShellConfig::new("echo hello");
274        let executor = ShellExecutor::new(&config);
275        let provider = create_test_provider();
276
277        let result = executor.execute(&provider).await;
278        assert!(result.is_ok());
279        let output = result.unwrap();
280        assert_eq!(output.output["exit_code"].as_i64().unwrap(), 0);
281        assert!(output.output["stdout"].as_str().unwrap().contains("hello"));
282    }
283
284    #[tokio::test]
285    async fn shell_nonzero_exit_returns_error() {
286        let config = ShellConfig::new("exit 1");
287        let executor = ShellExecutor::new(&config);
288        let provider = create_test_provider();
289
290        let result = executor.execute(&provider).await;
291        assert!(result.is_err());
292    }
293
294    #[tokio::test]
295    async fn shell_env_variables() {
296        let config = ShellConfig::new("echo $MY_VAR").env("MY_VAR", "test_value");
297        let executor = ShellExecutor::new(&config);
298        let provider = create_test_provider();
299
300        let result = executor.execute(&provider).await;
301        assert!(result.is_ok());
302        let output = result.unwrap();
303        assert!(
304            output.output["stdout"]
305                .as_str()
306                .unwrap()
307                .contains("test_value")
308        );
309    }
310
311    #[tokio::test]
312    async fn shell_step_output_has_structure() {
313        let config = ShellConfig::new("echo test");
314        let executor = ShellExecutor::new(&config);
315        let provider = create_test_provider();
316
317        let output = executor.execute(&provider).await.unwrap();
318        assert!(output.output.get("stdout").is_some());
319        assert!(output.output.get("stderr").is_some());
320        assert!(output.output.get("exit_code").is_some());
321        assert_eq!(output.cost_usd, Decimal::ZERO);
322        assert!(output.duration_ms < 5000);
323    }
324
325    #[tokio::test]
326    async fn shell_command_with_pipe() {
327        let config = ShellConfig::new("echo hello | grep hello");
328        let executor = ShellExecutor::new(&config);
329        let provider = create_test_provider();
330
331        let result = executor.execute(&provider).await;
332        assert!(result.is_ok());
333        let output = result.unwrap();
334        assert_eq!(output.output["exit_code"].as_i64().unwrap(), 0);
335        assert!(output.output["stdout"].as_str().unwrap().contains("hello"));
336    }
337
338    #[tokio::test]
339    async fn shell_streaming_emits_lines() {
340        let config = ShellConfig::new("echo line1 && echo line2");
341        let (sender, mut receiver) = crate::log_sender::channel();
342        let step_sender = StepLogSender::new(
343            sender,
344            uuid::Uuid::now_v7(),
345            uuid::Uuid::now_v7(),
346            "test".to_string(),
347        );
348        let executor = ShellExecutor::new(&config).with_log_sender(step_sender);
349        let provider = create_test_provider();
350
351        let result = executor.execute(&provider).await;
352        assert!(result.is_ok());
353
354        let output = result.unwrap();
355        assert!(output.output["stdout"].as_str().unwrap().contains("line1"));
356        assert!(output.output["stdout"].as_str().unwrap().contains("line2"));
357
358        let mut lines = Vec::new();
359        while let Ok(line) = receiver.try_recv() {
360            lines.push(line);
361        }
362        assert!(lines.len() >= 2);
363        assert_eq!(lines[0].stream, LogStream::Stdout);
364        assert_eq!(lines[0].line, "line1");
365        assert_eq!(lines[1].line, "line2");
366    }
367
368    #[tokio::test]
369    async fn shell_streaming_captures_stderr() {
370        let config = ShellConfig::new("echo err >&2");
371        let (sender, mut receiver) = crate::log_sender::channel();
372        let step_sender = StepLogSender::new(
373            sender,
374            uuid::Uuid::now_v7(),
375            uuid::Uuid::now_v7(),
376            "test".to_string(),
377        );
378        let executor = ShellExecutor::new(&config).with_log_sender(step_sender);
379        let provider = create_test_provider();
380
381        let result = executor.execute(&provider).await;
382        assert!(result.is_ok());
383
384        let mut stderr_lines = Vec::new();
385        while let Ok(line) = receiver.try_recv() {
386            if line.stream == LogStream::Stderr {
387                stderr_lines.push(line);
388            }
389        }
390        assert!(!stderr_lines.is_empty());
391        assert_eq!(stderr_lines[0].line, "err");
392    }
393
394    #[tokio::test]
395    async fn shell_streaming_nonzero_exit_returns_error() {
396        let config = ShellConfig::new("exit 42");
397        let (sender, _receiver) = crate::log_sender::channel();
398        let step_sender = StepLogSender::new(
399            sender,
400            uuid::Uuid::now_v7(),
401            uuid::Uuid::now_v7(),
402            "test".to_string(),
403        );
404        let executor = ShellExecutor::new(&config).with_log_sender(step_sender);
405        let provider = create_test_provider();
406
407        let result = executor.execute(&provider).await;
408        assert!(result.is_err());
409    }
410}
ironflow_engine/executor/shell.rs

ironflow_engine/executor/
shell.rs
