cargo_athena_core/

lib.rs

//! cargo-athena runtime.
//!
//! Every `#[workflow]`/`#[container]` becomes a unit-struct **type** that
//! implements [`Template`]. That type *is* the cross-crate wormhole: the
//! type system resolves a callee's Argo name/inputs across modules and
//! crates, and the generated `Template::collect` calls
//! `<Callee as Template>::collect` directly — a monomorphic call, so the
//! whole reachable closure is force-linked with no `inventory`/DCE games.
//!
//! Two worlds share one binary:
//!
//! * **Emit** — `main` calls [`entrypoint::<E>()`]; we walk the closure
//!   from `E` and print one Argo `WorkflowTemplate` document per template
//!   (cross-refs via `templateRef`) plus a runnable `Workflow` for `E`.
//! * **Run** — Argo invokes the binary with `--cargo-athena-template <name>`;
//!   we deserialize inputs, run the real container body, serialize outputs.
//!
//! `host!` declarations are still collected statically by the attribute
//! macros and resolved through the `#[fragment]` (`inventory`) closure here
//! — fragments are genuinely *called* by container bodies, so unlike
//! templates they have a real symbol reference and no DCE concern.

use std::collections::{HashMap, HashSet};

/// Argo API types (generated from protobuf).
pub use cargo_athena_api as api;
// Re-exported so macro-generated code has stable paths under `::cargo_athena`.
pub use inventory;
pub use serde_json;
// Maintained serde_yaml fork: YAML 1.1-aware emitter (quotes `n`/`yes`/
// `null`/… so Argo's Go YAML→JSON parser can't mis-type them). serde_yaml
// itself is archived/EOL.
pub use serde_norway;

/// Fan-out: `list.fan_out(|x| template(x, ..))` runs `template` once per
/// element (Argo `withParam`); the binding is the aggregated `Vec<U>` of
/// the per-element returns. This trait exists only so the ghost
/// type-checks the element type, the closure, and the resulting
/// `Vec<U>`; the macro lowers the call to Argo and it never runs.
pub trait AthenaList<T> {
    #[doc(hidden)]
    fn fan_out<U, F: FnOnce(T) -> U>(self, _f: F) -> Vec<U>
    where
        Self: Sized,
    {
        unimplemented!("athena ghost: never executed")
    }
}
impl<T> AthenaList<T> for Vec<T> {}
impl<T, const N: usize> AthenaList<T> for [T; N] {}

/// Marker for types that may be injected into a `#[container]`
/// attribute (`image = "repo:" + tag`). Restricted to `String`/`str`
/// and the primitive numbers: their `serde_json` form, unwrapped at
/// runtime by `{{=fromJSON(...)}}`, renders to the obvious raw scalar
/// (`"v"`→`v`, `7`→`7`). A `Display` bound would wrongly admit types
/// whose `Display` differs from their JSON round-trip. The macro emits
/// a hidden `Injectable`-bounded assertion against the real arg type.
#[doc(hidden)]
pub trait Injectable {}
macro_rules! __athena_injectable {
    ($($t:ty),* $(,)?) => { $( impl Injectable for $t {} )* };
}
__athena_injectable!(
    String, str, i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64,
);

/// `host!("/lit/path")` — declare a hostPath volume for the enclosing
/// container, evaluating to the (already-mounted) path at runtime.
///
/// Only valid inside a `#[cargo_athena::container]` or
/// `#[cargo_athena::fragment]` fn: those attribute macros rewrite the
/// invocations they can see into the private real macro below. This
/// public definition is therefore a *hard error* — it only ever expands
/// when `host!` is used somewhere the collector cannot see it (a plain
/// fn, a `#[workflow]`, or nested inside another macro's tokens), where a
/// silently-unmounted path would otherwise be a footgun.
#[macro_export]
macro_rules! host {
    ($($t:tt)*) => {
        ::core::compile_error!(
            "`host!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn \
             (not in a plain fn, a `#[workflow]`, or nested inside another \
             macro invocation)"
        )
    };
}

/// The real expansion. Private: only the attribute macros emit this path.
#[doc(hidden)]
#[macro_export]
macro_rules! __cargo_athena_host {
    ($path:literal) => {
        $crate::rt::host_path($path)
    };
    ($($t:tt)*) => {
        ::core::compile_error!("`host!` takes a single string-literal path")
    };
}

// --- artifact declaration macros (native Argo artifacts, no S3) ------------
//
// Each is a public (gated `compile_error!`) + private (real) pair, exactly
// like `host!`: only valid inside `#[container]`/`#[fragment]`, where the
// attribute macro rewrites the public form to the private one.

