aion-worker 0.27.1

//! A DECLARED command executed as an activity.
//!
//! [`DeclaredCommandAction`] is the sibling of [`super::ShellAction`], and the
//! difference is where the splitting happened. A `ShellAction` is handed a
//! command LINE and parses it into an argv here. A `DeclaredCommandAction` is
//! handed argvs that the AWL emitter already produced from a command
//! declaration — one argv per body line, the document's exported environment
//! and working directory beside them — so nothing on this side ever holds a
//! string that could be re-split.
//!
//! # The body ruling, honoured where the processes are
//!
//! A command body's lines run SEQUENTIALLY, in order; the first non-zero
//! exit fails the command; and the command's captured output is the
//! CONCATENATED stdout of every line in order. This is the one executor the
//! server's declared-body dispatcher and the `aion worker awl` executor
//! share, so the ruling cannot be answered differently by venue.
//!
//! Everything else is deliberately the same machinery as the string-body
//! executor: `execve` with no shell interposed, and the world of
//! [`super::world::place_in_declared_world`] — closed stdin, the host
//! environment cleared but for `PATH`, the document's exports applied in
//! declared order over it — plus process-group containment and line-by-line
//! transcript streaming. Two executors that agreed about the argv and
//! disagreed about the world the process runs in would be two different
//! bodies wearing one declaration.

use std::collections::BTreeMap;
use std::path::PathBuf;

use aion_package::{ArgumentValue, DeclaredCommandContract, RenderedCommand};
use tokio::process::Command;

use super::action::{ShellOutcome, trim_trailing_newline};
use super::exit::Ending;
use super::failure::{BodySite, ending_permits_retry, spawn_failure, unreadable_ending_clause};
use super::world::place_in_declared_world;
use crate::activity::ActivityFailure;
use crate::command_transcript::CommandTranscript;
use crate::context::ActivityContext;
use crate::process::{CancellableCommandOutput, run_cancellable_command};

/// A declared command, ready to run as an activity.
#[derive(Debug, Clone)]
pub struct DeclaredCommandAction {
    contract: DeclaredCommandContract,
    working_directory: Option<PathBuf>,
}

impl DeclaredCommandAction {
    /// Wrap an emitted command.
    #[must_use]
    pub const fn new(contract: DeclaredCommandContract) -> Self {
        Self {
            contract,
            working_directory: None,
        }
    }

    /// The working directory the DOCUMENT states, verbatim.
    ///
    /// Returned unexpanded because a `{workspace_root}` placeholder resolves
    /// against the executing host's own workspace, which this crate has no
    /// business knowing. A caller reads this, resolves it however its host
    /// resolves roots, and hands the answer back through
    /// [`Self::with_working_directory`].
    #[must_use]
    pub fn declared_working_directory(&self) -> Option<&str> {
        self.contract.cwd.as_deref()
    }

    /// Run the command in `directory`.
    #[must_use]
    pub fn with_working_directory(mut self, directory: impl Into<PathBuf>) -> Self {
        self.working_directory = Some(directory.into());
        self
    }

    /// The command's declared name, for a diagnostic.
    #[must_use]
    pub fn name(&self) -> &str {
        &self.contract.name
    }

    /// Render this command against `arguments`.
    ///
    /// `arguments` is the ACTION's whole input and may name parameters the
    /// command does not use — an action is free to declare more than one of
    /// its bodies needs. The command's own declared names select from it, so
    /// a surplus is not a refusal here; a command parameter the action cannot
    /// supply is refused at check time, before anything is deployed.
    ///
    /// # Errors
    ///
    /// Returns a terminal [`ActivityFailure`] when a value has no unambiguous
    /// argument form, when a declared parameter has neither a supplied value
    /// nor a default, or when an element in option position would open with
    /// leading-dash bytes. Every one of those fails identically on every
    /// retry, which is why none is retryable.
    pub fn render(
        &self,
        arguments: &BTreeMap<String, serde_json::Value>,
    ) -> Result<RenderedCommand, ActivityFailure> {
        let mut supplied = BTreeMap::new();
        for name in self.contract.parameter_names() {
            if let Some(value) = arguments.get(name) {
                supplied.insert(
                    name.to_owned(),
                    ArgumentValue::from_json(name, value)
                        .map_err(|error| ActivityFailure::terminal(error.to_string()))?,
                );
            }
        }
        self.contract
            .render(&supplied)
            .map_err(|error| ActivityFailure::terminal(error.to_string()))
    }

