aion-worker 0.29.0

//! What a declared command body actually does to a process.
//!
//! Its own file because its subjects are live process trees rather than
//! template shapes, and because [`super`] is already at the file-size law.

use std::collections::BTreeMap;

use aion_core::{ActivityId, RunId, WorkflowId};
use aion_package::{
    ArgvSlot, CommandLineContract, CommandParameterContract, DeclaredCommandContract,
    EnvBindingContract, FillPiece, FillTemplate,
};
use serde_json::json;

use super::DeclaredCommandAction;
use crate::activity::Classification;
use crate::context::{ActivityCancellationHandle, ActivityContext};

/// What a test returns. Every fallible step is carried rather than unwrapped,
/// because the workspace denies panicking accessors in test code as firmly as
/// in library code.
type TestResult = Result<(), Box<dyn std::error::Error>>;

fn context() -> (ActivityContext, ActivityCancellationHandle) {
    ActivityContext::new(
        WorkflowId::new_v4(),
        RunId::new_v4(),
        ActivityId::from_sequence_position(1),
        1,
    )
}

fn hole(parameter: &str) -> FillTemplate {
    FillTemplate {
        pieces: vec![FillPiece::Hole {
            parameter: parameter.to_owned(),
        }],
    }
}

fn slot(fill: FillTemplate, label: &str) -> ArgvSlot {
    ArgvSlot {
        fill,
        label: label.to_owned(),
        admits_leading_dash: true,
    }
}

/// One line from program words plus argument slots — the shape the AWL
/// emitter produces.
fn line(program: &[&str], args: Vec<ArgvSlot>) -> CommandLineContract {
    let mut slots: Vec<ArgvSlot> = program
        .iter()
        .map(|word| slot(FillTemplate::literal((*word).to_owned()), word))
        .collect();
    slots.extend(args);
    CommandLineContract { slots }
}

fn contract(program: &[&str], args: Vec<ArgvSlot>) -> DeclaredCommandContract {
    DeclaredCommandContract {
        name: "probe".to_owned(),
        parameters: Vec::new(),
        lines: vec![line(program, args)],
        env: Vec::new(),
        cwd: None,
        prior_form_refusal: None,
    }
}

fn parameter(name: &str) -> CommandParameterContract {
    CommandParameterContract {
        name: name.to_owned(),
        default: None,
    }
}

fn arguments(pairs: &[(&str, serde_json::Value)]) -> BTreeMap<String, serde_json::Value> {
    pairs
        .iter()
        .map(|(name, value)| ((*name).to_owned(), value.clone()))
        .collect()
}

#[tokio::test]
async fn a_declared_command_runs_and_returns_its_output() -> TestResult {
    let mut command = contract(&["echo"], vec![slot(hole("who"), "who")]);
    command.parameters = vec![parameter("who")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action
        .run(&arguments(&[("who", json!("world"))]), &context)
        .await?;
    assert_eq!(outcome.exit_code, 0);
    assert_eq!(outcome.stdout, "world");
    Ok(())
}

/// THE PROPERTY THE WHOLE SURFACE EXISTS FOR: a value that would be four
/// commands under a shell is one inert argv word here, because no shell ever
/// sees it. If this regresses, every other guarantee is decoration.
#[tokio::test]
async fn a_hostile_fill_arrives_as_exactly_one_literal_argv_word() -> TestResult {
    let mut command = contract(
        &["printf"],
        vec![
            slot(FillTemplate::literal("[%s]"), "format"),
            slot(hole("value"), "value"),
        ],
    );
    command.parameters = vec![parameter("value")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action
        .run(
            &arguments(&[("value", json!("$(boom); rm -rf / && echo PWNED"))]),
            &context,
        )
        .await?;
    assert_eq!(outcome.stdout, "[$(boom); rm -rf / && echo PWNED]");
    assert!(
        !outcome.stdout.contains("PWNED\n"),
        "the injected command must never have run"
    );
    Ok(())
}

/// THE BODY RULING: lines run sequentially, in order, and the outcome's
/// stdout is the concatenation of every line's stdout.
#[tokio::test]
async fn a_multi_line_body_runs_in_order_and_concatenates_stdout() -> TestResult {
    let mut command = contract(
        &["echo"],
        vec![slot(FillTemplate::literal("first"), "word")],
    );
    command.lines.push(line(
        &["echo"],
        vec![slot(FillTemplate::literal("second"), "word")],
    ));
    command.lines.push(line(
        &["echo"],
        vec![slot(FillTemplate::literal("third"), "word")],
    ));
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(outcome.exit_code, 0);
    assert_eq!(outcome.stdout, "first\nsecond\nthird");
    Ok(())
}

/// The first non-zero exit fails the command and no later line runs.
#[tokio::test]
async fn the_first_nonzero_exit_stops_the_body() -> TestResult {
    let directory = tempfile::tempdir()?;
    let scratch = directory.path().join("witness");
    let mut command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("echo before; exit 3"), "body"),
        ],
    );
    command.lines.push(line(
        &["touch"],
        vec![slot(
            FillTemplate::literal(scratch.display().to_string()),
            "witness",
        )],
    ));
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a non-zero exit must fail the command".into());
    };
    assert_eq!(failure.classification(), &Classification::Retryable);
    assert!(failure.message().contains('3'), "{}", failure.message());
    assert!(
        !scratch.exists(),
        "the line after a failing line must never run"
    );
    Ok(())
}

