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 `cargo_athena::entrypoint!(E)` (which calls
//!   [`entrypoint_impl::<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] {}

/// Error side of a `.continue_on(..)` binding.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ArgoError {
    /// Terminal Argo node phase (`"Failed"` / `"Error"`).
    pub status: String,
    /// Main-container exit code, when Argo reported one.
    pub exit_code: Option<i32>,
}

/// Ghost-only: types a `.continue_on` binding as `Result`.
#[doc(hidden)]
pub fn __athena_fallible<T>(_: T) -> Result<T, ArgoError> {
    unimplemented!("athena ghost: never executed")
}

/// 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 a literal `Path::new(...)` carrying
/// the precomputed mount path. This `macro_rules!` form is what
/// rust-analyzer / rustc see *before* the attribute macro expands — it
/// fires `compile_error!` as the misuse gate AND keeps a type-correct
/// stub expression so LSP hover / inlay hints / autocompletion on the
/// return value all work at the call site even when the attribute
/// macro hasn't expanded yet. The stub is never reached at runtime:
/// inside `#[container]`/`#[fragment]` the attribute macro replaces
/// this whole expression first; outside those, the `compile_error!`
/// blocks the build before any code runs.
#[macro_export]
macro_rules! host {
    ($path:literal) => {{
        ::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)"
        );
        ::std::path::Path::new($path)
    }};
    ($($t:tt)*) => {
        ::core::compile_error!("`host!` takes a single string-literal path")
    };
}

// None of the decl macros (`host!`, `load_artifact*!`, `save_artifact*!`,
// `secret*!`) have a private declarative form. The `#[container]` /
// `#[fragment]` attribute macros' `DeclRewrite` pass replaces every
// recognized invocation at the expression level with a direct call to
// the relevant `rt::*` helper, passing pre-baked strings (mount path,
// artifact file path, env var name) computed once at proc-macro
// expansion time. Invocations the attribute macro can't see — plain
// fns, `#[workflow]` bodies, nested inside other macro invocations —
// hit the public `compile_error!` gates below.