    /// Run the command with `arguments` bound to its declared parameters:
    /// every body line in order, stopping at the first line that does not exit
    /// zero, the stdout of the lines concatenated into one outcome.
    ///
    /// Cancellation reaches the line that is running through the same
    /// termination ladder the string-body executor uses — `SIGTERM` → grace →
    /// `SIGKILL` across the process group, with the verdict withheld until
    /// the group has been proven gone — and no later line starts: the
    /// cancellation is read at the top of the loop, before each line is handed
    /// over, so a cancellation that landed while the previous line was
    /// finishing stops the body instead of starting the next program.
    ///
    /// The residual window is stated rather than claimed away: a cancellation
    /// landing between that read and the `execve` still starts that one line,
    /// which is then terminated by the ladder above. There is no moment
    /// between "decide to run" and "the kernel has run it" that this side can
    /// hold, so what an operator may rely on is that a cancellation already
    /// standing when a line is reached runs no part of that line or any after
    /// it.
    ///
    /// That the ladder fires AT ALL for such a cancellation rests on
    /// [`ActivityContext::cancelled`], whose future is created here before the
    /// line is spawned and which registers its interest before it reads the
    /// flag: a cancellation landing anywhere after that future exists reaches
    /// the `select` that the ladder hangs off. A waiter that read the flag
    /// first would have a window in which the signal is spent on nobody, and
    /// this line would run to completion with the cancellation standing.
    ///
    /// # Errors
    ///
    /// Returns a terminal [`ActivityFailure`] for a command whose body renders
    /// no line at all, for a render refusal, for a line that cannot be spawned
    /// or observed, for a line whose ending this host cannot read (see
    /// [`super::failure::ending_permits_retry`]), and for a cancellation; and a
    /// retryable one for a line that exited non-zero or was ended by a signal
    /// — carrying how the line ended, THAT line's own standard error, what the
    /// lines before it wrote to standard error, and everything the body had
    /// printed before it stopped.
    pub async fn run(
        &self,
        arguments: &BTreeMap<String, serde_json::Value>,
        context: &ActivityContext,
    ) -> Result<ShellOutcome, ActivityFailure> {
        let rendered = self.render(arguments)?;
        // A body with no lines would otherwise run nothing and report success:
        // exit zero, empty capture, an action downstream reading that empty
        // capture as the answer. "Ran nothing, succeeded" is the one outcome
        // an operator cannot tell from a real one, so it is refused here.
        // The AWL checker refuses a command whose body states no line, so a
        // contract reaching this arm was deployed defective.
        //
        // The RENDERED lines are interrogated, not the contract's, because the
        // rendered lines are what the loop below iterates: a guard that reads
        // one field and a loop that spends another can disagree, and the one
        // that decides is the loop's.
        if rendered.argv_lines.is_empty() {
            return Err(ActivityFailure::terminal(format!(
                "command `{}` states no line to run, so nothing was executed; a command that ran \
                 nothing cannot be reported as having succeeded. Write the lines this command is \
                 meant to run underneath its header, then deploy the document again",
                self.contract.name
            )));
        }
        let total = rendered.argv_lines.len();
        let mut stdout = String::new();
        let mut stderr = String::new();
        for (position, argv) in rendered.argv_lines.iter().enumerate() {
            // Read BEFORE the line is handed over. A cancellation that landed
            // while the previous line was finishing must not start this one:
            // the whole point of cancelling a body halfway is that the rest of
            // it does not happen.
            if context.is_cancelled() {
                return Err(self.cancelled_between_lines(position, total, argv, &stdout, &stderr));
            }
            let site = BodySite {
                command: &self.contract.name,
                line: position + 1,
                total,
            };
            let (program, rest) = argv.split_first().ok_or_else(|| {
                // The AWL checker refuses a body line with no program word,
                // so reaching this arm means a defective contract was
                // deployed. Handled rather than indexed: a panic here would
                // take the worker down over a document that should have been
                // refused.
                ActivityFailure::terminal(format!(
                    "command `{name}`'s line {line} of {total} names no program to run, so \
                     nothing on that line could be executed; the deployed document is defective \
                     and no attempt at it can succeed",
                    name = self.contract.name,
                    line = site.line,
                ))
            })?;

            let mut command = Command::new(program);
            command.args(rest);
            // Closed stdin, a cleared host environment, PATH, then the
            // document's exports in declared order — the one world every
            // declared command runs in, established by the one function that
            // states it (see [`super::world`]), which is what stops this
            // executor and the string-body one drifting apart.
            place_in_declared_world(
                &mut command,
                rendered
                    .env
                    .iter()
                    .map(|(name, value)| (name.as_str(), value.as_str())),
            );
            if let Some(directory) = &self.working_directory {
                command.current_dir(directory);
            }

            // The transcript says WHICH line is speaking. A body of five lines
            // publishing five programs' output under one label leaves a reader
            // to guess which program wrote what.
            let transcript = CommandTranscript::for_body_line(context, site.line, total);
            match run_cancellable_command(command, context.cancelled(), &transcript).await {
                Ok(CancellableCommandOutput::Completed(output)) => {
                    // Where the EARLIER lines' standard error ends, marked
                    // before this line's own joins it: the failure below
                    // attributes what is on each side of this mark to the
                    // lines that actually wrote it. A mark rather than a copy,
                    // so a long-running body does not re-copy everything it
                    // has written once per line.
                    let earlier_stderr_ends = stderr.len();
                    let line_stderr = String::from_utf8_lossy(&output.stderr).into_owned();
                    stdout.push_str(&String::from_utf8_lossy(&output.stdout));
                    stderr.push_str(&line_stderr);
                    let ending = Ending::of(output.status);
                    if !ending.succeeded() {
                        return Err(self.line_failure(
                            program,
                            site,
                            ending,
                            &trim_trailing_newline(&stdout),
                            &trim_trailing_newline(&line_stderr),
                            &trim_trailing_newline(&stderr[..earlier_stderr_ends]),
                        ));
                    }
                }
                Ok(CancellableCommandOutput::Cancelled) => {
                    return Err(self.cancelled_mid_line(program, site, &stdout, &stderr));
                }
                Err(error) => return Err(spawn_failure(program, Some(site), &error)),
            }
        }
        Ok(ShellOutcome {
            // Every line exited zero — the loop returns at the first that did
            // not — so the command's own status is zero. It is stated rather
            // than carried forward from the last line, because "the last
            // line's code" and "the command succeeded" are the same number
            // only by accident.
            exit_code: 0,
            stdout: trim_trailing_newline(&stdout),
            stderr: trim_trailing_newline(&stderr),
        })
    }