#[tokio::test]
async fn a_surplus_action_parameter_is_not_a_refusal() -> TestResult {
    // An action may declare more than the command it serves needs; the
    // command's own names select from the input.
    let mut command = contract(&["echo"], vec![slot(hole("who"), "who")]);
    command.parameters = vec![parameter("who")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action
        .run(
            &arguments(&[("who", json!("world")), ("unused", json!("ignored"))]),
            &context,
        )
        .await?;
    assert_eq!(outcome.stdout, "world");
    Ok(())
}

#[tokio::test]
async fn a_declared_env_binding_reaches_the_child() -> TestResult {
    let mut command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(
                FillTemplate::literal("echo \"[$DECLARED_GREETING]\""),
                "body",
            ),
        ],
    );
    command.env = vec![EnvBindingContract {
        name: "DECLARED_GREETING".to_owned(),
        value: "exported".to_owned(),
    }];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(outcome.stdout, "[exported]");
    Ok(())
}

#[tokio::test]
async fn the_hosts_environment_does_not_cross_into_a_declared_command() -> TestResult {
    // A host process legitimately holds credentials. A deployed package must
    // not be able to read them by declaring a command that prints them. Reads
    // a variable the host ALREADY has rather than setting one, because
    // `std::env::set_var` is unsafe and the workspace denies unsafe outright.
    let Some(present) = std::env::vars_os()
        .filter_map(|(name, _)| name.into_string().ok())
        .find(|name| name != "PATH" && !name.is_empty() && !name.contains('='))
    else {
        tracing::info!(
            "skipping: the host has no environment variable besides PATH to prove \
             non-inheritance with"
        );
        return Ok(());
    };
    let command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(
                FillTemplate::literal(format!("echo \"[${{{present}:-absent}}]\"")),
                "body",
            ),
        ],
    );
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(
        outcome.stdout, "[absent]",
        "the host's `{present}` leaked into a declared command"
    );
    Ok(())
}

/// An `export PATH` binding replaces the inherited `PATH` for the child —
/// the document-level spelling of the old hardened path.
#[tokio::test]
async fn an_exported_path_wins_over_the_inherited_one() -> TestResult {
    let mut command = contract(
        &["/bin/sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("echo \"$PATH\""), "body"),
        ],
    );
    command.env = vec![EnvBindingContract {
        name: "PATH".to_owned(),
        value: "/usr/bin:/bin".to_owned(),
    }];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(outcome.stdout, "/usr/bin:/bin");
    Ok(())
}

#[tokio::test]
async fn a_command_runs_in_the_working_directory_it_was_given() -> TestResult {
    let command = contract(&["pwd"], Vec::new());
    let action = DeclaredCommandAction::new(command).with_working_directory("/");
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(outcome.stdout, "/");
    Ok(())
}