/// Declare an Argo *input* artifact port and read it (bytes) at
/// runtime. See the [`host!`] doc for the gate / LSP-stub pattern
/// shared by every decl macro.
#[macro_export]
macro_rules! load_artifact {
    ($name:literal) => {{
        ::core::compile_error!(
            "`load_artifact!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        );
        ::std::vec::Vec::<u8>::new()
    }};
    ($($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 {
    ($name:literal) => {{
        ::core::compile_error!(
            "`load_artifact_str!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        );
        ::std::string::String::new()
    }};
    ($($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 {
    ($name:literal, $data:expr) => {{
        ::core::compile_error!(
            "`save_artifact!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        );
        // Force the data expression through `AsRef<[u8]>` so LSP
        // catches bad-type passes at the call site (e.g. handing an
        // `i32` to a bytes-port).
        let _: &dyn ::core::convert::AsRef<[u8]> = &$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 {
    ($name:literal, $data:expr) => {{
        ::core::compile_error!(
            "`save_artifact_str!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        );
        let _: &dyn ::core::convert::AsRef<str> = &$data;
    }};
    ($($t:tt)*) => {
        ::core::compile_error!("save_artifact_str!(\"name\", data)")
    };
}

/// Declare a K8s Secret-sourced env var and read it back at runtime as
/// a `String`. Two literal args: `(secret_name, key)`. Panics at
/// runtime if the env var the macro plants isn't set; pair with
/// `secret_opt!` for the no-panic variant.
#[macro_export]
macro_rules! secret {
    ($name:literal, $key:literal) => {{
        ::core::compile_error!(
            "`secret!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        );
        ::std::string::String::new()
    }};
    ($($t:tt)*) => {
        ::core::compile_error!("secret!(\"secret-name\", \"key\")")
    };
}

/// Mount a PVC of type `T` (a unit struct that implements
/// [`Pvc`] via `#[ephemeral_pvc]` or `#[external_pvc]`) on this
/// container's pod, and evaluate to the (already-mounted) path —
/// `&'static Path`, picked deterministically from the type's
/// [`Pvc::MOUNT_PATH`] so emit-side and in-pod always agree.
///
/// Same gating as [`host!`]: only valid inside a
/// `#[cargo_athena::container]` or `#[cargo_athena::fragment]`. The
/// `macro_rules!` form below emits `compile_error!` AND a
/// type-correct stub so rust-analyzer can still infer `&'static Path`
/// at the call site before the attribute macro expands.
#[macro_export]
macro_rules! pvc {
    ($t:path) => {{
        ::core::compile_error!(
            "`pvc!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        );
        ::std::path::Path::new(<$t as $crate::Pvc>::MOUNT_PATH)
    }};
    ($($t:tt)*) => {
        ::core::compile_error!("`pvc!` takes a single PVC type: pvc!(MyPvc)")
    };
}

/// Same as [`secret!`] but returns `Option<String>` and emits
/// `optional: true` on the Argo `secretKeyRef` — Argo skips the env
/// entry instead of failing pod-start when the secret/key is missing.
#[macro_export]
macro_rules! secret_opt {
    ($name:literal, $key:literal) => {{
        ::core::compile_error!(
            "`secret_opt!` may only be used directly inside a \
             `#[cargo_athena::container]` or `#[cargo_athena::fragment]` fn"
        );
        ::core::option::Option::<::std::string::String>::None
    }};
    ($($t:tt)*) => {
        ::core::compile_error!("secret_opt!(\"secret-name\", \"key\")")
    };
}

/// Runtime shims referenced by the declaration macros. Artifact ports
/// are plain files at fixed paths; Argo moves them (no S3 from us).
///
/// Every `&str` argument is **pre-baked at proc-macro expansion time**:
/// the macros precompute the full artifact path / env var name from
/// the user's literal and pass it as a string literal. So these
/// helpers do no string-building of their own — just one I/O call plus
/// a panic on failure. The original-name args are kept purely for the
/// panic message context.
pub mod rt {
    // The host!, load_artifact!/_str, save_artifact!/_str, secret!,
    // and secret_opt! macros all precompute their derived strings at
    // expansion time and pass them through here. `super::host_mount
    // _path` and `super::secret_env_name` remain the single sources of
    // truth for the formulas (called once at emit time per template);
    // their proc-macro mirrors are pinned by the algorithm tests.

    // Where Argo drops/collects declared artifact ports inside the pod
    // — emit-side formats Argo template artifact paths from these, and
    // the proc macros bake them into `load_artifact!` / `save_artifact!`
    // call sites. Single source of truth in `api::munge`; re-exported
    // here so every existing `rt::IN_DIR` / `rt::OUT_DIR` caller keeps
    // resolving.
    pub use cargo_athena_api::munge::{ATHENA_IN_DIR as IN_DIR, ATHENA_OUT_DIR as OUT_DIR};

    pub fn load_artifact(path: &str, name: &str) -> Vec<u8> {
        std::fs::read(path)
            .unwrap_or_else(|e| panic!("athena: load_artifact!({name:?}) {path}: {e}"))
    }

    pub fn load_artifact_str(path: &str, name: &str) -> String {
        std::fs::read_to_string(path)
            .unwrap_or_else(|e| panic!("athena: load_artifact_str!({name:?}) {path}: {e}"))
    }

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

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

    /// `secret!(name, key)` runtime: read the precomputed env var. The
    /// macro passes `name`/`key` for the panic message context only;
    /// the env var name itself was already munged at expand time.
    pub fn secret_value(env_var: &str, name: &str, key: &str) -> String {
        std::env::var(env_var).unwrap_or_else(|e| {
            panic!("athena: secret!({name:?}, {key:?}) env var `{env_var}` missing: {e}")
        })
    }

    /// `secret_opt!(name, key)` runtime: `None` when the env var is
    /// unset (e.g. Argo skipped the entry because the secret/key is
    /// missing and `optional: true`). Pure `std::env::var().ok()` —
    /// kept here so the macros have a stable call target.
    pub fn secret_value_opt(env_var: &str) -> Option<String> {
        std::env::var(env_var).ok()
    }
}

// Re-export the shared name/path/env-var derivers from `api::munge`.
// Single source of truth — the proc-macro side imports them too.
pub use crate::api::munge::secret_env_name;

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

/// How a single Argo input or output flows: inline as a `parameter` (Argo
/// stores it in workflow status, sized like a JSON parameter) or via S3
/// as an `artifact` (DAG-wired via `outputs.artifacts` /
/// `arguments.artifacts.from`). The `#[container]` and `#[workflow]`
/// macros derive these per-slot from the function signature: a fn
/// argument or return type of `cargo_athena::Artifact<T>` is `Artifact`,
/// everything else is `Parameter`. Stamped into
/// [`Template::INPUT_KINDS`] / [`Template::OUTPUT_KIND`] and read at
/// emit-time by the workflow's per-task wiring (parallel to how
/// [`Template::INPUTS`] is read for parameter names).
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum IoKind {
    Parameter,
    Artifact,
}

/// A DAG-wired S3-backed value. Wrap a `#[container]` (or `#[workflow]`)
/// return in `Artifact<T>` to flow the value via Argo's native artifact
/// passing (`outputs.artifacts.return` + `arguments.artifacts.from`)
/// instead of the inline-parameter path (`outputs.parameters.return`).
/// Lifts the parameter-size ceiling and is the natural shape for
/// binary/large payloads. A consumer accepting `Artifact<T>` reads the
/// same value on the other side; the wire is the user's serialized `T`
/// (JSON, tar+gzip'd by Argo on the producer and untarred on the
/// consumer, transparent to user code).
///
/// The inner `T` is private by design: no field-access (`a.field`) on
/// an `Artifact<T>` binding, no `Deref<Target=T>`, no `AthenaList<_>`
/// impl. Those constraints fall out of Rust's own type rules in the
/// `#[workflow]` ghost (see `feedback-ghost-first.md` in agent
/// memory) — they are NOT enforced by bespoke macro checks. The only
/// public surface is `Artifact::new` / `Artifact::into_inner`.
#[derive(Debug)]
pub struct Artifact<T> {
    inner: T,
}

impl<T> Artifact<T> {
    pub fn new(v: T) -> Self {
        Self { inner: v }
    }

    pub fn into_inner(self) -> T {
        self.inner
    }
}

// `.clone()` in a `#[workflow]` body is the fan-out marker that lets
// the same producer feed multiple consumers (see WORKFLOW.md). The
// ghost type-checks it as a real `Clone` call so the binding has to
// implement `Clone`; the macro lowers it as "reference the same
// upstream task again", so no runtime clone happens in-pod.
impl<T: Clone> Clone for Artifact<T> {
    fn clone(&self) -> Self {
        Self::new(self.inner.clone())
    }
}

// Serde-transparent: the wire form is plain serialized `T`. The wrapper
// is purely a Rust type marker that the macros key on; on disk it is
// indistinguishable from `T` itself. Lets `#[container]`'s `run()` body
// write `serde_json::to_writer(File::create(...), &val)` after
// `.into_inner()` without any wrapper bytes leaking through.
impl<T: ::serde::Serialize> ::serde::Serialize for Artifact<T> {
    fn serialize<S: ::serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        self.inner.serialize(s)
    }
}

impl<'de, T: ::serde::Deserialize<'de>> ::serde::Deserialize<'de> for Artifact<T> {
    fn deserialize<D: ::serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        T::deserialize(d).map(Self::new)
    }
}

/// 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
    /// `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] = &[];
    /// Per-input I/O kind, parallel to [`Self::INPUTS`]. An empty slice
    /// (the backwards-compat default) means "all parameters" — what
    /// every template did before [`Artifact`] landed. The
    /// `#[container]` / `#[workflow]` macros set entries to
    /// [`IoKind::Artifact`] for any argument typed `Artifact<T>` and
    /// [`IoKind::Parameter`] for everything else. The workflow's
    /// per-task wiring reads this at emit-time to decide between
    /// `arguments.parameters[].value` and `arguments.artifacts[].from`.
    const INPUT_KINDS: &'static [IoKind] = &[];
    /// I/O kind of the function's return value. Default
    /// [`IoKind::Parameter`] keeps every existing template's
    /// `outputs.parameters.return` emission byte-identical; macros set
    /// this to [`IoKind::Artifact`] when the return type is
    /// `Artifact<T>` so the template emits `outputs.artifacts.return`
    /// (S3-backed) instead.
    const OUTPUT_KIND: IoKind = IoKind::Parameter;
    /// `true` for athena-synthesized templates (the `if`/`else`
    /// wrapper + per-arm sub-workflows). They're an implementation
    /// detail, so `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;
    /// Root-only `WorkflowSpec.nodeSelector`, from
    /// `#[workflow(node_selector_if_root = { … })]`. Argo's pod-build
    /// lookup is 3-tier: `tmpl.NodeSelector → boundary.NodeSelector →
    /// wfSpec.NodeSelector`, so this is the only knob that lands on
    /// EVERY pod in the run unless that pod (or its immediate enclosing
    /// dag/steps) overrides it (verified from `workflow/controller/
    /// workflowpod.go:928-958`). Same per-WT plumbing as
    /// `TTL`/`POD_GC`/`ACTIVE_DEADLINE_IF_ROOT`. Literal pairs only —
    /// workflow attrs have no injectable args (see `#[workflow(node
    /// _selector)]` for the rationale).
    const NODE_SELECTOR_IF_ROOT: &'static [(&'static str, &'static str)] = &[];
    /// Root-only `WorkflowSpec.synchronization.mutexes`, from
    /// `#[…(mutexes_if_root = [{ name = …, namespace = … }])]`. Each
    /// entry is `(name, namespace)`, already lowered to its final YAML
    /// string form (literal, or `{{=fromJSON(workflow.parameters[…])}}`
    /// for injected operands). `namespace == ""` ⇒ skip the field
    /// (Argo defaults to the workflow's own namespace per
    /// `workflow/sync/lock_name.go:58-67`). Same per-WT, root-only
    /// plumbing as `TTL` / `POD_GC` / `NODE_SELECTOR_IF_ROOT`; Argo's
    /// sync manager keys on `<ns>/Mutex/<name>` globally so two
    /// SEPARATE Workflow runs contend on the same name (empirically
    /// verified on v4.0.5 2026-05-25, holder key `<ns>/<wf>`).
    const MUTEXES_IF_ROOT: &'static [(&'static str, &'static str)] = &[];
    /// Root-only `WorkflowSpec.Tolerations` from
    /// `#[…(tolerations_if_root = [...])]`. Each entry is `(key,
    /// operator, value, effect, toleration_seconds)`. Strings are
    /// already lowered (literal verbatim, or
    /// `{{=fromJSON(workflow.parameters[..])}}` for injected operands).
    /// `toleration_seconds == 0` ⇒ Argo skip-serializes (treated as
    /// "unset" by k8s, applies forever).
    const TOLERATIONS_IF_ROOT: &'static [(
        &'static str,
        &'static str,
        &'static str,
        &'static str,
        i64,
    )] = &[];
    /// Root-only `WorkflowSpec.Affinity` from
    /// `#[…(affinity_if_root = "...")]` as an opaque YAML/JSON string.
    /// Parsed at emit time; user owns the schema (athena does NOT
    /// validate). None ⇒ skip.
    const AFFINITY_IF_ROOT: ::core::option::Option<&'static str> = None;
    /// Root-only `WorkflowSpec.PodSpecPatch`, from
    /// `#[workflow(pod_spec_patch_if_root = "...")]`. Already lowered
    /// to its final string form (literal verbatim, or with
    /// `{{=fromJSON(workflow.parameters[..])}}` operands spliced in
    /// for injected pieces). Same per-WT, root-only plumbing as
    /// `NODE_SELECTOR_IF_ROOT`. None ⇒ skip (existing goldens
    /// unaffected).
    const POD_SPEC_PATCH_IF_ROOT: ::core::option::Option<&'static str> = None;
    /// Root-only `WorkflowSpec.ImagePullSecrets` (Secret names) from
    /// `#[…(image_pull_secrets_if_root = ["regcred", ...])]`. K8s/
    /// Argo expose this only at workflow scope (no per-template
    /// knob); per-container needs go through `pod_spec_patch`. Same
    /// per-WT, root-only plumbing as `MUTEXES_IF_ROOT`.
    const IMAGE_PULL_SECRETS_IF_ROOT: &'static [&'static str] = &[];
    /// Root-only `WorkflowSpec.parallelism` (cap on total concurrent
    /// pods in the run), from `#[workflow(parallelism_if_root = N)]`.
    /// Same per-WT, root-only plumbing as the other `_IF_ROOT` consts.
    /// Template-level `Template.parallelism` from
    /// `#[workflow(parallelism = N)]` does NOT use this const — it's
    /// stamped directly into the `api::Template` literal by `build()`.
    const PARALLELISM_IF_ROOT: ::core::option::Option<i64> = None;

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

    /// Run-mode body. Argv is the function's positional parameters in
    /// `INPUTS` order, each JSON-encoded (string -> `"v"`, number/bool
    /// bare). Returns the function's JSON-encoded return value, which
    /// the entrypoint writes to `CARGO_ATHENA_OUTPUT`. Overridden by
    /// `#[container]`; never called on a `#[workflow]`.
    fn run(_argv: &[String]) -> String {
        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);
}

/// PVC lifecycle, exposed via [`Pvc::LIFECYCLE`].
///
/// * `Ephemeral` — athena emits a `WorkflowSpec.volumeClaimTemplates[]`
///   entry for this PVC on the submitted root; Argo creates a fresh
///   PVC at workflow start and deletes it at workflow end. The Pod
///   `claimName` is the [`Pvc::ARGO_NAME`].
/// * `External` — pre-existing PVC; athena emits nothing at the
///   workflow-spec level, just per-pod `volumes` + `volumeMounts`. The
///   Pod `claimName` is [`Pvc::CLAIM_NAME`] (the existing PVC's name
///   in the workflow's namespace).
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum PvcLifecycle {
    Ephemeral,
    External,
}

/// The cross-crate identity of a PVC, implemented by the unit struct
/// `#[ephemeral_pvc]` / `#[external_pvc]` generates.
///
/// Same wormhole pattern as [`Template`]: callers reference the type
/// (`<BuildCache as Pvc>::ARGO_NAME`), so name resolution is done by
/// the compiler — collision-proof across crates, and the reference
/// itself force-links the defining crate (incl. its `inventory::
/// submit!` of [`PvcReg`], which the emit-side reads to materialize
/// the full PVC spec by name).
///
/// Many of the consts below are lifecycle-conditional: `Ephemeral`
/// uses `SIZE`/`ACCESS_MODES`/`STORAGE_CLASS_NAME`; `External` uses
/// `CLAIM_NAME`/`READ_ONLY`. The attribute macros fill in only the
/// fields meaningful for the chosen lifecycle and leave the rest at
/// their `""`/empty defaults.
pub trait Pvc {
    /// Globally-unique Argo resource name (`<crate>-<type-kebab>` by
    /// default). Identifies the PVC across the wormhole AND drives
    /// the in-pod mount path hash ([`MOUNT_PATH`](Self::MOUNT_PATH))
    /// so emit and run agree.
    const ARGO_NAME: &'static str;
    /// `Ephemeral` or `External`.
    const LIFECYCLE: PvcLifecycle;
    /// `/athena/pvcs/<fnv-hash-of-ARGO_NAME>`. Stable across emit and
    /// run by construction (same hash both sides). Hidden from users:
    /// `pvc!(Type)` returns `&'static Path` pointing here.
    const MOUNT_PATH: &'static str;
    /// `Ephemeral`: storage request like `"10Gi"`. `""` for external.
    const SIZE: &'static str = "";
    /// `Ephemeral`: K8s `accessModes`, e.g.
    /// `&["ReadWriteOnce"]` or `&["ReadWriteMany"]`. `&[]` for external.
    const ACCESS_MODES: &'static [&'static str] = &[];
    /// `Ephemeral`: K8s `StorageClassName`. `""` = use cluster default.
    const STORAGE_CLASS_NAME: &'static str = "";
    /// `External`: the existing PVC's name in the workflow's
    /// namespace. `""` for ephemeral (Argo creates with `ARGO_NAME`).
    const CLAIM_NAME: &'static str = "";
    /// `External`: mount this PVC read-only on every consumer. Bakes
    /// into the volume source, so it's all-or-nothing per declaration.
    /// `false` for ephemeral.
    const READ_ONLY: bool = false;
}

/// Inventory-registered PVC spec, submitted by every `#[ephemeral_pvc]`
/// and `#[external_pvc]`. [`BuildCtx::collect`] loads them all into a
/// `argo_name → PvcReg` map so emit-side code can materialize the full
/// PVC spec from just a name string (which is what fragment closures
/// propagate).
pub struct PvcReg {
    pub argo_name: &'static str,
    pub mount_path: &'static str,
    pub lifecycle: PvcLifecycle,
    pub size: &'static str,
    pub access_modes: &'static [&'static str],
    pub storage_class_name: &'static str,
    pub claim_name: &'static str,
    pub read_only: bool,
}
inventory::collect!(PvcReg);

// PVC name / mount-path / volume-name derivers live in
// `api::munge` — re-exported so existing callers
// (`cargo_athena_core::pvc_mount_path` etc.) keep working.
pub use crate::api::munge::{pvc_mount_path, pvc_volume_name};

// ---- 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,
    #[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 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(),
    ]
}

/// Shown when no `athena.toml` can be located anywhere.
const CONFIG_NOT_FOUND: &str = "athena.toml not found. Run from your workflow \
    crate (or a parent dir), set $ATHENA_CONFIG, pass --config <FILE>, or create \
    a global config at ~/.config/cargo-athena/athena.toml (see `cargo athena init`).";

impl AthenaConfig {
    /// Resolve *which* `athena.toml` to load, in priority order:
    ///   1. `flag` (the `--config` flag),
    ///   2. `env_config` (`$ATHENA_CONFIG`),
    ///   3. the nearest `athena.toml` walking up from `cwd` (repo/dev mode),
    ///   4. `<xdg_dir>/athena.toml` (a global config, for source-free use).
    ///
    /// Pure: every input is passed in (no env / cwd reads of its own), so
    /// the ordering is unit-testable. `flag` / `env_config` are returned
    /// unconditionally (explicit intent; existence is checked at load);
    /// the walk-up and global candidates are returned only if the file
    /// exists. `xdg_dir` is supplied by the CLI (resolved via `etcetera`);
    /// the in-binary [`load`](Self::load) passes `None` and relies on the
    /// CLI having resolved + exported `ATHENA_CONFIG` first.
    pub fn resolve_config_path(
        flag: Option<&std::path::Path>,
        env_config: Option<&std::path::Path>,
        cwd: &std::path::Path,
        xdg_dir: Option<&std::path::Path>,
    ) -> Option<std::path::PathBuf> {
        if let Some(p) = flag {
            return Some(p.to_path_buf());
        }
        // A set-but-empty `$ATHENA_CONFIG` (`export ATHENA_CONFIG=`, a blank
        // `ATHENA_CONFIG=` in CI) is treated as unset, so it falls through
        // to walk-up / global rather than shadowing them and panicking on an
        // empty path later. Empty would otherwise silently defeat the global
        // fallback this resolver exists to provide.
        if let Some(p) = env_config.filter(|p| !p.as_os_str().is_empty()) {
            return Some(p.to_path_buf());
        }
        if let Some(p) = Self::find_upwards_from(cwd) {
            return Some(p);
        }
        if let Some(dir) = xdg_dir {
            let p = dir.join("athena.toml");
            if p.is_file() {
                return Some(p);
            }
        }
        None
    }

    /// `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`. The global (`~/.config`)
    /// fallback lives in the CLI, which resolves the effective path and
    /// exports `ATHENA_CONFIG` before this runs, so the same file loads
    /// here whether invoked in-process or in the spawned user binary.
    pub fn load() -> Self {
        Self::try_load().unwrap_or_else(|e| panic!("{e}"))
    }

    /// Fallible variant of [`Self::load`] for CLI callers: a missing or
    /// malformed `athena.toml` is a routine user mistake there, so it
    /// deserves a clean error message, not a panic with a backtrace
    /// hint. The panicking `load` stays for the in-binary emit path,
    /// where fail-loud is the documented contract.
    pub fn try_load() -> Result<Self, String> {
        let env_config = std::env::var_os("ATHENA_CONFIG").map(std::path::PathBuf::from);
        let cwd = std::env::current_dir().unwrap_or_default();
        let path = Self::resolve_config_path(None, env_config.as_deref(), &cwd, None)
            .ok_or_else(|| CONFIG_NOT_FOUND.to_string())?;
        let text =
            std::fs::read_to_string(&path).map_err(|e| format!("read {}: {e}", path.display()))?;
        toml::from_str(&text).map_err(|e| format!("parse {}: {e}", path.display()))
    }

    fn find_upwards_from(start: &std::path::Path) -> Option<std::path::PathBuf> {
        let mut d = start.to_path_buf();
        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";
// In-pod roots for `host!` / `pvc!` mounts live in `api::munge`
// (proc-macro and emit both reference them). Re-exported here so
// every existing `cargo_athena_core::ATHENA_*_DIR` caller keeps
// working.
pub use crate::api::munge::{ATHENA_MOUNTS_DIR, ATHENA_PVCS_DIR};
/// Where a *parameter-output* `#[container]` body writes its serialized
/// return value (read by the template's
/// `outputs.parameters.return.valueFrom.path`). The bootstrap exports
/// this as `CARGO_ATHENA_OUTPUT` for parameter-output templates.
pub const ATHENA_RESULT_FILE: &str = "/athena/result";
/// Where an *artifact-output* `#[container]` body writes its serialized
/// return value (read by the template's `outputs.artifacts.return.path`,
/// then tar+gzip'd and uploaded by Argo's executor). The bootstrap
/// exports this as `CARGO_ATHENA_OUTPUT` for artifact-output templates;
/// the run-side body has one write site (`CARGO_ATHENA_OUTPUT`) and
/// stays kind-agnostic.
pub const ATHENA_RESULT_ARTIFACT_FILE: &str = "/athena/result-artifact";
/// Where an `Artifact<T>`-typed input lands inside the container, one
/// file per arg, named after the arg. The template declares
/// `inputs.artifacts[].path = "/athena/in/<name>"`; the run-side body
/// reads + deserializes from there.
pub const ATHENA_INPUT_ARTIFACT_DIR: &str = "/athena/in";
/// 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@@` / `@@OUTPUT_PATH@@` 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 the in-pod entrypoint reads to pick which container template
/// to run. Argv (positional, in `INPUTS` order) carries the function's
/// own parameters.
pub const CARGO_ATHENA_TEMPLATE_ENV: &str = "CARGO_ATHENA_TEMPLATE";
/// Env var the in-pod entrypoint reads for where to write the body's
/// serialized return value. Cross-file contract: `bootstrap.sh` exports
/// it (`export CARGO_ATHENA_OUTPUT=@@OUTPUT_PATH@@`) before exec'ing
/// the binary — keep the two in sync.
pub const CARGO_ATHENA_OUTPUT_ENV: &str = "CARGO_ATHENA_OUTPUT";

/// `true` if this binary is dispatching a container body (in-pod,
/// started by Argo via the emitted bootstrap); `false` for every other
/// mode (`cargo athena emit` / `ls` / `describe` / `submit`-emit-JSON).
///
/// Use this in `main()` to gate one-time setup you only want to fire
/// in-pod -- a tracing/OTLP subscriber, a metrics exporter, anything
/// that costs network or has side effects. Without the gate, those
/// would also fire on every local `cargo athena emit` etc. that spawns
/// the binary to introspect templates.
///
/// ```ignore
/// fn main() {
///     let _otel = cargo_athena::is_container_run().then(|| {
///         tracing_subscriber::fmt().init();
///         OtelFlushGuard::new()        // drops at end of main()
///     });
///     cargo_athena::entrypoint!(MyRoot);
/// }
/// ```
pub fn is_container_run() -> bool {
    std::env::var_os(CARGO_ATHENA_TEMPLATE_ENV).is_some()
}

/// Resolved S3 coordinates for one artifact (creds are supplied
/// locally, e.g. via AWS env vars - `cargo athena emulate` 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,
}

impl S3Ref {
    /// A concrete object reference at `key` against the repo's bucket /
    /// endpoint. The one place the `S3Repo` -> `S3Ref` field copy lives,
    /// so `publish`, `prune`, and `doctor` can't drift on the mapping.
    /// (Distinct from `S3Repo::s3_loc`, which targets Argo's
    /// `api::S3Artifact` and also emits the secret-key selectors.)
    pub fn from_repo(repo: &S3Repo, key: String) -> Self {
        S3Ref {
            endpoint: repo.endpoint.clone(),
            bucket: repo.bucket.clone(),
            region: repo.region.clone(),
            insecure: repo.insecure,
            key,
        }
    }
}

/// 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 and its stringified Rust type (`""` if unknown,
/// e.g. synthetic templates). The position in the enclosing `params`
/// vector is the positional argv slot the parameter occupies in-pod.
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
pub struct ParamRef {
    pub name: 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 emulate` to realize the spec under
/// docker/podman locally, and by `cargo athena describe`.
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
pub struct ContainerRunMeta {
    /// Argo template name (`<crate>-<fn>`).
    pub name: String,
    /// Source crate (CARGO_PKG_NAME of the user binary). Used for the
    /// `PACKAGE` column in `cargo athena ls`; an empty
    /// string means the caller didn't fill it in.
    #[serde(default)]
    pub package: String,
    /// `"container"`, `"workflow"`, or `"other"`.
    pub kind: String,
    /// athena-synthesized template (an `if`/`else` wrapper or arm) —
    /// `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>,
}

/// Wire-format version of the binary's metadata modes (PROBE / LIST /
/// DESCRIBE / EMIT_JSON). The `cargo athena` CLI checks this on PROBE
/// before trusting the other modes. Bump ONLY on a breaking change to
/// those JSON shapes, NOT on every release, so an older CLI keeps
/// working with a newer binary at the same protocol. Within one release
/// line, a `ProbeInfo` struct mismatch at a matching protocol means the
/// workflow binary's library and the CLI were built from different
/// cargo-athena versions (a dev path-dependency skew) — `BinarySource::probe`
/// names that precisely rather than failing an opaque deserialize.
pub const ATHENA_PROTOCOL: u32 = 1;

/// Fixed marker in a [`ProbeInfo`] response (the `kind` field). Lets a
/// consumer positively identify a cargo-athena binary and reject stray
/// stdout from a non-athena executable with a clear error, rather than an
/// opaque deserialization failure.
pub const ATHENA_PROBE_KIND: &str = "cargo_athena_probe";

/// What a cargo-athena binary reports in `CARGO_ATHENA_PROBE` mode: a
/// handshake the CLI uses to (a) confirm the executable really is a
/// cargo-athena binary, (b) detect CLI/binary version skew, and (c)
/// learn the default/root template (so the workflow argument is optional).
/// Produced with NO `athena.toml` (no `BuildCtx::collect`): works
/// source-free / config-free.
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
pub struct ProbeInfo {
    /// Always [`ATHENA_PROBE_KIND`]. A consumer checks this first to tell a
    /// real probe response apart from an unrelated binary's stdout.
    pub kind: String,
    /// Metadata wire-format version ([`ATHENA_PROTOCOL`]).
    pub athena_protocol: u32,
    /// cargo-athena toolchain version this binary was built against (for
    /// skew diagnostics; distinct from the user crate `version` below).
    pub athena_version: String,
    /// The binary's root template (`entrypoint!(Root)`): a `#[workflow]` for
    /// the typical DAG, or a `#[container]` for a single-container binary.
    /// The consumer command uses it as the default when none is named.
    pub default_template: String,
    /// User crate name (`CARGO_PKG_NAME`).
    pub package: String,
    /// User crate version (`CARGO_PKG_VERSION`); the release-channel
    /// fallback for `version_tag`.
    pub version: String,
    /// User binary name (`CARGO_BIN_NAME`).
    pub bin: String,
    /// The build-time-sealed version tag that names this binary's
    /// `WorkflowTemplate`s and its S3 key (`{pkg}/<tag>/{bin}.tar.gz`):
    /// the baked `ATHENA_VERSION_TAG`, else `kebab` of `version`. The CLI
    /// uses it for the `publish` upload key and the resolved-tag feedback
    /// line — never recomputed from local Cargo.toml/git.
    pub version_tag: String,
    /// `release` (clean + on the release branch at build) or `dev`
    /// (off-road build), derived from `version_tag`.
    pub channel: String,
}

/// Drop the spaces `quote!` puts around `<` / `>` / `,` so a type
/// like `Vec < String >` round-trips as `Vec<String>` for human
/// display. `validate_args` already strips ALL whitespace; this is the
/// gentler variant for showing types verbatim (`Cow<'static, str>`
/// stays readable).
fn normalize_ty(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    let mut prev_open = false;
    for ch in s.chars() {
        match ch {
            '<' | '>' => {
                while out.ends_with(' ') {
                    out.pop();
                }
                out.push(ch);
                prev_open = ch == '<';
            }
            ',' => {
                while out.ends_with(' ') {
                    out.pop();
                }
                out.push(ch);
                prev_open = false;
            }
            ' ' if prev_open => {}
            ' ' if out.ends_with('<') || out.ends_with(',') => {}
            _ => {
                out.push(ch);
                prev_open = false;
            }
        }
    }
    out
}

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(),
            // Filled in by the caller from BuildCtx's artifact_key
            // (encoded as `{crate}/{version}/{bin}.tar.gz`).
            package: String::new(),
            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(),
                            // `quote!(#ty).to_string()` spaces `<` and
                            // `>`, e.g. `Vec < String >`. Strip those
                            // (only inside generics) for display.
                            ty: input_types
                                .get(idx)
                                .map(|s| normalize_ty(s))
                                .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.
///
/// The bootstrap forwards `"$@"` to the binary; the macro-generated
/// `Template::build` appends `--` then one positional argv slot per
/// fn parameter. Parameter args contribute `{{inputs.parameters.<n>}}`;
/// `#[inject("<expr>")]` args contribute the user's raw Argo expression
/// verbatim. The binary's runner reads all slots positionally in
/// declaration order. (We used to deliver each parameter via an
/// `ATHENA_PARAM_<n>` env var, but env vars are NOT eligible for
/// Argo's automatic large-args offload to a ConfigMap, so any large
/// parameter value would balloon the pod spec and could exceed the
/// `E2BIG` exec limit. Argo offloads `container.args` once the total
/// exceeds 128 KB.)
pub fn container_delivery(
    ctx: &BuildCtx,
    argv_elements: &[String],
    image_override: Option<&str>,
    output_kind: IoKind,
) -> 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"));
    }

    // The body's serialized return value lands at `CARGO_ATHENA_OUTPUT`.
    // Parameter-output containers route it to ATHENA_RESULT_FILE (which
    // the template's `outputs.parameters.return.valueFrom.path` reads).
    // Artifact-output containers route it to ATHENA_RESULT_ARTIFACT_FILE
    // (which the template's `outputs.artifacts.return.path` reads;
    // Argo's executor then tar+gzips and uploads to S3). The bootstrap
    // exports the right path before exec-ing the binary; the run-side
    // body has one write site (`CARGO_ATHENA_OUTPUT`) and stays kind-
    // agnostic.
    let output_path = match output_kind {
        IoKind::Parameter => ATHENA_RESULT_FILE,
        IoKind::Artifact => ATHENA_RESULT_ARTIFACT_FILE,
    };

    // 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("@@OUTPUT_PATH@@", output_path);

    // `sh -c "<script>" -- arg1 arg2 ...` puts "--" in $0 (placeholder)
    // and arg1/arg2 in "$@", which the bootstrap forwards to the binary.
    // Each element of `argv_elements` is a pre-formed positional argv
    // slot: parameter args contribute `{{inputs.parameters.<name>}}`,
    // `#[inject(...)]` args contribute the user's raw Argo expression.
    // Order matches the fn's declaration order so the run-side decode
    // (positional `__argv[i]`) lines up.
    let mut args = vec![script, "--".to_string()];
    args.extend(argv_elements.iter().cloned());

    // `archive: None` (NOT `archive: none`) lets Argo auto-detect the
    // input as a tarball and untar it into `path` (`ATHENA_BIN_DIR`).
    let artifact = api::Artifact {
        name: ATHENA_DIST_ARTIFACT.to_string(),
        path: ATHENA_BIN_DIR.to_string(),
        s3: Some(s3_loc(s3, ctx.artifact_key())),
        archive: None,
        mode: None,
        from: String::new(),
    };

    ContainerDelivery {
        image,
        command: vec!["/bin/sh".to_string(), "-c".to_string()],
        args,
        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],
    /// `(secret_name, key, optional)` triples from this fragment's
    /// `secret!`/`secret_opt!` declarations.
    pub secrets: &'static [(&'static str, &'static str, bool)],
    /// Argo names of `Pvc` types this fragment uses directly via
    /// `pvc!(Type)`. Container build-side resolves the transitive
    /// closure via [`BuildCtx::resolved_pvc_names`] (same callee
    /// walker as `resolved_host_paths`).
    pub pvc_argo_names: &'static [&'static str],
    pub callees: &'static [&'static str],
}
inventory::collect!(FragmentReg);

/// Field-wise equality for duplicate-registration detection in
/// [`BuildCtx::collect`] — an identical re-registration (same crate
/// linked twice) is harmless, a same-name/different-content pair is a
/// silent-collision bug.
fn same_fragment(a: &FragmentReg, b: &FragmentReg) -> bool {
    a.host_paths == b.host_paths
        && a.in_artifacts == b.in_artifacts
        && a.out_artifacts == b.out_artifacts
        && a.secrets == b.secrets
        && a.pvc_argo_names == b.pvc_argo_names
        && a.callees == b.callees
}

/// See [`same_fragment`].
fn same_pvc(a: &PvcReg, b: &PvcReg) -> bool {
    a.mount_path == b.mount_path
        && a.lifecycle == b.lifecycle
        && a.size == b.size
        && a.access_modes == b.access_modes
        && a.storage_class_name == b.storage_class_name
        && a.claim_name == b.claim_name
        && a.read_only == b.read_only
}

/// Fragment registry snapshot, passed to container `build`s.
pub struct BuildCtx {
    fragments: HashMap<&'static str, &'static FragmentReg>,
    /// `argo_name → PvcReg` of every PVC type linked into this
    /// binary. Populated from `inventory::iter::<PvcReg>` at
    /// [`BuildCtx::collect`] time. Used by emit-side code to
    /// materialize a full PVC spec (size, access_modes, claim_name,
    /// …) from just an argo name — which is what fragment closures
    /// propagate through.
    pvcs: HashMap<&'static str, &'static PvcReg>,
    config: AthenaConfig,
    /// Fully-resolved S3 object key for this binary's tarball
    /// (`{crate}/<tag>/{bin}.tar.gz`, where `<tag>` is [`version_tag`]).
    /// Built once by [`entrypoint_impl`] from the user binary's own
    /// `CARGO_PKG_*` / `CARGO_BIN_NAME` env vars (captured at the bin's
    /// compile time by the `entrypoint!` macro) plus the baked tag.
    /// `cargo athena publish` uploads to the same key (read off the
    /// freshly-built binary), so the upload and the emitted YAML always
    /// agree by construction — and a dev binary never overwrites a
    /// release tarball.
    ///
    /// [`version_tag`]: BuildCtx::version_tag
    artifact_key: String,
    /// User crate name (`CARGO_PKG_NAME`). Stamped as
    /// `cargo.athena/pkg`.
    krate: String,
    /// User crate version (`CARGO_PKG_VERSION`). Stamped as
    /// `cargo.athena/version` (version of the user's code, distinct from
    /// `cargo.athena/toolchain` which is athena's own version).
    version: String,
    /// User binary name (`CARGO_BIN_NAME`). Stamped as `cargo.athena/bin`.
    bin: String,
    /// The resolved, build-time-sealed version tag (the baked
    /// `ATHENA_VERSION_TAG`, else `kebab` of `version`). The single
    /// coordinate appended to every WT name, used as the S3 key segment,
    /// and stamped as `cargo.athena/tag`. NOT recomputed from local
    /// Cargo.toml/git — it is whatever was compiled into the binary.
    version_tag: String,
    /// Baked git short-sha (`ATHENA_GIT_COMMIT`), `None` for a plain
    /// `cargo build`. Stamped as `cargo.athena/commit`.
    commit: Option<String>,
    /// Baked dirty-tree flag (`ATHENA_GIT_DIRTY`), `None` for a plain
    /// `cargo build`. Stamped as `cargo.athena/dirty`.
    dirty: Option<bool>,
}

impl BuildCtx {
    /// Emit-only: gathers fragments AND loads `athena.toml`. Never called
    /// in run-mode, so the in-pod binary needs no `athena.toml`. The
    /// `version_tag`/`commit`/`dirty` come from the build-time-baked
    /// `option_env!` consts threaded through `entrypoint!` (all `None`
    /// for a plain `cargo build` → release tag = `kebab(version)`).
    pub fn collect(
        krate: &str,
        version: &str,
        bin: &str,
        version_tag: Option<&str>,
        commit: Option<&str>,
        dirty: Option<&str>,
    ) -> Self {
        // Both registries are keyed by a bare name (fragments by fn
        // ident, PVCs by argo name), so two same-named declarations in
        // different modules/crates would silently last-win — one
        // container would inherit the *other* declaration's mounts /
        // secrets / spec. Identical re-registrations (the same crate
        // linked in twice) are harmless and deduped; a same-name entry
        // with *different* contents is always a bug, so fail loud here
        // at emit rather than far away in-pod.
        let mut fragments: HashMap<&'static str, &'static FragmentReg> = HashMap::new();
        for f in inventory::iter::<FragmentReg> {
            if let Some(prev) = fragments.insert(f.rust_name, f)
                && !same_fragment(prev, f)
            {
                panic!(
                    "two different `#[fragment]` fns named `{}` are linked into this \
                     binary. Fragment propagation is keyed by the bare fn name, so \
                     they would silently collide — rename one of them.",
                    f.rust_name,
                );
            }
        }
        let mut pvcs: HashMap<&'static str, &'static PvcReg> = HashMap::new();
        for p in inventory::iter::<PvcReg> {
            if let Some(prev) = pvcs.insert(p.argo_name, p)
                && !same_pvc(prev, p)
            {
                panic!(
                    "two different PVC types share the Argo name {:?}. Rename one \
                     type or give one an explicit `#[ephemeral_pvc(name = \"…\")]` / \
                     `#[external_pvc(name = \"…\")]`.",
                    p.argo_name,
                );
            }
        }
        // Seal the version tag (baked value munged, else kebab(semver))
        // and build the S3 key from it. Both rules live in `api::munge`,
        // shared with the config-free probe path, so emit and probe can't
        // disagree on the sealed tag, and publish/prune can't drift on the
        // key layout.
        let version_tag = crate::api::munge::seal_tag(version_tag, version);
        Self {
            fragments,
            pvcs,
            config: AthenaConfig::load(),
            artifact_key: crate::api::munge::binary_key(krate, &version_tag, bin),
            krate: krate.to_string(),
            version: version.to_string(),
            bin: bin.to_string(),
            version_tag,
            commit: commit.map(str::to_string),
            dirty: dirty.map(|s| s == "true" || s == "1"),
        }
    }

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

    /// The S3 object key of this binary's tarball,
    /// `{crate}/<version_tag>/{bin}.tar.gz`.
    pub fn artifact_key(&self) -> &str {
        &self.artifact_key
    }

    /// The resolved, build-time-sealed version tag (kebab). Appended to
    /// every emitted `WorkflowTemplate` name, the S3 key segment, and the
    /// `cargo.athena/tag` label. Sealed in the binary — `emit`/`submit`
    /// read it, never recompute it from the local Cargo.toml/git.
    pub fn version_tag(&self) -> &str {
        &self.version_tag
    }

    /// `<base>-<tag>` cluster-resource name. The tag overlay is purely a
    /// cluster-identity concern: in-pod dispatch + every
    /// `templateRef.template` stay on `base` (the versioning invariant).
    /// Fails loud at emit if the result would exceed DNS-1123's 63-char
    /// limit — the user shortens the base with `#[…(name = "…")]`.
    pub fn versioned_name(&self, base: &str) -> String {
        let n = format!("{base}-{}", self.version_tag);
        assert!(
            n.len() <= 63,
            "versioned WorkflowTemplate name {n:?} is {} chars, over the \
             DNS-1123 63-char limit (base {base:?} + tag {:?}); shorten the \
             template name with #[workflow(name = \"…\")] / \
             #[container(name = \"…\")]",
            n.len(),
            self.version_tag,
        );
        // Fail loud at emit (not at cluster apply) if the tag ever carried
        // a non-DNS-1123 char. The tag is munged at every entry point
        // (BuildCtx::collect, gitinfo), so this is belt-and-suspenders.
        assert!(
            !n.is_empty()
                && !n.starts_with('-')
                && !n.ends_with('-')
                && n.chars()
                    .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-'),
            "versioned WorkflowTemplate name {n:?} is not DNS-1123-safe \
             (base {base:?} + tag {:?})",
            self.version_tag,
        );
        n
    }

    /// `release` (clean build on the release branch) or `dev` (off-road),
    /// derived from the sealed tag via the shared `api::munge::channel_of`.
    fn channel(&self) -> &'static str {
        crate::api::munge::channel_of(&self.version_tag)
    }

    /// Baseline provenance labels stamped onto every emitted
    /// `WorkflowTemplate` (and the `--with-workflow` runnable `Workflow`).
    /// All under the `cargo.athena/*` namespace - intentionally NOT in
    /// `app.kubernetes.io/*`, despite the convention's appeal. The shared
    /// namespace would clash with Helm-managed deploys (which stamp
    /// `managed-by: Helm`), ArgoCD's instance tracking, Backstage/IDP
    /// catalogs scraping it for service identity, etc. Argo Workflows
    /// itself made the same call: every Argo label lives under
    /// `workflows.argoproj.io/*`, with an explicit comment at
    /// `workflow/common/common.go:97` calling out that their `/component`
    /// is "intentionally similar to `app.kubernetes.io/component`" but
    /// staying in their own namespace.
    ///
    /// `cargo.athena/version` is the USER crate's version (matches
    /// reader intuition - "what version of this code emitted this WT?");
    /// athena's own toolchain version is `cargo.athena/toolchain` to
    /// avoid the ambiguity from PR #57's first cut.
    pub fn athena_labels(&self) -> std::collections::BTreeMap<String, String> {
        use crate::api::munge;
        let mut m = std::collections::BTreeMap::new();
        m.insert(munge::LABEL_PKG.to_string(), self.krate.clone());
        m.insert(munge::LABEL_VERSION.to_string(), self.version.clone());
        m.insert(munge::LABEL_BIN.to_string(), self.bin.clone());
        m.insert(
            munge::LABEL_TOOLCHAIN.to_string(),
            env!("CARGO_PKG_VERSION").to_string(),
        );
        // Build-time-sealed version coordinate + provenance. `tag` and
        // `channel` are always present (the tag falls back to
        // `kebab(version)` for a plain build); `commit`/`dirty` only when
        // `cargo athena build`/`publish` baked them. `tag` makes `cargo
        // athena prune` a pure label-selector delete; the rest describe
        // the *artifact*, not the deploy environment. Keys are the shared
        // `api::munge::LABEL_*` consts, so the prune selector can't drift.
        m.insert(munge::LABEL_TAG.to_string(), self.version_tag.clone());
        m.insert(munge::LABEL_CHANNEL.to_string(), self.channel().to_string());
        if let Some(c) = &self.commit {
            m.insert(munge::LABEL_COMMIT.to_string(), c.clone());
        }
        if let Some(d) = self.dirty {
            m.insert(munge::LABEL_DIRTY.to_string(), d.to_string());
        }
        m
    }

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

    /// PVCs referenced directly via `pvc!(Type)` ∪ the transitive
    /// closure through `#[fragment]` callees. Each entry is a
    /// [`PvcReg::argo_name`]; look up the full spec in
    /// [`BuildCtx::pvc`] (or just call [`Self::pvc_volumes`]).
    pub fn resolved_pvc_names(&self, own: &[&str], callees: &[&str]) -> Vec<String> {
        self.resolved(own, callees, |f| f.pvc_argo_names)
    }

    /// Look up a PVC's full spec by argo name (`""` if unknown — the
    /// macros only ever pass names backed by an actual `Pvc` impl, so
    /// `None` here means a wormhole link bug).
    pub fn pvc(&self, argo_name: &str) -> Option<&'static PvcReg> {
        self.pvcs.get(argo_name).copied()
    }

    /// `(volumes, volume_mounts)` for a resolved PVC name list (the
    /// output of [`Self::resolved_pvc_names`]). Each name produces
    /// one `Volume { persistent_volume_claim: { claim_name } }` plus
    /// one matching `VolumeMount` at the PVC's stable
    /// `/athena/pvcs/<hash>` mount path.
    ///
    /// For an `Ephemeral` PVC the volume's `claim_name` is the
    /// PVC's argo name itself — Argo creates the PVC under that name
    /// via the workflow spec's `volumeClaimTemplates`. For
    /// `External` PVCs the `claim_name` is the user-provided
    /// pre-existing PVC name.
    pub fn pvc_volumes(&self, names: &[String]) -> (Vec<api::Volume>, Vec<api::VolumeMount>) {
        let mut vols = Vec::new();
        let mut mounts = Vec::new();
        for n in names {
            // The macros only ever pass names backed by an actual `Pvc`
            // impl, so an unknown name is a wormhole link bug — emitting
            // a template *missing a mount* would fail far away in-pod
            // (the `pvc!` path wouldn't exist). Fail loud at emit.
            let Some(reg) = self.pvc(n) else {
                panic!(
                    "PVC {n:?} is not registered in this binary — \
                     `pvc!`/fragment wormhole link bug (please report it)"
                )
            };
            let vol_name = pvc_volume_name(reg.argo_name);
            let claim = match reg.lifecycle {
                PvcLifecycle::Ephemeral => reg.argo_name.to_string(),
                PvcLifecycle::External => reg.claim_name.to_string(),
            };
            vols.push(api::Volume {
                name: vol_name.clone(),
                persistent_volume_claim: Some(api::PersistentVolumeClaimVolumeSource {
                    claim_name: claim,
                    read_only: reg.read_only,
                }),
                ..Default::default()
            });
            mounts.push(api::VolumeMount {
                name: vol_name,
                mount_path: reg.mount_path.to_string(),
                read_only: reg.read_only,
            });
        }
        (vols, mounts)
    }

    /// Env-var-sourced K8s secrets: own `secret!`/`secret_opt!` decls
    /// ∪ the `#[fragment]` closure (deduped on the `(name, key)` pair).
    /// Same shape as `resolved`, but the data are triples not strings,
    /// so this is open-coded rather than going through the generic
    /// helper. Two extra rules the string kinds don't need:
    ///
    /// * **Required wins.** A pair declared by both `secret!` and
    ///   `secret_opt!` (traversal order is non-obvious to users) emits
    ///   `optional: false` — a missing secret then fails pod-start with
    ///   a clear K8s event instead of panicking mid-body in-pod.
    /// * **Env-name collisions fail loud.** `secret_env_name` flattens
    ///   `.`/`-`/`_` alike, so distinct pairs like `("a.b", "k")` and
    ///   `("a-b", "k")` map to one env var; emitting both would let one
    ///   silently shadow the other (host mounts are immune — they key
    ///   by hash). Panic at emit instead.
    pub fn resolved_secrets(
        &self,
        own: &[(&str, &str, bool)],
        own_callees: &[&str],
    ) -> Vec<(String, String, bool)> {
        let mut out: Vec<(String, String, bool)> = Vec::new();
        let mut idx: HashMap<(String, String), usize> = HashMap::new();
        let mut by_env: HashMap<String, (String, String)> = HashMap::new();
        let mut push = |n: &str, k: &str, opt: bool, out: &mut Vec<(String, String, bool)>| {
            let pair = (n.to_string(), k.to_string());
            if let Some(&i) = idx.get(&pair) {
                out[i].2 &= opt; // required wins over optional
                return;
            }
            let env = crate::api::munge::secret_env_name(n, k);
            if let Some((pn, pk)) = by_env.get(&env) {
                panic!(
                    "secrets ({pn:?}, {pk:?}) and ({n:?}, {k:?}) both map to the \
                     env var `{env}` — one would silently shadow the other in-pod. \
                     Rename one secret (or key) so they stay distinguishable."
                );
            }
            by_env.insert(env, pair.clone());
            idx.insert(pair, out.len());
            out.push((n.to_string(), k.to_string(), opt));
        };
        for (n, k, opt) in own {
            push(n, k, *opt, &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 (n, k, opt) in f.secrets {
                    push(n, k, *opt, &mut out);
                }
                queue.extend(f.callees.iter().copied());
            }
        }
        out
    }
}

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(),
            ..Default::default()
        }),
        secret_key_secret: Some(api::SecretKeySelector {
            name: s3.secret_key_secret.name.clone(),
            key: s3.secret_key_secret.key.clone(),
            ..Default::default()
        }),
    }
}