    /// The failure for a line that ran and did not exit zero.
    ///
    /// Carries four things an operator acts on: HOW the line ended (an exit
    /// code, or the signal that ended it — never a number standing in for a
    /// signal), THAT LINE'S OWN standard error, what the lines before it wrote
    /// to standard error, said of those lines rather than of this one, and the
    /// output the body had already produced.
    ///
    /// The attribution is the point. A body's stderr accumulates across lines,
    /// and a failure that quotes the whole accumulation as "it wrote" puts an
    /// earlier line's words in the failing program's mouth — which is how an
    /// operator comes to debug the wrong program. So the failing line's own
    /// standard error is quoted as its own, and the rest is attributed to the
    /// lines that produced it.
    ///
    /// Retryable when the ending is one this host can read: the common causes
    /// of a failing command — a busy resource, an unreachable host, a
    /// transient permission state — are the ones a second attempt clears.
    /// Terminal when it is not; see
    /// [`super::failure::ending_permits_retry`] for the one rationale both
    /// executors read.
    fn line_failure(
        &self,
        program: &str,
        site: BodySite<'_>,
        ending: Ending,
        stdout: &str,
        line_stderr: &str,
        earlier_stderr: &str,
    ) -> ActivityFailure {
        let mut sentences = vec![format!(
            "command `{name}` stopped at line {line} of {total}: `{program}` {ended}{unreadable}",
            name = self.contract.name,
            line = site.line,
            total = site.total,
            ended = ending.described(),
            unreadable = unreadable_ending_clause(ending),
        )];
        sentences.push(if line_stderr.is_empty() {
            "That line wrote nothing to standard error".to_owned()
        } else {
            format!("That line wrote to standard error: {line_stderr}")
        });
        // Only said when there WERE lines before it. A body that failed on its
        // first line has no earlier lines, and a sentence about them would be
        // a sentence about nothing.
        if site.line > 1 {
            let before = lines_phrase(site.line - 1, "before it");
            sentences.push(if earlier_stderr.is_empty() {
                format!("The {before} wrote nothing to standard error")
            } else {
                format!("The {before} wrote to standard error: {earlier_stderr}")
            });
        }
        sentences.push(if stdout.is_empty() {
            "The command had printed nothing before it stopped".to_owned()
        } else {
            format!("What the command had printed before it stopped: {stdout}")
        });
        let message = sentences.join(". ");
        if ending_permits_retry(ending) {
            ActivityFailure::retryable(message)
        } else {
            ActivityFailure::terminal(message)
        }
    }