/// Declare an Argo *input* artifact port and read it (bytes) at runtime.
#[macro_export]
macro_rules! load_artifact {
    ($($t:tt)*) => {
        ::core::compile_error!(
            "`load_artifact!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        )
    };
}
#[doc(hidden)]
#[macro_export]
macro_rules! __cargo_athena_load_artifact {
    ($name:literal) => {
        $crate::rt::load_artifact($name)
    };
    ($($t:tt)*) => {
        ::core::compile_error!("load_artifact!(\"name\")")
    };
}

/// Declare an Argo *input* artifact port and read it (UTF-8) at runtime.
#[macro_export]
macro_rules! load_artifact_str {
    ($($t:tt)*) => {
        ::core::compile_error!(
            "`load_artifact_str!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        )
    };
}
#[doc(hidden)]
#[macro_export]
macro_rules! __cargo_athena_load_artifact_str {
    ($name:literal) => {
        $crate::rt::load_artifact_str($name)
    };
    ($($t:tt)*) => {
        ::core::compile_error!("load_artifact_str!(\"name\")")
    };
}

/// Declare an Argo *output* artifact port and write bytes to it at runtime.
#[macro_export]
macro_rules! save_artifact {
    ($($t:tt)*) => {
        ::core::compile_error!(
            "`save_artifact!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        )
    };
}
#[doc(hidden)]
#[macro_export]
macro_rules! __cargo_athena_save_artifact {
    ($name:literal, $data:expr) => {
        $crate::rt::save_artifact($name, $data)
    };
    ($($t:tt)*) => {
        ::core::compile_error!("save_artifact!(\"name\", data)")
    };
}

/// Declare an Argo *output* artifact port and write a string at runtime.
#[macro_export]
macro_rules! save_artifact_str {
    ($($t:tt)*) => {
        ::core::compile_error!(
            "`save_artifact_str!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        )
    };
}
#[doc(hidden)]
#[macro_export]
macro_rules! __cargo_athena_save_artifact_str {
    ($name:literal, $data:expr) => {
        $crate::rt::save_artifact_str($name, $data)
    };
    ($($t:tt)*) => {
        ::core::compile_error!("save_artifact_str!(\"name\", data)")
    };
}

/// Runtime shims referenced by the declaration macros. Artifact ports are
/// plain files at fixed paths; Argo moves them (no S3 from us).
pub mod rt {
    use std::path::PathBuf;

    /// Identity: the volume is already mounted when the container runs.
    pub const fn host_path(path: &'static str) -> &'static str {
        path
    }

    /// Where Argo drops/collects declared artifact ports inside the pod.
    pub const IN_DIR: &str = "/athena/artifacts/in";
    pub const OUT_DIR: &str = "/athena/artifacts/out";

    fn in_path(name: &str) -> PathBuf {
        PathBuf::from(IN_DIR).join(name)
    }
    fn out_path(name: &str) -> PathBuf {
        PathBuf::from(OUT_DIR).join(name)
    }

    pub fn load_artifact(name: &str) -> Vec<u8> {
        let p = in_path(name);
        std::fs::read(&p).unwrap_or_else(|e| panic!("load_artifact({name:?}) {}: {e}", p.display()))
    }

    pub fn load_artifact_str(name: &str) -> String {
        let p = in_path(name);
        std::fs::read_to_string(&p)
            .unwrap_or_else(|e| panic!("load_artifact_str({name:?}) {}: {e}", p.display()))
    }

    pub fn save_artifact(name: &str, data: impl AsRef<[u8]>) {
        let p = out_path(name);
        if let Some(d) = p.parent() {
            std::fs::create_dir_all(d).expect("create artifact out dir");
        }
        std::fs::write(&p, data.as_ref())
            .unwrap_or_else(|e| panic!("save_artifact({name:?}) {}: {e}", p.display()));
    }

    pub fn save_artifact_str(name: &str, data: impl AsRef<str>) {
        save_artifact(name, data.as_ref().as_bytes());
    }
}

/// What kind of Argo template a type produces.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum TemplateKind {
    /// Leaf — real code in a pod (`#[container]`).
    Container,
    /// Composition — a DAG of other templates (`#[workflow]`).
    Workflow,
}

/// The cross-crate identity of a template, implemented by the unit struct
/// the `#[workflow]`/`#[container]` macros generate.
///
/// Callers never name the Argo string; they reference the *type*
/// (`<foo::ingest as Template>::ARGO_NAME`), so name/input resolution is
/// done by the compiler — collision-proof across crates, and the reference
/// itself force-links the defining crate.
pub trait Template {
    /// Globally-unique Argo resource name (`<crate>-<fn>` by default).
    const ARGO_NAME: &'static str;
    /// Declared input parameter names, in order.
    const INPUTS: &'static [&'static str];
    /// Stringified Rust types of [`Self::INPUTS`], same order — for
    /// `container emulate`'s pre-launch arg checking and the `ls`
    /// listings. Emitted by `#[container]` and `#[workflow]`; defaulted
    /// empty for synthetic/hand impls.
    const INPUT_TYPES: &'static [&'static str] = &[];
    /// `true` for athena-synthesized templates (the `if`/`else`
    /// wrapper + per-arm sub-workflows). They're an implementation
    /// detail, so `workflow ls` hides them unless `--include-synthetic`.
    const SYNTHETIC: bool = false;
    const KIND: TemplateKind;
    /// Whole-workflow exit-handler template name, from
    /// `#[workflow(on_exit_if_root=…)]` / `#[container(on_exit_if_root=…)]`.
    /// `emit` puts it on this template's own `spec.hooks.exit`; Argo
    /// fires exit hooks workflow-scoped, so it runs only when this
    /// workflow is the one submitted (inert as a nested templateRef).
    const ON_EXIT: Option<&'static str> = None;
    /// Workflow-scoped TTL GC, from `#[…(ttl(..))]`. `build_templates`
    /// puts it on this template's own `spec.ttlStrategy` (same per-WT
    /// plumbing as `ON_EXIT`; never on synthetic `if` wrappers).
    const TTL: ::core::option::Option<crate::api::TtlStrategy> = None;
    /// Workflow-scoped pod GC strategy, from `#[…(pod_gc(strategy=..))]`.
    /// `build_templates` puts it on this template's own `spec.podGC`.
    const POD_GC: ::core::option::Option<&'static str> = None;
    /// Root-only whole-workflow runtime cap (seconds), from
    /// `#[…(active_deadline_if_root=..)]`. `build_templates` stamps it
    /// on this template's own `spec.activeDeadlineSeconds` (same per-WT,
    /// root-only plumbing as `TTL`/`POD_GC`). This is the *only* working
    /// whole-workflow timeout: Argo applies neither `Template.timeout`
    /// nor `Template.activeDeadlineSeconds` to dag/steps templates.
    const ACTIVE_DEADLINE_IF_ROOT: ::core::option::Option<i64> = None;

    /// Build this template's inner Argo `template` object.
    fn build(ctx: &BuildCtx) -> api::Template;

    /// Run-mode body — overridden by `#[container]`; never called on a
    /// `#[workflow]`.
    fn run(_input: serde_json::Value) -> serde_json::Value {
        panic!(
            "`{}` is not a #[container]; nothing to run",
            Self::ARGO_NAME
        )
    }

    /// Push self + the transitive callee closure into `out`. The macro
    /// generates `<Callee as Template>::collect(out)` per callee, so the
    /// whole reachable set is linked by direct calls.
    fn collect(out: &mut Collector);
}

// ---- athena.toml ---------------------------------------------------------

/// `athena.toml` — required by `cargo athena` at emit time. Mirrors the
/// parts of Argo's S3 `ArtifactRepository` we inject, plus bootstrap config.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct AthenaConfig {
    pub artifact_repository: ArtifactRepository,
    pub artifact: ArtifactSpec,
    #[serde(default)]
    pub bootstrap: Bootstrap,
    #[serde(default)]
    pub defaults: Defaults,
}

#[derive(Debug, Clone, serde::Deserialize)]
pub struct Defaults {
    /// Kubernetes ServiceAccount the workflow pods run as (so users bind
    /// their own RBAC). Per-`#[container(service_account=...)]` overrides.
    #[serde(default = "default_service_account")]
    pub service_account: String,
    /// Default cargo package the `cargo athena` subcommands drive (so
    /// you don't repeat `--package`). The `--package`/`-p` flag wins.
    #[serde(default)]
    pub package: Option<String>,
    /// Default cargo bin within that package (multi-bin crates need
    /// it). The `--bin` flag wins.
    #[serde(default)]
    pub bin: Option<String>,
    /// Default Kubernetes namespace for `cargo athena submit`. Precedence:
    /// `-n/--namespace` → `$ARGO_NAMESPACE` → this → `"default"`.
    #[serde(default)]
    pub namespace: Option<String>,
}

impl Default for Defaults {
    fn default() -> Self {
        Self {
            service_account: default_service_account(),
            package: None,
            bin: None,
            namespace: None,
        }
    }
}

fn default_service_account() -> String {
    "default".to_string()
}

/// Resolve a container template's ServiceAccount: the
/// `#[container(service_account=...)]` override, else `[defaults]`.
pub fn service_account(ctx: &BuildCtx, over: Option<&str>) -> String {
    over.map(str::to_string)
        .unwrap_or_else(|| ctx.config().defaults.service_account.clone())
}

#[derive(Debug, Clone, serde::Deserialize)]
pub struct ArtifactRepository {
    pub s3: S3Repo,
}

#[derive(Debug, Clone, serde::Deserialize)]
pub struct S3Repo {
    pub endpoint: String,
    pub bucket: String,
    #[serde(default)]
    pub region: String,
    #[serde(default)]
    pub insecure: bool,
    pub access_key_secret: SecretRef,
    pub secret_key_secret: SecretRef,
}

#[derive(Debug, Clone, serde::Deserialize)]
pub struct SecretRef {
    pub name: String,
    pub key: String,
}

#[derive(Debug, Clone, serde::Deserialize)]
pub struct ArtifactSpec {
    /// Object key of the per-binary tarball (holds `app-<triple>` for every
    /// `bootstrap.targets`). `cargo athena build` fills any
    /// `{crate}`/`{version}`/`{bin}` placeholders before publish.
    pub key: String,
}

#[derive(Debug, Clone, serde::Deserialize)]
pub struct Bootstrap {
    /// Fallback image when a `#[container]` doesn't set its own. Per-
    /// container `image` always wins (arbitrary by design); this is just
    /// the small default for containers that don't care.
    #[serde(default = "default_image")]
    pub default_image: String,
    /// Cross-compile / `uname` target matrix.
    #[serde(default = "default_targets")]
    pub targets: Vec<String>,
}

impl Default for Bootstrap {
    fn default() -> Self {
        Self {
            default_image: default_image(),
            targets: default_targets(),
        }
    }
}

fn default_image() -> String {
    "busybox:1.36-musl".to_string()
}

fn default_targets() -> Vec<String> {
    vec![
        "x86_64-unknown-linux-musl".to_string(),
        "aarch64-unknown-linux-musl".to_string(),
    ]
}

impl AthenaConfig {
    /// `ATHENA_CONFIG` override, else the nearest `athena.toml` walking up
    /// from the cwd. Only ever called during emit — the in-pod binary
    /// (run-mode) never needs `athena.toml`.
    pub fn load() -> Self {
        let path = std::env::var_os("ATHENA_CONFIG")
            .map(std::path::PathBuf::from)
            .or_else(Self::find_upwards)
            .expect(
                "athena.toml not found: set ATHENA_CONFIG or add athena.toml \
                 to the workspace (required by `cargo athena`)",
            );
        let text = std::fs::read_to_string(&path)
            .unwrap_or_else(|e| panic!("read {}: {e}", path.display()));
        toml::from_str(&text).unwrap_or_else(|e| panic!("parse {}: {e}", path.display()))
    }

    fn find_upwards() -> Option<std::path::PathBuf> {
        let mut d = std::env::current_dir().ok()?;
        loop {
            let p = d.join("athena.toml");
            if p.is_file() {
                return Some(p);
            }
            if !d.pop() {
                return None;
            }
        }
    }
}

/// Pod-scoped scratch root, backed by an `emptyDir` on every container
/// template so all athena paths are writable regardless of the image
/// (distroless / read-only rootfs) and shared with Argo's init/wait
/// containers for artifact load/collect.
pub const ATHENA_DIR: &str = "/athena";
/// Where Argo's executor (init container) extracts the per-arch
/// binaries from our `.tar.gz` input artifact. We rely on Argo's
/// built-in tarball auto-extraction (no `archive: none`, no `tar` in
/// the main container's image — see `container_delivery`).
pub const ATHENA_BIN_DIR: &str = "/athena/bin";
/// The in-pod arch-resolving + exec bootstrap, kept in a separate
/// `bootstrap.sh` so it can be read, edited, and `shellcheck`'d as a
/// plain shell file rather than buried in a Rust `format!`. `@@ARMS@@`
/// / `@@BIN_DIR@@` / `@@TEMPLATE@@` are substituted at emit time in
/// `container_delivery`.
const BOOTSTRAP_TEMPLATE: &str = include_str!("bootstrap.sh");
/// Name of the scratch `emptyDir` volume.
pub const SCRATCH_VOLUME: &str = "athena-work";
/// Argo input-artifact name of the binary tarball `emit` injects.
pub const ATHENA_DIST_ARTIFACT: &str = "athena-dist";
/// Env-var prefix the in-pod bootstrap reads each input parameter from.
pub const ATHENA_PARAM_PREFIX: &str = "ATHENA_PARAM_";

/// Resolved S3 coordinates for one artifact (creds are supplied
/// locally, e.g. via AWS env vars — `cargo athena container run` uses
/// `object_store`; the in-cluster path uses the k8s Secret refs).
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
pub struct S3Ref {
    pub endpoint: String,
    pub bucket: String,
    pub region: String,
    pub insecure: bool,
    pub key: String,
}

/// One artifact bound into the container at `path`, backed by S3.
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
pub struct ArtifactRef {
    pub s3: S3Ref,
    pub path: String,
}

/// An input parameter, the env var the bootstrap reads it from, and its
/// stringified Rust type (`""` if unknown — synthetic templates).
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
pub struct ParamRef {
    pub name: String,
    pub env: String,
    pub ty: String,
}

/// Purpose-built introspection of one `#[container]`, derived from the
/// *same* `Template::build()` `emit` uses (so it never drifts), but
/// expressed in the runner's vocabulary instead of Argo's. Emitted as
/// JSON by the binary when `CARGO_ATHENA_DESCRIBE=<name>` is set;
/// consumed by `cargo athena container run` to realize the spec under
/// docker/podman locally. Also the basis for a future
/// `cargo athena container describe`.
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
pub struct ContainerRunMeta {
    /// Argo template name (`<crate>-<fn>`).
    pub name: String,
    /// `"container"`, `"workflow"`, or `"other"`.
    pub kind: String,
    /// athena-synthesized template (an `if`/`else` wrapper or arm) —
    /// `workflow ls` hides these unless `--include-synthetic`.
    pub synthetic: bool,
    /// Resolved container image.
    pub image: String,
    /// The injected bootstrap command + args, verbatim — run as-is so
    /// the local execution path is byte-identical to the pod's.
    pub command: Vec<String>,
    pub args: Vec<String>,
    /// Mount path of the pod-scoped scratch dir (the `emptyDir`, e.g.
    /// `/athena`); bind a host temp dir here to read `result_path` back.
    pub work_dir: String,
    /// Input parameters and the env var each is delivered through.
    pub params: Vec<ParamRef>,
    /// The binary tarball artifact (always present for a container).
    pub binary_artifact: Option<ArtifactRef>,
    /// `load_artifact!` input ports (excludes the binary tarball).
    pub input_artifacts: Vec<ArtifactRef>,
    /// `save_artifact!` output ports.
    pub output_artifacts: Vec<ArtifactRef>,
    /// `host!` paths (mounted at the same path in-pod; bind 1:1 locally).
    pub host_paths: Vec<String>,
    /// File the body writes its serialized return to
    /// (`outputs.parameters.return`); read it back from the bind mount.
    pub result_path: Option<String>,
}

impl ContainerRunMeta {
    /// Derive the runner metadata from one built Argo template.
    /// `input_types` is parallel to the template's input parameters
    /// (same order); empty when unknown.
    fn from_template(t: &api::Template, input_types: &[&str]) -> Self {
        let kind = if t.container.is_some() {
            "container"
        } else if t.dag.is_some() || !t.steps.is_empty() {
            "workflow"
        } else {
            "other"
        };
        let c = t.container.as_ref();
        let mount_path = |vol: &str| {
            c.and_then(|c| {
                c.volume_mounts
                    .iter()
                    .find(|m| m.name == vol)
                    .map(|m| m.mount_path.clone())
            })
        };
        let to_ref = |a: &api::Artifact| {
            a.s3.as_ref().map(|s| ArtifactRef {
                s3: S3Ref {
                    endpoint: s.endpoint.clone(),
                    bucket: s.bucket.clone(),
                    region: s.region.clone(),
                    insecure: s.insecure,
                    key: s.key.clone(),
                },
                path: a.path.clone(),
            })
        };
        let in_arts = t.inputs.as_ref().map(|i| &i.artifacts);
        ContainerRunMeta {
            name: t.name.clone(),
            kind: kind.to_string(),
            // set by the caller from the Collector (Template::SYNTHETIC
            // isn't visible through the type-erased builder fn here).
            synthetic: false,
            image: c.map(|c| c.image.clone()).unwrap_or_default(),
            command: c.map(|c| c.command.clone()).unwrap_or_default(),
            args: c.map(|c| c.args.clone()).unwrap_or_default(),
            work_dir: mount_path(SCRATCH_VOLUME).unwrap_or_else(|| ATHENA_DIR.to_string()),
            params: t
                .inputs
                .as_ref()
                .map(|i| {
                    i.parameters
                        .iter()
                        .enumerate()
                        .map(|(idx, p)| ParamRef {
                            name: p.name.clone(),
                            env: format!("{ATHENA_PARAM_PREFIX}{}", p.name),
                            ty: input_types
                                .get(idx)
                                .map(|s| (*s).to_string())
                                .unwrap_or_default(),
                        })
                        .collect()
                })
                .unwrap_or_default(),
            binary_artifact: in_arts
                .and_then(|a| a.iter().find(|a| a.name == ATHENA_DIST_ARTIFACT))
                .and_then(to_ref),
            input_artifacts: in_arts
                .map(|a| {
                    a.iter()
                        .filter(|a| a.name != ATHENA_DIST_ARTIFACT)
                        .filter_map(to_ref)
                        .collect()
                })
                .unwrap_or_default(),
            output_artifacts: t
                .outputs
                .as_ref()
                .map(|o| o.artifacts.iter().filter_map(to_ref).collect())
                .unwrap_or_default(),
            host_paths: t
                .volumes
                .iter()
                .filter_map(|v| v.host_path.as_ref().map(|h| h.path.clone()))
                .collect(),
            result_path: t.outputs.as_ref().and_then(|o| {
                o.parameters
                    .iter()
                    .find(|p| p.name == "return")
                    .and_then(|p| p.value_from.as_ref())
                    .map(|vf| vf.path.clone())
                    .filter(|p| !p.is_empty())
            }),
        }
    }
}

/// What [`container_delivery`] produces for one `#[container]` template.
pub struct ContainerDelivery {
    /// Resolved image: the `#[container(image=...)]` override, else
    /// `[bootstrap].default_image`.
    pub image: String,
    pub command: Vec<String>,
    pub args: Vec<String>,
    pub artifact: api::Artifact,
}

/// The arch-resolving bootstrap + the S3 binary artifact for one container
/// template. Called from macro-generated `Template::build` (emit only).
///
/// Runs inside the container's *arbitrary, user-chosen* image (it only
/// needs a POSIX `sh` and `uname` — **no `tar`**). The binary `.tar.gz`
/// is delivered as an Argo input artifact at [`ATHENA_BIN_DIR`]; with
/// the default `Archive` (no `archive: none`) Argo's executor init
/// container **auto-detects tarballs and untars them into the artifact
/// path** (proven from `workflow/executor/executor.go:262–289`,
/// `untar` ibid., real v4.0.5 source). So by the time our bootstrap
/// runs the per-arch `app-<triple>` files already exist under
/// [`ATHENA_BIN_DIR`]; the script just `uname`s, `chmod +x`'s, and
/// `exec`s — replacing the shell so the Rust binary is the container's
/// main process. Works whether the tarball has one entry or many.
pub fn container_delivery(
    ctx: &BuildCtx,
    argo_name: &str,
    image_override: Option<&str>,
) -> ContainerDelivery {
    let cfg = ctx.config();
    let s3 = &cfg.artifact_repository.s3;
    let image = image_override
        .map(str::to_string)
        .unwrap_or_else(|| cfg.bootstrap.default_image.clone());

    let mut arms = String::new();
    for triple in &cfg.bootstrap.targets {
        let arch = triple.split('-').next().unwrap_or(triple);
        let pat = if arch == "aarch64" {
            "aarch64|arm64"
        } else {
            arch
        };
        arms.push_str(&format!("  {pat}) __t={triple} ;;\n"));
    }

    // Argo's executor (init container) auto-extracts the `.tar.gz` into
    // ATHENA_BIN_DIR, so the bootstrap just picks + execs. The template
    // lives in a sibling `bootstrap.sh` file (legible / greppable /
    // shellcheck-able); we just substitute the @@-delimited slots.
    let script = BOOTSTRAP_TEMPLATE
        .replace("@@ARMS@@", &arms)
        .replace("@@BIN_DIR@@", ATHENA_BIN_DIR)
        .replace("@@TEMPLATE@@", argo_name);

    // `archive: None` (NOT `archive: none`) lets Argo auto-detect the
    // input as a tarball and untar it into `path` (`ATHENA_BIN_DIR`).
    // See `container_delivery` doc above.
    let artifact = api::Artifact {
        name: "athena-dist".to_string(),
        path: ATHENA_BIN_DIR.to_string(),
        s3: Some(s3_loc(s3, &cfg.artifact.key)),
        archive: None,
        mode: None,
    };

    ContainerDelivery {
        image,
        command: vec!["/bin/sh".to_string(), "-c".to_string()],
        args: vec![script],
        artifact,
    }
}

/// A `#[fragment]`: a plain helper carrying `host!` decls. Still
/// `inventory`-based — a container's real body actually *calls* its
/// fragments, so the symbol reference exists and DCE is not a concern.
pub struct FragmentReg {
    pub rust_name: &'static str,
    pub host_paths: &'static [&'static str],
    pub in_artifacts: &'static [&'static str],
    pub out_artifacts: &'static [&'static str],
    pub callees: &'static [&'static str],
}
inventory::collect!(FragmentReg);