#[test]
fn the_declared_working_directory_is_handed_back_unexpanded() {
    let mut command = contract(&["pwd"], Vec::new());
    command.cwd = Some("{workspace_root}/clones".to_owned());
    let action = DeclaredCommandAction::new(command);
    assert_eq!(
        action.declared_working_directory(),
        Some("{workspace_root}/clones"),
        "resolving the placeholder is the executing host's act, not this crate's"
    );
}

#[tokio::test]
async fn a_nonzero_exit_is_retryable_and_carries_the_exit_code_and_stderr() -> TestResult {
    let command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("echo trouble >&2; exit 3"), "body"),
        ],
    );
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a non-zero exit must fail the activity".into());
    };
    assert_eq!(failure.classification(), &Classification::Retryable);
    assert!(
        failure.message().contains("trouble"),
        "stderr must ride the failure: {}",
        failure.message()
    );
    assert!(
        failure.message().contains('3'),
        "the exit code is reported: {}",
        failure.message()
    );
    Ok(())
}

/// A BODY THAT STATES NO LINE RUNS NOTHING, and "ran nothing" must never be
/// reported as "succeeded" — an empty capture handed to a caller as an answer
/// is indistinguishable from a real one.
#[tokio::test]
async fn a_body_with_no_lines_refuses_rather_than_reporting_success() -> TestResult {
    let mut command = contract(&["echo"], Vec::new());
    command.lines.clear();
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a command with no lines must refuse rather than succeed".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        failure.message().contains("probe"),
        "the refusal must name the command: {}",
        failure.message()
    );
    assert!(
        failure.message().contains("no line"),
        "the refusal must say what is wrong with it: {}",
        failure.message()
    );
    Ok(())
}

/// THE FAILING LINE DOES NOT SWALLOW WHAT THE BODY ALREADY PRINTED, AND DOES
/// NOT CLAIM THE EARLIER LINES' WORDS AS ITS OWN.
///
/// A body that got a line in before failing has that line's story in it, and
/// discarding it leaves an operator reading only the last word. But a body's
/// standard error accumulates across lines, so quoting the whole accumulation
/// as what the failing program "wrote" puts an earlier line's complaint in the
/// failing program's mouth — which sends an operator to debug the wrong
/// program. The SPECIMEN is chosen for exactly that: the first line succeeds
/// while writing to standard error, so a message that attributes it to the
/// failing line is visible here and nowhere else.
#[tokio::test]
async fn a_failing_line_carries_the_output_the_earlier_lines_produced() -> TestResult {
    let mut command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(
                FillTemplate::literal("echo groundwork done; echo early trouble >&2"),
                "body",
            ),
        ],
    );
    command.lines.push(line(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("echo detail >&2; exit 4"), "body"),
        ],
    ));
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a non-zero exit must fail the command".into());
    };
    let message = failure.message();
    assert_eq!(failure.classification(), &Classification::Retryable);
    assert!(
        message.contains("What the command had printed before it stopped: groundwork done"),
        "the output of the lines that ran before the failure must ride the failure: {message}"
    );
    assert!(
        message.contains("That line wrote to standard error: detail"),
        "the FAILING line's own standard error is quoted as its own: {message}"
    );
    assert!(
        message.contains("The line before it wrote to standard error: early trouble"),
        "an earlier line's standard error is attributed to that line: {message}"
    );
    assert!(
        !message.contains("That line wrote to standard error: early trouble"),
        "the earlier line's words must NOT be put in the failing program's mouth: {message}"
    );
    assert!(
        message.contains("exited 4"),
        "the failing line's exit code must ride the failure: {message}"
    );
    assert!(
        message.contains("line 2 of 2"),
        "the failure must say WHICH line stopped the body: {message}"
    );
    Ok(())
}

/// The same attribution over MORE than one earlier line: the count is said in
/// words that agree with it, and every earlier line's standard error is
/// carried — none of it as the failing program's.
#[tokio::test]
async fn several_earlier_lines_stderr_is_carried_and_attributed_to_them() -> TestResult {
    let mut command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("echo first-warning >&2"), "body"),
        ],
    );
    command.lines.push(line(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("echo second-warning >&2"), "body"),
        ],
    ));
    command.lines.push(line(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("exit 5"), "body"),
        ],
    ));
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a non-zero exit must fail the command".into());
    };
    let message = failure.message();
    assert!(
        message.contains("line 3 of 3"),
        "the failure names the line that stopped the body: {message}"
    );
    assert!(
        message.contains("That line wrote nothing to standard error"),
        "the failing line wrote nothing, and the failure says so of THAT line: {message}"
    );
    assert!(
        message.contains(
            "The 2 lines before it wrote to standard error: first-warning\n\
                          second-warning"
        ),
        "both earlier lines' standard error is carried, attributed to them: {message}"
    );
    Ok(())
}

