tatara_process/

intent.rs

//! Intent — where the rendered artifacts come from.
//!
//! Exactly one field on `Intent` must be set. The reconciler's RENDER phase
//! selects a driver based on which variant is present:
//!   - `nix`:        tatara-engine `nix_eval` → resources
//!   - `flux`:       pass through an existing `GitRepository`
//!   - `lisp`:       tatara-lisp reader + macroexpander → resources
//!   - `container`:  emit Deployment/StatefulSet/etc directly (no Helm)
//!   - `aplicacao`:  emit a FluxCD `HelmRelease` for a pleme-io typed
//!                   Aplicacao chart (e.g. `lareira-demo-app`).
//!                   This is the canonical handoff from caixa-shaped
//!                   declarations to in-cluster reconciliation.
//!   - `guest`:      tatara-hospedeiro supervises a Linux VM or WASM
//!                   component. See `tatara/docs/declarative-guests.md`.
//!                   The GuestSpec itself is type-erased here (JSON value)
//!                   so tatara-process stays decoupled from tatara-vm;
//!                   hospedeiro re-parses the value as GuestSpec on boot.

use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

/// Intent — exactly one variant should be populated.
#[derive(Clone, Debug, Default, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct Intent {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub nix: Option<NixIntent>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub flux: Option<FluxIntent>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub lisp: Option<LispIntent>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub container: Option<ContainerIntent>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub aplicacao: Option<AplicacaoIntent>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub guest: Option<GuestIntent>,
}

/// Enum view over the populated variant — convenience for the reconciler.
#[derive(Clone, Debug)]
pub enum IntentVariant<'a> {
    Nix(&'a NixIntent),
    Flux(&'a FluxIntent),
    Lisp(&'a LispIntent),
    Container(&'a ContainerIntent),
    Aplicacao(&'a AplicacaoIntent),
    Guest(&'a GuestIntent),
}

impl IntentVariant<'_> {
    /// Reverse projection — every borrowed variant knows its
    /// `IntentKind` discriminator. Pairs with `IntentKind::select`
    /// so `IntentKind::select(intent).map(|v| v.kind())` round-trips
    /// the closed set; pinned by the substrate testkit
    /// [`crate::tagged_union::assert_variant_round_trip`] shared
    /// across every `<T: TaggedUnion>` implementor. The inherent
    /// method stays load-bearing (the `.kind()` calling convention
    /// pre-dates the trait lift; no consumer needs `use
    /// crate::tagged_union::VariantKind` to reach the reverse
    /// projection) while the trait impl below delegates to this body
    /// as the ground-truth arm-to-Kind mapping.
    pub fn kind(&self) -> IntentKind {
        match self {
            Self::Nix(_) => IntentKind::Nix,
            Self::Flux(_) => IntentKind::Flux,
            Self::Lisp(_) => IntentKind::Lisp,
            Self::Container(_) => IntentKind::Container,
            Self::Aplicacao(_) => IntentKind::Aplicacao,
            Self::Guest(_) => IntentKind::Guest,
        }
    }

    /// Canonical attestation-pillar bytes for the populated variant —
    /// the pre-lift `serde_json::to_vec(<inner>).unwrap_or_default()`
    /// shape every arm restated by hand now rides through the ONE
    /// substrate primitive [`crate::three_pillar::pillar_bytes`],
    /// peer of the four workload-render sites +
    /// `phase_machine::compute_intent_hash` + `identity::content_hash`
    /// consumers post-lift. Each arm names its inner payload ONCE;
    /// the fallback rule lives at the substrate owner. Adding a 7th
    /// intent variant requires only the arm here + one `pillar_bytes`
    /// delegation, not a per-arm fallback restatement.
    pub fn canonical_bytes(&self) -> Vec<u8> {
        match self {
            Self::Nix(n) => crate::three_pillar::pillar_bytes(n),
            Self::Flux(f) => crate::three_pillar::pillar_bytes(f),
            Self::Lisp(l) => crate::three_pillar::pillar_bytes(l),
            Self::Container(c) => crate::three_pillar::pillar_bytes(c),
            Self::Aplicacao(a) => crate::three_pillar::pillar_bytes(a),
            Self::Guest(g) => crate::three_pillar::pillar_bytes(g),
        }
    }
}

impl crate::tagged_union::VariantKind<IntentKind> for IntentVariant<'_> {
    fn variant_kind(&self) -> IntentKind {
        self.kind()
    }
}

/// Closed-set discriminator over `Intent`'s six tagged-union slots.
/// Single source of truth that drives `Intent::variant`'s ambiguity
/// + emptiness resolver, the `IntentError::Empty` message, and the
/// reverse `IntentVariant::kind` projection. Adding a 7th intent
/// variant lands at one `ALL` entry + one `as_str` arm + one
/// `select` arm + one `IntentVariant::kind` arm — exhaustively
/// checked by the compiler.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, tatara_closed_set::DeriveClosedSet)]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum IntentKind {
    Nix,
    Flux,
    Lisp,
    Container,
    Aplicacao,
    Guest,
}

impl IntentKind {
    /// The closed set of intent kinds — single source of truth that
    /// drives `Intent::variant`'s sweep so a variant added without
    /// an `ALL` entry never reaches the resolver.
    pub const ALL: [Self; 6] = [
        Self::Nix,
        Self::Flux,
        Self::Lisp,
        Self::Container,
        Self::Aplicacao,
        Self::Guest,
    ];

    /// Canonical lower-case wire-format key — matches the serde
    /// `rename_all = "camelCase"` field name on `Intent`. The
    /// `IntentError::Empty` message composes the human-readable
    /// list from this projection so a new variant lands in the
    /// operator-facing diagnostic automatically via the `ALL`
    /// sweep, not via hand-maintained error-string drift.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Nix => "nix",
            Self::Flux => "flux",
            Self::Lisp => "lisp",
            Self::Container => "container",
            Self::Aplicacao => "aplicacao",
            Self::Guest => "guest",
        }
    }

    /// Project an `Intent` borrow into the optional typed variant
    /// view for this kind. Returns `None` iff the matching slot is
    /// `None`. Composes the closed-set sweep `Intent::variant`
    /// loops over.
    pub fn select<'a>(self, intent: &'a Intent) -> Option<IntentVariant<'a>> {
        match self {
            Self::Nix => intent.nix.as_ref().map(IntentVariant::Nix),
            Self::Flux => intent.flux.as_ref().map(IntentVariant::Flux),
            Self::Lisp => intent.lisp.as_ref().map(IntentVariant::Lisp),
            Self::Container => intent.container.as_ref().map(IntentVariant::Container),
            Self::Aplicacao => intent.aplicacao.as_ref().map(IntentVariant::Aplicacao),
            Self::Guest => intent.guest.as_ref().map(IntentVariant::Guest),
        }
    }
}

crate::declare_tagged_union_error! {
    pub IntentError,
    empty = "intent has no variant set (one of {0} required)",
    ambiguous = "intent has multiple variants set; exactly one required",
}

/// Slash-joined list of every `IntentKind::as_str()` — composed once
/// at compile time so `IntentError::Empty`'s diagnostic carries the
/// closed-set summary without per-variant string drift. Pinned against
/// the canonical [`tatara_lisp::ClosedSet::labels_joined`] projection
/// by `intent_error_empty_lists_every_kind_in_canonical_order`, so a
/// regression that drifts this `&'static str` constant from the
/// `IntentKind::ALL × as_str` composition fails-loudly at the test
/// site without per-variant inline materialization.
pub(crate) const INTENT_KIND_LIST: &str = "nix/flux/lisp/container/aplicacao/guest";

// `impl FromStr for IntentKind` +
// `impl tatara_lisp::ClosedSet for IntentKind` +
// `impl fmt::Display for IntentKind` +
// `pub struct UnknownIntentKind(pub String)` are all generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` +
// `#[closed_set(via = "as_str", generate_unknown, display)]` on the
// enum declaration above. `label` delegates to the inherent
// `IntentKind::as_str` — the camelCase wire-vocabulary projection
// stays load-bearing (matches the serde `rename_all = "camelCase"`
// field names on `Intent` AND the `IntentVariant::canonical_bytes`
// per-variant arm), while generic `T: ClosedSet` consumers reach the
// STABLE workspace-wide name (`label`). The auto-derived carrier
// label "intent kind" matches the substrate-wide
// `#[error("unknown intent kind: {0}")]` shape every sibling
// closed-set carrier across `tatara-process` renders verbatim.
// Symmetric to [`crate::intent::WorkloadKind`] (the workload-axis
// sibling on the same `ProcessSpec` slice) and every other
// `#[derive(DeriveClosedSet)]` implementor across the crate.

crate::declare_tagged_union_impls! {
    parent = Intent,
    kind = IntentKind,
    variant = IntentVariant,
    error = IntentError,
    kind_list = INTENT_KIND_LIST,
}