/// Fragment registry snapshot, passed to container `build`s.
pub struct BuildCtx {
    fragments: HashMap<&'static str, &'static FragmentReg>,
    config: AthenaConfig,
}

impl BuildCtx {
    /// Emit-only: gathers fragments AND loads `athena.toml`. Never called
    /// in run-mode, so the in-pod binary needs no `athena.toml`.
    pub fn collect() -> Self {
        let mut fragments = HashMap::new();
        for f in inventory::iter::<FragmentReg> {
            fragments.insert(f.rust_name, f);
        }
        Self {
            fragments,
            config: AthenaConfig::load(),
        }
    }

    pub fn config(&self) -> &AthenaConfig {
        &self.config
    }

    /// Own literal decls ∪ the transitive `#[fragment]` closure for one
    /// kind of declaration (deduped, stable order). `select` picks which
    /// `FragmentReg` slice to pull from a reached fragment.
    fn resolved(
        &self,
        own: &[&str],
        own_callees: &[&str],
        select: impl Fn(&FragmentReg) -> &'static [&'static str],
    ) -> Vec<String> {
        let mut out: Vec<String> = Vec::new();
        let mut seen: HashSet<String> = HashSet::new();
        let mut push = |p: &str, out: &mut Vec<String>| {
            if seen.insert(p.to_string()) {
                out.push(p.to_string());
            }
        };
        for p in own {
            push(p, &mut out);
        }
        let mut queue: Vec<&str> = own_callees.to_vec();
        let mut visited: HashSet<&str> = HashSet::new();
        while let Some(c) = queue.pop() {
            if !visited.insert(c) {
                continue;
            }
            if let Some(f) = self.fragments.get(c) {
                for p in select(f) {
                    push(p, &mut out);
                }
                queue.extend(f.callees.iter().copied());
            }
        }
        out
    }

    /// hostPaths: own `host!`s ∪ fragment closure.
    pub fn resolved_host_paths(&self, own: &[&str], callees: &[&str]) -> Vec<String> {
        self.resolved(own, callees, |f| f.host_paths)
    }

    /// Input artifact ports: own `load_artifact*!`s ∪ fragment closure.
    pub fn resolved_in_artifacts(&self, own: &[&str], callees: &[&str]) -> Vec<String> {
        self.resolved(own, callees, |f| f.in_artifacts)
    }

    /// Output artifact ports: own `save_artifact*!`s ∪ fragment closure.
    pub fn resolved_out_artifacts(&self, own: &[&str], callees: &[&str]) -> Vec<String> {
        self.resolved(own, callees, |f| f.out_artifacts)
    }
}