/// 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
}

/// [`artifact_ident`] for each key, panicking when two distinct keys
/// flatten to the same ident: one template would declare two artifact
/// ports with the same name, which Argo rejects (or worse, last-wins).
/// The flattening is lossy (`a.b` and `a-b` → `a-b`), so this is the
/// fail-loud guard for it. The same key appearing twice is fine —
/// callers dedup on the exact key string.
fn artifact_idents_checked(keys: &[String]) -> Vec<String> {
    let mut by_ident: HashMap<String, String> = HashMap::new();
    keys.iter()
        .map(|k| {
            let id = artifact_ident(k);
            if let Some(prev) = by_ident.insert(id.clone(), k.clone())
                && prev != *k
            {
                panic!(
                    "artifact keys {prev:?} and {k:?} on one template both flatten \
                     to the Argo artifact name {id:?} — rename one so they stay \
                     distinguishable."
                );
            }
            id
        })
        .collect()
}

/// `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;
    let idents = artifact_idents_checked(keys);
    keys.iter()
        .zip(idents)
        .map(|(k, ident)| api::Artifact {
            name: ident,
            path: format!("{}/{k}", rt::IN_DIR),
            s3: Some(s3_loc(s3, k)),
            archive: Some(archive_none()),
            mode: None,
            from: String::new(),
        })
        .collect()
}