/// Nix-sourced intent — tatara-engine's nix_eval driver produces resources.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct NixIntent {
    /// Flake reference, e.g., `github:pleme-io/k8s?dir=shared/infrastructure`.
    pub flake_ref: String,
    /// Attribute path within the flake (e.g., `observability`).
    pub attribute: String,
    /// Target system. Defaults to the controller host's system.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub system: Option<String>,
    /// Attic cache to push the resulting store path into.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub attic_cache: Option<String>,
    /// Additional `nix build` arguments (e.g., `["--impure"]`).
    #[serde(default)]
    pub extra_args: Vec<String>,
    /// Delegate the actual build to a sibling NixBuild CRD
    /// (bridges to tatara-operator NATS bare-metal builder path).
    #[serde(default)]
    pub delegate_to_nix_build: bool,
}

/// FluxCD passthrough intent — reuse an existing GitRepository.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct FluxIntent {
    /// Name of an existing `GitRepository` (typically in `flux-system`).
    pub git_repository: String,
    /// Path inside the repository that the Kustomization will apply.
    pub path: String,
    /// Optional namespace of the GitRepository CR (defaults to `flux-system`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub git_repository_namespace: Option<String>,
    /// Optional target namespace for the emitted Kustomization.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target_namespace: Option<String>,
    /// SOPS decryption — defaults to true to match pleme-io conventions.
    #[serde(default = "default_true")]
    pub decrypt_sops: bool,
    /// If set, additionally emit a HelmRelease for this chart.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub helm_chart: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub helm_values: Option<BTreeMap<String, serde_json::Value>>,
}

fn default_true() -> bool {
    true
}

/// Lisp-sourced intent — tatara-lisp reader + macroexpander produces resources.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct LispIntent {
    /// Raw S-expression source, OR `include:<path>` / `configmap:<name>/<key>` pointer.
    pub source: String,
    /// Reader dialect / version tag.
    #[serde(default = "default_reader")]
    pub reader: String,
    /// Macro form version.
    #[serde(default = "default_version")]
    pub version: String,
    /// Symbols injected into the reader env (e.g., `cluster`, `region`).
    #[serde(default)]
    pub bindings: BTreeMap<String, serde_json::Value>,
}

fn default_reader() -> String {
    "tatara-lisp".to_string()
}
fn default_version() -> String {
    "v1".to_string()
}

/// Aplicacao intent — emit a FluxCD `HelmRelease` for a pleme-io
/// typed Aplicacao chart. The chart owns its own sub-chart DAG;
/// the reconciler only watches `HelmRelease.status.conditions[type=Ready]`.
///
/// This is the canonical handoff from caixa `(defaplicacao …)` declarations
/// (which the typescape renders to this Intent) into in-cluster
/// reconciliation. Closed-loop ephemeral test environments use this
/// variant with `:lifetime :ephemeral` on the surrounding ProcessSpec.
///
/// Example (Lisp):
/// ```lisp
/// :intent (:aplicacao
///           (:chart-ref "oci://ghcr.io/pleme-io/charts/lareira-demo-app"
///            :version "0.5.5"
///            :profile "all-in-one"
///            :values-overlay (:cluster (:name "ephemeral-test-01")
///                             :persistence false
///                             :compliance (:overlays []))))
/// ```
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct AplicacaoIntent {
    /// Helm chart reference. OCI (`oci://…`) or repo-relative (`pleme-io/lareira-demo-app`).
    pub chart_ref: String,
    /// Chart version (Helm semver constraint; `">=0.5.5"` allowed).
    pub version: String,
    /// Architecture profile from the chart's `values/*.yaml` family
    /// (e.g. `all-in-one`, `saas-internal`).
    /// Leave empty to use chart defaults.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub profile: String,
    /// Typed values overlay merged on top of the profile.
    /// Free-form JSON to keep tatara-process decoupled from chart schemas.
    #[serde(default)]
    #[schemars(schema_with = "crate::schema_helpers::preserve_unknown_object")]
    pub values_overlay: serde_json::Value,
    /// HelmRelease name override. Defaults to the Process's PID-derived name.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub release_name: Option<String>,
    /// Target namespace for the chart. Defaults to the Process's namespace.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target_namespace: Option<String>,
    /// Install timeout (`humantime` duration). Empty = chart-controller default.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub install_timeout: Option<String>,
}

/// Workspace-wide default for the `timeout` slot on a Flux
/// `HelmRelease.spec.{install,upgrade}` block, applied when the
/// operator did not populate [`AplicacaoIntent::install_timeout`].
/// Load-bearing on the reconciler's Helm-driven RENDER surface —
/// [`AplicacaoIntent::helm_lifecycle_policy`] substitutes this exact
/// string, and the reconciler's `render_aplicacao` byte-installs
/// the resulting policy into both install AND upgrade slots.
pub const HELM_LIFECYCLE_DEFAULT_TIMEOUT: &str = "25m";

/// Workspace-wide default for the `remediation.retries` slot on a
/// Flux `HelmRelease.spec.{install,upgrade}` block. Constant across
/// both slots today; a future two-slot split (e.g. distinct retry
/// budgets for a first install vs a rolling upgrade) lands as two
/// consts here + a two-slot [`HelmLifecyclePolicy`] shape, not at
/// the render callsite.
pub const HELM_LIFECYCLE_DEFAULT_RETRIES: u8 = 3;

/// Workspace-wide default for the reconcile-loop cadence on both Flux
/// resources a Helm-driven `AplicacaoIntent` publishes today: the
/// `OCIRepository.spec.interval` on the source side (how often the
/// source-controller re-pulls the chart from OCI) and the
/// `HelmRelease.spec.interval` on the release side (how often the
/// helm-controller re-reconciles the release against the chart).
/// The fleet convention ties both cadences to the same `5m` string
/// today, so the substrate exposes ONE named const rather than two
/// literals sprayed across `render_aplicacao`.
///
/// Peer to [`HELM_LIFECYCLE_DEFAULT_TIMEOUT`] on the same
/// AplicacaoIntent-facing "workspace-wide Flux default" axis. A
/// future per-slot divergence (`SOURCE_INTERVAL` vs `RELEASE_INTERVAL`
/// as two consts, or a two-slot method returning a
/// `FluxReconcileIntervals { source, release }` shape) lands here,
/// NOT at the two render callsites.
///
/// Load-bearing wire-format string: the byte-exact `5m` shape is
/// what the Flux source- and helm-controllers parse via `humantime`;
/// a regression that renamed it to any other duration would silently
/// throttle or hammer every Helm-driven Process's reconciliation
/// loop. Pinned at
/// [`tests::flux_helm_default_interval_is_pinned_to_5m`].
pub const FLUX_HELM_DEFAULT_INTERVAL: &str = "5m";

/// Typed shape of one Flux `HelmRelease.spec.{install,upgrade}` slot
/// — the substrate's projection of the "how long may Helm take, and
/// how many retries after a failed run" contract every Helm-driven
/// Process publishes on both slots. Pre-lift the reconciler's
/// `render_aplicacao` hand-authored the shape via TWO adjacent
/// identical `json!({"timeout": …, "remediation": {"retries": …}})`
/// blocks past the ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold
/// — one for install, one for upgrade, each restating the same
/// three-slot literal with the same `Option::unwrap_or_else` fallback
/// on the timeout. Post-lift the shape lives at ONE named typed
/// struct here whose serde projection matches Flux HelmRelease v2's
/// `install` / `upgrade` block schema byte-identically, and the
/// reconciler composes both slots off ONE
/// [`AplicacaoIntent::helm_lifecycle_policy`] call.
///
/// A future addition — a `wait: bool` slot, a `crds:
/// CreateReplace` slot, a `disableOpenAPIValidation: bool` slot,
/// a two-slot split that lets install carry a longer timeout than
/// upgrade — lands at ONE struct here and every downstream
/// consumer (the render surface, snapshot tests, an operator-
/// facing dashboard column, a future validating webhook) inherits
/// the upgrade mechanically.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
pub struct HelmLifecyclePolicy {
    /// Chart-controller timeout (`humantime` duration). Set from
    /// [`AplicacaoIntent::install_timeout`] when present; otherwise
    /// [`HELM_LIFECYCLE_DEFAULT_TIMEOUT`].
    pub timeout: String,
    /// Retry budget for the slot.
    pub remediation: HelmRemediationPolicy,
}

/// Typed shape of one `HelmLifecyclePolicy::remediation` slot.
/// A named struct rather than an inline `{retries: u8}` map so
/// downstream consumers can talk about "one Helm remediation
/// policy" as a nameable handle rather than an unnamed nested
/// object.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
pub struct HelmRemediationPolicy {
    /// Number of times Flux's helm-controller retries a failed
    /// install / upgrade before surfacing the failure to the parent
    /// Process's boundary evaluator.
    pub retries: u8,
}