fn archive_none() -> api::ArchiveStrategy {
    api::ArchiveStrategy {
        none: Some(api::NoneStrategy {}),
    }
}

/// Build an Argo S3 location (artifact-repository creds from `athena.toml`)
/// for an exact object `key`.
pub fn s3_loc(s3: &S3Repo, key: &str) -> api::S3Artifact {
    api::S3Artifact {
        endpoint: s3.endpoint.clone(),
        bucket: s3.bucket.clone(),
        region: s3.region.clone(),
        insecure: s3.insecure,
        key: key.to_string(),
        access_key_secret: Some(api::SecretKeySelector {
            name: s3.access_key_secret.name.clone(),
            key: s3.access_key_secret.key.clone(),
        }),
        secret_key_secret: Some(api::SecretKeySelector {
            name: s3.secret_key_secret.name.clone(),
            key: s3.secret_key_secret.key.clone(),
        }),
    }
}

/// A valid Argo artifact identifier derived from an S3 key (which may
/// contain `/`, `.`). The key itself is preserved in `s3.key`.
fn artifact_ident(key: &str) -> String {
    let mut s: String = key
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() { c } else { '-' })
        .collect();
    s = s.trim_matches('-').to_ascii_lowercase();
    if s.is_empty() {
        s.push('a');
    }
    s
}