/// A CANCELLED BODY CARRIES WHAT ITS FINISHED LINES LEFT BEHIND — both
/// streams. An operator deciding what to clean up needs to see how far the body
/// got AND what it complained about on the way; a report that carried only
/// stdout would drop the half that says something was already going wrong.
#[tokio::test]
async fn a_cancelled_body_carries_the_stderr_its_finished_lines_produced() -> TestResult {
    let mut command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(
                FillTemplate::literal("echo groundwork done; echo warned >&2"),
                "body",
            ),
        ],
    );
    command.lines.push(line(
        &["sleep"],
        vec![slot(FillTemplate::literal("30"), "seconds")],
    ));
    let action = DeclaredCommandAction::new(command);
    let (context, handle) = context();
    let run = tokio::spawn(async move { action.run(&BTreeMap::new(), &context).await });
    // Long enough that the first line has finished and the second is genuinely
    // running, so this is the mid-line cancellation and not the between-lines
    // one.
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    handle.cancel();
    let Err(failure) = run.await? else {
        return Err("a cancelled command must fail the activity".into());
    };
    let message = failure.message();
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        message.contains("line 2 of 2"),
        "the failure names the line that was running: {message}"
    );
    assert!(
        message.contains("What the line that had finished printed: groundwork done"),
        "the finished line's output rides the cancellation: {message}"
    );
    assert!(
        message.contains("That line wrote to standard error: warned"),
        "the finished line's standard error rides the cancellation: {message}"
    );
    assert!(
        message.contains("The cancelled line's own output is not captured here"),
        "the report says what it does NOT carry rather than implying it carries everything: \
         {message}"
    );
    Ok(())
}

/// A LINE ENDED BY A SIGNAL IS REPORTED AS ONE, and the signal is the signal
/// that actually ended it. Reported as a fixed exit code, a segmentation fault
/// reads as an out-of-memory kill and sends an operator after the wrong thing.
#[tokio::test]
async fn a_line_ended_by_a_signal_names_that_signal_rather_than_a_fixed_code() -> TestResult {
    let command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("kill -SEGV $$"), "body"),
        ],
    );
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a line killed by a signal must fail the command".into());
    };
    let message = failure.message();
    assert!(
        message.contains("signal 11"),
        "the signal that ended the line must be named: {message}"
    );
    assert!(
        message.contains("SIGSEGV"),
        "the signal must be named in words an operator recognises: {message}"
    );
    assert!(
        !message.contains("exited 137") && !message.contains("exited 139"),
        "a signal death must not be reported as an exit: {message}"
    );
    Ok(())
}

