wdl-engine 0.13.2

//! Implementation of workflow and task inputs.

use std::collections::BTreeSet;
use std::collections::HashMap;
use std::collections::HashSet;
use std::fs::File;
use std::io::BufReader;
use std::path::Path;

use anyhow::Context;
use anyhow::Result;
use anyhow::bail;
use indexmap::IndexMap;
use serde::Serialize;
use serde::ser::SerializeMap;
use serde_json::Value as JsonValue;
use serde_yaml_ng::Value as YamlValue;
use wdl_analysis::Document;
use wdl_analysis::document::Input;
use wdl_analysis::document::Task;
use wdl_analysis::document::Workflow;
use wdl_analysis::types::CallKind;
use wdl_analysis::types::Coercible as _;
use wdl_analysis::types::Optional;
use wdl_analysis::types::PrimitiveType;
use wdl_analysis::types::display_types;
use wdl_analysis::types::v1::task_hint_types;
use wdl_analysis::types::v1::task_requirement_types;

use crate::Array;
use crate::Coercible;
use crate::CompoundValue;
use crate::EvaluationPath;
use crate::Value;

/// A type alias to a JSON map (object).
pub type JsonMap = serde_json::Map<String, JsonValue>;

/// Checks that an input value matches the type of the input.
fn check_input_type(_document: &Document, name: &str, input: &Input, value: &Value) -> Result<()> {
    // We accept optional values for the input even if the input's type is
    // non-optional; if the runtime value is `None` for a non-optional input,
    // the default expression will be evaluated instead.
    let expected_ty = if !input.required() {
        input.ty().optional()
    } else {
        input.ty().clone()
    };

    let ty = value.ty();
    if !ty.is_coercible_to(&expected_ty) {
        bail!("expected {expected_ty:#} for input `{name}`, but found {ty:#}");
    }

    Ok(())
}

/// Resolves paths in a value using per-element origins.
///
/// When `origins` contains multiple entries and the value is an array, each
/// element is resolved against its corresponding origin. Otherwise, all paths
/// are resolved against the first (and only) origin.
async fn resolve_with_origins(
    value: Value,
    ty: &wdl_analysis::types::Type,
    origins: &[EvaluationPath],
) -> Result<Value> {
    if origins.len() > 1
        && let Value::Compound(CompoundValue::Array(ref array)) = value
    {
        let arr_ty = ty.as_array().expect("should be an array type");
        assert_eq!(
            origins.len(),
            array.as_slice().len(),
            "the number of origins should match the number of array elements"
        );
        let optional = arr_ty.element_type().is_optional();
        let mut resolved = Vec::with_capacity(array.as_slice().len());
        for (elem, base_dir) in array.as_slice().iter().zip(origins) {
            resolved.push(
                elem.resolve_paths(optional, None, None, &|p| p.expand(base_dir))
                    .await?,
            );
        }
        return Ok(Value::Compound(CompoundValue::Array(Array::new_unchecked(
            arr_ty.clone(),
            resolved,
        ))));
    }

    let base_dir = &origins[0];
    value
        .resolve_paths(ty.is_optional(), None, None, &|p| p.expand(base_dir))
        .await
}

/// Represents inputs to a task.
#[derive(Default, Debug, Clone)]
pub struct TaskInputs {
    /// The task input values.
    inputs: IndexMap<String, Value>,
    /// The overridden requirements section values.
    requirements: HashMap<String, Value>,
    /// The overridden hints section values.
    hints: HashMap<String, Value>,
}