/// `(key, operator, value, effect, toleration_seconds)` — the lowered
/// shape the macro produces for each toleration entry, threaded through
/// `Template::TOLERATIONS_IF_ROOT` and into `WorkflowSpec.Tolerations`
/// by `Collector::stamp_spec`.
pub type TolerationTuple = (String, String, String, String, i64);

/// `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;
    let idents = artifact_idents_checked(keys);
    keys.iter()
        .zip(idents)
        .map(|(k, ident)| api::Artifact {
            name: ident,
            path: format!("{}/{k}", rt::OUT_DIR),
            s3: Some(s3_loc(s3, k)),
            archive: Some(archive_none()),
            mode: None,
            from: String::new(),
        })
        .collect()
}

/// Accumulates the reachable templates (as `WorkflowTemplate`s) and the
/// run-mode dispatch table while `Template::collect` walks the closure.
pub struct Collector {
    /// argo name → the registering type's identity (+ its rust path for
    /// diagnostics). A `HashMap` (not a `HashSet`) so `enter` can tell a
    /// harmless re-visit of the same template (diamond reachability)
    /// from two *different* fns colliding on one Argo name — the latter
    /// would silently emit one template and dispatch both call sites to
    /// whichever body registered, so it fails loud instead.
    seen: HashMap<String, (std::any::TypeId, &'static str)>,
    /// Deferred so `athena.toml` is read only at emit, never run-mode.
    builders: Vec<fn(&BuildCtx) -> api::Template>,
    runners: HashMap<String, fn(&[String]) -> String>,
    /// `<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> -> nodeSelector key-value pairs` for every template
    /// that declares `#[workflow(node_selector_if_root = …)]`. Stamped
    /// onto that WT's `spec.nodeSelector` (the only nodeSelector knob
    /// Argo cascades over every pod in the run, root-only).
    node_selector_if_root: HashMap<String, Vec<(String, String)>>,
    /// `<argo name> -> mutexes` for every template declaring
    /// `#[…(mutexes_if_root = […])]`. Each entry is `(name, namespace)`
    /// already lowered to the final YAML form (literal, or
    /// `{{=fromJSON(workflow.parameters[…])}}` for injected operands);
    /// `namespace == ""` means "skip the field" (defaults to the wf's
    /// own ns). Stamped onto that WT's `spec.synchronization.mutexes`.
    mutexes_if_root: HashMap<String, Vec<(String, String)>>,
    /// `<argo name> -> tolerations` for every template declaring
    /// `#[…(tolerations_if_root = [...])]`. Strings already lowered.
    /// Stamped onto that WT's `spec.tolerations`.
    tolerations_if_root: HashMap<String, Vec<TolerationTuple>>,
    /// `<argo name> -> affinity YAML string` for every template with
    /// `#[…(affinity_if_root = "...")]`. Parsed at emit time and
    /// stuffed into `spec.affinity` as a `serde_norway::Value`.
    affinity_if_root: HashMap<String, String>,
    /// `<argo name> -> pod-spec strategic-merge patch (string)` for
    /// every template with `#[workflow(pod_spec_patch_if_root = "...")]`.
    /// Already lowered to its final form (literal, or with
    /// `{{=fromJSON(workflow.parameters[..])}}` injection operands).
    /// Stamped onto that WT's `spec.podSpecPatch`.
    pod_spec_patch_if_root: HashMap<String, String>,
    /// `<argo name> -> Secret names` for every template declaring
    /// `#[…(image_pull_secrets_if_root = [...])]`. Stamped onto that
    /// WT's `spec.imagePullSecrets` as `[{name}]` k8s references.
    image_pull_secrets_if_root: HashMap<String, Vec<String>>,
    /// `<argo name> -> WorkflowSpec.parallelism` for every template
    /// declaring `#[workflow(parallelism_if_root = N)]`. Stamped onto
    /// that WT's `spec.parallelism`.
    parallelism_if_root: HashMap<String, i64>,
    /// `<argo name> -> stringified input types` (parallel to the
    /// template's INPUTS), for `emulate` arg type-checking.
    types: HashMap<String, &'static [&'static str]>,
    /// Argo names of athena-synthesized templates (`Template::SYNTHETIC`)
    /// so `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: HashMap::new(),
            builders: Vec::new(),
            runners: HashMap::new(),
            exits: HashMap::new(),
            ttl: HashMap::new(),
            pod_gc: HashMap::new(),
            active_deadline: HashMap::new(),
            node_selector_if_root: HashMap::new(),
            mutexes_if_root: HashMap::new(),
            tolerations_if_root: HashMap::new(),
            affinity_if_root: HashMap::new(),
            pod_spec_patch_if_root: HashMap::new(),
            image_pull_secrets_if_root: HashMap::new(),
            parallelism_if_root: HashMap::new(),
            types: HashMap::new(),
            synthetic: HashSet::new(),
        }
    }

    /// Returns `false` if `T` was already collected (generated `collect`
    /// impls return early in that case — dedup + cycle guard).
    ///
    /// Panics when a *different* type already claimed `T::ARGO_NAME`:
    /// without the check, only the first registrant would emit a
    /// template and every call site would dispatch to it by name —
    /// silently running the wrong body. Reachable with two same-named
    /// fns in different modules/crates, or two identical explicit
    /// `name = "…"` overrides.
    pub fn enter<T: Template + 'static>(&mut self) -> bool {
        let id = std::any::TypeId::of::<T>();
        match self.seen.entry(T::ARGO_NAME.to_string()) {
            std::collections::hash_map::Entry::Occupied(e) => {
                let (prev_id, prev_ty) = *e.get();
                if prev_id != id {
                    panic!(
                        "duplicate Argo template name {:?}: `{}` and `{}` both map to \
                         it. Rename one fn, or give one an explicit \
                         `#[container(name = \"…\")]` / `#[workflow(name = \"…\")]`.",
                        T::ARGO_NAME,
                        prev_ty,
                        std::any::type_name::<T>(),
                    );
                }
                false
            }
            std::collections::hash_map::Entry::Vacant(v) => {
                v.insert((id, std::any::type_name::<T>()));
                true
            }
        }
    }

    /// 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::NODE_SELECTOR_IF_ROOT.is_empty() {
            self.node_selector_if_root.insert(
                T::ARGO_NAME.to_string(),
                T::NODE_SELECTOR_IF_ROOT
                    .iter()
                    .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
                    .collect(),
            );
        }
        if !T::MUTEXES_IF_ROOT.is_empty() {
            self.mutexes_if_root.insert(
                T::ARGO_NAME.to_string(),
                T::MUTEXES_IF_ROOT
                    .iter()
                    .map(|(n, ns)| ((*n).to_string(), (*ns).to_string()))
                    .collect(),
            );
        }
        if !T::TOLERATIONS_IF_ROOT.is_empty() {
            self.tolerations_if_root.insert(
                T::ARGO_NAME.to_string(),
                T::TOLERATIONS_IF_ROOT
                    .iter()
                    .map(|(k, op, v, eff, secs)| {
                        (
                            (*k).to_string(),
                            (*op).to_string(),
                            (*v).to_string(),
                            (*eff).to_string(),
                            *secs,
                        )
                    })
                    .collect(),
            );
        }
        if let Some(a) = T::AFFINITY_IF_ROOT {
            self.affinity_if_root
                .insert(T::ARGO_NAME.to_string(), a.to_string());
        }
        if let Some(p) = T::POD_SPEC_PATCH_IF_ROOT {
            self.pod_spec_patch_if_root
                .insert(T::ARGO_NAME.to_string(), p.to_string());
        }
        if !T::IMAGE_PULL_SECRETS_IF_ROOT.is_empty() {
            self.image_pull_secrets_if_root.insert(
                T::ARGO_NAME.to_string(),
                T::IMAGE_PULL_SECRETS_IF_ROOT
                    .iter()
                    .map(|n| (*n).to_string())
                    .collect(),
            );
        }
        if let Some(p) = T::PARALLELISM_IF_ROOT {
            self.parallelism_if_root.insert(T::ARGO_NAME.to_string(), p);
        }
        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(&[String]) -> String) {
        self.runners.insert(argo_name.to_string(), run);
    }

    /// 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, ctx: &BuildCtx) -> Vec<api::WorkflowTemplate> {
        let labels = ctx.athena_labels();
        let mut tpls: Vec<api::WorkflowTemplate> = self
            .builders
            .iter()
            .map(|b| {
                let inner = b(ctx);
                let mut wt = wrap_workflow_template(inner.name.clone(), inner);
                if let Some(meta) = wt.metadata.as_mut() {
                    meta.labels
                        .extend(labels.iter().map(|(k, v)| (k.clone(), v.clone())));
                }
                wt
            })
            .collect();
        tpls.sort_by_key(name_of);

        // Stamp the `*_if_root` family onto each declaring template's
        // own `spec` — `on_exit_if_root` becomes `spec.hooks.exit`
        // (`templateRef`; legacy `spec.onExit` name-string can't cross
        // the one-WT-per-template wormhole), `ttl_if_root`/
        // `pod_gc_if_root`/`active_deadline_if_root`/
        // `node_selector_if_root` land on their matching `spec` fields.
        // Argo fires/applies them workflow-scoped (only for the
        // SUBMITTED root), so per-WT stamping is the correct model: a
        // templateRef'd sub-workflow stays inert when nested but fires
        // on direct submission. Single source of truth lives in
        // `stamp_spec` — the runnable Workflow path in `emit` calls the
        // same method so the two sites can't drift.
        for t in tpls.iter_mut() {
            let name = name_of(t);
            if let Some(spec) = t.spec.as_mut() {
                self.stamp_spec(&name, spec, ctx);
                // Version overlay, applied AFTER stamp_spec so its
                // base-name keying above is untouched: append the
                // build-time tag to this WT's resource identity — every
                // `templateRef.name`. The `metadata.name` rename below
                // completes the pair. `entrypoint` / inner
                // `templates[].name` / `templateRef.template` stay base
                // (the versioning invariant), so in-pod dispatch and
                // Argo's within-WT template lookup are unaffected.
                version_spec_refs(spec, ctx);
            }
            if let Some(meta) = t.metadata.as_mut() {
                meta.name = ctx.versioned_name(&meta.name);
            }
        }
        tpls
    }

    /// Apply every per-template spec-scoped attribute (`on_exit_if_root`,
    /// `ttl_if_root`, `pod_gc_if_root`, `active_deadline_if_root`,
    /// `node_selector_if_root`) for `name` onto `spec`. The single
    /// source of truth for the `*_if_root` family — called once per
    /// emitted WT in `build_templates`, and once for the convenience
    /// runnable Workflow's root in `emit`. Adding a new spec-scoped
    /// attribute means adding one map field, one populate line in
    /// `add::<T>()`, and one `if let Some` block here — both
    /// stamping sites pick it up automatically.
    ///
    /// Also overlays `spec.volume_claim_templates` from every
    /// `#[ephemeral_pvc]` reachable through inventory (binary-wide).
    /// Argo only honors the submitted root's spec, so over-inclusion
    /// on non-root WTs is inert — the bloat is the cost of a simple
    /// inventory iteration vs. tracking per-WT reachability. (See
    /// the followup item in the PVC PR notes for the precision
    /// improvement.)
    fn stamp_spec(&self, name: &str, spec: &mut api::WorkflowSpec, ctx: &BuildCtx) {
        if let Some(handler) = self.exits.get(name) {
            spec.hooks
                .insert("exit".to_string(), exit_hook_ref(handler));
        }
        if let Some(ttl) = self.ttl.get(name) {
            spec.ttl_strategy = Some(ttl.clone());
        }
        if let Some(s) = self.pod_gc.get(name) {
            spec.pod_gc = Some(api::PodGc {
                strategy: s.clone(),
            });
        }
        if let Some(s) = self.active_deadline.get(name) {
            spec.active_deadline_seconds = Some(*s);
        }
        if let Some(ns) = self.node_selector_if_root.get(name) {
            for (k, v) in ns {
                spec.node_selector.insert(k.clone(), v.clone());
            }
        }
        if let Some(mtx) = self.mutexes_if_root.get(name) {
            let sync = spec
                .synchronization
                .get_or_insert_with(api::Synchronization::default);
            for (mname, mns) in mtx {
                sync.mutexes.push(api::Mutex {
                    name: mname.clone(),
                    namespace: mns.clone(),
                });
            }
        }
        if let Some(tols) = self.tolerations_if_root.get(name) {
            for (k, op, v, eff, secs) in tols {
                spec.tolerations.push(api::Toleration {
                    key: k.clone(),
                    operator: op.clone(),
                    value: v.clone(),
                    effect: eff.clone(),
                    toleration_seconds: if *secs == 0 { None } else { Some(*secs) },
                });
            }
        }
        if let Some(s) = self.affinity_if_root.get(name) {
            spec.affinity = Some(
                serde_norway::from_str(s)
                    .unwrap_or_else(|e| panic!("affinity_if_root: invalid YAML/JSON: {e}")),
            );
        }
        if let Some(p) = self.pod_spec_patch_if_root.get(name) {
            spec.pod_spec_patch = Some(p.clone());
        }
        if let Some(ipss) = self.image_pull_secrets_if_root.get(name) {
            for n in ipss {
                spec.image_pull_secrets
                    .push(api::LocalObjectReference { name: n.clone() });
            }
        }
        if let Some(p) = self.parallelism_if_root.get(name) {
            spec.parallelism = Some(*p);
        }
        // Every `#[ephemeral_pvc]` linked into this binary becomes a
        // `spec.volume_claim_templates` entry on every emitted WT.
        // Argo creates each PVC at workflow start (with
        // `metadata.name = <argo-name>`, the same `claim_name` each
        // pod's `volumes[]` references) and deletes it at workflow
        // end.
        //
        // Caveat: this over-includes for multi-workflow binaries.
        // Submitting workflow A in a binary that also defines an
        // unrelated workflow B causes Argo to create B's PVCs too,
        // since both are linked into the same binary and stamped on
        // every WT. The simplest fix is keeping one workflow per
        // binary (the recommended layout). Per-WT precision is a
        // possible follow-up if this hurts in practice.
        //
        // Sorted by argo name for deterministic emit.
        let mut ephemeral: Vec<&PvcReg> = ctx
            .pvcs
            .values()
            .copied()
            .filter(|r| r.lifecycle == PvcLifecycle::Ephemeral)
            .collect();
        ephemeral.sort_by_key(|r| r.argo_name);
        for reg in ephemeral {
            let mut requests = std::collections::BTreeMap::new();
            requests.insert("storage".to_string(), reg.size.to_string());
            spec.volume_claim_templates
                .push(api::PersistentVolumeClaim {
                    metadata: Some(api::ObjectMeta {
                        name: reg.argo_name.to_string(),
                        ..Default::default()
                    }),
                    spec: Some(api::PersistentVolumeClaimSpec {
                        access_modes: reg.access_modes.iter().map(|s| s.to_string()).collect(),
                        resources: Some(api::VolumeResourceRequirements { requests }),
                        storage_class_name: reg.storage_class_name.to_string(),
                    }),
                });
        }
    }

    /// Emit the multi-doc YAML stream: one `WorkflowTemplate` per
    /// reachable template. `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 -`.
    pub fn emit<E: Template>(&self, ctx: &BuildCtx, with_workflow: bool) -> String {
        let tpls = self.build_templates(ctx);
        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");
        }

        // Start from a default + the two fields specific to the runnable
        // Workflow (`workflowTemplateRef → root`, default SA from
        // athena.toml), then `stamp_spec` overlays every `*_if_root`
        // attribute for the root — same path `build_templates` uses
        // per-WT, so the two stamping sites can never drift when a new
        // `*_if_root` attribute is added.
        let mut spec = api::WorkflowSpec {
            workflow_template_ref: Some(api::WorkflowTemplateRef {
                // Points at the versioned root WT resource.
                name: ctx.versioned_name(E::ARGO_NAME),
                cluster_scope: false,
            }),
            service_account_name: ctx.config().defaults.service_account.clone(),
            ..Default::default()
        };
        // stamp_spec keys on the BASE name (the `_if_root` maps) and may
        // add an exit-hook templateRef; version_spec_refs then bumps that
        // ref to the versioned handler WT. Same order as build_templates.
        self.stamp_spec(E::ARGO_NAME, &mut spec, ctx);
        version_spec_refs(&mut spec, ctx);
        let wf = api::Workflow {
            api_version: api::API_VERSION.to_string(),
            kind: api::KIND_WORKFLOW.to_string(),
            metadata: Some(api::ObjectMeta {
                // Versioned, to match the workflow_template_ref above and
                // `cargo athena submit`'s generateName (both use the
                // versioned root) — the two run-creation paths stay in step.
                generate_name: format!("{}-", ctx.versioned_name(E::ARGO_NAME)),
                labels: ctx.athena_labels(),
                ..Default::default()
            }),
            spec: Some(spec),
        };
        docs.push(serde_norway::to_string(&wf).expect("Workflow is serializable"));
        docs.join("---\n")
    }
}