/// `load_artifact!("key")` input ports: Argo pulls the exact S3 object
/// `key` from the configured repo into the pod (raw, `archive: none`).
pub fn artifact_inputs(ctx: &BuildCtx, keys: &[String]) -> Vec<api::Artifact> {
    let s3 = &ctx.config().artifact_repository.s3;
    keys.iter()
        .map(|k| api::Artifact {
            name: artifact_ident(k),
            path: format!("{}/{k}", rt::IN_DIR),
            s3: Some(s3_loc(s3, k)),
            archive: Some(archive_none()),
            mode: None,
        })
        .collect()
}

/// `save_artifact!("key")` output ports: Argo pushes the written file to
/// the exact S3 object `key` in the configured repo (raw, `archive: none`).
pub fn artifact_outputs(ctx: &BuildCtx, keys: &[String]) -> Vec<api::Artifact> {
    let s3 = &ctx.config().artifact_repository.s3;
    keys.iter()
        .map(|k| api::Artifact {
            name: artifact_ident(k),
            path: format!("{}/{k}", rt::OUT_DIR),
            s3: Some(s3_loc(s3, k)),
            archive: Some(archive_none()),
            mode: None,
        })
        .collect()
}

/// Accumulates the reachable templates (as `WorkflowTemplate`s) and the
/// run-mode dispatch table while `Template::collect` walks the closure.
pub struct Collector {
    seen: HashSet<String>,
    /// Deferred so `athena.toml` is read only at emit, never run-mode.
    builders: Vec<fn(&BuildCtx) -> api::Template>,
    runners: HashMap<String, fn(serde_json::Value) -> serde_json::Value>,
    /// `<argo name> -> <on_exit handler argo name>` for *every* template
    /// with an `on_exit` (not just the root). Each WorkflowTemplate
    /// carries its own `spec.hooks.exit`; Argo only fires the hook of
    /// the workflow that is actually submitted (workflow-scoped), so
    /// submitting a sub-workflow's template directly runs its own hook.
    exits: HashMap<String, &'static str>,
    /// `<argo name> -> ttlStrategy` for every template with `ttl(..)`.
    /// Stamped onto that WorkflowTemplate's `spec.ttlStrategy` (same
    /// per-WT, workflow-scoped semantics as `exits`).
    ttl: HashMap<String, crate::api::TtlStrategy>,
    /// `<argo name> -> podGC strategy` for every template with
    /// `pod_gc(..)`. Stamped onto that WT's `spec.podGC`.
    pod_gc: HashMap<String, String>,
    /// `<argo name> -> activeDeadlineSeconds` for every template with
    /// `active_deadline_if_root(..)`. Stamped onto that WT's
    /// `spec.activeDeadlineSeconds` (same per-WT, root-only as `ttl`).
    active_deadline: HashMap<String, i64>,
    /// `<argo name> -> stringified input types` (parallel to the
    /// template's INPUTS), for `container emulate` arg type-checking.
    types: HashMap<String, &'static [&'static str]>,
    /// Argo names of athena-synthesized templates (`Template::SYNTHETIC`)
    /// so `workflow ls` can hide them by default.
    synthetic: HashSet<String>,
}