impl TaskInputs {
    /// Iterates the inputs to the task.
    pub fn iter(&self) -> impl Iterator<Item = (&str, &Value)> + use<'_> {
        self.inputs.iter().map(|(k, v)| (k.as_str(), v))
    }

    /// Determines if the inputs are empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Gets the length of the inputs.
    ///
    /// This includes the count of inputs, requirements, and hints.
    pub fn len(&self) -> usize {
        self.inputs.len() + self.requirements.len() + self.hints.len()
    }

    /// Gets an input by name.
    pub fn get(&self, name: &str) -> Option<&Value> {
        self.inputs.get(name)
    }

    /// Sets a task input.
    ///
    /// Returns the previous value, if any.
    pub fn set(&mut self, name: impl Into<String>, value: impl Into<Value>) -> Option<Value> {
        self.inputs.insert(name.into(), value.into())
    }

    /// Gets an overridden requirement by name.
    pub fn requirement(&self, name: &str) -> Option<&Value> {
        self.requirements.get(name)
    }

    /// Overrides a requirement by name.
    pub fn override_requirement(&mut self, name: impl Into<String>, value: impl Into<Value>) {
        self.requirements.insert(name.into(), value.into());
    }

    /// Gets an overridden hint by name.
    pub fn hint(&self, name: &str) -> Option<&Value> {
        self.hints.get(name)
    }

    /// Overrides a hint by name.
    pub fn override_hint(&mut self, name: impl Into<String>, value: impl Into<Value>) {
        self.hints.insert(name.into(), value.into());
    }

    /// Replaces any `File` or `Directory` input values with joining the
    /// specified path with the value.
    ///
    /// This method will attempt to coerce matching input values to their
    /// expected types.
    pub async fn join_paths<'a>(
        &mut self,
        task: &Task,
        path: impl Fn(&str) -> Result<&'a [EvaluationPath]>,
    ) -> Result<()> {
        for (name, value) in self.inputs.iter_mut() {
            let Some(ty) = task.inputs().get(name).map(|input| input.ty().clone()) else {
                bail!("could not find an expected type for input {name}");
            };

            let origins = path(name)?;

            if let Ok(v) = value.coerce(None, &ty) {
                *value = resolve_with_origins(v, &ty, origins).await?;
            }
        }
        Ok(())
    }

    /// Validates the inputs for the given task.
    ///
    /// The `specified` set of inputs are those that are present, but may not
    /// have values available at validation.
    pub fn validate(
        &self,
        document: &Document,
        task: &Task,
        specified: Option<&HashSet<String>>,
    ) -> Result<()> {
        let version = document.version().context("missing document version")?;

        // Start by validating all the specified inputs and their types
        for (name, value) in &self.inputs {
            let input = task
                .inputs()
                .get(name)
                .with_context(|| format!("unknown input `{name}`"))?;

            check_input_type(document, name, input, value)?;
        }

        // Next check for missing required inputs
        for (name, input) in task.inputs() {
            if input.required()
                && !self.inputs.contains_key(name)
                && specified.map(|s| !s.contains(name)).unwrap_or(true)
            {
                bail!(
                    "missing required input `{name}` to task `{task}`",
                    task = task.name()
                );
            }
        }

        // Check the types of the specified requirements
        for (name, value) in &self.requirements {
            let ty = value.ty();
            if let Some(expected) = task_requirement_types(version, name.as_str()) {
                if !expected.iter().any(|target| ty.is_coercible_to(target)) {
                    bail!(
                        "expected {expected:#} for requirement `{name}`, but found {ty:#}",
                        expected = display_types(expected),
                    );
                }

                continue;
            }

            bail!("unsupported requirement `{name}`");
        }

        // Check the types of the specified hints
        for (name, value) in &self.hints {
            let ty = value.ty();
            if let Some(expected) = task_hint_types(version, name.as_str(), false)
                && !expected.iter().any(|target| ty.is_coercible_to(target))
            {
                bail!(
                    "expected {expected:#} for hint `{name}`, but found {ty:#}",
                    expected = display_types(expected),
                );
            }
        }

        Ok(())
    }

    /// Sets a value with dotted path notation.
    ///
    /// If the provided `value` is a [`PrimitiveType`] other than
    /// [`PrimitiveType::String`] and the `path` is to an input which is of
    /// type [`PrimitiveType::String`], `value` will be converted to a string
    /// and accepted as valid.
    ///
    /// Returns `true` if the given path was for an input or `false` if the
    /// given path was for a requirement or hint.
    fn set_path_value(
        &mut self,
        document: &Document,
        task: &Task,
        path: &str,
        value: Value,
    ) -> Result<bool> {
        let version = document.version().expect("document should have a version");

        match path.split_once('.') {
            // The path might contain a requirement or hint
            Some((key, remainder)) => {
                let (must_match, matched) = match key {
                    "runtime" => (
                        false,
                        task_requirement_types(version, remainder)
                            .map(|types| (true, types))
                            .or_else(|| {
                                task_hint_types(version, remainder, false)
                                    .map(|types| (false, types))
                            }),
                    ),
                    "requirements" => (
                        true,
                        task_requirement_types(version, remainder).map(|types| (true, types)),
                    ),
                    "hints" => (
                        false,
                        task_hint_types(version, remainder, false).map(|types| (false, types)),
                    ),
                    _ => {
                        bail!(
                            "task `{task}` does not have an input named `{path}`",
                            task = task.name()
                        );
                    }
                };

                if let Some((requirement, expected)) = matched {
                    for ty in expected {
                        if value.ty().is_coercible_to(ty) {
                            if requirement {
                                self.requirements.insert(remainder.to_string(), value);
                            } else {
                                self.hints.insert(remainder.to_string(), value);
                            }
                            return Ok(false);
                        }
                    }

                    bail!(
                        "expected {expected:#} for {key} key `{remainder}`, but found {ty:#}",
                        expected = display_types(expected),
                        ty = value.ty()
                    );
                } else if must_match {
                    bail!("unsupported {key} key `{remainder}`");
                } else {
                    Ok(false)
                }
            }
            // The path is to an input
            None => {
                let input = task.inputs().get(path).with_context(|| {
                    format!(
                        "task `{name}` does not have an input named `{path}`",
                        name = task.name()
                    )
                })?;

                // Allow primitive values to implicitly convert to string
                let actual = value.ty();
                let expected = input.ty();
                if let Some(PrimitiveType::String) = expected.as_primitive()
                    && let Some(actual) = actual.as_primitive()
                    && actual != PrimitiveType::String
                {
                    self.inputs
                        .insert(path.to_string(), value.to_string().into());
                    return Ok(true);
                }

                // Auto-wrap a non-array value in a single-element array when the
                // expected type is an array and the value is coercible to the
                // element type.
                let value = if let Some(arr_ty) = expected.as_array()
                    && !matches!(&value, Value::Compound(CompoundValue::Array(_)))
                    && value.ty().is_coercible_to(arr_ty.element_type())
                {
                    Value::Compound(CompoundValue::Array(Array::new_unchecked(
                        expected.clone(),
                        vec![value],
                    )))
                } else {
                    value
                };

                check_input_type(document, path, input, &value)?;
                self.inputs.insert(path.to_string(), value);
                Ok(true)
            }
        }
    }
}