/// A `spec.hooks.exit` `LifecycleHook` referencing the named handler
/// template. The legacy `spec.onExit: <name>` (a name-string) can't
/// cross the one-WT-per-template wormhole on real Argo v4.0.5 — only
/// the structured `templateRef` form survives, hence this single
/// construction reused by every stamping site.
fn exit_hook_ref(handler: &str) -> api::LifecycleHook {
    api::LifecycleHook {
        template_ref: Some(api::TemplateRef {
            name: handler.to_string(),
            template: handler.to_string(),
            cluster_scope: false,
        }),
        ..Default::default()
    }
}

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

/// Append the build-time version tag to every cluster-resource name
/// REFERENCE in `spec`: each `templateRef.name` on DAG tasks (and their
/// hooks), step tasks (and their hooks), and the workflow-scoped
/// `spec.hooks` (e.g. the `on_exit_if_root` handler). Deliberately NOT
/// `templateRef.template` — that's the callee's inner template name,
/// which stays on the base (Argo resolves it within the target WT, whose
/// single template keeps the base name). The tag is uniform (the one
/// root-binary version), so all refs get the same suffix and cross-WT
/// links stay consistent. The versioning analog of `Collector::stamp_spec`,
/// run per WT in `build_templates` and once for the `--with-workflow`
/// Workflow in `emit`.
fn version_spec_refs(spec: &mut api::WorkflowSpec, ctx: &BuildCtx) {
    fn bump(tr: &mut Option<api::TemplateRef>, ctx: &BuildCtx) {
        if let Some(r) = tr.as_mut() {
            r.name = ctx.versioned_name(&r.name);
        }
    }
    for t in spec.templates.iter_mut() {
        if let Some(dag) = t.dag.as_mut() {
            for task in dag.tasks.iter_mut() {
                bump(&mut task.template_ref, ctx);
                for h in task.hooks.values_mut() {
                    bump(&mut h.template_ref, ctx);
                }
            }
        }
        for group in t.steps.iter_mut() {
            for step in group.iter_mut() {
                bump(&mut step.template_ref, ctx);
                for h in step.hooks.values_mut() {
                    bump(&mut h.template_ref, ctx);
                }
            }
        }
    }
    for h in spec.hooks.values_mut() {
        bump(&mut h.template_ref, ctx);
    }
}

