1use std::collections::BTreeMap;
10
11use serde::{Deserialize, Serialize};
12use tokio::process::Command;
13
14use super::exit::Ending;
15use super::failure::{spawn_failure, unreadable_ending_clause};
16use super::template::{CommandTemplate, SubstitutionError, TemplateError};
17use super::world::place_in_declared_world;
18use crate::activity::ActivityFailure;
19use crate::command_transcript::CommandTranscript;
20use crate::context::ActivityContext;
21use crate::process::{CancellableCommandOutput, run_cancellable_command};
22
23#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
25pub struct ShellOutcome {
26 pub exit_code: i32,
28 pub stdout: String,
31 pub stderr: String,
35}
36
37#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
39pub enum FailureMode {
40 #[default]
46 Retryable,
47 Terminal,
51}
52
53#[derive(Debug, Clone)]
55pub struct ShellAction {
56 template: CommandTemplate,
57 failure_mode: FailureMode,
58 environment: BTreeMap<String, String>,
59 working_directory: Option<std::path::PathBuf>,
60}
61
62impl ShellAction {
63 pub fn new(command: &str) -> Result<Self, TemplateError> {
71 Ok(Self {
72 template: CommandTemplate::parse(command)?,
73 failure_mode: FailureMode::default(),
74 environment: BTreeMap::new(),
75 working_directory: None,
76 })
77 }
78
79 #[must_use]
81 pub const fn with_failure_mode(mut self, failure_mode: FailureMode) -> Self {
82 self.failure_mode = failure_mode;
83 self
84 }
85
86 #[must_use]
94 pub fn with_environment(mut self, environment: BTreeMap<String, String>) -> Self {
95 self.environment = environment;
96 self
97 }
98
99 #[must_use]
108 pub fn with_working_directory(mut self, directory: impl Into<std::path::PathBuf>) -> Self {
109 self.working_directory = Some(directory.into());
110 self
111 }
112
113 #[must_use]
118 pub fn referenced_parameters(&self) -> Vec<String> {
119 self.template.referenced_parameters()
120 }
121
122 pub async fn run(
150 &self,
151 arguments: &BTreeMap<String, serde_json::Value>,
152 context: &ActivityContext,
153 ) -> Result<ShellOutcome, ActivityFailure> {
154 if context.is_cancelled() {
161 return Err(ActivityFailure::terminal(
162 "this command was cancelled before it started, so it never ran",
163 ));
164 }
165 let argv = self
166 .template
167 .render(arguments)
168 .map_err(|error| substitution_failure(&error))?;
169 let (program, rest) = argv
170 .split_first()
171 .ok_or_else(|| ActivityFailure::terminal("the declared command rendered no program"))?;
175
176 let mut command = Command::new(program);
177 command.args(rest);
178 place_in_declared_world(
183 &mut command,
184 self.environment
185 .iter()
186 .map(|(name, value)| (name.as_str(), value.as_str())),
187 );
188 if let Some(directory) = &self.working_directory {
189 command.current_dir(directory);
190 }
191
192 let transcript = CommandTranscript::new(context);
193 match run_cancellable_command(command, context.cancelled(), &transcript).await {
194 Ok(CancellableCommandOutput::Completed(output)) => {
195 let ending = Ending::of(output.status);
196 let stdout = trim_trailing_newline(&String::from_utf8_lossy(&output.stdout));
197 let stderr = trim_trailing_newline(&String::from_utf8_lossy(&output.stderr));
198 let Some(exit_code) = ending.reported_code() else {
199 return Err(ActivityFailure::terminal(format!(
206 "the command `{program}` {ended}{unreadable}",
207 ended = ending.described(),
208 unreadable = unreadable_ending_clause(ending),
209 )));
210 };
211 let outcome = ShellOutcome {
212 exit_code,
213 stdout,
214 stderr,
215 };
216 if ending.succeeded() {
217 Ok(outcome)
218 } else {
219 Err(self.exit_failure(program, ending, &outcome))
220 }
221 }
222 Ok(CancellableCommandOutput::Cancelled) => Err(ActivityFailure::terminal(format!(
223 "the command `{program}` was cancelled: its process group was terminated and \
224 proven gone, so nothing it started is still running"
225 ))),
226 Err(error) => Err(spawn_failure(program, None, &error)),
227 }
228 }
229
230 fn exit_failure(
232 &self,
233 program: &str,
234 ending: Ending,
235 outcome: &ShellOutcome,
236 ) -> ActivityFailure {
237 let ended = ending.described();
243 let message = if outcome.stderr.is_empty() {
244 format!("the command `{program}` {ended} and wrote nothing to standard error")
245 } else {
246 format!(
247 "the command `{program}` {ended} and wrote to standard error: {}",
248 outcome.stderr
249 )
250 };
251 match self.failure_mode {
252 FailureMode::Retryable => ActivityFailure::retryable(message),
253 FailureMode::Terminal => ActivityFailure::terminal(message),
254 }
255 }
256}
257
258fn substitution_failure(error: &SubstitutionError) -> ActivityFailure {
261 ActivityFailure::terminal(error.to_string())
262}
263
264pub(super) fn trim_trailing_newline(text: &str) -> String {
271 text.strip_suffix('\n')
272 .map_or(text, |trimmed| {
273 trimmed.strip_suffix('\r').unwrap_or(trimmed)
274 })
275 .to_owned()
276}
277
278#[cfg(test)]
279mod tests {
280 use super::{FailureMode, ShellAction, trim_trailing_newline};
281 use crate::activity::Classification;
282 use crate::context::ActivityContext;
283 use aion_core::{ActivityId, RunId, WorkflowId};
284 use std::collections::BTreeMap;
285
286 fn context() -> (ActivityContext, crate::context::ActivityCancellationHandle) {
287 ActivityContext::new(
288 WorkflowId::new_v4(),
289 RunId::new_v4(),
290 ActivityId::from_sequence_position(1),
291 1,
292 )
293 }
294
295 fn arguments(pairs: &[(&str, serde_json::Value)]) -> BTreeMap<String, serde_json::Value> {
296 pairs
297 .iter()
298 .map(|(name, value)| ((*name).to_owned(), value.clone()))
299 .collect()
300 }
301
302 type TestResult = Result<(), Box<dyn std::error::Error>>;
306
307 #[tokio::test]
308 async fn a_succeeding_command_returns_its_output() -> TestResult {
309 let action = ShellAction::new("echo hello")?;
310 let (context, _handle) = context();
311 let outcome = action.run(&BTreeMap::new(), &context).await?;
312 assert_eq!(outcome.exit_code, 0);
313 assert_eq!(outcome.stdout, "hello");
314 assert_eq!(outcome.stderr, "");
315 Ok(())
316 }
317
318 #[tokio::test]
319 async fn a_parameter_value_reaches_the_program_as_one_argument() -> TestResult {
320 let action = ShellAction::new("echo {{greeting}}")?;
321 let (context, _handle) = context();
322 let outcome = action
323 .run(
324 &arguments(&[("greeting", serde_json::json!("hello there world"))]),
325 &context,
326 )
327 .await?;
328 assert_eq!(outcome.stdout, "hello there world");
329 Ok(())
330 }
331
332 #[tokio::test]
333 async fn a_hostile_value_is_inert_because_no_shell_ever_sees_it() -> TestResult {
334 let action = ShellAction::new("echo {{value}}")?;
337 let (context, _handle) = context();
338 let outcome = action
339 .run(
340 &arguments(&[("value", serde_json::json!("hi; echo PWNED"))]),
341 &context,
342 )
343 .await?;
344 assert_eq!(outcome.stdout, "hi; echo PWNED");
345 assert!(
346 !outcome.stdout.contains("PWNED\n"),
347 "the injected command must never have run"
348 );
349 Ok(())
350 }
351
352 #[tokio::test]
353 async fn a_nonzero_exit_is_retryable_by_default_and_carries_stderr() -> TestResult {
354 let action = ShellAction::new("sh -c 'echo trouble >&2; exit 3'")?;
355 let (context, _handle) = context();
356 let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
357 return Err("a non-zero exit must fail the activity".into());
358 };
359 assert_eq!(failure.classification(), &Classification::Retryable);
360 assert!(
361 failure.message().contains("trouble"),
362 "the failure must carry the command's own words: {}",
363 failure.message()
364 );
365 assert!(failure.message().contains('3'), "the exit code is reported");
366 Ok(())
367 }
368
369 #[tokio::test]
370 async fn a_nonzero_exit_is_terminal_when_the_action_says_so() -> TestResult {
371 let action = ShellAction::new("sh -c 'exit 1'")?.with_failure_mode(FailureMode::Terminal);
372 let (context, _handle) = context();
373 let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
374 return Err("a non-zero exit must fail the activity".into());
375 };
376 assert_eq!(failure.classification(), &Classification::Terminal);
377 Ok(())
378 }
379
380 #[tokio::test]
381 async fn a_missing_parameter_fails_terminally_before_anything_runs() -> TestResult {
382 let action = ShellAction::new("echo {{absent}}")?;
383 let (context, _handle) = context();
384 let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
385 return Err("a missing parameter must fail the activity".into());
386 };
387 assert_eq!(failure.classification(), &Classification::Terminal);
388 assert!(failure.message().contains("absent"));
389 Ok(())
390 }
391
392 #[tokio::test]
393 async fn an_absent_program_fails_terminally() -> TestResult {
394 let action = ShellAction::new("aion-no-such-program-exists-anywhere")?;
395 let (context, _handle) = context();
396 let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
397 return Err("an absent program must fail the activity".into());
398 };
399 assert_eq!(failure.classification(), &Classification::Terminal);
400 Ok(())
401 }
402
403 #[tokio::test]
404 async fn cancellation_stops_the_command_and_fails_terminally() -> TestResult {
405 let action = ShellAction::new("sleep 30")?;
406 let (context, handle) = context();
407 let run = tokio::spawn(async move {
408 let (context, _keep) = (context, ());
409 action.run(&BTreeMap::new(), &context).await
410 });
411 tokio::time::sleep(std::time::Duration::from_millis(200)).await;
414 handle.cancel();
415 let Err(failure) = run.await? else {
416 return Err("a cancelled command must fail the activity".into());
417 };
418 assert_eq!(failure.classification(), &Classification::Terminal);
419 assert!(failure.message().contains("cancelled"));
420 Ok(())
421 }
422
423 #[tokio::test]
431 async fn a_cancellation_already_standing_runs_nothing_at_all() -> TestResult {
432 let directory = tempfile::tempdir()?;
433 let witness = directory.path().join("cancelled");
434 let control = directory.path().join("control");
435
436 let (cancelled_context, handle) = context();
439 let (control_context, _control_handle) = context();
440
441 let action = ShellAction::new(&format!("touch {}", witness.display()))?;
442 handle.cancel();
443 let Err(failure) = action.run(&BTreeMap::new(), &cancelled_context).await else {
444 return Err("a cancelled command must fail the activity".into());
445 };
446 assert_eq!(failure.classification(), &Classification::Terminal);
447 assert!(
448 failure.message().contains("cancelled"),
449 "a cancellation must be reported as one: {}",
450 failure.message()
451 );
452 assert!(
453 !witness.exists(),
454 "a cancellation that had already landed still executed the program"
455 );
456
457 let control_action = ShellAction::new(&format!("touch {}", control.display()))?;
458 control_action
459 .run(&BTreeMap::new(), &control_context)
460 .await?;
461 assert!(
462 control.exists(),
463 "the control failed: `touch` did not run even without a cancellation, so the first \
464 arm proves nothing"
465 );
466 Ok(())
467 }
468
469 #[tokio::test]
470 async fn the_hosts_environment_does_not_cross_into_a_declared_command() -> TestResult {
471 let Some(present) = std::env::vars_os()
481 .filter_map(|(name, _)| name.into_string().ok())
482 .find(|name| name != "PATH" && !name.is_empty() && !name.contains('='))
483 else {
484 tracing::info!(
485 "skipping: the host has no environment variable besides PATH to prove \
486 non-inheritance with"
487 );
488 return Ok(());
489 };
490 let action = ShellAction::new(&format!("sh -c 'echo \"[${{{present}:-absent}}]\"'"))?;
491 let (context, _handle) = context();
492 let outcome = action.run(&BTreeMap::new(), &context).await?;
493 assert_eq!(
494 outcome.stdout, "[absent]",
495 "the host's `{present}` leaked into a declared command"
496 );
497 Ok(())
498 }
499
500 #[tokio::test]
501 async fn an_operator_supplied_variable_does_reach_the_command() -> TestResult {
502 let mut environment = BTreeMap::new();
504 environment.insert("DECLARED_GREETING".to_owned(), "supplied".to_owned());
505 let action = ShellAction::new("sh -c 'echo \"[$DECLARED_GREETING]\"'")?
506 .with_environment(environment);
507 let (context, _handle) = context();
508 let outcome = action.run(&BTreeMap::new(), &context).await?;
509 assert_eq!(outcome.stdout, "[supplied]");
510 Ok(())
511 }
512
513 #[tokio::test]
514 async fn a_command_runs_in_its_declared_working_directory() -> TestResult {
515 let action = ShellAction::new("pwd")?.with_working_directory("/");
516 let (context, _handle) = context();
517 let outcome = action.run(&BTreeMap::new(), &context).await?;
518 assert_eq!(outcome.stdout, "/");
519 Ok(())
520 }
521
522 #[tokio::test]
523 async fn a_command_reading_stdin_gets_end_of_file_rather_than_the_hosts() -> TestResult {
524 let action = ShellAction::new("cat")?;
527 let (context, _handle) = context();
528 let outcome = action.run(&BTreeMap::new(), &context).await?;
529 assert_eq!(outcome.exit_code, 0);
530 assert_eq!(outcome.stdout, "");
531 Ok(())
532 }
533
534 #[tokio::test]
535 async fn awkward_values_survive_as_exactly_one_argument_each() -> TestResult {
536 for (value, expected) in [
540 ("two\nlines", "[two\nlines]"),
541 ("", "[]"),
542 (" leading and trailing ", "[ leading and trailing ]"),
543 ("tab\there", "[tab\there]"),
544 ("quote\"inside", "[quote\"inside]"),
545 ("single'inside", "[single'inside]"),
546 ("back\\slash", "[back\\slash]"),
547 ] {
548 let action = ShellAction::new("printf [%s] {{value}}")?;
549 let (context, _handle) = context();
550 let outcome = action
551 .run(&arguments(&[("value", serde_json::json!(value))]), &context)
552 .await?;
553 assert_eq!(
554 outcome.stdout, expected,
555 "value {value:?} did not arrive as exactly one argument"
556 );
557 }
558 Ok(())
559 }
560
561 #[tokio::test]
562 async fn a_value_that_looks_like_a_flag_is_still_passed_as_a_value() -> TestResult {
563 let exposed = ShellAction::new("printf %s {{value}}")?;
569 let (exposed_context, _exposed_handle) = context();
570 let outcome = exposed
571 .run(
572 &arguments(&[("value", serde_json::json!("-n"))]),
573 &exposed_context,
574 )
575 .await;
576 assert!(
579 outcome.is_ok(),
580 "the executor must deliver the value and let the program parse it"
581 );
582
583 let guarded = ShellAction::new("printf -- [%s] {{value}}")?;
586 let (guarded_context, _guarded_handle) = context();
587 let guarded_outcome = guarded
588 .run(
589 &arguments(&[("value", serde_json::json!("-n"))]),
590 &guarded_context,
591 )
592 .await?;
593 assert_eq!(guarded_outcome.stdout, "[-n]");
594 Ok(())
595 }
596
597 #[tokio::test]
607 async fn a_declared_command_streams_both_streams_before_it_exits() -> TestResult {
608 use aion_core::{RunId, WorkflowId};
609
610 let (sender, mut events) = tokio::sync::mpsc::unbounded_channel();
611 let (context, cancellation) = ActivityContext::with_transcript(
612 WorkflowId::new_v4(),
613 RunId::new_v4(),
614 ActivityId::from_sequence_position(9),
615 3,
616 sender,
617 );
618 let action = ShellAction::new("sh -c 'echo working; echo warning >&2; sleep 30'")?;
619 let no_arguments = BTreeMap::new();
620 let run = action.run(&no_arguments, &context);
621 tokio::pin!(run);
622
623 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
624 let mut observed: Vec<(String, String)> = Vec::new();
625 while !observed.iter().any(|(role, _)| role.contains("stdout"))
626 || !observed.iter().any(|(role, _)| role.contains("stderr"))
627 {
628 tokio::select! {
629 biased;
630 outcome = &mut run => {
631 drop(outcome);
632 return Err(format!(
633 "the command completed before its output reached the transcript: \
634 {observed:?}"
635 )
636 .into());
637 }
638 event = events.recv() => {
639 let event = event.ok_or("the transcript seam closed mid-command")?;
640 assert_eq!(event.attempt, 3, "the event carries the attempt it belongs to");
641 assert_eq!(
642 event.activity_id,
643 ActivityId::from_sequence_position(9),
644 "the event is keyed to the activity that is running"
645 );
646 let aion_core::ActivityEventKind::Message { text, .. } = event.kind else {
647 return Err("an output line must be a Message".into());
648 };
649 observed.push((event.agent_role, text));
650 }
651 () = tokio::time::sleep_until(deadline) => {
652 return Err(format!(
653 "the running command's output never reached the transcript: {observed:?}"
654 )
655 .into());
656 }
657 }
658 }
659
660 assert!(
661 observed.contains(&("command stdout".to_owned(), "working".to_owned())),
662 "stdout must arrive labelled by its stream: {observed:?}"
663 );
664 assert!(
665 observed.contains(&("command stderr".to_owned(), "warning".to_owned())),
666 "stderr must arrive labelled by its stream: {observed:?}"
667 );
668
669 cancellation.cancel();
670 let Err(failure) = run.await else {
671 return Err("a cancelled command must fail the activity".into());
672 };
673 assert_eq!(failure.classification(), &Classification::Terminal);
674 Ok(())
675 }
676
677 #[test]
678 fn only_one_trailing_newline_is_trimmed() {
679 assert_eq!(trim_trailing_newline("hello\n"), "hello");
680 assert_eq!(trim_trailing_newline("hello\r\n"), "hello");
681 assert_eq!(trim_trailing_newline("hello\n\n"), "hello\n");
682 assert_eq!(trim_trailing_newline("hello"), "hello");
683 assert_eq!(trim_trailing_newline(""), "");
684 }
685}