impl Default for Collector {
    fn default() -> Self {
        Self::new()
    }
}

impl Collector {
    pub fn new() -> Self {
        Self {
            seen: HashSet::new(),
            builders: Vec::new(),
            runners: HashMap::new(),
            exits: HashMap::new(),
            ttl: HashMap::new(),
            pod_gc: HashMap::new(),
            active_deadline: HashMap::new(),
            types: HashMap::new(),
            synthetic: HashSet::new(),
        }
    }

    /// Returns `false` if `argo_name` was already collected (generated
    /// `collect` impls return early in that case — dedup + cycle guard).
    pub fn enter(&mut self, argo_name: &str) -> bool {
        self.seen.insert(argo_name.to_string())
    }

    /// Register a template's `build` fn (invoked lazily at emit).
    pub fn add_builder(&mut self, build: fn(&BuildCtx) -> api::Template) {
        self.builders.push(build);
    }

    /// Register a template by type: its `build` fn plus, if it sets
    /// `on_exit_if_root`, its exit handler keyed by Argo name (so
    /// `emit` can put `spec.hooks.exit` on *that* WorkflowTemplate).
    pub fn add<T: Template>(&mut self) {
        self.builders.push(T::build);
        if let Some(handler) = T::ON_EXIT {
            self.exits.insert(T::ARGO_NAME.to_string(), handler);
        }
        if let Some(t) = T::TTL {
            self.ttl.insert(T::ARGO_NAME.to_string(), t);
        }
        if let Some(s) = T::POD_GC {
            self.pod_gc.insert(T::ARGO_NAME.to_string(), s.to_string());
        }
        if let Some(s) = T::ACTIVE_DEADLINE_IF_ROOT {
            self.active_deadline.insert(T::ARGO_NAME.to_string(), s);
        }
        if !T::INPUT_TYPES.is_empty() {
            self.types.insert(T::ARGO_NAME.to_string(), T::INPUT_TYPES);
        }
        if T::SYNTHETIC {
            self.synthetic.insert(T::ARGO_NAME.to_string());
        }
    }

    pub fn add_runner(&mut self, argo_name: &str, run: fn(serde_json::Value) -> serde_json::Value) {
        self.runners.insert(argo_name.to_string(), run);
    }