#[tokio::test]
async fn cancellation_stops_the_command_and_fails_terminally() -> TestResult {
    let command = contract(
        &["sleep"],
        vec![slot(FillTemplate::literal("30"), "seconds")],
    );
    let action = DeclaredCommandAction::new(command);
    let (context, handle) = context();
    let run = tokio::spawn(async move { action.run(&BTreeMap::new(), &context).await });
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    handle.cancel();
    let Err(failure) = run.await? else {
        return Err("a cancelled command must fail the activity".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        failure.message().contains("cancelled"),
        "a cancellation must be reported as one: {}",
        failure.message()
    );
    Ok(())
}

/// A CANCELLED BODY STARTS NO LATER LINE. The line that was running is killed
/// with its whole process group, and the line after it — which would have
/// changed the world in its own right — never runs at all.
#[tokio::test]
async fn cancelling_a_multi_line_body_leaves_the_later_lines_unrun() -> TestResult {
    // A `TempDir` cleans up by being dropped, so the witness is removed on
    // every exit path from this test — including the failing ones, where a
    // hand-rolled `remove_file` would either be skipped or have its `Result`
    // swallowed to keep the test compiling.
    let directory = tempfile::tempdir()?;
    let witness = directory.path().join("witness");
    let mut command = contract(
        &["sleep"],
        vec![slot(FillTemplate::literal("30"), "seconds")],
    );
    command.lines.push(line(
        &["touch"],
        vec![slot(
            FillTemplate::literal(witness.display().to_string()),
            "witness",
        )],
    ));
    let action = DeclaredCommandAction::new(command);
    let (context, handle) = context();
    let run = tokio::spawn(async move { action.run(&BTreeMap::new(), &context).await });
    // Long enough that the first line is genuinely running, so this exercises
    // termination rather than the pre-start guard the next test covers.
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    handle.cancel();
    let outcome = run.await?;
    let ran_later_line = witness.exists();

    let Err(failure) = outcome else {
        return Err("a cancelled command must fail the activity".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        failure.message().contains("cancelled"),
        "a cancellation must be reported as one: {}",
        failure.message()
    );
    assert!(
        !ran_later_line,
        "the line after the cancelled one must never start"
    );
    Ok(())
}

/// A CANCELLATION THAT LANDED BEFORE THE BODY STARTED RUNS NOTHING AT ALL —
/// not the first line, not any line. The window this closes is the one where a
/// cancellation arrives as a line finishes: without the check at the top of
/// the loop the next line is handed to `execve` and the program runs before
/// anything kills it.
#[tokio::test]
async fn a_cancellation_already_standing_runs_no_line_of_the_body() -> TestResult {
    let directory = tempfile::tempdir()?;
    let first = directory.path().join("first");
    let second = directory.path().join("second");
    let mut command = contract(
        &["touch"],
        vec![slot(
            FillTemplate::literal(first.display().to_string()),
            "witness",
        )],
    );
    command.lines.push(line(
        &["touch"],
        vec![slot(
            FillTemplate::literal(second.display().to_string()),
            "witness",
        )],
    ));
    let action = DeclaredCommandAction::new(command);
    let (context, handle) = context();
    handle.cancel();
    let outcome = action.run(&BTreeMap::new(), &context).await;
    let ran_first = first.exists();
    let ran_second = second.exists();

    let Err(failure) = outcome else {
        return Err("a cancelled command must fail the activity".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        failure.message().contains("cancelled") && failure.message().contains("probe"),
        "the failure must say the command was cancelled and name it: {}",
        failure.message()
    );
    assert!(!ran_first, "the first line ran despite a standing cancel");
    assert!(!ran_second, "the second line ran despite a standing cancel");
    Ok(())
}

#[tokio::test]
async fn a_missing_parameter_fails_terminally_before_anything_runs() -> TestResult {
    let mut command = contract(&["echo"], vec![slot(hole("who"), "who")]);
    command.parameters = vec![parameter("who")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a missing parameter must fail the activity".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(failure.message().contains("who"), "{}", failure.message());
    Ok(())
}

#[test]
fn a_capture_decides_the_result_and_only_the_result() -> TestResult {
    use aion_package::contract::CommandBodyCapture;

    let outcome = super::ShellOutcome {
        exit_code: 0,
        stdout: "{\"name\":\"world\"}".to_owned(),
        stderr: String::new(),
    };
    assert_eq!(
        super::shape_command_result("greet", CommandBodyCapture::Text, outcome.clone())?,
        json!("{\"name\":\"world\"}")
    );
    assert_eq!(
        super::shape_command_result("greet", CommandBodyCapture::Json, outcome)?,
        json!({ "name": "world" })
    );
    Ok(())
}

#[test]
fn a_json_capture_over_output_that_is_not_json_refuses_terminally() -> TestResult {
    use aion_package::contract::CommandBodyCapture;

    let outcome = super::ShellOutcome {
        exit_code: 0,
        stdout: "not json".to_owned(),
        stderr: String::new(),
    };
    let Err(failure) = super::shape_command_result("greet", CommandBodyCapture::Json, outcome)
    else {
        return Err("output that is not JSON must refuse a `json` capture".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(failure.message().contains("greet"), "{}", failure.message());
    Ok(())
}