/// 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` / `host_mount_path` / `host_volume_name` live in
// `api::munge`. The proc-macro and emit-side both call into the
// single source. Re-exported here so existing
// `cargo_athena_core::kebab` / `host_mount_path` call sites keep
// resolving.
use crate::api::munge::host_volume_name as volume_name;
pub use crate::api::munge::{host_mount_path, kebab};

/// `volumes` + `volumeMounts` for a set of hostPaths (from `host!`).
/// Each mounts at [`host_mount_path`] (`/athena/mounts/<munged>`),
/// NOT at the host's own path — safe-by-construction.
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: host_mount_path(p),
            read_only: false,
        });
    }
    (vols, mounts)
}

/// Every container template's volumes/mounts: the always-present
/// `emptyDir` scratch at [`ATHENA_DIR`] + the declared hostPaths. Two
/// hostPath sources:
///
/// - `host_paths` from `host!` — safe-by-construction, mounted at
///   [`host_mount_path`] (`/athena/mounts/<munged>`).
/// - `host_mounts` from `#[container(host_mount = [{…}])]` — explicit
///   `host_path` + `mount_path` + `read_only`, the user's escape hatch
///   for chosen mount paths (`/dev/shm`, sidecar data dirs, …).
///
/// If the same `host_path` appears in both, the `host_mount` entry
/// wins — same Volume, explicit `mount_path`/`read_only`. Keeps the
/// emit free of duplicate Volume names while preserving the user's
/// "I asked for it explicitly" intent.
pub fn container_volumes(
    host_paths: &[String],
    host_mounts: &[(String, String, bool)],
) -> (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,
    }];
    // host_mount wins over host! on a shared host_path.
    let overridden: HashSet<&str> = host_mounts.iter().map(|(h, _, _)| h.as_str()).collect();
    for p in host_paths {
        if overridden.contains(p.as_str()) {
            continue;
        }
        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: host_mount_path(p),
            read_only: false,
        });
    }
    for (host_path, mount_path, read_only) in host_mounts {
        let name = volume_name(host_path);
        vols.push(api::Volume {
            name: name.clone(),
            host_path: Some(api::HostPathVolumeSource {
                path: host_path.clone(),
                r#type: String::new(),
            }),
            ..Default::default()
        });
        mounts.push(api::VolumeMount {
            name,
            mount_path: mount_path.clone(),
            read_only: *read_only,
        });
    }
    (vols, mounts)
}