impl HelmLifecyclePolicy {
    /// The workspace-wide default policy — used when the operator
    /// omitted [`AplicacaoIntent::install_timeout`]. Named projection
    /// of the two `HELM_LIFECYCLE_DEFAULT_*` consts so a future
    /// consumer wanting "the substrate's fresh-out-of-the-box Helm
    /// lifecycle policy" pulls the pair through ONE call rather than
    /// composing the struct by hand at every callsite.
    pub fn workspace_default() -> Self {
        Self {
            timeout: HELM_LIFECYCLE_DEFAULT_TIMEOUT.to_string(),
            remediation: HelmRemediationPolicy {
                retries: HELM_LIFECYCLE_DEFAULT_RETRIES,
            },
        }
    }
}

impl AplicacaoIntent {
    /// Chart-pointer-only composer — the canonical minimal
    /// `AplicacaoIntent` every fixture and default-shape callsite
    /// restated pre-lift by binding only `(chart_ref, version)` and
    /// leaving every remaining slot at its K8s-schema-default value
    /// (`profile = ""`, `values_overlay = Value::Null`,
    /// `release_name = target_namespace = install_timeout = None`).
    ///
    /// Pre-lift the 7-slot struct-literal was hand-authored at 14
    /// workspace-wide sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication threshold, each restating the SAME 5-slot default
    /// tail after a caller-varying `(chart_ref, version)` pair:
    ///
    /// * `tatara-process` — 10 sites across `lib.rs` (3× `empty_template`
    ///   in the ephemeral-spec / matrix / observer test blocks),
    ///   `lifetime_clock.rs` (2× the ephemeral / permanent Process
    ///   composer helpers), `tagged_union.rs` (1× the
    ///   `IntentKind::Aplicacao` sample-intent arm), `pool.rs` (1×
    ///   `empty_template`), and `intent.rs` (3× the `helm_intent`
    ///   helper + the `aplicacao_plus_flux_is_ambiguous` fixture + the
    ///   `IntentKind::Aplicacao` sample-intent-for arm).
    /// * `tatara-pool-reconciler` — 4 sites across `router.rs`,
    ///   `pool_decide.rs`, `desired.rs`, `allocation_decide.rs`
    ///   (all `empty_template` fixtures seeding the pool + allocation
    ///   convergence-decision test batteries).
    /// * `tatara-reconciler` — 1 site in `render.rs`
    ///   (`helmrepository_chartref_for_non_oci` production-shape
    ///   fixture stamping the non-OCI chart-ref render path).
    ///
    /// All 14 sites walked the SAME 5-slot default tail — differing
    /// only in the caller-varying `chart_ref` / `version` values
    /// (`"oci://x"` / `"1"` on the majority test-fixture slice,
    /// `"oci://ghcr.io/x"` / `"0.1.0"` on the `IntentKind` sample,
    /// `"pleme-io/lareira-demo-app"` / `"0.5.5"` on the non-OCI
    /// render pin). Post-lift each callsite reads
    /// `AplicacaoIntent::chart_only(chart, version)` and the 5-slot
    /// default tail lives at ONE substrate owner.
    ///
    /// The `impl Into<String>` argument form matches the pre-lift
    /// call shape — every site that spelled `"oci://x".into()` +
    /// `"1".into()` in the struct-literal continues to compile
    /// unchanged, and callers with a live `String` (e.g. reading
    /// from a caller-supplied fixture parameter) pass it through
    /// without a `.to_string()` re-wrap.
    ///
    /// Return-form axis: `AplicacaoIntent` — the owned typed value
    /// every consumer's downstream `Intent { aplicacao: Some(...), ..
    /// }` / `EphemeralSpec { aplicacao: ..., .. }` binding stamps
    /// verbatim, matching the pre-lift 7-slot struct-literal's return
    /// shape exactly. The five default-tail slots reify the K8s
    /// schema's own defaults (`profile` empty ⇒ chart profile default;
    /// `values_overlay = Null` ⇒ pass-through; three `None` slots ⇒
    /// server / chart-computed) so no consumer inherits a semantic
    /// change from the lift.
    ///
    /// A future normalization of the "minimal AplicacaoIntent" default
    /// — an added struct field with its own K8s-schema default, a
    /// tightening of a `None` slot to a substrate-owned default value,
    /// a per-workspace-default `install_timeout` override — lands at
    /// THIS ONE substrate primitive and every downstream fixture /
    /// default-shape consumer inherits the upgrade mechanically — no
    /// per-site edit at any of the 14 listed callers or at future
    /// consumers (a new pool-shard-flavor fixture, a new intent-axis
    /// convergence probe, a per-tenant AplicacaoIntent minimal seed).
    ///
    /// Peer to [`Self::helm_lifecycle_policy`] +
    /// [`Self::flux_reconcile_interval`] on the (COMPOSE, DERIVE)
    /// axis: `chart_only` is the WRITE-side composer (stamp a minimal
    /// `AplicacaoIntent`); the lifecycle + interval methods are the
    /// READ-side derivers (project a substrate-default Flux policy /
    /// cadence off an `AplicacaoIntent`). The three primitives
    /// partition the `AplicacaoIntent` compose × derive surface at
    /// the substrate.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the 5-slot default tail recurred at 14 hand-authored sites
    /// past the ★★ PRIME-DIRECTIVE ≥ 2 duplication trigger, spanning
    /// three workspace crates, and is lifted onto the ONE workspace-
    /// wide substrate owner here). THEORY.md §II.1 invariant 5
    /// (composition preserves proofs — the pin block below binds the
    /// primitive at fail-before-pass-after granularity so a
    /// regression that drifted any of the five default-tail slots
    /// surfaces at THESE pins rather than as silent fixture skew
    /// across the 14 downstream consumers).
    #[must_use]
    pub fn chart_only(chart_ref: impl Into<String>, version: impl Into<String>) -> Self {
        Self {
            chart_ref: chart_ref.into(),
            version: version.into(),
            profile: String::new(),
            values_overlay: serde_json::Value::Null,
            release_name: None,
            target_namespace: None,
            install_timeout: None,
        }
    }

    /// Effective HelmRelease name — the operator-supplied
    /// [`Self::release_name`] override when set, else the caller-
    /// supplied fallback (canonically the Process's PID-derived name
    /// at the reconciler's `render_aplicacao` site, and the enclosing
    /// matrix env's `NamedEphemeral.name` at the matrix
    /// `breathe_bands` site — both are the Process's own `name`).
    ///
    /// Pre-lift the pattern was hand-authored at 2 workspace-wide
    /// sites past the ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold,
    /// each restating the SAME `Option::clone().unwrap_or_else(||
    /// fallback.into())` chain differing only in the caller-varying
    /// fallback slot:
    ///
    /// * `tatara-reconciler::render::render_aplicacao` — the
    ///   `HelmRelease.spec.releaseName` slot on the Flux-owned
    ///   release the reconciler emits per Helm-driven Process.
    /// * `tatara-process::matrix::EnvMatrixSpec::breathe_bands` —
    ///   the `spec.targetRef.name` slot on the breathe Band CRs the
    ///   matrix sweep emits per generated env × dimension.
    ///
    /// Post-lift both sites read `a.release_name_or(fallback)` and
    /// the fallback-composition shape lives at ONE substrate owner.
    /// A future tightening — a workspace-wide default seeded off the
    /// PID, a fallback shape derived off `Self::chart_ref`, a
    /// validation that the fallback is a DNS-1123 label — lands at
    /// THIS ONE primitive.
    ///
    /// Peer to [`Self::target_namespace_or`] on the same (release,
    /// namespace) axis: both project one operator-optional
    /// override-or-fallback slot the reconciler + matrix consume in
    /// lock-step.
    #[must_use]
    pub fn release_name_or(&self, fallback: &str) -> String {
        self.release_name.clone().unwrap_or_else(|| fallback.into())
    }

    /// Effective HelmRelease target namespace — the operator-supplied
    /// [`Self::target_namespace`] override when set, else the caller-
    /// supplied fallback (canonically the Process's own namespace at
    /// the reconciler's `render_aplicacao` site, and the enclosing
    /// matrix env's `NamedEphemeral.name` at the matrix
    /// `breathe_bands` site).
    ///
    /// Peer to [`Self::release_name_or`] — same substrate-owner
    /// motivation, same 2-site collapse. Post-lift both consumers
    /// read `a.target_namespace_or(fallback)` and the
    /// `Option::clone().unwrap_or_else(|| fallback.into())` shape
    /// lives at ONE primitive here.
    #[must_use]
    pub fn target_namespace_or(&self, fallback: &str) -> String {
        self.target_namespace
            .clone()
            .unwrap_or_else(|| fallback.into())
    }