impl<S, V> FromIterator<(S, V)> for TaskInputs
where
    S: Into<String>,
    V: Into<Value>,
{
    fn from_iter<T: IntoIterator<Item = (S, V)>>(iter: T) -> Self {
        Self {
            inputs: iter
                .into_iter()
                .map(|(k, v)| (k.into(), v.into()))
                .collect(),
            requirements: Default::default(),
            hints: Default::default(),
        }
    }
}

impl Serialize for TaskInputs {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        let mut map = serializer.serialize_map(Some(self.len()))?;

        for (k, v) in &self.inputs {
            let v = crate::ValueSerializer::new(None, v, true);
            map.serialize_entry(k, &v)?;
        }

        for (k, v) in &self.requirements {
            let v = crate::ValueSerializer::new(None, v, true);
            map.serialize_entry(&format!("requirements.{k}"), &v)?;
        }

        for (k, v) in &self.hints {
            let v = crate::ValueSerializer::new(None, v, true);
            map.serialize_entry(&format!("hints.{k}"), &v)?;
        }

        map.end()
    }
}

/// Represents inputs to a workflow.
#[derive(Default, Debug, Clone)]
pub struct WorkflowInputs {
    /// The workflow input values.
    inputs: IndexMap<String, Value>,
    /// The nested call inputs.
    calls: HashMap<String, Inputs>,
}

impl WorkflowInputs {
    /// Determines if there are any nested inputs in the workflow inputs.
    ///
    /// Returns `true` if the inputs contains nested inputs or `false` if it
    /// does not.
    pub fn has_nested_inputs(&self) -> bool {
        self.calls.values().any(|inputs| match inputs {
            Inputs::Task(task) => !task.inputs.is_empty(),
            Inputs::Workflow(workflow) => workflow.has_nested_inputs(),
        })
    }

