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::{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            output_tokens: None,
129            model: None,
130            debug_messages: None,
131        })
132    }
133
134    /// Streaming execution: reads stdout/stderr line-by-line and forwards
135    /// each line to the [`StepLogSender`] in real time.
136    async fn execute_streaming(&self, sender: StepLogSender) -> Result<StepOutput, EngineError> {
137        let start = Instant::now();
138
139        let mut cmd = Command::new("sh");
140        cmd.arg("-c").arg(&self.config.command);
141        cmd.stdout(Stdio::piped())
142            .stderr(Stdio::piped())
143            .kill_on_drop(true);
144
145        if self.config.clean_env {
146            cmd.env_clear();
147        }
148        if let Some(ref dir) = self.config.dir {
149            cmd.current_dir(dir);
150        }
151        for (key, value) in &self.config.env {
152            cmd.env(key, value);
153        }
154
155        let mut child = cmd.spawn().map_err(|e| {
156            EngineError::Operation(OperationError::Shell {
157                exit_code: -1,
158                stderr: format!("failed to spawn shell: {e}"),
159            })
160        })?;
161
162        let stdout_pipe = child.stdout.take().expect("stdout piped");
163        let stderr_pipe = child.stderr.take().expect("stderr piped");
164
165        let stdout_task = spawn(read_and_stream(
166            stdout_pipe,
167            sender.clone(),
168            LogStream::Stdout,
169        ));
170        let stderr_task = spawn(read_and_stream(stderr_pipe, sender, LogStream::Stderr));
171
172        let timeout_dur = self
173            .config
174            .timeout_secs
175            .map(Duration::from_secs)
176            .unwrap_or(DEFAULT_SHELL_TIMEOUT);
177
178        let status = match tokio::time::timeout(timeout_dur, child.wait()).await {
179            Ok(Ok(status)) => status,
180            Ok(Err(e)) => {
181                return Err(EngineError::Operation(OperationError::Shell {
182                    exit_code: -1,
183                    stderr: format!("failed to wait for shell: {e}"),
184                }));
185            }
186            Err(_) => {
187                child.kill().await.ok();
188                return Err(EngineError::Operation(OperationError::Timeout {
189                    step: self.config.command.clone(),
190                    limit: timeout_dur,
191                }));
192            }
193        };
194
195        let raw_stdout = stdout_task.await.unwrap_or_default();
196        let raw_stderr = stderr_task.await.unwrap_or_default();
197
198        let stdout = truncate_output(raw_stdout.as_bytes(), "shell stdout");
199        let stderr = truncate_output(raw_stderr.as_bytes(), "shell stderr");
200
201        let exit_code = status.code().unwrap_or(-1);
202        let duration_ms = start.elapsed().as_millis() as u64;
203
204        info!(
205            step_kind = "shell",
206            command = %self.config.command,
207            exit_code,
208            duration_ms,
209            streaming = true,
210            "shell step completed"
211        );
212
213        self.record_metrics(duration_ms);
214
215        if exit_code != 0 {
216            return Err(EngineError::Operation(OperationError::Shell {
217                exit_code,
218                stderr: stderr.clone(),
219            }));
220        }
221
222        Ok(StepOutput {
223            output: json!({
224                "stdout": stdout,
225                "stderr": stderr,
226                "exit_code": exit_code,
227            }),
228            duration_ms,
229            cost_usd: Decimal::ZERO,
230            input_tokens: None,
231            output_tokens: None,
232            model: None,
233            debug_messages: None,
234        })
235    }
236
237    #[allow(unused_variables)]
238    fn record_metrics(&self, duration_ms: u64) {
239        #[cfg(feature = "prometheus")]
240        {
241            use ironflow_core::metric_names::{
242                SHELL_DURATION_SECONDS, SHELL_TOTAL, STATUS_SUCCESS,
243            };
244            use metrics::{counter, histogram};
245            counter!(SHELL_TOTAL, "status" => STATUS_SUCCESS).increment(1);
246            histogram!(SHELL_DURATION_SECONDS).record(duration_ms as f64 / 1000.0);
247        }
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use super::*;
254    use ironflow_core::providers::claude::ClaudeCodeProvider;
255    use ironflow_core::providers::record_replay::RecordReplayProvider;
256
257    fn create_test_provider() -> Arc<dyn AgentProvider> {
258        let inner = ClaudeCodeProvider::new();
259        Arc::new(RecordReplayProvider::replay(
260            inner,
261            "/tmp/ironflow-fixtures",
262        ))
263    }
264
265    #[tokio::test]
266    async fn shell_simple_command() {
267        let config = ShellConfig::new("echo hello");
268        let executor = ShellExecutor::new(&config);
269        let provider = create_test_provider();
270
271        let result = executor.execute(&provider).await;
272        assert!(result.is_ok());
273        let output = result.unwrap();
274        assert_eq!(output.output["exit_code"].as_i64().unwrap(), 0);
275        assert!(output.output["stdout"].as_str().unwrap().contains("hello"));
276    }
277
278    #[tokio::test]
279    async fn shell_nonzero_exit_returns_error() {
280        let config = ShellConfig::new("exit 1");
281        let executor = ShellExecutor::new(&config);
282        let provider = create_test_provider();
283
284        let result = executor.execute(&provider).await;
285        assert!(result.is_err());
286    }
287
288    #[tokio::test]
289    async fn shell_env_variables() {
290        let config = ShellConfig::new("echo $MY_VAR").env("MY_VAR", "test_value");
291        let executor = ShellExecutor::new(&config);
292        let provider = create_test_provider();
293
294        let result = executor.execute(&provider).await;
295        assert!(result.is_ok());
296        let output = result.unwrap();
297        assert!(
298            output.output["stdout"]
299                .as_str()
300                .unwrap()
301                .contains("test_value")
302        );
303    }
304
305    #[tokio::test]
306    async fn shell_step_output_has_structure() {
307        let config = ShellConfig::new("echo test");
308        let executor = ShellExecutor::new(&config);
309        let provider = create_test_provider();
310
311        let output = executor.execute(&provider).await.unwrap();
312        assert!(output.output.get("stdout").is_some());
313        assert!(output.output.get("stderr").is_some());
314        assert!(output.output.get("exit_code").is_some());
315        assert_eq!(output.cost_usd, Decimal::ZERO);
316        assert!(output.duration_ms < 5000);
317    }
318
319    #[tokio::test]
320    async fn shell_command_with_pipe() {
321        let config = ShellConfig::new("echo hello | grep hello");
322        let executor = ShellExecutor::new(&config);
323        let provider = create_test_provider();
324
325        let result = executor.execute(&provider).await;
326        assert!(result.is_ok());
327        let output = result.unwrap();
328        assert_eq!(output.output["exit_code"].as_i64().unwrap(), 0);
329        assert!(output.output["stdout"].as_str().unwrap().contains("hello"));
330    }
331
332    #[tokio::test]
333    async fn shell_streaming_emits_lines() {
334        let config = ShellConfig::new("echo line1 && echo line2");
335        let (sender, mut receiver) = crate::log_sender::channel();
336        let step_sender = StepLogSender::new(
337            sender,
338            uuid::Uuid::now_v7(),
339            uuid::Uuid::now_v7(),
340            "test".to_string(),
341        );
342        let executor = ShellExecutor::new(&config).with_log_sender(step_sender);
343        let provider = create_test_provider();
344
345        let result = executor.execute(&provider).await;
346        assert!(result.is_ok());
347
348        let output = result.unwrap();
349        assert!(output.output["stdout"].as_str().unwrap().contains("line1"));
350        assert!(output.output["stdout"].as_str().unwrap().contains("line2"));
351
352        let mut lines = Vec::new();
353        while let Ok(line) = receiver.try_recv() {
354            lines.push(line);
355        }
356        assert!(lines.len() >= 2);
357        assert_eq!(lines[0].stream, LogStream::Stdout);
358        assert_eq!(lines[0].line, "line1");
359        assert_eq!(lines[1].line, "line2");
360    }
361
362    #[tokio::test]
363    async fn shell_streaming_captures_stderr() {
364        let config = ShellConfig::new("echo err >&2");
365        let (sender, mut receiver) = crate::log_sender::channel();
366        let step_sender = StepLogSender::new(
367            sender,
368            uuid::Uuid::now_v7(),
369            uuid::Uuid::now_v7(),
370            "test".to_string(),
371        );
372        let executor = ShellExecutor::new(&config).with_log_sender(step_sender);
373        let provider = create_test_provider();
374
375        let result = executor.execute(&provider).await;
376        assert!(result.is_ok());
377
378        let mut stderr_lines = Vec::new();
379        while let Ok(line) = receiver.try_recv() {
380            if line.stream == LogStream::Stderr {
381                stderr_lines.push(line);
382            }
383        }
384        assert!(!stderr_lines.is_empty());
385        assert_eq!(stderr_lines[0].line, "err");
386    }
387
388    #[tokio::test]
389    async fn shell_streaming_nonzero_exit_returns_error() {
390        let config = ShellConfig::new("exit 42");
391        let (sender, _receiver) = crate::log_sender::channel();
392        let step_sender = StepLogSender::new(
393            sender,
394            uuid::Uuid::now_v7(),
395            uuid::Uuid::now_v7(),
396            "test".to_string(),
397        );
398        let executor = ShellExecutor::new(&config).with_log_sender(step_sender);
399        let provider = create_test_provider();
400
401        let result = executor.execute(&provider).await;
402        assert!(result.is_err());
403    }
404}
ironflow_engine/executor/shell.rs

ironflow_engine/executor/
shell.rs