/// De-reference one argv slot if Argo's emissary rewrote it as a
/// `@/tmp/argo_arg_<i>.txt` sentinel.
///
/// When `c.Args` exceeds 128 KB the controller offloads the whole vector
/// to a `ConfigMap` and clears `c.Args`. The emissary then re-hydrates
/// the args from `$ARGO_CONTAINER_ARGS_FILE` and, separately, replaces
/// any single arg whose length still exceeds 128 KB with a sentinel
/// `@/tmp/argo_arg_<i>.txt` whose contents are the real value
/// (`cmd/argoexec/commands/emissary.go` PR #15265). The container is
/// expected to know that convention and read the file.
///
/// Gating on `$ARGO_CONTAINER_ARGS_FILE` (set by the controller only on
/// offload) keeps this safe: a Regime-B parameter value never starts
/// with `@` (string literals start with `"`, numbers with a digit/sign,
/// bools with `t`/`f`), so even if a user shell wires up a workflow
/// outside Argo we won't mis-interpret a literal `@`-prefixed argv.
fn deref_offloaded_arg(raw: String) -> String {
    if std::env::var_os("ARGO_CONTAINER_ARGS_FILE").is_none() {
        return raw;
    }
    let Some(path) = raw.strip_prefix('@') else {
        return raw;
    };
    if !path.starts_with("/tmp/argo_arg_") {
        return raw;
    }
    std::fs::read_to_string(path)
        .unwrap_or_else(|e| panic!("failed reading offloaded arg {path}: {e}"))
}