    /// Iterates the inputs to the workflow.
    pub fn iter(&self) -> impl Iterator<Item = (&str, &Value)> + use<'_> {
        self.inputs.iter().map(|(k, v)| (k.as_str(), v))
    }

    /// Determines if the inputs are empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Gets the length of the workflow inputs.
    ///
    /// This includes the workflow inputs plus the lengths of all nested inputs.
    pub fn len(&self) -> usize {
        self.inputs.len() + self.calls.values().map(Inputs::len).sum::<usize>()
    }

    /// Gets an input by name.
    pub fn get(&self, name: &str) -> Option<&Value> {
        self.inputs.get(name)
    }

    /// Gets the nested call inputs.
    pub fn calls(&self) -> &HashMap<String, Inputs> {
        &self.calls
    }

    /// Gets the nested call inputs.
    pub fn calls_mut(&mut self) -> &mut HashMap<String, Inputs> {
        &mut self.calls
    }

    /// Sets a workflow input.
    ///
    /// Returns the previous value, if any.
    pub fn set(&mut self, name: impl Into<String>, value: impl Into<Value>) -> Option<Value> {
        self.inputs.insert(name.into(), value.into())
    }

    /// Checks if the inputs contain a value with the specified name.
    ///
    /// This does not check nested call inputs.
    pub fn contains(&self, name: &str) -> bool {
        self.inputs.contains_key(name)
    }

    /// Replaces any `File` or `Directory` input values with joining the
    /// specified path with the value.
    ///
    /// This method will attempt to coerce matching input values to their
    /// expected types.
    pub async fn join_paths<'a>(
        &mut self,
        workflow: &Workflow,
        path: impl Fn(&str) -> Result<&'a [EvaluationPath]>,
    ) -> Result<()> {
        for (name, value) in self.inputs.iter_mut() {
            let Some(ty) = workflow.inputs().get(name).map(|input| input.ty().clone()) else {
                bail!("could not find an expected type for input {name}");
            };

            let origins = path(name)?;

            if let Ok(v) = value.coerce(None, &ty) {
                *value = resolve_with_origins(v, &ty, origins).await?;
            }
        }
        Ok(())
    }

    /// Validates the inputs for the given workflow.
    ///
    /// The `specified` set of inputs are those that are present, but may not
    /// have values available at validation.
    pub fn validate(
        &self,
        document: &Document,
        workflow: &Workflow,
        specified: Option<&HashSet<String>>,
    ) -> Result<()> {
        // Start by validating all the specified inputs and their types
        for (name, value) in &self.inputs {
            let input = workflow
                .inputs()
                .get(name)
                .with_context(|| format!("unknown input `{name}`"))?;
            check_input_type(document, name, input, value)?;
        }

        // Next check for missing required inputs
        for (name, input) in workflow.inputs() {
            if input.required()
                && !self.inputs.contains_key(name)
                && specified.map(|s| !s.contains(name)).unwrap_or(true)
            {
                bail!(
                    "missing required input `{name}` to workflow `{workflow}`",
                    workflow = workflow.name()
                );
            }
        }

        // Check that the workflow allows nested inputs
        if !workflow.allows_nested_inputs() && self.has_nested_inputs() {
            bail!(
                "cannot specify a nested call input for workflow `{name}` as it does not allow \
                 nested inputs",
                name = workflow.name()
            );
        }

        // Check the inputs to the specified calls
        for (name, inputs) in &self.calls {
            let call = workflow.calls().get(name).with_context(|| {
                format!(
                    "workflow `{workflow}` does not have a call named `{name}`",
                    workflow = workflow.name()
                )
            })?;

            // Resolve the target document; the namespace is guaranteed to be present in the
            // document.
            let document = call
                .namespace()
                .map(|ns| {
                    document
                        .namespace(ns)
                        .expect("namespace should be present")
                        .document()
                })
                .unwrap_or(document);

            // Validate the call's inputs
            let inputs = match call.kind() {
                CallKind::Task => {
                    let task = document
                        .task_by_name(call.name())
                        .expect("task should be present");

                    let task_inputs = inputs.as_task_inputs().with_context(|| {
                        format!("`{name}` is a call to a task, but workflow inputs were supplied")
                    })?;

                    task_inputs.validate(document, task, Some(call.specified()))?;
                    &task_inputs.inputs
                }
                CallKind::Workflow => {
                    let workflow = document.workflow().expect("should have a workflow");
                    assert_eq!(
                        workflow.name(),
                        call.name(),
                        "call name does not match workflow name"
                    );
                    let workflow_inputs = inputs.as_workflow_inputs().with_context(|| {
                        format!("`{name}` is a call to a workflow, but task inputs were supplied")
                    })?;

                    workflow_inputs.validate(document, workflow, Some(call.specified()))?;
                    &workflow_inputs.inputs
                }
            };

            for input in inputs.keys() {
                if call.specified().contains(input) {
                    bail!(
                        "cannot specify nested input `{input}` for call `{call}` as it was \
                         explicitly specified in the call itself",
                        call = call.name(),
                    );
                }
            }
        }

        // Finally, check for missing call arguments
        if workflow.allows_nested_inputs() {
            for (call, ty) in workflow.calls() {
                let inputs = self.calls.get(call);

                for (input, _) in ty
                    .inputs()
                    .iter()
                    .filter(|(n, i)| i.required() && !ty.specified().contains(*n))
                {
                    if !inputs.map(|i| i.get(input).is_some()).unwrap_or(false) {
                        bail!("missing required input `{input}` for call `{call}`");
                    }
                }
            }
        }

        Ok(())
    }

    /// Sets a value with dotted path notation.
    ///
    /// If the provided `value` is a [`PrimitiveType`] other than
    /// [`PrimitiveType::String`] and the `path` is to an input which is of
    /// type [`PrimitiveType::String`], `value` will be converted to a string
    /// and accepted as valid.
    ///
    /// Returns `true` if the path was to an input or `false` if it was not.
    fn set_path_value(
        &mut self,
        document: &Document,
        workflow: &Workflow,
        path: &str,
        value: Value,
    ) -> Result<bool> {
        match path.split_once('.') {
            Some((name, remainder)) => {
                // Resolve the call by name
                let call = workflow.calls().get(name).with_context(|| {
                    format!(
                        "workflow `{workflow}` does not have a call named `{name}`",
                        workflow = workflow.name()
                    )
                })?;

                // Insert the inputs for the call
                let inputs =
                    self.calls
                        .entry(name.to_string())
                        .or_insert_with(|| match call.kind() {
                            CallKind::Task => Inputs::Task(Default::default()),
                            CallKind::Workflow => Inputs::Workflow(Default::default()),
                        });

                // Resolve the target document; the namespace is guaranteed to be present in the
                // document.
                let document = call
                    .namespace()
                    .map(|ns| {
                        document
                            .namespace(ns)
                            .expect("namespace should be present")
                            .document()
                    })
                    .unwrap_or(document);

                let next = remainder
                    .split_once('.')
                    .map(|(n, _)| n)
                    .unwrap_or(remainder);
                if call.specified().contains(next) {
                    bail!(
                        "cannot specify nested input `{next}` for call `{name}` as it was \
                         explicitly specified in the call itself",
                    );
                }

                // Recurse on the call's inputs to set the value
                let input = match call.kind() {
                    CallKind::Task => {
                        let task = document
                            .task_by_name(call.name())
                            .expect("task should be present");
                        inputs
                            .as_task_inputs_mut()
                            .expect("should be a task input")
                            .set_path_value(document, task, remainder, value)?
                    }
                    CallKind::Workflow => {
                        let workflow = document.workflow().expect("should have a workflow");
                        assert_eq!(
                            workflow.name(),
                            call.name(),
                            "call name does not match workflow name"
                        );
                        inputs
                            .as_workflow_inputs_mut()
                            .expect("should be a task input")
                            .set_path_value(document, workflow, remainder, value)?
                    }
                };

                if input && !workflow.allows_nested_inputs() {
                    bail!(
                        "cannot specify a nested call input for workflow `{workflow}` as it does \
                         not allow nested inputs",
                        workflow = workflow.name()
                    );
                }

                Ok(input)
            }
            None => {
                let input = workflow.inputs().get(path).with_context(|| {
                    format!(
                        "workflow `{workflow}` does not have an input named `{path}`",
                        workflow = workflow.name()
                    )
                })?;

                // Allow primitive values to implicitly convert to string
                let actual = value.ty();
                let expected = input.ty();
                if let Some(PrimitiveType::String) = expected.as_primitive()
                    && let Some(actual) = actual.as_primitive()
                    && actual != PrimitiveType::String
                {
                    self.inputs
                        .insert(path.to_string(), value.to_string().into());
                    return Ok(true);
                }

                // Auto-wrap a non-array value in a single-element array when
                // the expected type is an array and the value is coercible to
                // the element type.
                let value = if let Some(arr_ty) = expected.as_array()
                    && !matches!(&value, Value::Compound(CompoundValue::Array(_)))
                    && value.ty().is_coercible_to(arr_ty.element_type())
                {
                    Value::Compound(CompoundValue::Array(Array::new_unchecked(
                        expected.clone(),
                        vec![value],
                    )))
                } else {
                    value
                };

                check_input_type(document, path, input, &value)?;
                self.inputs.insert(path.to_string(), value);
                Ok(true)
            }
        }
    }
}

