aion-worker 0.26.0

//! 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 an argv that the AWL emitter already produced from a typed
//! `command` declaration — program words, one templated slot per argument,
//! declared environment bindings, a working directory and a timeout — so
//! nothing on this side ever holds a string that could be re-split.
//!
//! Everything else is deliberately the same machinery: `execve` with no shell
//! interposed, the host environment cleared but for `PATH`, closed stdin,
//! 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 std::time::Duration;

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

use super::action::{INHERITED_VARIABLE, ShellOutcome, trim_trailing_newline};
use crate::activity::ActivityFailure;
use crate::command_transcript::CommandTranscript;
use crate::context::ActivityContext;
use crate::process::{CancellableCommandOutput, ProcessGroupError, 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 DECLARATION 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 timeout the declaration states, if it states one.
    ///
    /// `Ok(None)` means the declaration stated NONE, which is a legal declared
    /// state: nothing here substitutes a ceiling.
    ///
    /// # Errors
    ///
    /// Returns a terminal [`ActivityFailure`] when the contract carries a
    /// millisecond count no duration can hold — a negative one, say. The AWL
    /// checker refuses those before deploy, so reaching this arm means a
    /// defective contract was deployed; it is REFUSED rather than treated as
    /// "no ceiling", because the direction of that failure is a command
    /// running unbounded on the operator's machine while the declaration, the
    /// `--check` report and the named owner all say a bound was set.
    pub fn declared_timeout(&self) -> Result<Option<Duration>, ActivityFailure> {
        let Some(millis) = self.contract.timeout_ms else {
            return Ok(None);
        };
        u64::try_from(millis)
            .map(Duration::from_millis)
            .map_or_else(
                |_| {
                    Err(ActivityFailure::terminal(format!(
                        "declared command `{name}` states a timeout of {millis}ms, which is not a \
                     duration; the deployed contract is defective and running the command \
                     unbounded would contradict the ceiling `{owner}` declared",
                        name = self.contract.name,
                        owner = self.timeout_owner().unwrap_or("its author"),
                    )))
                },
                |bound| Ok(Some(bound)),
            )
    }

    /// Who owns the declared timeout, when there is one.
    #[must_use]
    pub fn timeout_owner(&self) -> Option<&str> {
        self.contract.timeout_owner.as_deref()
    }

    /// 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 operand would carry leading-dash bytes into a
    /// position the program still reads options in. 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.
    ///
    /// Two things end the command early and both reach the same termination
    /// ladder the string-body executor uses — `SIGTERM` → grace → `SIGKILL`
    /// across the whole process group, with the verdict withheld until the
    /// group has been proven gone. The first is the activity being cancelled.
    /// The second is the command's OWN declared timeout, which is enforced
    /// here, where the process is: a deadline applied only to the waiting side
    /// stops a caller waiting and leaves a process running.
    ///
    /// # Errors
    ///
    /// Returns a terminal [`ActivityFailure`] for a render refusal, a command
    /// that cannot be spawned or observed, a cancellation, or a declared
    /// timeout; and a retryable one for a non-zero exit, carrying the exit
    /// code and the command's own standard error.
    pub async fn run(
        &self,
        arguments: &BTreeMap<String, serde_json::Value>,
        context: &ActivityContext,
    ) -> Result<ShellOutcome, ActivityFailure> {
        let rendered = self.render(arguments)?;
        let (program, rest) = rendered.argv.split_first().ok_or_else(|| {
            // The AWL checker refuses a declaration with no `program` clause,
            // 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!(
                "declared command `{}` rendered no program to execute; the deployed contract \
                 is defective",
                self.contract.name
            ))
        })?;

        let mut command = Command::new(program);
        command.args(rest);
        // stdin is CLOSED, never inherited: a declared command that read
        // standard input would otherwise read the host's and block forever.
        command.stdin(std::process::Stdio::null());
        // The host's environment does not cross into a declared command. Only
        // `PATH` does, and only so a bare program name resolves — see
        // `INHERITED_VARIABLE`.
        command.env_clear();
        if let Some(path) = std::env::var_os(INHERITED_VARIABLE) {
            command.env(INHERITED_VARIABLE, path);
        }
        for (name, value) in &rendered.env {
            command.env(name, value);
        }
        // The hardened PATH is applied LAST, so it wins over both the
        // inherited value and any `env PATH` binding. A `hardening` block that
        // could be overridden by an ordinary binding elsewhere in the same
        // declaration would harden nothing.
        if let Some(path) = &rendered.hardened_path {
            command.env(INHERITED_VARIABLE, path);
        }
        if let Some(directory) = &self.working_directory {
            command.current_dir(directory);
        }

        let bound = self.declared_timeout()?;
        let expired = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let stop = {
            let expired = std::sync::Arc::clone(&expired);
            async move {
                let Some(bound) = bound else {
                    context.cancelled().await;
                    return;
                };
                // BIASED, and cancellation first. When a real cancel and the
                // declared bound become ready in the same poll, an unbiased
                // select would report a cancelled attempt as a timeout half
                // the time — sending an operator to the ceiling's owner over a
                // stop somebody else asked for. Both outcomes are terminal, so
                // the only thing at stake is which one the failure names, and
                // the cancel is the truthful cause when both hold.
                tokio::select! {
                    biased;
                    () = context.cancelled() => {}
                    () = tokio::time::sleep(bound) => {
                        expired.store(true, std::sync::atomic::Ordering::Release);
                    }
                }
            }
        };

        let transcript = CommandTranscript::new(context);
        match run_cancellable_command(command, stop, &transcript).await {
            Ok(CancellableCommandOutput::Completed(output)) => {
                let outcome = ShellOutcome {
                    exit_code: output.status.code().unwrap_or(EXIT_CODE_SIGNALLED),
                    stdout: trim_trailing_newline(&String::from_utf8_lossy(&output.stdout)),
                    stderr: trim_trailing_newline(&String::from_utf8_lossy(&output.stderr)),
                };
                if output.status.success() {
                    Ok(outcome)
                } else {
                    Err(self.exit_failure(program, &outcome))
                }
            }
            Ok(CancellableCommandOutput::Cancelled) => {
                if expired.load(std::sync::atomic::Ordering::Acquire) {
                    Err(self.timeout_failure(program, bound))
                } else {
                    Err(ActivityFailure::terminal(format!(
                        "the declared command `{program}` was cancelled and its process group \
                         was terminated"
                    )))
                }
            }
            Err(error) => Err(spawn_failure(program, &error)),
        }
    }

    /// The failure for a command that ran and exited non-zero.
    ///
    /// Retryable, and carrying the command's own standard error: the common
    /// causes of a failing command — a busy resource, an unreachable host, a
    /// transient permission state — are the ones a second attempt clears, and
    /// a failure that said only "exited non-zero" would leave an operator with
    /// nothing to act on.
    fn exit_failure(&self, program: &str, outcome: &ShellOutcome) -> ActivityFailure {
        let stderr = if outcome.stderr.is_empty() {
            " with no standard error output".to_owned()
        } else {
            format!(": {}", outcome.stderr)
        };
        ActivityFailure::retryable(format!(
            "the declared command `{name}` (`{program}`) exited {code}{stderr}",
            name = self.contract.name,
            code = outcome.exit_code,
        ))
    }

    /// The failure for a command stopped on its own declared timeout.
    ///
    /// Terminal, and it names the owner: a declared ceiling is a number
    /// somebody chose, and an operator meeting it needs to know who to ask.
    fn timeout_failure(&self, program: &str, bound: Option<Duration>) -> ActivityFailure {
        let owner = self
            .timeout_owner()
            .map_or_else(String::new, |owner| format!(", owned by `{owner}`"));
        ActivityFailure::terminal(format!(
            "the declared command `{name}` (`{program}`) outlived its declared timeout of \
             {bound:?}{owner}; its process group was terminated",
            name = self.contract.name,
            bound = bound.unwrap_or_default(),
        ))
    }
}

/// Exit code reported when a command was ended by a signal and so has none.
const EXIT_CODE_SIGNALLED: i32 = 137;

/// 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}"
                ))
            }),
    }
}

/// Classify a spawn or observation failure.
///
/// Terminal in every arm: a program that is absent, unrunnable, or whose
/// containment could not be established will be exactly as absent on the next
/// attempt, and retrying a command whose termination could NOT be confirmed
/// risks running it twice concurrently.
fn spawn_failure(program: &str, error: &ProcessGroupError) -> ActivityFailure {
    ActivityFailure::terminal(format!(
        "the declared command `{program}` could not be run to completion: {error}"
    ))
}

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