/// 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).
///
/// `krate`/`version`/`bin` identify the *current binary* and are baked
/// into every emitted container's S3 artifact key
/// (`{krate}/{version}/{bin}.tar.gz`). They're captured at the user
/// binary's compile time by the `entrypoint!` macro (the facade
/// crate) from `CARGO_PKG_NAME`/`CARGO_PKG_VERSION`/`CARGO_BIN_NAME`,
/// so `cargo athena publish` (which derives the same key from
/// `cargo metadata` + `--bin`) and the emitted YAML always agree by
/// construction, with no `[artifact]` config field needed.
pub fn entrypoint_impl<E: Template>(
    krate: &str,
    version: &str,
    bin: &str,
    version_tag: Option<&str>,
    commit: Option<&str>,
    dirty: Option<&str>,
) {
    let mut collector = Collector::new();
    E::collect(&mut collector);

    // Run-mode: `CARGO_ATHENA_TEMPLATE=<name>` selects which template's
    // body to run; the function's parameters arrive as positional argv
    // in INPUTS order (JSON-encoded). The selector lives in env so the
    // pod spec doesn't carry it as a per-template argv string, and so
    // argv is 100% function data, eligible for Argo's automatic offload
    // of large `container.args` to a ConfigMap (env vars are not).
    if let Ok(t) = std::env::var(CARGO_ATHENA_TEMPLATE_ENV) {
        let run = *collector
            .runners
            .get(&t)
            .unwrap_or_else(|| panic!("no runnable container template named {t:?}"));
        let argv: Vec<String> = std::env::args().skip(1).map(deref_offloaded_arg).collect();
        let output = run(&argv);
        if let Ok(path) = std::env::var(CARGO_ATHENA_OUTPUT_ENV) {
            std::fs::write(path, &output).expect("write CARGO_ATHENA_OUTPUT");
        } else {
            println!("{output}");
        }
        return;
    }

    // `cargo athena` consumer commands run this FIRST to confirm the
    // binary is a cargo-athena binary and to agree on the metadata
    // wire-format version before trusting LIST / DESCRIBE / EMIT_JSON.
    // Deliberately needs NO `athena.toml` (no `BuildCtx::collect`) so it
    // works source-free, and it reports the root template the consumer
    // command defaults to when none is named.
    if std::env::var_os("CARGO_ATHENA_PROBE").is_some() {
        // Resolve the sealed tag + channel without a BuildCtx (probe is
        // config-free), via the SAME `api::munge` helpers `BuildCtx::collect`
        // uses, so probe and emit can't disagree on either.
        let tag = crate::api::munge::seal_tag(version_tag, version);
        let channel = crate::api::munge::channel_of(&tag);
        let info = ProbeInfo {
            kind: ATHENA_PROBE_KIND.to_string(),
            athena_protocol: ATHENA_PROTOCOL,
            athena_version: env!("CARGO_PKG_VERSION").to_string(),
            default_template: E::ARGO_NAME.to_string(),
            package: krate.to_string(),
            version: version.to_string(),
            bin: bin.to_string(),
            version_tag: tag,
            channel: channel.to_string(),
        };
        println!(
            "{}",
            serde_json::to_string(&info).expect("ProbeInfo is serializable")
        );
        return;
    }

    // `cargo athena 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(krate, version, bin, version_tag, commit, dirty);
        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.package = krate.to_string();
                m.synthetic = collector.synthetic.contains(&t.name);
                m
            })
            .collect();
        println!(
            "{}",
            serde_json::to_string(&all).expect("ContainerRunMeta is serializable")
        );
        return;
    }

    // `cargo athena 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(krate, version, bin, version_tag, commit, dirty);
        // The default emitted name is `<crate>-<fn>`; the CLI already
        // shows package + short name as separate columns, so accept
        // either the full name or the short form (and fall back to the
        // full name in error messages so the user sees what we tried).
        let full = format!("{krate}-{name}");
        let tpl = collector
            .builders
            .iter()
            .map(|b| b(&ctx))
            .find(|t| t.name == name || t.name == full)
            .unwrap_or_else(|| panic!("no template named {name:?} (or {full:?})"));
        let resolved = tpl.name.clone();
        let input_types = collector.types.get(&resolved).copied().unwrap_or(&[]);
        let mut meta = ContainerRunMeta::from_template(&tpl, input_types);
        meta.package = krate.to_string();
        meta.synthetic = collector.synthetic.contains(&resolved);
        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() {
        let ctx = BuildCtx::collect(krate, version, bin, version_tag, commit, dirty);
        println!(
            "{}",
            serde_json::to_string(&collector.build_templates(&ctx))
                .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");
    let ctx = BuildCtx::collect(krate, version, bin, version_tag, commit, dirty);
    print!("{}", collector.emit::<E>(&ctx, with_workflow));
}

#[cfg(test)]
mod tests {
    use super::AthenaConfig;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU32, Ordering};

    // `resolve_config_path` is pure (takes its inputs as params, reads no
    // env / cwd of its own), so these exercise the precedence order
    // without mutating the process environment, safe under parallel test
    // runs. Filesystem-touching cases use a unique temp dir each.
    static SEQ: AtomicU32 = AtomicU32::new(0);

    fn tmpdir(tag: &str) -> PathBuf {
        let n = SEQ.fetch_add(1, Ordering::Relaxed);
        let d =
            std::env::temp_dir().join(format!("athena-cfg-test-{}-{tag}-{n}", std::process::id()));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    fn touch(dir: &Path, name: &str) -> PathBuf {
        let p = dir.join(name);
        std::fs::write(&p, "").unwrap();
        p
    }

    // temp_dir may be a symlink (e.g. macOS /tmp -> /private/tmp); compare
    // canonicalized so the walk-up result matches the file we created.
    fn canon(p: PathBuf) -> PathBuf {
        std::fs::canonicalize(p).unwrap()
    }

    #[test]
    fn flag_wins_over_everything() {
        let flag = PathBuf::from("/tmp/explicit-athena.toml");
        let env = PathBuf::from("/tmp/env-athena.toml");
        let got = AthenaConfig::resolve_config_path(
            Some(&flag),
            Some(&env),
            Path::new("/nonexistent"),
            Some(Path::new("/nonexistent")),
        );
        assert_eq!(got.as_deref(), Some(flag.as_path()));
    }

    #[test]
    fn env_wins_over_walkup_and_global() {
        let env = PathBuf::from("/tmp/env-athena.toml");
        let got = AthenaConfig::resolve_config_path(
            None,
            Some(&env),
            Path::new("/nonexistent"),
            Some(Path::new("/nonexistent")),
        );
        assert_eq!(got.as_deref(), Some(env.as_path()));
    }

    #[test]
    fn empty_env_is_treated_as_unset() {
        // A set-but-empty `$ATHENA_CONFIG` must fall through, not shadow the
        // walk-up / global fallback (the source-free path).
        let cwd = tmpdir("empty-env-cwd"); // no athena.toml on the walk-up
        let xdg = tmpdir("empty-env-xdg");
        let cfg = touch(&xdg, "athena.toml");
        let got = AthenaConfig::resolve_config_path(None, Some(Path::new("")), &cwd, Some(&xdg));
        assert_eq!(got.map(canon), Some(canon(cfg)));
        let _ = std::fs::remove_dir_all(&cwd);
        let _ = std::fs::remove_dir_all(&xdg);
    }

    #[test]
    fn walkup_finds_athena_toml_in_ancestor() {
        let root = tmpdir("walkup");
        let cfg = touch(&root, "athena.toml");
        let nested = root.join("a").join("b");
        std::fs::create_dir_all(&nested).unwrap();
        // No flag/env; the global must be ignored because walk-up hits first.
        let got =
            AthenaConfig::resolve_config_path(None, None, &nested, Some(Path::new("/nonexistent")));
        assert_eq!(got.map(canon), Some(canon(cfg)));
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn global_fallback_when_no_repo_config() {
        let cwd = tmpdir("global-cwd"); // no athena.toml anywhere on the walk-up
        let xdg = tmpdir("global-xdg");
        let cfg = touch(&xdg, "athena.toml");
        let got = AthenaConfig::resolve_config_path(None, None, &cwd, Some(&xdg));
        assert_eq!(got.map(canon), Some(canon(cfg)));
        let _ = std::fs::remove_dir_all(&cwd);
        let _ = std::fs::remove_dir_all(&xdg);
    }

    #[test]
    fn none_when_nothing_resolves() {
        let cwd = tmpdir("none");
        let got = AthenaConfig::resolve_config_path(
            None,
            None,
            &cwd,
            Some(Path::new("/definitely/not/here")),
        );
        assert_eq!(got, None);
        let _ = std::fs::remove_dir_all(&cwd);
    }

    // ---- fail-loud guards --------------------------------------------

    /// Inventory is empty in this test binary, so `collect` gives a
    /// fragment-free ctx — enough to exercise `resolved_secrets`' own
    /// merge/collision rules.
    fn empty_ctx() -> super::BuildCtx {
        super::BuildCtx::collect("t", "0.0.0", "t", None, None, None)
    }

    #[test]
    fn resolved_secrets_required_wins_over_optional() {
        let ctx = empty_ctx();
        // Optional seen first, required later: the emitted entry must be
        // required, else a missing secret panics mid-body in-pod instead
        // of failing pod-start.
        let got = ctx.resolved_secrets(&[("s", "k", true), ("s", "k", false)], &[]);
        assert_eq!(got, vec![("s".to_string(), "k".to_string(), false)]);
        // And the reverse order too.
        let got = ctx.resolved_secrets(&[("s", "k", false), ("s", "k", true)], &[]);
        assert_eq!(got, vec![("s".to_string(), "k".to_string(), false)]);
    }

    #[test]
    #[should_panic(expected = "both map to the env var")]
    fn resolved_secrets_env_collision_fails_loud() {
        // "a.b" and "a-b" are distinct valid K8s secret names but
        // flatten to the same ATHENA_SEC_A_B__K env var.
        let ctx = empty_ctx();
        ctx.resolved_secrets(&[("a.b", "k", false), ("a-b", "k", false)], &[]);
    }

    #[test]
    fn artifact_idents_pass_distinct_keys() {
        let keys = vec!["data/in.json".to_string(), "data/out.json".to_string()];
        assert_eq!(
            super::artifact_idents_checked(&keys),
            vec!["data-in-json".to_string(), "data-out-json".to_string()],
        );
        // The same key twice is the caller-dedup contract, not a clash.
        let dup = vec!["a.b".to_string(), "a.b".to_string()];
        super::artifact_idents_checked(&dup);
    }

    #[test]
    #[should_panic(expected = "both flatten to the Argo artifact name")]
    fn artifact_ident_collision_fails_loud() {
        let keys = vec!["a.b".to_string(), "a-b".to_string()];
        super::artifact_idents_checked(&keys);
    }

    #[test]
    #[should_panic(expected = "duplicate Argo template name")]
    fn collector_enter_rejects_same_name_different_type() {
        struct A;
        struct B;
        impl super::Template for A {
            const ARGO_NAME: &'static str = "dup-name";
            const INPUTS: &'static [&'static str] = &[];
            const KIND: super::TemplateKind = super::TemplateKind::Container;
            fn build(_: &super::BuildCtx) -> crate::api::Template {
                Default::default()
            }
            fn collect(_: &mut super::Collector) {}
        }
        impl super::Template for B {
            const ARGO_NAME: &'static str = "dup-name";
            const INPUTS: &'static [&'static str] = &[];
            const KIND: super::TemplateKind = super::TemplateKind::Container;
            fn build(_: &super::BuildCtx) -> crate::api::Template {
                Default::default()
            }
            fn collect(_: &mut super::Collector) {}
        }
        let mut c = super::Collector::new();
        assert!(c.enter::<A>());
        assert!(!c.enter::<A>()); // same type re-visit: fine, deduped
        c.enter::<B>(); // different type, same name: panics
    }
}