    /// Derive the Flux `HelmRelease.spec.{install,upgrade}` policy
    /// this intent publishes on BOTH slots. Pre-lift the reconciler's
    /// `render_aplicacao` restated the shape by hand via two adjacent
    /// identical `json!` blocks (install and upgrade); post-lift both
    /// slots ride through this ONE composer. A future two-slot split
    /// (distinct install vs upgrade policies) lands as a two-method
    /// pair here, not at the render callsite.
    pub fn helm_lifecycle_policy(&self) -> HelmLifecyclePolicy {
        HelmLifecyclePolicy {
            timeout: self
                .install_timeout
                .clone()
                .unwrap_or_else(|| HELM_LIFECYCLE_DEFAULT_TIMEOUT.to_string()),
            remediation: HelmRemediationPolicy {
                retries: HELM_LIFECYCLE_DEFAULT_RETRIES,
            },
        }
    }

    /// Derive the Flux reconcile-loop cadence this intent publishes on
    /// BOTH `OCIRepository.spec.interval` (source-controller poll) and
    /// `HelmRelease.spec.interval` (helm-controller re-reconcile).
    /// Pre-lift the reconciler's `render_aplicacao` restated the value
    /// via two adjacent hand-authored `"5m"` string literals past the
    /// ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold; post-lift both
    /// slots ride through this ONE composer. A future divergence
    /// (distinct per-slot cadences, a per-intent override field, a
    /// two-slot method returning a `FluxReconcileIntervals` shape)
    /// lands at ONE method here, not at the render callsites.
    ///
    /// Sibling composer to [`Self::helm_lifecycle_policy`]: both
    /// return the substrate-default shape a Helm-driven Process
    /// publishes on the Flux resources `render_aplicacao` emits,
    /// keyed off the same `AplicacaoIntent`.
    pub fn flux_reconcile_interval(&self) -> String {
        FLUX_HELM_DEFAULT_INTERVAL.to_string()
    }
}

/// Container intent — direct Deployment/StatefulSet/etc, no Helm.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ContainerIntent {
    pub image: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub replicas: Option<i32>,
    #[serde(default)]
    pub command: Vec<String>,
    #[serde(default)]
    pub args: Vec<String>,
    #[serde(default)]
    pub env: BTreeMap<String, String>,
    #[serde(default)]
    pub workload_kind: WorkloadKind,
}

/// K8s workload kind the `container` intent renders into. PascalCase
/// values match the K8s `kind:` field on the emitted manifest verbatim,
/// so `as_str` doubles as the canonical `kind:` projection at render time.
#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    Hash,
    Serialize,
    Deserialize,
    JsonSchema,
    Default,
    tatara_closed_set::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum WorkloadKind {
    #[default]
    Deployment,
    StatefulSet,
    DaemonSet,
    Job,
    CronJob,
}

impl WorkloadKind {
    /// The closed set of workload kinds — single source of truth that
    /// drives the `as_str` / Display / `FromStr` triad and the typed
    /// `api_version` / `is_batch` projections. Adding a sixth variant
    /// lands at one `ALL` entry + one `as_str` arm + one arm in each
    /// projection — exhaustively checked by the compiler (the `[Self; 5]`
    /// array literal forces the arity).
    ///
    /// Sibling closed-set lifts on the same `ProcessSpec` axis:
    /// [`crate::encapsulates::EncapsulationMode::ALL`],
    /// [`crate::export::ExportTrigger::ALL`],
    /// [`crate::export::ReportFormat::ALL`],
    /// [`crate::lifetime::TeardownPolicy::ALL`],
    /// [`crate::intent::IntentKind::ALL`],
    /// [`crate::lifetime::LifetimeKind::ALL`],
    /// [`crate::boundary::ConditionKind::ALL`],
    /// [`crate::phase::ProcessPhase::ALL`],
    /// [`crate::signal::ProcessSignal::ALL`].
    pub const ALL: [Self; 5] = [
        Self::Deployment,
        Self::StatefulSet,
        Self::DaemonSet,
        Self::Job,
        Self::CronJob,
    ];

    /// Canonical PascalCase wire-format projection — matches the serde
    /// `rename_all = "PascalCase"` output verbatim AND the K8s manifest
    /// `kind:` field the `container` intent's future renderer will emit.
    /// Used by Display (single source of truth), by `FromStr` to identify
    /// the variant from its annotation / status-field representation, and
    /// by operator-facing reason strings without reaching for `{:?}` Debug
    /// formatting. Pinned by `workload_kind_as_str_matches_serde`.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Deployment => "Deployment",
            Self::StatefulSet => "StatefulSet",
            Self::DaemonSet => "DaemonSet",
            Self::Job => "Job",
            Self::CronJob => "CronJob",
        }
    }

    /// Canonical K8s `apiVersion:` projection — `apps/v1` for the
    /// long-running workload trio, `batch/v1` for the batch pair.
    /// Single source of truth for the apiVersion the `container` intent
    /// renderer will stamp on the emitted manifest; pinned by
    /// `workload_kind_projection_truth_table` so a future variant lands
    /// at one arm here, not at every render site that previously
    /// hand-rolled `match kind { Job | CronJob => "batch/v1", _ => … }`.
    ///
    /// Closed-set match (not `matches!`) so adding a sixth variant
    /// triggers the compiler's exhaustiveness check at this site
    /// rather than silently defaulting to either group.
    pub const fn api_version(self) -> &'static str {
        match self {
            Self::Deployment | Self::StatefulSet | Self::DaemonSet => "apps/v1",
            Self::Job | Self::CronJob => "batch/v1",
        }
    }

    /// True iff the workload kind is a batch (terminating) workload —
    /// `Job` or `CronJob`. Drives the future container renderer's
    /// decision between persistent / one-shot retry semantics and lets
    /// the lifetime clock distinguish "naturally terminates" from "runs
    /// until SIGTERM" without re-deriving the partition from
    /// `api_version() == "batch/v1"`.
    ///
    /// Closed-set match (not `matches!`) so adding a sixth variant
    /// triggers the compiler's exhaustiveness check at this site.
    pub const fn is_batch(self) -> bool {
        match self {
            Self::Job | Self::CronJob => true,
            Self::Deployment | Self::StatefulSet | Self::DaemonSet => false,
        }
    }
}

// `impl FromStr for WorkloadKind` +
// `impl tatara_lisp::ClosedSet for WorkloadKind` +
// `impl fmt::Display for WorkloadKind` +
// `pub struct UnknownWorkloadKind(pub String)` are all generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` +
// `#[closed_set(via = "as_str", generate_unknown, display)]` on the
// enum declaration above. `label` delegates to the inherent
// `WorkloadKind::as_str` — the PascalCase wire-vocabulary projection
// stays load-bearing (matches the serde `rename_all = "PascalCase"`
// output AND the K8s manifest `kind:` field verbatim), while generic
// `T: ClosedSet` consumers reach the STABLE workspace-wide name
// (`label`). The auto-derived carrier label "workload kind" matches
// the prior hand-rolled `#[error("unknown workload kind: {0}")]`
// annotation byte-for-byte. Symmetric to every other
// `#[derive(DeriveClosedSet)]` implementor across the crate.