    /// Emit the multi-document stream: one `WorkflowTemplate` per template
    /// plus a runnable `Workflow` for the entrypoint `E`. Builds the
    /// `BuildCtx` (and reads `athena.toml`) here — emit only.
    /// Emit the multi-doc YAML. `with_workflow` appends a convenience
    /// runnable `Workflow` (`generateName`, `workflowTemplateRef` →
    /// root) — off by default: the deterministic, stable-named
    /// `WorkflowTemplate`s are the artifact you register/GitOps, and
    /// runs are triggered with `argo submit --from
    /// workflowtemplate/<root>`. The convenience Workflow is opt-in for
    /// quick demos / `kubectl create -f -`.
    /// The deterministic `WorkflowTemplate` set `emit` serializes —
    /// every reachable template, sorted, with each `on_exit_if_root`
    /// hook stamped on its own template. Shared by YAML emit and the
    /// `CARGO_ATHENA_EMIT_JSON` mode `cargo athena submit` consumes for
    /// its register/drift checks.
    pub fn build_templates(&self) -> Vec<api::WorkflowTemplate> {
        let ctx = BuildCtx::collect();
        let mut tpls: Vec<api::WorkflowTemplate> = self
            .builders
            .iter()
            .map(|b| {
                let inner = b(&ctx);
                wrap_workflow_template(inner.name.clone(), inner)
            })
            .collect();
        tpls.sort_by_key(name_of);

        // `on_exit_if_root`: EVERY template that declares it carries the
        // exit hook on its OWN WorkflowTemplate's `spec.hooks.exit`
        // (`templateRef` — the legacy `spec.onExit` name-string can't
        // cross the one-WT-per-template model). Argo only fires the hook
        // of the workflow actually submitted (workflow-scoped, proven on
        // real Argo v4.0.5 via both `argo submit --from` and a
        // `workflowTemplateRef` Workflow); a templateRef'd sub-workflow's
        // own hook stays inert when nested — but submit that sub-WT
        // directly and its own hook fires.
        for t in tpls.iter_mut() {
            let name = name_of(t);
            if let Some(handler) = self.exits.get(&name)
                && let Some(spec) = t.spec.as_mut()
            {
                spec.hooks.insert(
                    "exit".to_string(),
                    api::LifecycleHook {
                        template_ref: Some(api::TemplateRef {
                            name: handler.to_string(),
                            template: handler.to_string(),
                            cluster_scope: false,
                        }),
                        ..Default::default()
                    },
                );
            }
            // `ttl(..)`/`pod_gc(..)` are WorkflowSpec-scoped: stamped on
            // each declaring template's OWN WorkflowTemplate (Argo runs
            // GC workflow-scoped, same per-WT model as the exit hook).
            if let Some(ttl) = self.ttl.get(&name)
                && let Some(spec) = t.spec.as_mut()
            {
                spec.ttl_strategy = Some(ttl.clone());
            }
            if let Some(s) = self.pod_gc.get(&name)
                && let Some(spec) = t.spec.as_mut()
            {
                spec.pod_gc = Some(api::PodGc {
                    strategy: s.clone(),
                });
            }
            // `active_deadline_if_root` is the genuine whole-workflow
            // runtime cap (Argo applies `Template.timeout` /
            // `Template.activeDeadlineSeconds` to neither dag nor steps);
            // root-only, same per-WT plumbing as `ttl`/`pod_gc`.
            if let Some(s) = self.active_deadline.get(&name)
                && let Some(spec) = t.spec.as_mut()
            {
                spec.active_deadline_seconds = Some(*s);
            }
        }
        tpls
    }

    pub fn emit<E: Template>(&self, with_workflow: bool) -> String {
        let tpls = self.build_templates();
        let mut docs: Vec<String> = tpls
            .iter()
            .map(|t| serde_norway::to_string(t).expect("WorkflowTemplate is serializable"))
            .collect();

        if !with_workflow {
            return docs.join("---\n");
        }

        let ctx = BuildCtx::collect();
        let wf = api::Workflow {
            api_version: api::API_VERSION.to_string(),
            kind: api::KIND_WORKFLOW.to_string(),
            metadata: Some(api::ObjectMeta {
                generate_name: format!("{}-", E::ARGO_NAME),
                ..Default::default()
            }),
            spec: Some(api::WorkflowSpec {
                workflow_template_ref: Some(api::WorkflowTemplateRef {
                    name: E::ARGO_NAME.to_string(),
                    cluster_scope: false,
                }),
                service_account_name: ctx.config().defaults.service_account.clone(),
                // Only the emit root's on_exit reaches a runnable Workflow,
                // and it must be `hooks.exit` with a templateRef (the
                // legacy `spec.onExit` name-string can't cross the
                // one-WT-per-template wormhole).
                hooks: E::ON_EXIT
                    .map(|n| {
                        let mut m = std::collections::BTreeMap::new();
                        m.insert(
                            "exit".to_string(),
                            api::LifecycleHook {
                                template_ref: Some(api::TemplateRef {
                                    name: n.to_string(),
                                    template: n.to_string(),
                                    cluster_scope: false,
                                }),
                                ..Default::default()
                            },
                        );
                        m
                    })
                    .unwrap_or_default(),
                // Only the emit root's ttl/pod_gc/active_deadline reaches
                // the runnable Workflow (same workflow-scoped reasoning
                // as the hook).
                ttl_strategy: E::TTL,
                active_deadline_seconds: E::ACTIVE_DEADLINE_IF_ROOT,
                pod_gc: E::POD_GC.map(|s| api::PodGc {
                    strategy: s.to_string(),
                }),
                ..Default::default()
            }),
        };
        docs.push(serde_norway::to_string(&wf).expect("Workflow is serializable"));
        docs.join("---\n")
    }
}

fn name_of(t: &api::WorkflowTemplate) -> String {
    t.metadata
        .as_ref()
        .map(|m| m.name.clone())
        .unwrap_or_default()
}

/// Wrap one inner Argo `template` as a standalone `WorkflowTemplate` whose
/// resource name == the inner template name == its entrypoint.
pub fn wrap_workflow_template(name: String, inner: api::Template) -> api::WorkflowTemplate {
    api::WorkflowTemplate {
        api_version: api::API_VERSION.to_string(),
        kind: api::KIND_WORKFLOW_TEMPLATE.to_string(),
        metadata: Some(api::ObjectMeta {
            name: name.clone(),
            ..Default::default()
        }),
        spec: Some(api::WorkflowSpec {
            entrypoint: name,
            templates: vec![inner],
            arguments: None,
            workflow_template_ref: None,
            ..Default::default()
        }),
    }
}

/// kebab-case an Argo identifier (DNS-1123-ish) from a Rust ident.
pub fn kebab(s: &str) -> String {
    s.replace('_', "-").to_ascii_lowercase()
}

fn volume_name(path: &str) -> String {
    let mut n: String = path
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() { c } else { '-' })
        .collect();
    n = n.trim_matches('-').to_string();
    if n.is_empty() {
        n.push('v');
    }
    format!("host-{n}")
}