impl<S, V> FromIterator<(S, V)> for WorkflowInputs
where
    S: Into<String>,
    V: Into<Value>,
{
    fn from_iter<T: IntoIterator<Item = (S, V)>>(iter: T) -> Self {
        Self {
            inputs: iter
                .into_iter()
                .map(|(k, v)| (k.into(), v.into()))
                .collect(),
            calls: Default::default(),
        }
    }
}

impl Serialize for WorkflowInputs {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        let mut map = serializer.serialize_map(Some(self.len()))?;
        for (k, v) in &self.inputs {
            let serialized_value = crate::ValueSerializer::new(None, v, true);
            map.serialize_entry(k, &serialized_value)?;
        }

        for (k, v) in &self.calls {
            let serialized = serde_json::to_value(v).map_err(|_| {
                serde::ser::Error::custom(format!("failed to serialize inputs for call `{k}`"))
            })?;
            let mut map = serde_json::Map::new();
            if let JsonValue::Object(obj) = serialized {
                for (inner, value) in obj {
                    map.insert(format!("{k}.{inner}"), value);
                }
            }
        }

        map.end()
    }
}

/// Represents inputs to a WDL workflow or task.
#[derive(Debug, Clone)]
pub enum Inputs {
    /// The inputs are to a task.
    Task(TaskInputs),
    /// The inputs are to a workflow.
    Workflow(WorkflowInputs),
}