    /// The failure for a cancellation that landed between two lines.
    ///
    /// Terminal, and it names the line that did NOT start: an operator reading
    /// it needs to know how much of the body ran, because the part that ran
    /// has already changed whatever it changes — so what those lines printed
    /// and what they wrote to standard error ride with it.
    fn cancelled_between_lines(
        &self,
        position: usize,
        total: usize,
        argv: &[String],
        stdout: &str,
        stderr: &str,
    ) -> ActivityFailure {
        let next = argv.first().map_or_else(
            || "the next line".to_owned(),
            |program| format!("`{program}`"),
        );
        ActivityFailure::terminal(format!(
            "command `{name}` was cancelled after {ran} of its {total} lines had run, so {next} \
             and everything after it never started. {story}",
            name = self.contract.name,
            ran = position,
            story = finished_lines_story(position, stdout, stderr),
        ))
    }

    /// The failure for a cancellation that landed while a line was running.
    ///
    /// Terminal, and it carries what the finished lines left behind — both
    /// streams: a cancelled command leaves whatever its finished lines already
    /// did behind, and an operator deciding what to clean up needs to see how
    /// far it got and what it complained about on the way.
    ///
    /// The CANCELLED line's own output is not among it. A cancelled run
    /// returns no capture at all (see
    /// [`crate::process::CancellableCommandOutput::Cancelled`]); what that line
    /// wrote before it was stopped went out line by line as it wrote it,
    /// through the transcript observer, and the failure says so rather than
    /// implying the bytes were lost.
    fn cancelled_mid_line(
        &self,
        program: &str,
        site: BodySite<'_>,
        stdout: &str,
        stderr: &str,
    ) -> ActivityFailure {
        ActivityFailure::terminal(format!(
            "command `{name}` was cancelled while line {line} of {total} (`{program}`) was \
             running: that line's process group was terminated and proven gone, so nothing it \
             started is still running, and no later line ran. {story}. The cancelled line's own \
             output is not captured here — it was streamed line by line as it was written",
            name = self.contract.name,
            line = site.line,
            total = site.total,
            story = finished_lines_story(site.line - 1, stdout, stderr),
        ))
    }
}

/// `line before it` / `3 lines before it` — a count said in words that agree
/// with it, so a failure never reads "the 1 lines".
fn lines_phrase(count: usize, relation: &str) -> String {
    if count == 1 {
        format!("line {relation}")
    } else {
        format!("{count} lines {relation}")
    }
}

/// What the lines that had FINISHED left behind, said of those lines and of
/// nothing else.
///
/// `finished` is how many lines of the body had run to completion; `stdout`
/// and `stderr` are what those lines — and only those lines — produced.
fn finished_lines_story(finished: usize, stdout: &str, stderr: &str) -> String {
    if finished == 0 {
        return "No line of the body had finished".to_owned();
    }
    let printed = trim_trailing_newline(stdout);
    let wrote = trim_trailing_newline(stderr);
    let subject = lines_phrase(finished, "that had finished");
    let pronoun = if finished == 1 {
        "That line".to_owned()
    } else {
        format!("Those {finished} lines")
    };
    let printed_sentence = if printed.is_empty() {
        format!("The {subject} printed nothing")
    } else {
        format!("What the {subject} printed: {printed}")
    };
    let wrote_sentence = if wrote.is_empty() {
        format!("{pronoun} wrote nothing to standard error")
    } else {
        format!("{pronoun} wrote to standard error: {wrote}")
    };
    format!("{printed_sentence}. {wrote_sentence}")
}

/// Shape a successful command's outcome into the action's declared result.
///
/// The one place a `runs command` body's capture is honoured, so the server
/// and the `aion worker awl` executor cannot answer differently about what an
/// action returns. Failure classification is NOT here: it belongs to the run
/// and is the same for both captures, which is what stops two forms drifting
/// into two failure vocabularies.
///
/// # Errors
///
/// Returns a terminal [`ActivityFailure`] when a `json` capture's command
/// printed output that is not valid JSON. Re-running it would print the same
/// bytes, so nothing is gained by a retry.
pub fn shape_command_result(
    action: &str,
    capture: aion_package::contract::CommandBodyCapture,
    outcome: ShellOutcome,
) -> Result<serde_json::Value, ActivityFailure> {
    match capture {
        aion_package::contract::CommandBodyCapture::Text => {
            Ok(serde_json::Value::String(outcome.stdout))
        }
        aion_package::contract::CommandBodyCapture::Json => serde_json::from_str(&outcome.stdout)
            .map_err(|error| {
                ActivityFailure::terminal(format!(
                    "action `{action}` declares a `runs json command` body and its command \
                     printed output that is not valid JSON: {error}"
                ))
            }),
    }
}

#[cfg(test)]
#[path = "declared_tests.rs"]
mod tests;