/// `volumes` + `volumeMounts` for a set of hostPaths (from `host!`).
pub fn host_path_volumes(paths: &[String]) -> (Vec<api::Volume>, Vec<api::VolumeMount>) {
    let mut vols = Vec::new();
    let mut mounts = Vec::new();
    for p in paths {
        let name = volume_name(p);
        vols.push(api::Volume {
            name: name.clone(),
            host_path: Some(api::HostPathVolumeSource {
                path: p.clone(),
                r#type: String::new(),
            }),
            ..Default::default()
        });
        mounts.push(api::VolumeMount {
            name,
            mount_path: p.clone(),
            read_only: false,
        });
    }
    (vols, mounts)
}

/// Every container template's volumes/mounts: the always-present
/// `emptyDir` scratch at [`ATHENA_DIR`] (binary tarball, in/out artifact
/// ports, extraction, result — all writable on any image) followed by the
/// declared hostPaths.
pub fn container_volumes(host_paths: &[String]) -> (Vec<api::Volume>, Vec<api::VolumeMount>) {
    let mut vols = vec![api::Volume {
        name: SCRATCH_VOLUME.to_string(),
        empty_dir: Some(api::EmptyDirVolumeSource {}),
        ..Default::default()
    }];
    let mut mounts = vec![api::VolumeMount {
        name: SCRATCH_VOLUME.to_string(),
        mount_path: ATHENA_DIR.to_string(),
        read_only: false,
    }];
    let (hv, hm) = host_path_volumes(host_paths);
    vols.extend(hv);
    mounts.extend(hm);
    (vols, mounts)
}

/// The entrypoint a user's `main` calls, parameterised by the root
/// workflow type. Referencing `E` force-links the entire reachable
/// closure (each `collect` calls callees' `collect` directly).
pub fn entrypoint<E: Template>() {
    let mut collector = Collector::new();
    E::collect(&mut collector);

    let args: Vec<String> = std::env::args().collect();
    let target = args
        .iter()
        .position(|a| a == "--cargo-athena-template")
        .and_then(|i| args.get(i + 1).cloned())
        .or_else(|| std::env::var("CARGO_ATHENA_TEMPLATE").ok());

    if let Some(t) = target {
        let run = *collector
            .runners
            .get(&t)
            .unwrap_or_else(|| panic!("no runnable container template named {t:?}"));
        // Base inputs: `cargo athena run --input` (local), then overlay
        // Argo-supplied params delivered as `ATHENA_PARAM_<name>` env vars
        // (set on the container template from `{{inputs.parameters.*}}`).
        let mut input = std::env::var("CARGO_ATHENA_INPUT")
            .ok()
            .map(|s| serde_json::from_str(&s).expect("CARGO_ATHENA_INPUT must be JSON"))
            .unwrap_or(serde_json::Value::Object(Default::default()));
        if let serde_json::Value::Object(map) = &mut input {
            for (k, v) in std::env::vars() {
                if let Some(name) = k.strip_prefix("ATHENA_PARAM_") {
                    // A param is JSON if it parses, else a bare string.
                    let val =
                        serde_json::from_str(&v).unwrap_or(serde_json::Value::String(v.clone()));
                    map.insert(name.to_string(), val);
                }
            }
        }
        let output = run(input);
        if let Ok(path) = std::env::var("CARGO_ATHENA_OUTPUT") {
            std::fs::write(path, output.to_string()).expect("write CARGO_ATHENA_OUTPUT");
        } else {
            println!("{output}");
        }
        return;
    }

    // `cargo athena container ls` sets this to enumerate every reachable
    // template's metadata as a JSON array (same per-template derivation
    // as describe — so names/params shown are exactly what runs).
    if std::env::var_os("CARGO_ATHENA_LIST").is_some() {
        let ctx = BuildCtx::collect();
        let all: Vec<ContainerRunMeta> = collector
            .builders
            .iter()
            .map(|b| {
                let t = b(&ctx);
                let it = collector.types.get(&t.name).copied().unwrap_or(&[]);
                let mut m = ContainerRunMeta::from_template(&t, it);
                m.synthetic = collector.synthetic.contains(&t.name);
                m
            })
            .collect();
        println!(
            "{}",
            serde_json::to_string(&all).expect("ContainerRunMeta is serializable")
        );
        return;
    }

    // `cargo athena container emulate/describe` sets this to fetch ONE
    // template's metadata as JSON (it then realizes that exact spec
    // locally via docker/podman). Reusing `Template::build` here is what
    // makes the local run identical to Argo by construction — same
    // image, bootstrap, env, volumes, and artifacts as `emit`.
    if let Ok(name) = std::env::var("CARGO_ATHENA_DESCRIBE") {
        let ctx = BuildCtx::collect();
        let tpl = collector
            .builders
            .iter()
            .map(|b| b(&ctx))
            .find(|t| t.name == name)
            .unwrap_or_else(|| panic!("no template named {name:?}"));
        let input_types = collector.types.get(&name).copied().unwrap_or(&[]);
        let mut meta = ContainerRunMeta::from_template(&tpl, input_types);
        meta.synthetic = collector.synthetic.contains(&name);
        println!(
            "{}",
            serde_json::to_string(&meta).expect("ContainerRunMeta is serializable")
        );
        return;
    }

    // `cargo athena submit` sets this to get the deterministic
    // `WorkflowTemplate` set as a JSON array (structured — for its
    // register-if-missing / drift-detect / apply checks), instead of
    // re-parsing the YAML `emit` prints.
    if std::env::var_os("CARGO_ATHENA_EMIT_JSON").is_some() {
        println!(
            "{}",
            serde_json::to_string(&collector.build_templates())
                .expect("WorkflowTemplate is serializable")
        );
        return;
    }

    // `cargo athena emit --with-workflow` sets this on the child so the
    // convenience runnable Workflow is appended (default: templates
    // only — deterministic, `kubectl apply`-able, GitOps-clean).
    let with_workflow = std::env::var_os("CARGO_ATHENA_WITH_WORKFLOW").is_some_and(|v| v == "1");
    print!("{}", collector.emit::<E>(with_workflow));
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn kebab_lowercases_and_hyphenates() {
        assert_eq!(kebab("run_a_container"), "run-a-container");
        assert_eq!(kebab("RunFoo"), "runfoo");
    }

    #[test]
    fn volume_name_is_dns_safe() {
        assert_eq!(volume_name("/etc/myapp"), "host-etc-myapp");
        assert_eq!(volume_name("/var/lib/extra"), "host-var-lib-extra");
    }
}