impl Inputs {
    /// Parses an inputs file from the given file path.
    ///
    /// The format (JSON or YAML) is determined by the file extension:
    ///
    /// - `.json` for JSON format
    /// - `.yml` or `.yaml` for YAML format
    ///
    /// The parse uses the provided document to validate the input keys within
    /// the file.
    ///
    /// Returns `Ok(Some(_))` if the inputs are not empty.
    ///
    /// Returns `Ok(None)` if the inputs are empty.
    pub fn parse(document: &Document, path: impl AsRef<Path>) -> Result<Option<(String, Self)>> {
        let path = path.as_ref();

        match path.extension().and_then(|ext| ext.to_str()) {
            Some("json") => Self::parse_json(document, path),
            Some("yml") | Some("yaml") => Self::parse_yaml(document, path),
            ext => bail!(
                "unsupported file extension: `{ext}`; the supported formats are JSON (`.json`) \
                 and YAML (`.yaml` and `.yml`)",
                ext = ext.unwrap_or("")
            ),
        }
        .with_context(|| format!("failed to parse input file `{path}`", path = path.display()))
    }

    /// Parses a JSON inputs file from the given file path.
    ///
    /// The parse uses the provided document to validate the input keys within
    /// the file.
    ///
    /// Returns `Ok(Some(_))` if the inputs are not empty.
    ///
    /// Returns `Ok(None)` if the inputs are empty.
    pub fn parse_json(
        document: &Document,
        path: impl AsRef<Path>,
    ) -> Result<Option<(String, Self)>> {
        let path = path.as_ref();

        let file = File::open(path).with_context(|| {
            format!("failed to open input file `{path}`", path = path.display())
        })?;

        // Parse the JSON (should be an object)
        let reader = BufReader::new(file);

        let map = std::mem::take(
            serde_json::from_reader::<_, JsonValue>(reader)?
                .as_object_mut()
                .with_context(|| {
                    format!(
                        "expected input file `{path}` to contain a JSON object",
                        path = path.display()
                    )
                })?,
        );

        Self::parse_json_object(document, map)
    }