/// Guest intent — the Process is a Linux VM or WASM component supervised
/// by `tatara-hospedeiro`. See `tatara/docs/declarative-guests.md`.
///
/// The actual `GuestSpec` is stored as a serde JSON value to keep
/// `tatara-process` decoupled from `tatara-vm`. Hospedeiro re-parses
/// the value as the concrete `tatara_vm::GuestSpec` at boot time; a
/// round-trip test on the tatara-vm side guarantees the shape.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct GuestIntent {
    /// The (defguest …) spec as JSON. Shape matches `tatara_vm::GuestSpec`.
    #[schemars(schema_with = "crate::schema_helpers::preserve_unknown_object")]
    pub spec: serde_json::Value,

    /// Where to write per-guest state on the host (logs, socket, PID file).
    /// Defaults to `~/.local/state/tatara/guests/<name>/`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub state_dir: Option<String>,

    /// Whether hospedeiro is allowed to pull guest artifacts from a remote
    /// transport (Attic, ssh-ng) if not already present locally. The
    /// default is taken from the GuestSpec's `buildOn` field; setting
    /// this explicitly overrides at the intent layer.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub allow_remote_build: Option<bool>,
}

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

    #[test]
    fn empty_intent_errors() {
        let i = Intent::default();
        match i.variant().unwrap_err() {
            IntentError::Empty(list) => assert_eq!(list, INTENT_KIND_LIST),
            other => panic!("expected Empty, got {other:?}"),
        }
    }

    #[test]
    fn exactly_one_ok() {
        let i = Intent {
            nix: Some(NixIntent {
                flake_ref: "github:a/b".into(),
                attribute: "x".into(),
                system: None,
                attic_cache: None,
                extra_args: vec![],
                delegate_to_nix_build: false,
            }),
            ..Intent::default()
        };
        assert!(matches!(i.variant().unwrap(), IntentVariant::Nix(_)));
    }

    /// AMBIGUOUS-PATH CONTRACT: when two slots are populated the
    /// resolver yields `Ambiguous`, exhaustively across every pair in
    /// `ALL × ALL` (excluding the diagonal). Routes through the
    /// substrate primitive
    /// [`crate::tagged_union::assert_two_slots_ambiguous`] shared with
    /// the sibling
    /// `encapsulation_kind_two_slots_is_ambiguous_across_every_pair`
    /// / `artifact_source_two_slots_is_ambiguous_across_every_pair`
    /// / `vector_channel_two_slots_is_ambiguous_across_every_pair`
    /// sites. Subsumes the pre-lift hand-authored two-pair probes
    /// (`nix + flux`, `nix + guest`) with exhaustive `6 × 5 = 30`
    /// coverage — every off-diagonal pair on `IntentKind` is pinned.
    #[test]
    fn intent_two_slots_is_ambiguous_across_every_pair() {
        crate::tagged_union::assert_two_slots_ambiguous::<Intent, _>(two_slot_intent);
    }

    #[test]
    fn guest_intent_selects_its_variant() {
        let i = Intent {
            guest: Some(GuestIntent {
                spec: serde_json::json!({
                    "name": "fast-fn",
                    "kind": { "kind": "wasm", "runtime": "wasmtime",
                              "wasiPreview": "p2",
                              "component": { "kind": "flake",
                                             "value": {"url":"github:x/y","attr":"wasi"} },
                              "features": { "simd": true } },
                    "cmdline": []
                }),
                state_dir: None,
                allow_remote_build: Some(true),
            }),
            ..Intent::default()
        };
        match i.variant().unwrap() {
            IntentVariant::Guest(g) => {
                assert_eq!(g.spec["name"], "fast-fn");
                assert_eq!(g.allow_remote_build, Some(true));
            }
            other => panic!("expected Guest, got {other:?}"),
        }
    }

    #[test]
    fn aplicacao_intent_selects_its_variant() {
        let i = Intent {
            aplicacao: Some(AplicacaoIntent {
                chart_ref: "oci://ghcr.io/pleme-io/charts/lareira-demo-app".into(),
                version: "0.5.5".into(),
                profile: "all-in-one".into(),
                values_overlay: serde_json::json!({ "cluster": { "name": "test-01" } }),
                release_name: None,
                target_namespace: None,
                install_timeout: Some("25m".into()),
            }),
            ..Intent::default()
        };
        match i.variant().unwrap() {
            IntentVariant::Aplicacao(a) => {
                assert_eq!(a.profile, "all-in-one");
                assert_eq!(a.version, "0.5.5");
                assert_eq!(a.install_timeout.as_deref(), Some("25m"));
            }
            other => panic!("expected Aplicacao, got {other:?}"),
        }
    }

    /// Structural well-formedness of [`IntentKind`] as a
    /// [`tatara_lisp::ClosedSet`] implementor — the workspace-wide
    /// testkit lift that pins all structural invariants (`ALL` is
    /// non-empty, every variant round-trips through `label ↔
    /// parse_label`, labels are pairwise distinct, `""` is outside
    /// the closed set, the `UnknownIntentKind` carrier's Display
    /// renders the substrate-wide `"unknown intent kind: <input>"`
    /// shape, `labels()` equals the natural `ALL × label` projection,
    /// `parse_label_with_hint` composes `parse_label` +
    /// `suggest_closest` verbatim) at ONE call site. Replaces the
    /// hand-derived `intent_kind_all_is_unique_and_complete` —
    /// clause (1)+(3) of the testkit subsume the uniqueness +
    /// non-emptiness sweep that test pinned independently.
    #[test]
    fn intent_kind_is_well_formed_closed_set() {
        tatara_closed_set::assert_closed_set_well_formed::<IntentKind>();
    }

    /// The Display impl IS `as_str` — pinning this lets future callers
    /// reach for either projection without drift. Symmetric to the
    /// sibling `workload_kind_display_matches_as_str` invariant; if a
    /// reviewer accidentally re-introduces an inline match in Display,
    /// this test would fail the moment a variant rename touches one
    /// site but not the other.
    ///
    /// Routes through the substrate primitive
    /// [`crate::tagged_union::assert_display_matches_label`], which
    /// composes `<T as ClosedSet>::label` against `T::to_string`
    /// byte-identically for every `<T: ClosedSet + Display>`
    /// implementor — the Display-alignment testkit shared with every
    /// sibling `X_display_matches_as_str` site across the crate.
    /// Pre-lift the 27 bodies each restated the same
    /// `for k in K::ALL { assert_eq!(k.to_string(), k.as_str()) }`
    /// two-line probe at the test surface; post-lift the projection
    /// lives at ONE substrate primitive and every site binds through
    /// a single call.
    #[test]
    fn intent_kind_display_matches_as_str() {
        crate::tagged_union::assert_display_matches_label::<IntentKind>();
    }

    /// CANONICAL-KEY CONTRACT: each variant's `as_str()` matches the
    /// camelCase serde field name on `Intent`. A future rename of
    /// any field lands here at one site — and the `Empty` diagnostic
    /// composed from `INTENT_KIND_LIST` stays coherent with the
    /// wire format.
    ///
    /// Routes through the substrate primitive
    /// [`crate::tagged_union::assert_single_slot_key_matches_label`],
    /// which pins the exactly-one-key + name-equality projection
    /// byte-identically for every `<T: TaggedUnion + Serialize>`
    /// implementor — the wire-alignment testkit shared with the sibling
    /// `encapsulation_target_as_str_matches_field_name` /
    /// `artifact_kind_as_str_matches_field_name` /
    /// `channel_kind_as_str_matches_field_name` sites. Pre-lift the
    /// four bodies each restated the same serialize-and-inspect sweep
    /// at the test surface (three through a weaker YAML-substring
    /// check; this site alone through the strong JSON-object exactly-
    /// one form); post-lift the projection lives at ONE substrate
    /// primitive and every site binds through a single call — the
    /// three YAML sites simultaneously upgrade to the strong exactly-
    /// one form.
    #[test]
    fn intent_kind_as_str_matches_intent_field_name() {
        crate::tagged_union::assert_single_slot_key_matches_label::<Intent, _>(single_slot_intent);
    }

    /// ROUND-TRIP CONTRACT: `IntentKind::select(intent).map(|v|
    /// v.kind()) == Some(kind)`. The reverse `IntentVariant::kind`
    /// projection composes the closed set in both directions — a
    /// regression that misroutes a select arm (e.g. `Self::Nix =>
    /// intent.flux.as_ref()...`) fails loudly here.
    ///
    /// Routes through the substrate primitive
    /// [`crate::tagged_union::assert_variant_round_trip`], which
    /// composes [`crate::tagged_union::VariantSelector::select`]
    /// (forward) with [`crate::tagged_union::VariantKind::variant_kind`]
    /// (reverse) byte-identically for every `<T: TaggedUnion>`
    /// implementor — the round-trip testkit shared with the sibling
    /// `artifact_kind_round_trips_through_variant_kind` /
    /// `channel_kind_round_trips_through_variant_kind` /
    /// `encapsulation_target_round_trips_through_variant_target`
    /// sites. Pre-lift the four bodies each restated the same
    /// two-arm round-trip probe at the test surface; post-lift the
    /// projection lives at ONE substrate primitive and every site
    /// binds through a single call.
    #[test]
    fn intent_kind_round_trips_through_variant_kind() {
        crate::tagged_union::assert_variant_round_trip::<Intent, _>(single_slot_intent);
    }

    /// PRESENCE-PROBE WIRE CONTRACT: the `intent-<kind>` require-tag
    /// dispatcher in `tatara-check` (`bin/tatara-check.rs`) parses
    /// each suffix via `IntentKind::from_str` and dispatches through
    /// the substrate primitive `Intent::has` (a one-line inherent
    /// forwarder over [`crate::tagged_union::TaggedUnion::has`]).
    /// Pre-lift the dispatcher restated five hand-authored
    /// `spec.intent.<field>.is_some()` arms whose per-field addressing
    /// drifted from `IntentKind::ALL` (the sixth variant `Guest` had
    /// no `intent-guest` arm at all); post-lift adding a seventh
    /// variant to `IntentKind` lands the corresponding `intent-<kind>`
    /// tag automatically — the sweep here pins that every
    /// `IntentKind` roundtrips through the `intent-{as_str}` wire
    /// key, and that `Intent::has(k)` fires exactly on the populated
    /// slot addressed by `k`.
    #[test]
    fn intent_has_dispatches_through_wire_key_across_every_kind() {
        for populated in IntentKind::ALL {
            let intent = single_slot_intent(populated);
            for probed in IntentKind::ALL {
                let wire_key = format!("intent-{}", probed.as_str());
                let parsed: IntentKind = wire_key
                    .strip_prefix("intent-")
                    .expect("wire key composes as intent-<as_str>")
                    .parse()
                    .expect("as_str→from_str round trip pinned by DeriveClosedSet");
                assert_eq!(parsed, probed);
                let expected = probed == populated;
                assert_eq!(
                    intent.has(probed),
                    expected,
                    "Intent::has drift — populated={populated:?} probed={probed:?}",
                );
            }
        }
    }

    /// EMPTY-DIAGNOSTIC CONTRACT: the closed-set kind list embedded
    /// in `IntentError::Empty` echoes the canonical join of every
    /// `IntentKind::as_str()` projection. A variant added without
    /// updating `INTENT_KIND_LIST` (or a renamed variant) shows up
    /// here as a mismatch.
    ///
    /// Routes through the substrate primitive
    /// [`crate::tagged_union::assert_kind_list_matches_closed_set`],
    /// which composes `<T::Kind as ClosedSet>::labels_joined("/")`
    /// against `<T as TaggedUnion>::KIND_LIST` byte-identically for
    /// every implementor — the diagnostic-stability testkit shared
    /// with the sibling `artifact_error_empty_lists_every_kind_in_canonical_order`
    /// / `channel_error_empty_lists_every_kind_in_canonical_order`
    /// / `encapsulation_kind_error_empty_lists_every_target_in_canonical_order`
    /// sites. Pre-lift the four bodies each restated the same
    /// two-argument `assert_eq!(<XxxKind as ClosedSet>::labels_joined("/"),
    /// XXX_KIND_LIST)` comparison at the test surface; post-lift
    /// the projection lives at ONE substrate primitive and every
    /// site binds through a single call.
    #[test]
    fn intent_error_empty_lists_every_kind_in_canonical_order() {
        crate::tagged_union::assert_kind_list_matches_closed_set::<Intent>();
    }

    /// CANONICAL-BYTES CONTRACT: every populated variant yields the
    /// SAME bytes as `serde_json::to_vec` on the inner reference.
    /// Pins the lift of the parallel observe-mode match in
    /// `tatara-reconciler::render` to this single method.
    #[test]
    fn intent_variant_canonical_bytes_matches_inner_serialize() {
        for kind in IntentKind::ALL {
            let i = single_slot_intent(kind);
            let v = i.variant().expect("exactly-one variant");
            let via_method = v.canonical_bytes();
            let expected: Vec<u8> = match &v {
                IntentVariant::Nix(n) => serde_json::to_vec(n).unwrap_or_default(),
                IntentVariant::Flux(f) => serde_json::to_vec(f).unwrap_or_default(),
                IntentVariant::Lisp(l) => serde_json::to_vec(l).unwrap_or_default(),
                IntentVariant::Container(c) => serde_json::to_vec(c).unwrap_or_default(),
                IntentVariant::Aplicacao(a) => serde_json::to_vec(a).unwrap_or_default(),
                IntentVariant::Guest(g) => serde_json::to_vec(g).unwrap_or_default(),
            };
            assert_eq!(
                via_method, expected,
                "canonical_bytes mismatch for {kind:?}"
            );
            assert!(!via_method.is_empty(), "{kind:?} produced empty bytes");
        }
    }

    /// Construct an `Intent` with two slots populated — drives the
    /// pairwise `Ambiguous` sweep through the substrate primitive
    /// [`crate::tagged_union::assert_two_slots_ambiguous`]. Composes
    /// the single-slot constructor on top of itself per-field so ONE
    /// source of truth for per-variant inner payloads is preserved.
    /// Mirrors `two_slot_source` / `two_slot_channel` / `two_slot_kind`
    /// in shape across `ProcessSpec`'s tagged-union axis.
    fn two_slot_intent(a: IntentKind, b: IntentKind) -> Intent {
        let ia = single_slot_intent(a);
        let ib = single_slot_intent(b);
        Intent {
            nix: ia.nix.or(ib.nix),
            flux: ia.flux.or(ib.flux),
            lisp: ia.lisp.or(ib.lisp),
            container: ia.container.or(ib.container),
            aplicacao: ia.aplicacao.or(ib.aplicacao),
            guest: ia.guest.or(ib.guest),
        }
    }

    /// Construct an `Intent` with exactly the given kind's slot
    /// populated by a minimal valid inner spec. Shared across the
    /// closed-set property tests so they each cover every variant
    /// without restating the construction table.
    fn single_slot_intent(kind: IntentKind) -> Intent {
        match kind {
            IntentKind::Nix => Intent {
                nix: Some(NixIntent {
                    flake_ref: "github:a/b".into(),
                    attribute: "x".into(),
                    system: None,
                    attic_cache: None,
                    extra_args: vec![],
                    delegate_to_nix_build: false,
                }),
                ..Intent::default()
            },
            IntentKind::Flux => Intent {
                flux: Some(FluxIntent {
                    git_repository: "g".into(),
                    path: "p".into(),
                    git_repository_namespace: None,
                    target_namespace: None,
                    decrypt_sops: true,
                    helm_chart: None,
                    helm_values: None,
                }),
                ..Intent::default()
            },
            IntentKind::Lisp => Intent {
                lisp: Some(LispIntent {
                    source: "()".into(),
                    reader: "tatara-lisp".into(),
                    version: "v1".into(),
                    bindings: BTreeMap::new(),
                }),
                ..Intent::default()
            },
            IntentKind::Container => Intent {
                container: Some(ContainerIntent {
                    image: "ghcr.io/x:1".into(),
                    replicas: Some(1),
                    command: vec![],
                    args: vec![],
                    env: BTreeMap::new(),
                    workload_kind: WorkloadKind::default(),
                }),
                ..Intent::default()
            },
            IntentKind::Aplicacao => Intent {
                aplicacao: Some(AplicacaoIntent::chart_only("oci://ghcr.io/x", "0.1.0")),
                ..Intent::default()
            },
            IntentKind::Guest => Intent {
                guest: Some(GuestIntent {
                    spec: serde_json::json!({"name": "guest-1"}),
                    state_dir: None,
                    allow_remote_build: None,
                }),
                ..Intent::default()
            },
        }
    }

    // ── closed-set algebra for WorkloadKind (ALL × as_str × Display ×
    //    FromStr × api_version × is_batch) ─────────────────────────────

    /// Structural well-formedness of [`WorkloadKind`] as a
    /// [`tatara_lisp::ClosedSet`] implementor — the workspace-wide
    /// testkit lift that pins all three structural invariants (`ALL`
    /// is non-empty, every variant round-trips through `label ↔
    /// parse_label`, labels are pairwise distinct, `""` is outside the
    /// closed set) at ONE call site. Replaces the hand-derived
    /// `workload_kind_all_is_unique_and_complete` +
    /// `workload_kind_roundtrip_via_as_str` + the empty-input arm of
    /// `unknown_workload_kind_errors`. `FromStr` delegates to
    /// `<Self as tatara_closed_set::ClosedSet>::parse_label`, so this helper
    /// exercises the same code path the reconciler hits when parsing a
    /// K8s `kind:`-shaped value back to the typed workload kind.
    #[test]
    fn workload_kind_is_well_formed_closed_set() {
        tatara_closed_set::assert_closed_set_well_formed::<WorkloadKind>();
    }

    /// CANONICAL-KEY CONTRACT: every variant's `as_str()` matches serde's
    /// PascalCase output verbatim. A future variant rename (or an
    /// `as_str` arm typo) lands at one site, instead of drifting
    /// between the typed surface, the K8s `kind:` manifest field, and
    /// the YAML wire format the reconciler / operator both read.
    #[test]
    fn workload_kind_as_str_matches_serde() {
        crate::tagged_union::assert_label_matches_serde_serialization::<WorkloadKind>();
    }

    /// The Display impl IS `as_str` — pinning this lets future callers
    /// reach for either projection without drift. If a reviewer
    /// accidentally re-introduces an inline match in Display, this
    /// test would fail the moment a variant rename touches one site
    /// but not the other.
    #[test]
    fn workload_kind_display_matches_as_str() {
        crate::tagged_union::assert_display_matches_label::<WorkloadKind>();
    }

    /// `FromStr` rejects strings that aren't in the canonical
    /// projection — lowercased / typo / unrelated — and the error
    /// echoes the input verbatim so the operator-facing diagnostic
    /// carries the offending value, not a normalized form. The
    /// empty-input arm is pinned by
    /// [`workload_kind_is_well_formed_closed_set`] via the
    /// `tatara_lisp::ClosedSet` testkit; the cases here pin the
    /// verbatim-echo contract on the [`UnknownWorkloadKind`]
    /// newtype, which the trait's `make_unknown` can't see.
    #[test]
    fn unknown_workload_kind_errors() {
        use std::str::FromStr;
        for bad in ["deployment", "JOB", "ReplicaSet", "Pod"] {
            let err = WorkloadKind::from_str(bad).unwrap_err();
            assert_eq!(err.0, bad, "error payload should echo input verbatim");
        }
    }

    #[test]
    fn workload_kind_default_is_deployment() {
        assert_eq!(WorkloadKind::default(), WorkloadKind::Deployment);
    }

    /// TRUTH-TABLE CONTRACT: `api_version` / `is_batch` agree with the
    /// documented (kind) -> (apiVersion, is_batch) table for every
    /// variant. A new variant in `WorkloadKind` without extending
    /// either projection's match is caught by the compiler (closed-set
    /// match in each method); adding a variant without extending its
    /// truth row is caught here. Also pins the invariant
    /// `is_batch <=> api_version == "batch/v1"`, so a future renderer
    /// can route on either projection without re-deriving the partition.
    #[test]
    fn workload_kind_projection_truth_table() {
        let table: &[(WorkloadKind, &str, bool)] = &[
            // (kind, api_version, is_batch)
            (WorkloadKind::Deployment, "apps/v1", false),
            (WorkloadKind::StatefulSet, "apps/v1", false),
            (WorkloadKind::DaemonSet, "apps/v1", false),
            (WorkloadKind::Job, "batch/v1", true),
            (WorkloadKind::CronJob, "batch/v1", true),
        ];
        assert_eq!(table.len(), WorkloadKind::ALL.len());
        for (kind, api, batch) in table {
            assert_eq!(kind.api_version(), *api, "api_version drift for {kind:?}");
            assert_eq!(kind.is_batch(), *batch, "is_batch drift for {kind:?}");
            assert_eq!(
                kind.is_batch(),
                kind.api_version() == "batch/v1",
                "is_batch / api_version partition disagrees for {kind:?}"
            );
        }
    }

    #[test]
    fn aplicacao_plus_flux_is_ambiguous() {
        let i = Intent {
            aplicacao: Some(AplicacaoIntent::chart_only("x", "1")),
            flux: Some(FluxIntent {
                git_repository: "g".into(),
                path: "p".into(),
                git_repository_namespace: None,
                target_namespace: None,
                decrypt_sops: true,
                helm_chart: None,
                helm_values: None,
            }),
            ..Intent::default()
        };
        assert_eq!(i.variant().unwrap_err(), IntentError::Ambiguous);
    }

    // ── Helm lifecycle policy — install / upgrade slot substrate ────

    fn helm_intent(install_timeout: Option<&str>) -> AplicacaoIntent {
        AplicacaoIntent {
            install_timeout: install_timeout.map(str::to_string),
            ..AplicacaoIntent::chart_only("oci://ghcr.io/pleme-io/charts/lareira-demo-app", "0.5.5")
        }
    }

    /// The workspace-wide default timeout const is pinned to `25m`.
    /// A regression that renamed it to any other duration would
    /// silently misroute every Helm-driven Process's default retry
    /// budget, so pin the byte-exact spelling here rather than at
    /// every consumer's own callsite.
    #[test]
    fn helm_lifecycle_default_timeout_is_pinned_to_25m() {
        assert_eq!(HELM_LIFECYCLE_DEFAULT_TIMEOUT, "25m");
    }

    /// The workspace-wide default retries const is pinned to `3`.
    /// Peer to the `_timeout` pin; same rationale.
    #[test]
    fn helm_lifecycle_default_retries_is_pinned_to_three() {
        assert_eq!(HELM_LIFECYCLE_DEFAULT_RETRIES, 3);
    }

    /// Fallback branch of the primitive: an intent that omitted
    /// `install_timeout` picks up the workspace-wide default
    /// (`25m` + retries `3`). Pin binds the "no override" shape
    /// every render / snapshot / dashboard consumer sees today.
    #[test]
    fn helm_lifecycle_policy_defaults_when_install_timeout_is_none() {
        let policy = helm_intent(None).helm_lifecycle_policy();
        assert_eq!(policy.timeout, HELM_LIFECYCLE_DEFAULT_TIMEOUT);
        assert_eq!(policy.remediation.retries, HELM_LIFECYCLE_DEFAULT_RETRIES);
    }

    /// Override branch of the primitive: when the operator populated
    /// `install_timeout`, the primitive substitutes that string
    /// verbatim (no normalization, no trimming) — the reconciler
    /// hands the exact `humantime` shape to Flux, and any parse
    /// error surfaces from the chart-controller, not from here.
    #[test]
    fn helm_lifecycle_policy_uses_install_timeout_when_present() {
        for shape in ["10m", "1h30m", "5s", "25m", "0s"] {
            let policy = helm_intent(Some(shape)).helm_lifecycle_policy();
            assert_eq!(
                policy.timeout, shape,
                "override shape {shape} not substituted verbatim"
            );
            // Retries stay at the workspace default regardless of timeout.
            assert_eq!(policy.remediation.retries, HELM_LIFECYCLE_DEFAULT_RETRIES);
        }
    }

    /// Coherence axis: the retries slot is invariant across every
    /// timeout shape the operator might publish — a regression that
    /// coupled the two slots (e.g. "when timeout is short, retry
    /// more") surfaces here rather than at every consumer.
    #[test]
    fn helm_lifecycle_policy_retries_are_invariant_across_timeout_shapes() {
        let seen: std::collections::BTreeSet<u8> = [None, Some("1m"), Some("25m"), Some("2h")]
            .into_iter()
            .map(|t| helm_intent(t).helm_lifecycle_policy().remediation.retries)
            .collect();
        assert_eq!(
            seen.len(),
            1,
            "retries should be constant across timeout shapes"
        );
        assert_eq!(
            seen.into_iter().next(),
            Some(HELM_LIFECYCLE_DEFAULT_RETRIES)
        );
    }

    /// Wire-shape pin: the serde projection matches Flux
    /// `HelmRelease.spec.{install,upgrade}` v2 byte-identically —
    /// `{"timeout": <string>, "remediation": {"retries": <int>}}`
    /// with no extra keys, no field renames, no camelCase surprises.
    /// A regression that added a slot to `HelmLifecyclePolicy` or
    /// renamed one would fail here rather than as a Flux CR
    /// rejection at every deployment.
    #[test]
    fn helm_lifecycle_policy_serializes_to_flux_hr_v2_install_upgrade_shape() {
        let policy = helm_intent(Some("10m")).helm_lifecycle_policy();
        let json = serde_json::to_value(&policy).unwrap();
        assert_eq!(
            json,
            serde_json::json!({
                "timeout": "10m",
                "remediation": { "retries": 3 },
            }),
        );
    }

    /// Coherence axis: `HelmLifecyclePolicy::workspace_default()`
    /// composes byte-identically to the intent-derived policy of an
    /// intent with `install_timeout: None` — the two paths to the
    /// substrate default (via the `Aplicacao` intent's own resolver
    /// vs the standalone workspace-default constructor) yield the
    /// same shape. Binds the "workspace_default IS the fallback"
    /// invariant so a future divergence (e.g. workspace_default
    /// changes but the intent resolver's inline fallback does not)
    /// surfaces here rather than as a silent drift at every render
    /// callsite.
    #[test]
    fn helm_lifecycle_policy_workspace_default_matches_intent_fallback_branch() {
        let default_policy = HelmLifecyclePolicy::workspace_default();
        let intent_policy = helm_intent(None).helm_lifecycle_policy();
        assert_eq!(default_policy, intent_policy);
    }

    // ── Flux reconcile interval — OCIRepository + HelmRelease shared ─

    /// The workspace-wide default Flux reconcile-interval const is
    /// pinned to `5m`. A regression that renamed it would silently
    /// throttle or hammer every Helm-driven Process's OCIRepository
    /// pull cadence AND its HelmRelease reconcile cadence, so pin
    /// the byte-exact spelling here rather than at the two render
    /// callsites the primitive owns.
    #[test]
    fn flux_helm_default_interval_is_pinned_to_5m() {
        assert_eq!(FLUX_HELM_DEFAULT_INTERVAL, "5m");
    }

    /// The intent-side composer returns the workspace-wide default
    /// verbatim today. A regression that hand-authored some other
    /// string here (or that stopped routing through the const)
    /// would surface at this pin.
    #[test]
    fn flux_reconcile_interval_returns_workspace_default() {
        assert_eq!(
            helm_intent(None).flux_reconcile_interval(),
            FLUX_HELM_DEFAULT_INTERVAL,
        );
    }

    /// Coherence axis: the reconcile interval is invariant across
    /// every `install_timeout` shape the operator publishes today.
    /// Pre-lift the two slots were siblings hand-authored with the
    /// same `"5m"` value regardless of any other AplicacaoIntent
    /// shape; post-lift the same invariance holds through the
    /// composer. A future coupling (e.g. "when timeout is short,
    /// reconcile more often") lands at the composer's shape, not
    /// silently at any render callsite.
    #[test]
    fn flux_reconcile_interval_is_invariant_across_install_timeout_shapes() {
        let seen: std::collections::BTreeSet<String> =
            [None, Some("10m"), Some("1h30m"), Some("25m"), Some("2h")]
                .into_iter()
                .map(|t| helm_intent(t).flux_reconcile_interval())
                .collect();
        assert_eq!(
            seen.len(),
            1,
            "reconcile interval should be constant across install_timeout shapes"
        );
        assert_eq!(
            seen.into_iter().next().as_deref(),
            Some(FLUX_HELM_DEFAULT_INTERVAL),
        );
    }

    // ─── AplicacaoIntent::chart_only substrate pins ─────────────────
    //
    // Bind the chart-pointer-only composer at fail-before-pass-after
    // granularity so a regression that drifted any of the five
    // default-tail slots (profile → non-empty, values_overlay → non-
    // `Null`, any of the three `Option<String>` slots → `Some`),
    // reshaped the two-argument surface, or swapped the positional
    // slot order surfaces HERE rather than as silent fixture skew at
    // the 14 downstream consumers.

    #[test]
    fn chart_only_binds_two_caller_slots_and_defaults_the_other_five() {
        // Primary shape asserted end-to-end: the returned value
        // carries the caller-supplied `(chart_ref, version)` and the
        // K8s-schema-default `("", Null, None, None, None)` tail. A
        // regression that swapped the two positional slots would
        // land `"1"` in `chart_ref` and `"oci://x"` in `version`;
        // the byte-equality pin below catches that.
        let a = AplicacaoIntent::chart_only("oci://x", "1");
        assert_eq!(a.chart_ref, "oci://x");
        assert_eq!(a.version, "1");
        assert_eq!(a.profile, "");
        assert_eq!(a.values_overlay, serde_json::Value::Null);
        assert!(a.release_name.is_none());
        assert!(a.target_namespace.is_none());
        assert!(a.install_timeout.is_none());
    }

    #[test]
    fn chart_only_matches_hand_authored_pre_lift_struct_literal_shape() {
        // Byte-identical parity with the pre-lift 7-slot struct-
        // literal every one of the 14 hand-authored sites restated.
        // A regression that drifted the composer would surface HERE
        // rather than as silent fixture skew at every downstream
        // `empty_template` / `sample_intent_for` / `helm_intent`
        // consumer. Swept across the three representative
        // `(chart_ref, version)` shape families the pre-lift sites
        // used (fixture stub `oci://x`/`1`; sample `oci://ghcr.io/x`
        // / `0.1.0`; production non-OCI `pleme-io/lareira-demo-app`
        // / `0.5.5`).
        for (chart_ref, version) in [
            ("oci://x", "1"),
            ("oci://ghcr.io/x", "0.1.0"),
            ("pleme-io/lareira-demo-app", "0.5.5"),
            ("x", "1"),
        ] {
            let composed = AplicacaoIntent::chart_only(chart_ref, version);
            let hand_authored = AplicacaoIntent {
                chart_ref: chart_ref.into(),
                version: version.into(),
                profile: String::new(),
                values_overlay: serde_json::Value::Null,
                release_name: None,
                target_namespace: None,
                install_timeout: None,
            };
            assert_eq!(
                serde_json::to_value(&composed).unwrap(),
                serde_json::to_value(&hand_authored).unwrap(),
                "composed and hand-authored must agree for ({chart_ref}, {version})"
            );
        }
    }

    #[test]
    fn chart_only_accepts_string_and_str_uniformly() {
        // The `impl Into<String>` argument form matches both the
        // pre-lift `"literal".into()` shape AND callers with a live
        // `String` (e.g. a fixture parameter). A regression that
        // narrowed the argument type to `&str` or `String` would
        // break one of the two shapes; this pin binds both.
        let owned_chart = String::from("oci://y");
        let owned_version = String::from("2");
        let via_string = AplicacaoIntent::chart_only(owned_chart.clone(), owned_version.clone());
        let via_str = AplicacaoIntent::chart_only("oci://y", "2");
        assert_eq!(
            serde_json::to_value(&via_string).unwrap(),
            serde_json::to_value(&via_str).unwrap(),
        );
    }

    // ── AplicacaoIntent::{release_name,target_namespace}_or pins ───
    //
    // Bind the (override, fallback) fallback-composer pair at
    // fail-before-pass-after granularity so a regression that flipped
    // the branch order (fallback wins when override is Some), dropped
    // the `.clone()` on the override, dropped the lazy branch on the
    // fallback, or swapped the two field slots surfaces HERE rather
    // than as silent drift at the two production consumers
    // (`tatara-reconciler::render::render_aplicacao` and
    // `tatara-process::matrix::EnvMatrixSpec::breathe_bands`).

    #[test]
    fn release_name_or_returns_override_when_set() {
        let mut a = AplicacaoIntent::chart_only("oci://x", "1");
        a.release_name = Some("operator-picked".into());
        assert_eq!(a.release_name_or("fallback-pid"), "operator-picked");
    }

    #[test]
    fn release_name_or_returns_fallback_when_unset() {
        let a = AplicacaoIntent::chart_only("oci://x", "1");
        assert!(a.release_name.is_none());
        assert_eq!(a.release_name_or("fallback-pid"), "fallback-pid");
    }

    #[test]
    fn target_namespace_or_returns_override_when_set() {
        let mut a = AplicacaoIntent::chart_only("oci://x", "1");
        a.target_namespace = Some("operator-picked".into());
        assert_eq!(a.target_namespace_or("fallback-ns"), "operator-picked");
    }

    #[test]
    fn target_namespace_or_returns_fallback_when_unset() {
        let a = AplicacaoIntent::chart_only("oci://x", "1");
        assert!(a.target_namespace.is_none());
        assert_eq!(a.target_namespace_or("fallback-ns"), "fallback-ns");
    }

    #[test]
    fn release_name_and_target_namespace_or_are_independent_across_all_four_shapes() {
        // Coherence axis: the two fallback slots are independent —
        // any of the four (release_name, target_namespace) ∈
        // {None, Some} shapes projects the expected pair with no
        // cross-slot bleed. A regression that keyed one field off
        // the other's `Option` state would surface HERE.
        for (rn, tn) in [
            (None, None),
            (Some("r".to_string()), None),
            (None, Some("t".to_string())),
            (Some("r".to_string()), Some("t".to_string())),
        ] {
            let mut a = AplicacaoIntent::chart_only("oci://x", "1");
            a.release_name = rn.clone();
            a.target_namespace = tn.clone();
            let got_rn = a.release_name_or("rn-fb");
            let got_tn = a.target_namespace_or("tn-fb");
            let want_rn = rn.clone().unwrap_or_else(|| "rn-fb".into());
            let want_tn = tn.clone().unwrap_or_else(|| "tn-fb".into());
            assert_eq!(got_rn, want_rn, "release_name_or for ({rn:?}, {tn:?})");
            assert_eq!(got_tn, want_tn, "target_namespace_or for ({rn:?}, {tn:?})");
        }
    }

    #[test]
    fn release_name_or_matches_hand_authored_pre_lift_option_clone_unwrap_or_else_shape() {
        // Byte-identical parity with the two pre-lift call shapes.
        // A regression that drifted the composer would surface HERE
        // rather than as silent skew at either production site.
        for override_ in [None, Some("op".to_string())] {
            let mut a = AplicacaoIntent::chart_only("oci://x", "1");
            a.release_name = override_.clone();
            let via_primitive = a.release_name_or("fb");
            let hand_authored = a.release_name.clone().unwrap_or_else(|| "fb".into());
            assert_eq!(via_primitive, hand_authored);
        }
        for override_ in [None, Some("op".to_string())] {
            let mut a = AplicacaoIntent::chart_only("oci://x", "1");
            a.target_namespace = override_.clone();
            let via_primitive = a.target_namespace_or("fb");
            let hand_authored = a.target_namespace.clone().unwrap_or_else(|| "fb".into());
            assert_eq!(via_primitive, hand_authored);
        }
    }

    #[test]
    fn chart_only_composes_downstream_through_helm_lifecycle_policy_default_branch() {
        // Cross-primitive coherence pin: an `AplicacaoIntent` built
        // through `chart_only` has `install_timeout = None` and
        // therefore rides the workspace-default branch of
        // `helm_lifecycle_policy`. A regression that flipped
        // `install_timeout` to `Some(_)` at the composer would
        // silently un-default every downstream Helm policy; this
        // pin binds the default-branch composition end-to-end.
        let a = AplicacaoIntent::chart_only("oci://x", "1");
        let policy = a.helm_lifecycle_policy();
        assert_eq!(policy.timeout, HELM_LIFECYCLE_DEFAULT_TIMEOUT);
        assert_eq!(policy.remediation.retries, HELM_LIFECYCLE_DEFAULT_RETRIES);
    }
}