    /// Parses a YAML inputs file from the given file path.
    ///
    /// The parse uses the provided document to validate the input keys within
    /// the file.
    ///
    /// Returns `Ok(Some(_))` if the inputs are not empty.
    ///
    /// Returns `Ok(None)` if the inputs are empty.
    pub fn parse_yaml(
        document: &Document,
        path: impl AsRef<Path>,
    ) -> Result<Option<(String, Self)>> {
        let path = path.as_ref();

        let file = File::open(path).with_context(|| {
            format!("failed to open input file `{path}`", path = path.display())
        })?;

        // Parse the YAML
        let reader = BufReader::new(file);
        let yaml = serde_yaml_ng::from_reader::<_, YamlValue>(reader)?;

        // Convert YAML to JSON format
        let mut json = serde_json::to_value(yaml).with_context(|| {
            format!(
                "failed to convert YAML to JSON for processing `{path}`",
                path = path.display()
            )
        })?;

        let object = std::mem::take(json.as_object_mut().with_context(|| {
            format!(
                "expected input file `{path}` to contain a YAML mapping",
                path = path.display()
            )
        })?);

        Self::parse_json_object(document, object)
    }

    /// Determines if the inputs are empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Gets the length of all inputs.
    ///
    /// For task inputs, this include the inputs, requirements, and hints.
    ///
    /// For workflow inputs, this includes the inputs and nested inputs.
    pub fn len(&self) -> usize {
        match self {
            Self::Task(inputs) => inputs.len(),
            Self::Workflow(inputs) => inputs.len(),
        }
    }

    /// Gets an input value.
    pub fn get(&self, name: &str) -> Option<&Value> {
        match self {
            Self::Task(t) => t.inputs.get(name),
            Self::Workflow(w) => w.inputs.get(name),
        }
    }

    /// Sets an input value.
    ///
    /// Returns the previous value, if any.
    pub fn set(&mut self, name: impl Into<String>, value: impl Into<Value>) -> Option<Value> {
        match self {
            Self::Task(inputs) => inputs.set(name, value),
            Self::Workflow(inputs) => inputs.set(name, value),
        }
    }

    /// Gets the task inputs.
    ///
    /// Returns `None` if the inputs are for a workflow.
    pub fn as_task_inputs(&self) -> Option<&TaskInputs> {
        match self {
            Self::Task(inputs) => Some(inputs),
            Self::Workflow(_) => None,
        }
    }

    /// Gets a mutable reference to task inputs.
    ///
    /// Returns `None` if the inputs are for a workflow.
    pub fn as_task_inputs_mut(&mut self) -> Option<&mut TaskInputs> {
        match self {
            Self::Task(inputs) => Some(inputs),
            Self::Workflow(_) => None,
        }
    }

    /// Unwraps the inputs as task inputs.
    ///
    /// # Panics
    ///
    /// Panics if the inputs are for a workflow.
    pub fn unwrap_task_inputs(self) -> TaskInputs {
        match self {
            Self::Task(inputs) => inputs,
            Self::Workflow(_) => panic!("inputs are for a workflow"),
        }
    }

    /// Gets the workflow inputs.
    ///
    /// Returns `None` if the inputs are for a task.
    pub fn as_workflow_inputs(&self) -> Option<&WorkflowInputs> {
        match self {
            Self::Task(_) => None,
            Self::Workflow(inputs) => Some(inputs),
        }
    }

    /// Gets a mutable reference to workflow inputs.
    ///
    /// Returns `None` if the inputs are for a task.
    pub fn as_workflow_inputs_mut(&mut self) -> Option<&mut WorkflowInputs> {
        match self {
            Self::Task(_) => None,
            Self::Workflow(inputs) => Some(inputs),
        }
    }

    /// Unwraps the inputs as workflow inputs.
    ///
    /// # Panics
    ///
    /// Panics if the inputs are for a task.
    pub fn unwrap_workflow_inputs(self) -> WorkflowInputs {
        match self {
            Self::Task(_) => panic!("inputs are for a task"),
            Self::Workflow(inputs) => inputs,
        }
    }

    /// Parses the root object in a [`JsonMap`].
    ///
    /// Returns `Ok(Some(_))` if the inputs are not empty.
    ///
    /// Returns `Ok(None)` if the inputs are empty.
    pub fn parse_json_object(
        document: &Document,
        object: JsonMap,
    ) -> Result<Option<(String, Self)>> {
        // If the object is empty, treat it as an invocation without any inputs.
        if object.is_empty() {
            return Ok(None);
        }

        // Otherwise, build a set of candidate targets from the prefixes of each input
        // key.
        let mut target_candidates = BTreeSet::new();
        for key in object.keys() {
            let Some((prefix, _)) = key.split_once('.') else {
                bail!(
                    "invalid input key `{key}`: expected the key to be prefixed with the workflow \
                     or task name",
                )
            };
            target_candidates.insert(prefix);
        }

        // If every prefix is the same, there will be only one candidate. If not, report
        // an error.
        let target_name = match target_candidates
            .iter()
            .take(2)
            .collect::<Vec<_>>()
            .as_slice()
        {
            [] => panic!("no target candidates for inputs; report this as a bug"),
            [target_name] => target_name.to_string(),
            _ => bail!(
                "invalid inputs: expected each input key to be prefixed with the same workflow or \
                 task name, but found multiple prefixes: {target_candidates:?}",
            ),
        };

        let inputs = match (document.task_by_name(&target_name), document.workflow()) {
            (Some(task), _) => Self::parse_task_inputs(document, task, object)?,
            (None, Some(workflow)) if workflow.name() == target_name => {
                Self::parse_workflow_inputs(document, workflow, object)?
            }
            _ => bail!(
                "invalid inputs: a task or workflow named `{target_name}` does not exist in the \
                 document"
            ),
        };
        Ok(Some((target_name, inputs)))
    }

    /// Parses the inputs for a task.
    fn parse_task_inputs(document: &Document, task: &Task, object: JsonMap) -> Result<Self> {
        let mut inputs = TaskInputs::default();
        for (key, value) in object {
            // Convert from serde_json::Value to crate::Value
            let value = serde_json::from_value(value)
                .with_context(|| format!("invalid input value for key `{key}`"))?;

            match key.split_once(".") {
                Some((prefix, remainder)) if prefix == task.name() => {
                    inputs
                        .set_path_value(document, task, remainder, value)
                        .with_context(|| format!("invalid input key `{key}`"))?;
                }
                _ => {
                    // This should be caught by the initial check of the prefixes in
                    // `parse_json_object()`, but we retain a friendly error message in case this
                    // function gets called from another context in the future.
                    bail!(
                        "invalid input key `{key}`: expected key to be prefixed with `{task}`",
                        task = task.name()
                    );
                }
            }
        }

        Ok(Inputs::Task(inputs))
    }

    /// Parses the inputs for a workflow.
    fn parse_workflow_inputs(
        document: &Document,
        workflow: &Workflow,
        object: JsonMap,
    ) -> Result<Self> {
        let mut inputs = WorkflowInputs::default();
        for (key, value) in object {
            // Convert from serde_json::Value to crate::Value
            let value = serde_json::from_value(value)
                .with_context(|| format!("invalid input value for key `{key}`"))?;

            match key.split_once(".") {
                Some((prefix, remainder)) if prefix == workflow.name() => {
                    inputs
                        .set_path_value(document, workflow, remainder, value)
                        .with_context(|| format!("invalid input key `{key}`"))?;
                }
                _ => {
                    // This should be caught by the initial check of the prefixes in
                    // `parse_json_object()`, but we retain a friendly error message in case this
                    // function gets called from another context in the future.
                    bail!(
                        "invalid input key `{key}`: expected key to be prefixed with `{workflow}`",
                        workflow = workflow.name()
                    );
                }
            }
        }

        Ok(Inputs::Workflow(inputs))
    }
}

impl From<TaskInputs> for Inputs {
    fn from(inputs: TaskInputs) -> Self {
        Self::Task(inputs)
    }
}

impl From<WorkflowInputs> for Inputs {
    fn from(inputs: WorkflowInputs) -> Self {
        Self::Workflow(inputs)
    }
}

impl Serialize for Inputs {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        match self {
            Self::Task(inputs) => inputs.serialize(serializer),
            Self::Workflow(inputs) => inputs.serialize(serializer),
        }
    }
}