tatara_process/

status.rs

//! `ProcessStatus` sub-structures — conditions, checked boundaries, Flux refs.

use chrono::{DateTime, Utc};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::boundary::Condition;
use crate::condition_type::ProcessConditionType;
use crate::crd::Process;
use crate::json_object::ValueGetExt;
use crate::k8s_condition::K8sConditionStatus;

/// Standard K8s Condition (shape of `metav1.Condition`).
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ProcessCondition {
    #[serde(rename = "type")]
    pub type_: String,
    pub status: String,
    pub last_transition_time: DateTime<Utc>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
}

impl ProcessCondition {
    /// Compose a [`ProcessCondition`] stamped at "observed now" from a
    /// typed [`ProcessConditionType`] + [`K8sConditionStatus`] pair,
    /// a reason slug, and an optional message.
    ///
    /// Owns the 5-slot wire-shape backbone
    ///
    /// ```text
    /// Self {
    ///     type_:                <ConditionType>.as_wire_str().into(),
    ///     status:               <StatusEnum>.as_wire_str().into(),
    ///     last_transition_time: Utc::now(),
    ///     reason:               Some(<reason>.into()),
    ///     message:              <opt>,
    /// }
    /// ```
    ///
    /// that every sibling constructor on this impl block ([`Self::ready`],
    /// [`Self::not_ready`], [`Self::attested`]) hand-authored pre-lift.
    ///
    /// Pre-lift the SAME 5-slot struct-literal skeleton recurred at
    /// THREE sibling constructor bodies past the ★★ PRIME-DIRECTIVE
    /// ≥ 2 duplication threshold, each restating (a) the typed-enum →
    /// wire-string projection at the `type_` slot, (b) the peer typed-
    /// enum → wire-string projection at the `status` slot, (c) the
    /// `Utc::now()` clock stamp at the `last_transition_time` slot,
    /// (d) the `Some(<reason>.into())` wrap at the `reason` slot. Post-
    /// lift the three siblings name their two typed variants + reason +
    /// message opt ONCE and route through this ONE composer; the shared
    /// backbone lives at ONE substrate site so a future normalization
    /// of the K8s Condition wire shape (adding `observed_generation:
    /// Option<i64>` to match `metav1.Condition` v2, injecting a
    /// `Clock` for deterministic tests replacing the direct
    /// `Utc::now()` call, adding a `severity:` slot for K8s-Condition-
    /// v3 style diagnostics) lands at ONE composer body and all three
    /// sibling constructors — plus every future `ProcessCondition::*`
    /// variant on this impl block — inherit the upgrade mechanically.
    ///
    /// The typed enum inputs bind the wire-form projection axis
    /// structurally: a regression that swapped `K8sConditionStatus::
    /// True` for `False` at ONE sibling (a mechanical copy-paste error
    /// the pre-lift hand-authored `status: "True".into()` /
    /// `status: "False".into()` string-literal restatements would not
    /// have caught at compile time) is caught by the enum's closed
    /// set, and the wire-string projection itself rides through the
    /// SAME `as_wire_str()` projection every reader/writer pair in the
    /// workspace already routes through — so a future spelling change
    /// at ONE of those enums propagates mechanically to all three
    /// sibling constructors without a per-site edit.
    ///
    /// Sibling to the pre-existing `K8sConditionStatus` /
    /// `ProcessConditionType` closed-set variants on the (typed
    /// wire-form × K8s-Condition-slot) axis: those closed sets own
    /// the type-variant enumeration + wire-form projection; this
    /// composer owns the K8s-Condition struct-literal composition
    /// that binds two typed variants onto the wire.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the 5-slot struct-literal backbone recurred at 3 hand-authored
    /// sibling constructor sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication trigger and is lifted onto the ONE substrate
    /// composer here). THEORY.md §II.1 invariant 5 (composition
    /// preserves proofs — the pin block below binds the composer at
    /// fail-before-pass-after granularity, so a regression that drifted
    /// any slot's default — `Utc::now()` swapped for a fixed-anchor
    /// stamp, `reason: Some(<reason>.into())` narrowed to a bare
    /// `<reason>.into()` opting out of the K8s-Condition-mandated
    /// `Option`-wrap, the typed-enum → wire-string projection at the
    /// `type_` or `status` slot bypassed by an inline literal that
    /// silently drifts off the closed-set owner — surfaces at
    /// `status::tests::new_at_now_*` rather than as silent operator-
    /// facing skew across the three sibling constructor callsites).
    ///
    /// # Delegation to [`Self::new_at`]
    ///
    /// The 5-slot struct-literal body lives at the clock-injectable
    /// substrate peer [`Self::new_at`]; this composer supplies
    /// `Utc::now()` as the `at` slot and delegates. The wall-clock
    /// projection lives at ONE substrate site, so a future clock swap
    /// (a monotonic-clock cross-check, a per-fleet skew tolerance,
    /// promotion to an injectable `Clock` trait) lands at the SINGLE
    /// `Utc::now()` call on this delegation body rather than at every
    /// hand-authored `last_transition_time: Utc::now()` stamp — no
    /// per-sibling edit at [`Self::ready`], [`Self::not_ready`], or
    /// [`Self::attested`], and no edit at any future
    /// `ProcessCondition::*` sibling that inherits the composer.
    fn new_at_now(
        type_: ProcessConditionType,
        status: K8sConditionStatus,
        reason: impl Into<String>,
        message: Option<String>,
    ) -> Self {
        Self::new_at(type_, status, reason, message, Utc::now())
    }

    /// Clock-injectable substrate peer of [`Self::new_at_now`] — the
    /// ONE substrate owner of the 5-slot wire-shape backbone
    ///
    /// ```text
    /// Self {
    ///     type_:                <ConditionType>.as_wire_str().into(),
    ///     status:               <StatusEnum>.as_wire_str().into(),
    ///     last_transition_time: <at>,
    ///     reason:               Some(<reason>.into()),
    ///     message:              <opt>,
    /// }
    /// ```
    ///
    /// with the wall-clock stamp lifted out onto an explicit `at:
    /// DateTime<Utc>` slot so a caller (typically a deterministic-
    /// clock test on this module or a future controller-side callsite
    /// that already carries a wall-clock anchor threaded through its
    /// own decide-tick) supplies the anchor rather than reading it
    /// from `Utc::now()` implicitly.
    ///
    /// # Why it exists
    ///
    /// [`Self::new_at_now`]'s pre-lift body read `Utc::now()` inline
    /// at the composer body — a shape the module's own tests already
    /// bracket with `let before = Utc::now(); ... let after = Utc::
    /// now(); assert!(<c>.last_transition_time >= before && <c>.
    /// last_transition_time <= after)` to pin the wall-clock projection
    /// non-deterministically. That non-determinism is a proof shape
    /// every peer clock-anchored composer family in this crate has
    /// already collapsed onto a `_now` + `_at` peer pair — see
    /// [`crate::pool::PoolStatus::observed`] (`at`-slot substrate) +
    /// [`crate::pool::PoolStatus::observed_now`] (wall-clock-anchored
    /// peer), [`crate::allocation::AllocationStatus::transition`] +
    /// [`crate::allocation::AllocationStatus::transition_now`],
    /// [`crate::lifetime_clock::evaluate`] +
    /// [`crate::lifetime_clock::evaluate_now`] — all binding the
    /// `Utc::now()` read at exactly ONE substrate site (the `_now`
    /// peer's body) with the `_at` peer owning the pure struct-literal
    /// composition. The docstring on [`Self::new_at_now`] itself
    /// called out the trajectory (`"a promotion of the direct
    /// Utc::now() call to an injectable Clock for deterministic tests"`)
    /// as its next compounding step; this method opens exactly that
    /// step for the `ProcessCondition` composer family, closing the
    /// pattern uniformity across the four peer axes.
    ///
    /// # Invariants
    ///
    /// - **Same slots as [`Self::new_at_now`]:** the four non-clock
    ///   slots (`type_`, `status`, `reason`, `message`) ride through
    ///   the SAME projections — typed enum → wire string at `type_` +
    ///   `status`, `Some(<reason>.into())` wrap at `reason`, verbatim
    ///   `Option<String>` pass-through at `message`. The peer pair
    ///   differs only at the `last_transition_time` slot's projection.
    /// - **`at` slot verbatim:** the caller-supplied `DateTime<Utc>`
    ///   binds `last_transition_time` bytewise, with no accidental
    ///   `Utc::now()` clamp / `Duration::seconds(0)` rounding / offset
    ///   normalization at the composer body. Pinned by
    ///   [`tests::new_at_binds_at_slot_to_supplied_datetime_verbatim`].
    /// - **Delegation invariant:** [`Self::new_at_now`] composes
    ///   through this primitive with `Utc::now()` at the `at` slot —
    ///   a regression that re-inlined the 5-slot literal at
    ///   [`Self::new_at_now`] (bypassing this substrate) surfaces at
    ///   [`tests::new_at_now_routes_through_new_at_with_utc_now_stamp_bytewise`]
    ///   rather than as silent skew between the wall-clock-anchored
    ///   peer and any future `new_at` consumer that pinned a fixed
    ///   anchor.
    ///
    /// # `#[must_use]`
    ///
    /// Every consumer feeds the returned [`ProcessCondition`] into a
    /// `ProcessStatus.conditions` slot or a peer status-patch call.
    /// Dropping the return means the condition composed for no
    /// observable reason — the attribute surfaces that as a warning
    /// at every call site.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the 5-slot struct-literal backbone that recurred at three
    /// sibling constructors' pre-lift bodies is now the SINGLE
    /// substrate site every clock-injected AND every wall-clock-
    /// anchored composer routes through). THEORY.md §II.1 invariant 5
    /// (composition preserves proofs — the pair `Self::new_at` +
    /// `Self::new_at_now` mirrors the four peer clock-anchored
    /// composer families across the crate, so the workspace's
    /// `<K8sWireResource>Condition` / `<CRD>Status` composer family
    /// stays uniform on the (clock-injectable, wall-clock-anchored)
    /// peer axis; a future primitive on the same axis inherits the
    /// convention structurally).
    #[must_use]
    fn new_at(
        type_: ProcessConditionType,
        status: K8sConditionStatus,
        reason: impl Into<String>,
        message: Option<String>,
        at: DateTime<Utc>,
    ) -> Self {
        Self {
            type_: type_.as_wire_str().into(),
            status: status.as_wire_str().into(),
            last_transition_time: at,
            reason: Some(reason.into()),
            message,
        }
    }

    pub fn ready(reason: impl Into<String>, message: Option<String>) -> Self {
        Self::new_at_now(
            ProcessConditionType::Ready,
            K8sConditionStatus::True,
            reason,
            message,
        )
    }

    pub fn not_ready(reason: impl Into<String>, message: impl Into<String>) -> Self {
        Self::new_at_now(
            ProcessConditionType::Ready,
            K8sConditionStatus::False,
            reason,
            Some(message.into()),
        )
    }

    pub fn attested(root: &str) -> Self {
        Self::new_at_now(
            ProcessConditionType::Attested,
            K8sConditionStatus::True,
            "AttestationWritten",
            Some(format!("composed_root={root}")),
        )
    }
}

/// Reference to a FluxCD resource emitted as part of this Process.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct FluxResourceRef {
    pub api_version: String,
    pub kind: String,
    pub name: String,
    pub namespace: String,
    #[serde(default)]
    pub ready: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_check: Option<DateTime<Utc>>,
}

impl FluxResourceRef {
    /// Pure typed projection of the four fetch coordinates
    /// `(namespace, api_version, kind, name)` every consumer that
    /// dispatches this persisted reference through kube-rs's dynamic-
    /// object surface splats by hand pre-lift. The 4-tuple binds the
    /// slot order at ONE typed accessor so a copy-paste at any downstream
    /// consumer cannot swap two adjacent `&str` slots in the fetch call.
    ///
    /// Peer projection to
    /// [`crate::k8s_wire_identity::K8sWireIdentity`] on the static-
    /// identity axis: [`K8sWireIdentity`] carries a
    /// `(&'static str, &'static str)` closed-set variant's pair for
    /// emit-time (RENDER phase) composition; this method carries the
    /// full `(ns, apiVersion, kind, name)` 4-slot borrow for fetch-time
    /// (VERIFY / ATTEST-heartbeat) composition where the ref's payload
    /// comes back off the persisted `ProcessStatus.flux_resources`
    /// slice with owned `String`s rather than static literals. The two
    /// primitives partition the fetch axis by whether the caller starts
    /// from a closed-set variant (emit-time) or a persisted status
    /// slice (fetch-time).
    ///
    /// Pre-lift the 5-slot `ssapply::fetch(client, &r.namespace,
    /// &r.api_version, &r.kind, &r.name)` splat was hand-authored at
    /// TWO sites past the ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold
    /// in `tatara-reconciler::phase_machine`:
    /// * `handle_running` — the VERIFY-phase per-ref readiness probe
    ///   that populates the updated `FluxResourceRef` slice with
    ///   `ready` + `message` + `last_check`.
    /// * `handle_attested` — the ATTEST-heartbeat drift detector that
    ///   short-circuits on the first non-Ready ref.
    ///
    /// Both sites splatted the SAME four `&r.X` field borrows in the
    /// SAME order into raw `ssapply::fetch`. A copy-paste that swapped
    /// two adjacent `&str` slots (`&r.api_version` and `&r.kind` are
    /// both strings that look interchangeable to a mechanical
    /// substitution) would silently 404 at wire time and diagnose as a
    /// broken CRD rather than as slot skew at the callsite. Post-lift
    /// each site names the ref ONCE and unpacks it through this ONE
    /// projection; the slot order binds structurally at the tuple
    /// return so a caller cannot desync one axis.
    ///
    /// A future addition (a case-fold normalization on the group, a
    /// virtual-cluster prefix rewrite for multi-tenancy, a
    /// `generateName` fallback on the name slot, a cluster-cache
    /// short-circuit inserted between the projection and the fetch
    /// call) lands at this ONE method and every downstream fetch
    /// consumer inherits the upgrade mechanically — no per-site edit
    /// at `handle_running` / `handle_attested` / any future kenshi-
    /// runner / mirror-audit / drift-probe consumer that grows a third
    /// consumer.
    ///
    /// Return-order pin lives at
    /// [`tests::flux_resource_ref_fetch_coords_binds_slots_by_position`]
    /// so a regression that swapped `namespace` and `api_version`
    /// (both `String`, same type) inside the tuple constructor fails-
    /// loudly here rather than as a silent wire-time 404 at every
    /// downstream fetch consumer.
    ///
    /// Theory grounding: THEORY.md §II.1 invariant 5 (composition
    /// preserves proofs — the 4-tuple slot order binds at ONE typed
    /// projection so a regression across the two fields of the same
    /// `String` type fails at the projection's positional pin rather
    /// than at every downstream fetch consumer). THEORY.md §VI.1
    /// (generation over composition — the 5-slot splat recurred at
    /// two hand-authored sites past the ≥ 2 duplication trigger, and
    /// is lifted to ONE typed borrow-projection here).
    pub fn fetch_coords(&self) -> (&str, &str, &str, &str) {
        (&self.namespace, &self.api_version, &self.kind, &self.name)
    }

    /// Compose a `FluxResourceRef` stamped at "observed now" — the
    /// `last_check` slot is set to `Some(Utc::now())` at ONE substrate
    /// owner, and the four coordinate slots + `ready` + `message`
    /// are bound positionally so a slot-swap regression surfaces at
    /// the constructor's positional pin rather than as silent drift
    /// at every downstream `ProcessStatus.flux_resources` writer.
    ///
    /// Pre-lift the 7-slot `FluxResourceRef { …, last_check:
    /// Some(chrono::Utc::now()) }` struct-literal was hand-authored
    /// at TWO sites past the ★★ PRIME-DIRECTIVE ≥ 2 duplication
    /// threshold in `tatara-reconciler::phase_machine`:
    /// * `handle_running` — the VERIFY-phase per-ref rebuild that
    ///   restamps each polled ref with fresh `ready` + `message` +
    ///   `last_check`.
    /// * `flux_ref_from_json` — the post-SSA initial-state seeder
    ///   that stamps a freshly-applied ref as `ready = false`,
    ///   `message = Some("applied; awaiting reconciliation")`,
    ///   `last_check = Some(Utc::now())`.
    ///
    /// Both sites restated the SAME seven field bindings in the
    /// SAME order, and both restated the SAME `Some(chrono::Utc::
    /// now())` stamp. A copy-paste that swapped two adjacent
    /// `String` slots (`api_version` and `kind`, `kind` and `name`,
    /// or `name` and `namespace` are all mechanically
    /// indistinguishable at the type level) would silently persist
    /// a slot-inverted ref that the downstream Flux fetch consumer
    /// (via [`Self::fetch_coords`]) would then 404 on. Post-lift
    /// both sites name the six inputs ONCE and route through this
    /// ONE composer; the seventh slot (`last_check`) is stamped at
    /// the composer's body so a future injection point (a fake
    /// clock for testing, a monotonic-clock cross-check, a per-
    /// fleet skew tolerance) lands at ONE substrate site rather
    /// than at every hand-authored `Some(chrono::Utc::now())` stamp.
    ///
    /// Return-order pin lives at
    /// [`tests::flux_resource_ref_observed_binds_slots_by_position`]
    /// so a regression that swapped `api_version` and `kind` (both
    /// `String`, same type) inside the constructor's argument list
    /// fails-loudly here rather than as a silent wire-time 404 at
    /// every downstream fetch consumer.
    ///
    /// Theory grounding: THEORY.md §II.1 invariant 5 (composition
    /// preserves proofs — the 6-slot positional binding + the
    /// `last_check` stamp compose at ONE typed owner, so a
    /// regression across the four `String` coordinate slots fails
    /// at the composer's positional pin rather than at every
    /// downstream Flux status writer). THEORY.md §VI.1 (generation
    /// over composition — the 7-slot struct-literal recurred at two
    /// hand-authored sites past the ≥ 2 duplication trigger, and is
    /// lifted to ONE typed composer here).
    pub fn observed(
        api_version: String,
        kind: String,
        name: String,
        namespace: String,
        ready: bool,
        message: Option<String>,
    ) -> Self {
        Self {
            api_version,
            kind,
            name,
            namespace,
            ready,
            message,
            last_check: Some(Utc::now()),
        }
    }

    /// Compose a `FluxResourceRef` in the pre-observation shape — the
    /// 4-slot coordinate binding with the three status slots defaulted
    /// (`ready: false`, `message: None`, `last_check: None`). The
    /// deterministic-fixture peer of [`Self::observed`] on the same
    /// `→ FluxResourceRef` composer axis: `observed` reads the wall
    /// clock and takes 6 args (a live post-fetch stamp), `pending`
    /// reads no clock and takes 4 args (a pre-observation fixture
    /// seed, and the natural base for `..base.clone()` spread updates
    /// that vary a single slot for a per-corner test sweep).
    ///
    /// Pre-lift the SAME 7-slot `FluxResourceRef { api_version, kind,
    /// name, namespace, ready: false, message: None, last_check: None
    /// }` struct-literal was hand-authored at THREE workspace-wide
    /// fixture sites past the ★★ PRIME-DIRECTIVE ≥ 2 duplication
    /// threshold:
    ///
    /// * [`crate::crd`]
    ///   `crd::observed_flux_resources_tests::sample_flux_ref(name)`
    ///   — the shared `Kustomization`/`flux-system` fixture the
    ///   `Process::observed_flux_resources` pin family destructures
    ///   for its `flux_resources`-populated corners.
    /// * `tatara-reconciler::ssapply::tests::sample_flux_ref_for_diag`
    ///   — the `HelmRelease`/`flux-system` fixture the
    ///   `flux_ref_fetch_error_context` diagnostic-wording pin
    ///   family destructures for its (kind, name) slot-coverage
    ///   sweep.
    /// * `tatara-reconciler::ssapply::tests::
    ///   flux_ref_fetch_error_context_matches_pre_lift_hand_authored_wording`
    ///   — the inline 7-slot literal inside the cross-substrate
    ///   coherence pin's per-case sweep over three distinct
    ///   `(api_version, kind, name, namespace)` tuples.
    ///
    /// All THREE sites restated the SAME seven field bindings in the
    /// SAME order and the SAME three defaulted status slots (`ready:
    /// false, message: None, last_check: None`), differing only in
    /// the four coordinate `String` values. Post-lift each callsite
    /// reads `FluxResourceRef::pending(<api_version>, <kind>, <name>,
    /// <namespace>)` and the four-slot bind + three-slot default
    /// sinks live at ONE substrate owner.
    ///
    /// The `impl Into<String>` signature accepts BOTH `&'static str`
    /// (the fixture-helper sites that spell coordinate literals
    /// inline) AND owned `String` (a future callsite handing off a
    /// dynamically-derived coordinate) without widening. Matches the
    /// discipline of the sibling substrate composers
    /// [`crate::pool::PoolMember::unallocated`] +
    /// [`crate::allocation::AllocationRef::new`] on the identity-slot
    /// axis.
    ///
    /// A future normalization (a case-fold on the group, a
    /// virtual-cluster prefix rewrite for multi-tenancy, a stricter
    /// kind gate, a `generateName` fallback on the name slot, a
    /// canonical rename of one of the three defaulted status slots
    /// to a typed `PreObservation` marker) lands at THIS ONE
    /// substrate primitive and every downstream fixture / helper
    /// inherits the upgrade mechanically — no per-site edit at any
    /// of the THREE listed callers or at future consumers (a
    /// stable-name claim-arbiter's pending-ref seed, a kenshi-runner
    /// pre-observation fixture, a mirror-audit drift-probe test
    /// helper).
    ///
    /// Theory grounding: THEORY.md §II.1 invariant 5 (composition
    /// preserves proofs — the 4-slot positional binding + the three
    /// defaulted status slots compose at ONE typed owner, so a
    /// regression across the four `String` coordinate slots fails at
    /// the composer's positional pin rather than at every downstream
    /// fixture consumer). THEORY.md §VI.1 (generation over
    /// composition — the 7-slot struct-literal recurred at three
    /// hand-authored fixture sites past the ≥ 2 duplication trigger,
    /// and is lifted to ONE typed composer here).
    #[must_use]
    pub fn pending(
        api_version: impl Into<String>,
        kind: impl Into<String>,
        name: impl Into<String>,
        namespace: impl Into<String>,
    ) -> Self {
        Self {
            api_version: api_version.into(),
            kind: kind.into(),
            name: name.into(),
            namespace: namespace.into(),
            ready: false,
            message: None,
            last_check: None,
        }
    }
}

/// Identifying coordinates of a rendered K8s resource — the
/// `(apiVersion, kind, metadata.name, metadata.namespace)` 4-tuple
/// every consumer that walks a rendered `serde_json::Value` resource
/// unwraps by hand pre-lift.
///
/// The three K8s API-path segments (`apiVersion`, `kind`,
/// `metadata.name`) are REQUIRED — a rendered resource missing any
/// of them cannot be applied via kube-rs's dynamic API surface, so
/// the extraction fails fast at the boundary rather than as a
/// downstream `Api::patch` panic. `metadata.namespace` is
/// intentionally kept as `Option<String>` because different consumers
/// resolve the fallback differently: `apply_owned` uses the
/// caller-supplied `namespace: &str` argument (the reconciler already
/// resolved the target namespace upstream), while `flux_ref_from_json`
/// records the K8s canonical `"default"` fallback into the persisted
/// `FluxResourceRef.namespace` slot. The peer method
/// [`Self::namespace_or_default`] applies the K8s canonical fallback
/// (`Process::DEFAULT_NAMESPACE = "default"`) for consumers wanting
/// the same shape [`FluxResourceRef.namespace`] carries.
///
/// Pre-lift the 3+1 slot extraction was hand-authored at TWO sites
/// past the ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold across
/// `tatara-reconciler`:
/// * `tatara-reconciler::phase_machine::flux_ref_from_json` — the
///   post-SSA `FluxResourceRef` builder that persists into
///   `ProcessStatus.flux_resources`; namespace half fallback-
///   defaulted to `"default"`.
/// * `tatara-reconciler::ssapply::apply_owned` — the SSA entry
///   point that extracts (apiVersion, kind, name) for the
///   [`kube::Api::patch`] call; namespace half discarded (the
///   `namespace: &str` argument comes from the caller upstream).
///
/// Both callsites restated the same three
/// `.get(K).and_then(|v| v.as_str()).ok_or_else(|| anyhow!(...))?
/// .to_string()` incantations with subtly different error wording
/// (`"resource missing X"` vs `"rendered resource missing X"`); post-
/// lift both route through this ONE substrate owner with the
/// canonical `"rendered resource missing X"` wording. A future
/// addition (case-fold on the group, a rename of the namespace
/// fallback, a stricter kind gate, a Unicode-safe collation step,
/// support for `metadata.generateName` as a name fallback) lands at
/// the primitive's body on the substrate, not at 2 independent
/// hand-writes across 2 reconciler files.
///
/// Namespace fallback const is shared with
/// [`Process::DEFAULT_NAMESPACE`] — a rename of the K8s canonical
/// default namespace lands at that ONE workspace-wide const, not at
/// per-primitive local literals that would drift silently.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RenderedResourceCoords {
    /// `apiVersion` — the group+version pair kube-rs uses to resolve
    /// the `ApiResource` for the SSA call.
    pub api_version: String,
    /// `kind` — the resource kind (Kustomization, HelmRelease, …).
    pub kind: String,
    /// `metadata.name` — the API-path leaf segment.
    pub name: String,
    /// `metadata.namespace` — raw from the resource, `None` when the
    /// slot is absent (a cluster-scoped resource, or a namespaced
    /// resource whose namespace was left for the API server to
    /// substitute). Consumers apply their own fallback:
    /// [`Self::namespace_or_default`] applies the K8s canonical
    /// `"default"` (matching what [`FluxResourceRef.namespace`]
    /// records); other consumers substitute a caller-supplied string
    /// (see `tatara-reconciler::ssapply::apply_owned`).
    pub namespace: Option<String>,
}

impl RenderedResourceCoords {
    /// Extract the 4-tuple from a rendered K8s resource JSON `Value`.
    ///
    /// Fails with a canonical `"rendered resource missing X"` message
    /// when any of the three required slots (`apiVersion`, `kind`,
    /// `metadata.name`) is absent or non-string; `metadata.namespace`
    /// is optional and captured as `None` when absent.
    ///
    /// The error wording is pinned by
    /// [`tests::rendered_resource_coords_error_wording_is_canonical`]
    /// so a regression that reshaped the message surfaces at the test
    /// surface rather than as silent drift between the two pre-lift
    /// call sites (which used subtly different wording — `"resource
    /// missing X"` in `apply_owned` vs `"rendered resource missing
    /// X"` in `flux_ref_from_json`).
    pub fn from_json(res: &Value) -> anyhow::Result<Self> {
        // The three REQUIRED-slot extracts (`apiVersion`, `kind`,
        // `metadata.name`) route through the ONE substrate primitive
        // `Self::required_str` — the required-extract sibling of
        // `crate::json_object::ValueGetExt::get_str` on the same
        // rendered-resource axis. A future normalization (Unicode
        // NFC-fold, whitespace trim, empty-string rejection) lands
        // at the primitive body and every downstream consumer of the
        // canonical `"rendered resource missing X"` wire form inherits
        // it mechanically. The optional `metadata.namespace` slot
        // routes directly through the ONE substrate primitive
        // `crate::json_object::ValueGetExt::get_str` on the
        // `Option<&Value>` receiver arm — the outer-optionality
        // widening added alongside the `Value` + `Map<String, Value>`
        // impls. Pre-lift this callsite hand-authored the outer-
        // optionality closure `metadata.and_then(|m| m.get_str
        // ("namespace"))` past the ★★ PRIME-DIRECTIVE ≥ 2 duplication
        // threshold (this site + `Self::required_str`'s body); post-
        // lift the closure disappears and the axis-family method-call
        // surface stays identical to the `Value` / `Map` receiver
        // callers.
        let api_version = Self::required_str(Some(res), "apiVersion", "apiVersion")?;
        let kind = Self::required_str(Some(res), "kind", "kind")?;
        let metadata = res.get("metadata");
        let name = Self::required_str(metadata, "name", "metadata.name")?;
        let namespace = metadata.get_str("namespace").map(str::to_string);
        Ok(Self {
            api_version,
            kind,
            name,
            namespace,
        })
    }

    /// Diagnostic prefix stamped ahead of every required-slot label in
    /// the canonical error wire form. Owned in ONE workspace-wide place
    /// so a rename (a fleet-wide switch to `"resource is missing"` /
    /// `"missing rendered-resource field"`) lands here and every
    /// downstream `.to_string()`-consumer + operator-facing log grep
    /// inherits the rename mechanically, not at 3 hand-authored
    /// `anyhow!(…)` restatements.
    pub const MISSING_MESSAGE_PREFIX: &'static str = "rendered resource missing";

    /// Required-slot extract on a rendered-resource JSON `Value` — the
    /// substrate owner of the paired `.get_str(<key>).ok_or_else(||
    /// anyhow!("rendered resource missing <slot>"))?.to_string()`
    /// four-link chain every REQUIRED slot on a rendered `Value`
    /// walks pre-lift.
    ///
    /// The primitive accepts an `Option<&Value>` receiver so BOTH
    /// shallow reads (top-level `apiVersion` / `kind` on the resource
    /// root, callers thread `Some(res)`) AND one-level-nested reads
    /// (`metadata.name` walking through `res.get("metadata")`,
    /// callers thread the `Option<&Value>` handle the `.get()` step
    /// returns) reach the same owner. The `key` slot is the wire-form
    /// name the underlying [`ValueGetExt::get_str`] looks up on the
    /// object; the `error_slot` slot is the diagnostic label stamped
    /// into the error's `Display` output. The two are decoupled so
    /// `metadata.name` can look up `"name"` on the `metadata` sub-
    /// object while reporting the dotted `"metadata.name"` path an
    /// operator bisecting a fault sees in the log.
    ///
    /// Ok arm returns `String` (owned) rather than the borrowed
    /// `&str` [`ValueGetExt::get_str`] returns — every downstream
    /// slot on the [`RenderedResourceCoords`] struct is an owned
    /// `String`, so the primitive absorbs the `str::to_string`
    /// coerce that pre-lift lived at three hand-authored callsites.
    /// Err arm carries an `anyhow::Error` whose `Display` reads
    /// exactly `"<Self::MISSING_MESSAGE_PREFIX> <error_slot>"` —
    /// byte-identical to the pre-lift hand-authored `anyhow!(
    /// "rendered resource missing {slot}")` wire form.
    ///
    /// ### Fires on all four absent-shape corners
    ///
    /// The primitive returns `Err` on ALL four ways a required
    /// slot can miss:
    ///
    /// 1. Receiver is `None` — the `metadata.name` corner when the
    ///    top-level `metadata` object itself is absent (the caller
    ///    threaded `res.get("metadata")` which returned `None`).
    /// 2. Slot is absent — the receiver is present but does not
    ///    carry a value at `key`.
    /// 3. Slot is present but non-string — a fixture bug that
    ///    stamped a JSON number / bool / object / array at the
    ///    slot; the `get_str` step falls through and the primitive
    ///    reports the slot as missing (matching the pre-lift
    ///    behavior where every non-string variant surfaced as the
    ///    same `"missing"` diagnostic — pinning "cannot be applied
    ///    via kube-rs's dynamic API surface" as the shared
    ///    failure mode).
    /// 4. Receiver is non-object — a resource authored as a JSON
    ///    array / string / null at any of the levels the primitive
    ///    walks (the `get_str` step returns `None` verbatim).
    ///
    /// All four corners produce the SAME wire form so an operator's
    /// `rg "rendered resource missing"` sweep hits exactly one
    /// footprint per faulted slot, not four differently-worded
    /// diagnostics per absent-shape variant.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the 4-link `.get_str(<key>).ok_or_else(|| anyhow!("rendered
    /// resource missing <slot>"))?.to_string()` shape recurred at 3
    /// hand-authored sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication trigger, and is lifted to ONE substrate owner
    /// here). THEORY.md §II.1 invariant 5 (composition preserves
    /// proofs — a regression that drifted the diagnostic prefix
    /// wording at ONE site would silently pass the two sibling
    /// pins and fail HERE; post-lift the wire form is owned once
    /// at [`Self::MISSING_MESSAGE_PREFIX`] and every downstream
    /// composition inherits the rename mechanically).
    fn required_str(
        v: Option<&Value>,
        key: &'static str,
        error_slot: &'static str,
    ) -> anyhow::Result<String> {
        // The `Option<&Value>` outer-optionality unwrap-then-project
        // step rides through the substrate primitive
        // `crate::json_object::ValueGetExt::get_str` on the
        // `Option<&Value>` receiver arm — the outer-optionality
        // widening sibling of the `Value` / `Map<String, Value>`
        // impls. Pre-lift this body hand-authored the outer-
        // optionality closure `v.and_then(|x| x.get_str(key))` past
        // the ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold (this
        // helper's body + the sibling `metadata.namespace` walk in
        // `Self::from_json`); post-lift the closure disappears and
        // the axis-family method-call surface stays identical to the
        // `Value` / `Map` receiver callers.
        v.get_str(key).map(str::to_string).ok_or_else(|| {
            anyhow::anyhow!(
                "{prefix} {slot}",
                prefix = Self::MISSING_MESSAGE_PREFIX,
                slot = error_slot,
            )
        })
    }

    /// `metadata.namespace` slice with the K8s canonical `"default"`
    /// fallback applied — matching what [`Process::DEFAULT_NAMESPACE`]
    /// spells for the `Process`-borne coordinate primitive family
    /// and what [`FluxResourceRef.namespace`] records into
    /// `ProcessStatus.flux_resources`.
    pub fn namespace_or_default(&self) -> &str {
        self.namespace
            .as_deref()
            .unwrap_or(Process::DEFAULT_NAMESPACE)
    }
}

/// A boundary condition paired with its current satisfaction state.
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct CheckedCondition {
    #[serde(flatten)]
    pub condition: Condition,
    pub satisfied: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_check: Option<DateTime<Utc>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
}

impl CheckedCondition {
    /// True iff every [`CheckedCondition`] in the slice has
    /// `satisfied == true` — the ONE-line collapse of the paired
    /// `checked.iter().all(|c| c.satisfied)` incantation the
    /// reconciler's precondition + postcondition boundary gates both
    /// spelled by hand pre-lift.
    ///
    /// Pre-lift the SAME `.iter().all(|c| c.satisfied)` chain was
    /// hand-authored at TWO sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication threshold in `tatara-reconciler::phase_machine`,
    /// each walking the SAME `Vec<CheckedCondition>` → `bool`
    /// projection to gate a phase transition on a boundary predicate:
    /// * `handle_execing` — the PROVE-phase precondition gate that
    ///   stays in Execing (heartbeat requeue) while any precondition
    ///   remains unsatisfied and proceeds to RENDER only when every
    ///   precondition holds.
    /// * `handle_running` — the VERIFY-phase postcondition gate that
    ///   stays in Running (heartbeat requeue) while any postcondition
    ///   remains unsatisfied and advances to Attested only when every
    ///   postcondition holds.
    ///
    /// Both sites walked the SAME `Iterator::all` short-circuit on the
    /// SAME `bool` slot of the SAME struct. Post-lift both consumers
    /// name the slice ONCE and route through this ONE primitive; the
    /// vacuous-truth corner (empty slice → `true`, matching
    /// [`Iterator::all`]'s empty-input identity) sits at ONE substrate
    /// site so a future normalization (a per-slot weight overlay, a
    /// per-kind override that treats `Warn`-severity failures as
    /// satisfied, a compliance-baseline gate that requires N-of-M
    /// rather than all-of-M) lands at ONE substrate function and both
    /// downstream phase gates inherit the upgrade mechanically.
    ///
    /// Return-form axis: `bool` — the exact type each phase gate
    /// pre-lift bound at `let all_pass = <chain>;` and immediately
    /// consumed in a `!all_pass` short-circuit + a `message` slot's
    /// ternary branch. The `&[Self]` argument accepts every pre-lift
    /// slice provenance verbatim: a `&Vec<CheckedCondition>` (both
    /// pre-lift sites had the `Vec` on the stack from
    /// [`crate::phase_machine::evaluate_conditions`]'s owned return)
    /// coerces through auto-deref, so no callsite has to change its
    /// upstream provenance to route through the primitive.
    ///
    /// Peer to the sibling projection [`Self::satisfied`] on the (row
    /// scope × predicate) axis pair: `satisfied` is the per-row
    /// projection; `all_satisfied` is the slice-wide fold of the same
    /// bit. Both live on `CheckedCondition` so a future rename or
    /// per-slot normalization travels through the same owner without
    /// splitting between "per-row" and "slice-wide" call sinks.
    ///
    /// Return-shape pin lives at
    /// [`tests::checked_condition_all_satisfied_matches_pre_lift_iter_all_chain_shape`]
    /// so a regression that flipped the fold direction (`any` for
    /// `all`), inverted the bit (`!c.satisfied`), or reshaped the
    /// return form (an owned `Vec<bool>` instead of the folded `bool`)
    /// fails-loudly here rather than as silent operator-facing skew
    /// between the pre-lift `if !all_pass { requeue }` gate and the
    /// post-lift call — every downstream consumer would still
    /// short-circuit but on inverted semantics.
    ///
    /// Theory grounding: THEORY.md §VI.1 (generation over composition
    /// — the 1-line `.iter().all(...)` chain recurred at two hand-
    /// authored sites past the ≥ 2 duplication trigger, and is lifted
    /// to ONE typed fold here). THEORY.md §II.1 invariant 5
    /// (composition preserves proofs — the empty-slice vacuous-truth
    /// corner + the fold direction + the projected bit's polarity all
    /// bind at ONE substrate site, so a regression across any of the
    /// three surfaces at [`tests::checked_condition_all_satisfied_*`]
    /// pin rather than as silent gate-flip at every downstream phase
    /// handler).
    #[must_use]
    pub fn all_satisfied(checked: &[Self]) -> bool {
        checked.iter().all(|c| c.satisfied)
    }
}

/// Summary of boundary verification.
#[derive(Clone, Debug, Default, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct BoundaryStatus {
    #[serde(default)]
    pub preconditions: Vec<CheckedCondition>,
    #[serde(default)]
    pub postconditions: Vec<CheckedCondition>,
    /// Absolute deadline for VERIFY (derived from `spec.boundary.timeout`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub deadline: Option<DateTime<Utc>>,
}

/// Summary of compliance checks at the latest attestation.
#[derive(Clone, Debug, Default, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ComplianceStatus {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub baseline: Option<String>,
    pub satisfied: u32,
    pub violated: u32,
    pub total: u32,
    #[serde(default)]
    pub violations: Vec<String>,
}

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

    // ─── ProcessCondition writer / K8sConditionStatus substrate pins ─
    //
    // Byte-shape parity pins between `ProcessCondition::{ready,
    // not_ready, attested}` writer output and the pre-lift hand-
    // authored `"True"` / `"False"` `status` slot literals every
    // downstream K8s API server + K8s-Condition-reading peer depends
    // on. Post-lift the writers compose through
    // `K8sConditionStatus::<V>.as_wire_str()`; these pins catch a
    // regression at the substrate primitive (a lower-case drift, a
    // whitespace prefix, a `serde(rename)` addition on the enum)
    // that would silently reshape every emitted Process
    // `status.conditions[]` slot away from the K8s wire form.

    /// Fail-before-pass-after: `ProcessCondition::ready` emits the
    /// exact-case ASCII `"True"` on the `status` slot the K8s API
    /// server accepts, byte-identical to the pre-lift hand-authored
    /// `status: "True".into()` literal. A regression at the substrate
    /// primitive (a `to_lowercase` pass, a case-drifted variant
    /// literal in `k8s_condition::K8sConditionStatus::as_wire_str`)
    /// surfaces HERE, not as silent operator-facing wire-form skew
    /// on the emitted Process CRD.
    #[test]
    fn ready_writer_status_slot_matches_pre_lift_true_literal_bytewise() {
        let c = ProcessCondition::ready("ObservedRunning", Some("healthy".into()));
        assert_eq!(c.type_, "Ready");
        assert_eq!(c.status, "True");
    }

    /// Fail-before-pass-after: `ProcessCondition::not_ready` emits
    /// the exact-case ASCII `"False"` on the `status` slot, byte-
    /// identical to the pre-lift hand-authored `status: "False".
    /// into()` literal.
    #[test]
    fn not_ready_writer_status_slot_matches_pre_lift_false_literal_bytewise() {
        let c = ProcessCondition::not_ready("ObservedFailed", "boom");
        assert_eq!(c.type_, "Ready");
        assert_eq!(c.status, "False");
    }

    /// Fail-before-pass-after: `ProcessCondition::attested` emits
    /// the exact-case ASCII `"True"` on the `status` slot with the
    /// `"Attested"` type row, byte-identical to the pre-lift hand-
    /// authored `type_: "Attested".into()` + `status: "True".into()`
    /// pair.
    #[test]
    fn attested_writer_status_slot_matches_pre_lift_true_literal_bytewise() {
        let c = ProcessCondition::attested("blake3:abc123");
        assert_eq!(c.type_, "Attested");
        assert_eq!(c.status, "True");
        // Message body preserves the composed_root diagnostic
        // wording verbatim — the substrate lift only reshaped the
        // `status` slot, not the human-facing message.
        assert_eq!(c.message.as_deref(), Some("composed_root=blake3:abc123"));
        assert_eq!(c.reason.as_deref(), Some("AttestationWritten"));
    }

    /// Writer/reader wire-form parity: the `status` slot every
    /// writer here emits is the SAME byte-shape the
    /// `K8sConditionStatus::from_wire_str` reader in
    /// `tatara-reconciler::ssapply::ready_condition_value` accepts.
    /// A regression that drifted `as_wire_str` at ONE variant would
    /// silently desynchronize every writer/reader pair in the
    /// workspace; this pin surfaces the drift at the substrate.
    #[test]
    fn writer_output_round_trips_through_from_wire_str() {
        let ready = ProcessCondition::ready("R", None);
        assert_eq!(
            K8sConditionStatus::from_wire_str(&ready.status),
            Some(K8sConditionStatus::True),
        );
        let not_ready = ProcessCondition::not_ready("R", "why");
        assert_eq!(
            K8sConditionStatus::from_wire_str(&not_ready.status),
            Some(K8sConditionStatus::False),
        );
        let attested = ProcessCondition::attested("blake3:zzz");
        assert_eq!(
            K8sConditionStatus::from_wire_str(&attested.status),
            Some(K8sConditionStatus::True),
        );
    }

    // ─── ProcessCondition::new_at_now backbone-composer substrate pins ───
    //
    // The composer [`ProcessCondition::new_at_now`] owns the 5-slot wire-
    // shape backbone every sibling constructor on this impl block
    // ([`ProcessCondition::ready`], [`ProcessCondition::not_ready`],
    // [`ProcessCondition::attested`]) hand-authored pre-lift. These pins
    // bind the observable slots (typed-enum → wire-string projection on
    // `type_` + `status`, `Utc::now()` stamp on `last_transition_time`,
    // `Some(<reason>.into())` wrap on `reason`, verbatim message opt on
    // `message`) at fail-before-pass-after granularity so a regression
    // that drifted any slot's projection (a typed-enum arm bypassed by
    // an inline literal, the `Some(...)`-wrap on `reason` narrowed to
    // a bare `.into()`, the `last_transition_time` stamp swapped for a
    // fixed anchor) surfaces HERE rather than as silent operator-facing
    // K8s-Condition wire-shape skew across the three sibling
    // constructor sites.

    /// Fail-before-pass-after: the composer's 5-slot output is byte-
    /// identical (modulo the `Utc::now()` monotonic clock stamp) to
    /// the pre-lift hand-authored struct literal every sibling
    /// constructor restated. A regression that drifted the projection
    /// at any of the four fixed slots surfaces HERE, not as silent
    /// wire-form skew on the emitted Process CRD.
    #[test]
    fn new_at_now_composes_typed_enums_reason_option_wrap_and_message_slot_verbatim() {
        let before = Utc::now();
        let c = ProcessCondition::new_at_now(
            ProcessConditionType::Ready,
            K8sConditionStatus::True,
            "ObservedRunning",
            Some("healthy".into()),
        );
        let after = Utc::now();
        assert_eq!(c.type_, ProcessConditionType::Ready.as_wire_str());
        assert_eq!(c.status, K8sConditionStatus::True.as_wire_str());
        assert_eq!(c.reason.as_deref(), Some("ObservedRunning"));
        assert_eq!(c.message.as_deref(), Some("healthy"));
        assert!(
            c.last_transition_time >= before && c.last_transition_time <= after,
            "last_transition_time rides Utc::now() at the composer body",
        );
    }

    /// Fail-before-pass-after: the composer preserves an explicit
    /// `None` message-slot verbatim — a K8s Condition with an absent
    /// message vs an empty-string message are distinct wire shapes
    /// the K8s API server treats differently, so the composer must
    /// route `None` through without an accidental `Some(String::
    /// new())` wrap.
    #[test]
    fn new_at_now_preserves_none_message_slot_verbatim() {
        let c = ProcessCondition::new_at_now(
            ProcessConditionType::Ready,
            K8sConditionStatus::False,
            "R",
            None,
        );
        assert!(
            c.message.is_none(),
            "message-slot rides `None` verbatim — an empty-Some wrap is a distinct wire shape",
        );
    }

    /// Byte-shape parity witness — `ProcessCondition::ready` routes
    /// through `new_at_now(Ready, True, reason, message)`, matching
    /// the pre-lift hand-authored 5-slot struct literal on every
    /// observable slot except the monotonic `last_transition_time`
    /// stamp.
    #[test]
    fn ready_routes_through_new_at_now_composer_slots_match_pre_lift_literal() {
        let via_public = ProcessCondition::ready("ObservedRunning", Some("healthy".into()));
        let via_composer = ProcessCondition::new_at_now(
            ProcessConditionType::Ready,
            K8sConditionStatus::True,
            "ObservedRunning",
            Some("healthy".into()),
        );
        assert_eq!(via_public.type_, via_composer.type_);
        assert_eq!(via_public.status, via_composer.status);
        assert_eq!(via_public.reason, via_composer.reason);
        assert_eq!(via_public.message, via_composer.message);
    }

    /// Byte-shape parity witness — `ProcessCondition::not_ready`
    /// routes through `new_at_now(Ready, False, reason, Some(message.
    /// into()))`, matching the pre-lift hand-authored 5-slot struct
    /// literal on every observable slot.
    #[test]
    fn not_ready_routes_through_new_at_now_composer_slots_match_pre_lift_literal() {
        let via_public = ProcessCondition::not_ready("ObservedFailed", "boom");
        let via_composer = ProcessCondition::new_at_now(
            ProcessConditionType::Ready,
            K8sConditionStatus::False,
            "ObservedFailed",
            Some("boom".into()),
        );
        assert_eq!(via_public.type_, via_composer.type_);
        assert_eq!(via_public.status, via_composer.status);
        assert_eq!(via_public.reason, via_composer.reason);
        assert_eq!(via_public.message, via_composer.message);
    }

    /// Byte-shape parity witness — `ProcessCondition::attested`
    /// routes through `new_at_now(Attested, True, "AttestationWritten",
    /// Some(format!("composed_root={root}")))`, matching the pre-lift
    /// hand-authored 5-slot struct literal on every observable slot.
    #[test]
    fn attested_routes_through_new_at_now_composer_slots_match_pre_lift_literal() {
        let via_public = ProcessCondition::attested("blake3:abc123");
        let via_composer = ProcessCondition::new_at_now(
            ProcessConditionType::Attested,
            K8sConditionStatus::True,
            "AttestationWritten",
            Some("composed_root=blake3:abc123".into()),
        );
        assert_eq!(via_public.type_, via_composer.type_);
        assert_eq!(via_public.status, via_composer.status);
        assert_eq!(via_public.reason, via_composer.reason);
        assert_eq!(via_public.message, via_composer.message);
    }

    /// Routing pin — the `type_` slot at the composer's `Self { …
    /// type_: type_.as_wire_str().into() }` binding rides through the
    /// [`ProcessConditionType::as_wire_str`] projection rather than an
    /// inline `format!("{:?}", type_)` or hand-authored per-variant
    /// literal. A regression that inlined the wire-string projection
    /// (drifting the composer off the closed-set owner + reopening the
    /// typo-drift surface a hand-authored `"ready"` / `"Ready "` /
    /// `"READY"` spelling would fall into silently) surfaces HERE
    /// rather than as silent per-variant wire-form skew where the
    /// composed Condition's `type_` slot disagrees with the K8s API
    /// server's expected literal.
    #[test]
    fn new_at_now_routes_type_slot_through_process_condition_type_as_wire_str() {
        for type_ in ProcessConditionType::ALL {
            let via_composer =
                ProcessCondition::new_at_now(type_, K8sConditionStatus::True, "R", None);
            assert_eq!(
                via_composer.type_,
                type_.as_wire_str(),
                "type_ slot must route through ProcessConditionType::as_wire_str for {type_:?}",
            );
        }
    }

    /// Routing pin — the `status` slot at the composer's `Self { …
    /// status: status.as_wire_str().into() }` binding rides through
    /// the [`K8sConditionStatus::as_wire_str`] projection rather than
    /// an inline hand-authored per-variant literal. A regression that
    /// inlined the wire-string projection surfaces HERE rather than
    /// as silent per-variant wire-form skew where the composed
    /// Condition's `status` slot disagrees with the K8s API server's
    /// expected `"True"` / `"False"` / `"Unknown"` closed set.
    #[test]
    fn new_at_now_routes_status_slot_through_k8s_condition_status_as_wire_str() {
        for status in K8sConditionStatus::ALL {
            let via_composer =
                ProcessCondition::new_at_now(ProcessConditionType::Ready, status, "R", None);
            assert_eq!(
                via_composer.status,
                status.as_wire_str(),
                "status slot must route through K8sConditionStatus::as_wire_str for {status:?}",
            );
        }
    }

    // ─── ProcessCondition::new_at clock-injectable substrate pins ────
    //
    // The clock-injectable substrate peer [`ProcessCondition::new_at`]
    // owns the 5-slot wire-shape backbone with `last_transition_time`
    // sourced from a caller-supplied `DateTime<Utc>` slot instead of
    // an inline `Utc::now()` read. [`ProcessCondition::new_at_now`]
    // delegates through it with `Utc::now()`. These pins bind:
    // (a) the `at` slot is stamped bytewise (no clamp/round/offset
    //     drift at the composer body),
    // (b) [`ProcessCondition::new_at_now`] composes through the
    //     substrate with a wall-clock third arg (a regression that
    //     re-inlined the 5-slot literal at `new_at_now` bypasses the
    //     substrate and surfaces here rather than as silent skew
    //     between `new_at`-anchored callers and `new_at_now`),
    // (c) the four non-clock slots ride through the SAME projections
    //     `new_at_now`'s existing pins already bind — closed-set
    //     `as_wire_str` at type_/status, `Some(<r>.into())` wrap at
    //     reason, verbatim `Option<String>` pass-through at message.

    /// Fail-before-pass-after: the clock-injectable substrate peer
    /// [`ProcessCondition::new_at`] stamps the caller-supplied
    /// `DateTime<Utc>` at the `last_transition_time` slot bytewise —
    /// a deterministic anchor rides through with no wall-clock read,
    /// no clamp, no `Duration::seconds(0)` rounding, no offset
    /// normalization. Peer of the `[before, after]` bracket pin on
    /// [`ProcessCondition::new_at_now`]; this pin owns the
    /// deterministic-anchor half of the (wall-clock, deterministic-
    /// anchor) axis pair. A regression that inserted an implicit
    /// `Utc::now()` fallback at the composer body (silently ignoring
    /// the caller's anchor) surfaces HERE rather than as a hidden
    /// non-determinism at every future
    /// `ProcessCondition::new_at`-anchored test / controller-side
    /// callsite that pinned a specific anchor.
    #[test]
    fn new_at_binds_at_slot_to_supplied_datetime_verbatim() {
        let anchor: DateTime<Utc> = crate::time::at_epoch_second(1_700_000_000);
        let c = ProcessCondition::new_at(
            ProcessConditionType::Ready,
            K8sConditionStatus::True,
            "R",
            Some("m".into()),
            anchor,
        );
        assert_eq!(
            c.last_transition_time, anchor,
            "last_transition_time must ride the caller-supplied `at` slot bytewise — \
             a regression that inserted an implicit `Utc::now()` fallback would surface here",
        );
        // Cross-anchor pin — a distant future anchor + the epoch also
        // ride through bytewise, so the invariant holds at both
        // extremes of the wall-clock axis rather than only at the
        // 1.7-billion-second-past-epoch mid-range corner above. The
        // upper bound stays inside `chrono`'s valid `DateTime<Utc>`
        // range (year 2262 ceiling for nanosecond-precision anchors
        // the substrate carries) so the epoch composer succeeds.
        for anchor in [
            crate::time::at_epoch_second(0),
            crate::time::at_epoch_second(4_000_000_000),
        ] {
            let c = ProcessCondition::new_at(
                ProcessConditionType::Attested,
                K8sConditionStatus::True,
                "R",
                None,
                anchor,
            );
            assert_eq!(c.last_transition_time, anchor);
        }
    }

    /// Cross-primitive coherence pin — [`ProcessCondition::new_at_now`]
    /// composes through [`ProcessCondition::new_at`] with `Utc::now()`
    /// at the `at` slot. Every non-clock slot on the two composers'
    /// outputs is byte-identical for the SAME (type_, status, reason,
    /// message) input tuple; only the `last_transition_time` differs
    /// (the wall-clock peer's slot falls in `[before, after]`, and the
    /// deterministic peer's slot is exactly the caller's anchor). A
    /// regression that re-inlined the 5-slot literal at
    /// [`ProcessCondition::new_at_now`] (bypassing the substrate) or
    /// swapped `Utc::now()` for a fixed anchor at the delegation body
    /// surfaces HERE rather than as silent skew between the two peer
    /// composers' consumers.
    #[test]
    fn new_at_now_routes_through_new_at_with_utc_now_stamp_bytewise() {
        let before = Utc::now();
        let via_wall = ProcessCondition::new_at_now(
            ProcessConditionType::Ready,
            K8sConditionStatus::False,
            "ObservedFailed",
            Some("boom".into()),
        );
        let after = Utc::now();
        // Anchor a peer of `via_wall` at a fixed deterministic slot
        // and pin non-clock parity; the two composers must agree on
        // every slot except `last_transition_time`.
        let via_at = ProcessCondition::new_at(
            ProcessConditionType::Ready,
            K8sConditionStatus::False,
            "ObservedFailed",
            Some("boom".into()),
            crate::time::at_epoch_second(0),
        );
        assert_eq!(via_wall.type_, via_at.type_);
        assert_eq!(via_wall.status, via_at.status);
        assert_eq!(via_wall.reason, via_at.reason);
        assert_eq!(via_wall.message, via_at.message);
        assert!(
            via_wall.last_transition_time >= before && via_wall.last_transition_time <= after,
            "new_at_now must stamp Utc::now() at the delegation body — a regression that \
             swapped it for a fixed anchor surfaces here",
        );
        // Delegation identity — for every closed-set (type_, status)
        // pair, composing through `new_at` with an anchor threaded
        // through `new_at_now`'s bracket yields identical non-clock
        // slots on both peers. Sweep the two closed-set axes
        // exhaustively so a future variant addition inherits the pin.
        for type_ in ProcessConditionType::ALL {
            for status in K8sConditionStatus::ALL {
                let via_wall = ProcessCondition::new_at_now(type_, status, "R", None);
                let via_at = ProcessCondition::new_at(
                    type_,
                    status,
                    "R",
                    None,
                    via_wall.last_transition_time,
                );
                assert_eq!(via_wall.type_, via_at.type_);
                assert_eq!(via_wall.status, via_at.status);
                assert_eq!(via_wall.reason, via_at.reason);
                assert_eq!(via_wall.message, via_at.message);
                assert_eq!(via_wall.last_transition_time, via_at.last_transition_time);
            }
        }
    }

    /// Fail-before-pass-after: the clock-injectable substrate peer
    /// preserves an explicit `None` message-slot verbatim — mirror of
    /// the sibling pin on [`ProcessCondition::new_at_now`]. A K8s
    /// Condition with an absent message vs an empty-string message
    /// are distinct wire shapes the K8s API server treats
    /// differently, so a regression at the substrate body that
    /// wrapped `None` into `Some(String::new())` would surface HERE
    /// as well as at the wall-clock peer's pin.
    #[test]
    fn new_at_preserves_none_message_slot_verbatim() {
        let c = ProcessCondition::new_at(
            ProcessConditionType::Ready,
            K8sConditionStatus::False,
            "R",
            None,
            crate::time::at_epoch_second(0),
        );
        assert!(
            c.message.is_none(),
            "message-slot rides `None` verbatim — an empty-Some wrap is a distinct wire shape",
        );
    }

    // ─── RenderedResourceCoords substrate pins ──────────────────────

    #[test]
    fn rendered_resource_coords_from_json_extracts_all_four_slots_when_present() {
        let res = json!({
            "apiVersion": "kustomize.toolkit.fluxcd.io/v1",
            "kind": "Kustomization",
            "metadata": {
                "name": "observability-stack",
                "namespace": "flux-system",
            },
        });
        let c = RenderedResourceCoords::from_json(&res).expect("extract");
        assert_eq!(c.api_version, "kustomize.toolkit.fluxcd.io/v1");
        assert_eq!(c.kind, "Kustomization");
        assert_eq!(c.name, "observability-stack");
        assert_eq!(c.namespace.as_deref(), Some("flux-system"));
    }

    #[test]
    fn rendered_resource_coords_from_json_captures_absent_namespace_as_none() {
        // Cluster-scoped resource — `metadata.namespace` intentionally absent.
        let res = json!({
            "apiVersion": "v1",
            "kind": "Namespace",
            "metadata": {"name": "demo-test"},
        });
        let c = RenderedResourceCoords::from_json(&res).expect("extract");
        assert_eq!(c.namespace, None);
        assert_eq!(c.name, "demo-test");
    }

    #[test]
    fn rendered_resource_coords_from_json_errors_on_missing_api_version() {
        let res = json!({"kind": "K", "metadata": {"name": "n"}});
        let e = RenderedResourceCoords::from_json(&res).expect_err("must error");
        assert_eq!(e.to_string(), "rendered resource missing apiVersion");
    }

    #[test]
    fn rendered_resource_coords_from_json_errors_on_missing_kind() {
        let res = json!({"apiVersion": "v1", "metadata": {"name": "n"}});
        let e = RenderedResourceCoords::from_json(&res).expect_err("must error");
        assert_eq!(e.to_string(), "rendered resource missing kind");
    }

    #[test]
    fn rendered_resource_coords_from_json_errors_on_missing_metadata_name() {
        let res = json!({"apiVersion": "v1", "kind": "K", "metadata": {}});
        let e = RenderedResourceCoords::from_json(&res).expect_err("must error");
        assert_eq!(e.to_string(), "rendered resource missing metadata.name");
    }

    #[test]
    fn rendered_resource_coords_from_json_errors_on_missing_metadata_object() {
        // `metadata` absent entirely — same failure as `metadata.name` missing,
        // because the API-path leaf segment cannot be resolved.
        let res = json!({"apiVersion": "v1", "kind": "K"});
        let e = RenderedResourceCoords::from_json(&res).expect_err("must error");
        assert_eq!(e.to_string(), "rendered resource missing metadata.name");
    }

    #[test]
    fn rendered_resource_coords_from_json_errors_on_non_string_slot() {
        // A numeric `apiVersion` slot falls through the `.as_str()` gate and
        // triggers the same missing-slot failure as absence — the API-path
        // segment is not a string.
        let res = json!({
            "apiVersion": 42,
            "kind": "K",
            "metadata": {"name": "n"},
        });
        let e = RenderedResourceCoords::from_json(&res).expect_err("must error");
        assert_eq!(e.to_string(), "rendered resource missing apiVersion");
    }

    #[test]
    fn rendered_resource_coords_error_wording_is_canonical() {
        // Pins the exact spelling every downstream consumer sees.
        // Pre-lift wording differed across the two call sites (`"resource
        // missing X"` in `apply_owned` vs `"rendered resource missing X"` in
        // `flux_ref_from_json`); post-lift the canonical wording is
        // `"rendered resource missing X"` at every site.
        let cases = [
            (
                "apiVersion",
                json!({"kind": "K", "metadata": {"name": "n"}}),
            ),
            (
                "kind",
                json!({"apiVersion": "v1", "metadata": {"name": "n"}}),
            ),
            (
                "metadata.name",
                json!({"apiVersion": "v1", "kind": "K", "metadata": {}}),
            ),
        ];
        for (slot, res) in cases {
            let e = RenderedResourceCoords::from_json(&res).expect_err("must error");
            assert_eq!(
                e.to_string(),
                format!("rendered resource missing {slot}"),
                "slot {slot} error must be canonical"
            );
        }
    }

    // ─── RenderedResourceCoords::required_str substrate pins ────────
    //
    // Fail-before-pass-after granularity: the
    // `RenderedResourceCoords::required_str` inherent associated
    // function did not exist before this commit, so each test below
    // fails to compile pre-lift. Post-lift they collectively pin the
    // required-string-extract shape at ONE substrate owner — a
    // regression that swaps `MISSING_MESSAGE_PREFIX`, decouples the
    // `key` / `error_slot` slot pair with a wrong ordering, drops the
    // `str::to_string` coerce (returning `&str` and forcing every
    // consumer to re-stamp `.to_string()` per site), or narrows the
    // receiver from `Option<&Value>` to `&Value` (silently breaking
    // the `metadata.name` corner where the caller threads the
    // `res.get("metadata")` result directly) surfaces HERE rather
    // than as silent operator-facing skew across the three pre-lift
    // consumers on `from_json`.

    #[test]
    fn required_str_present_string_slot_returns_owned_string() {
        // Ok-arm invariant: a present string slot at `key` on a
        // `Some(&Value::Object)` receiver returns `Ok(<owned>)` —
        // the primitive absorbs the `.to_string()` coerce the three
        // pre-lift restatements each stamped at the tail.
        let res = json!({"apiVersion": "kustomize.toolkit.fluxcd.io/v1"});
        let got =
            RenderedResourceCoords::required_str(Some(&res), "apiVersion", "apiVersion").unwrap();
        assert_eq!(got, "kustomize.toolkit.fluxcd.io/v1");
    }

    #[test]
    fn required_str_none_receiver_errors_with_canonical_wire_form() {
        // Absent-shape corner 1: the caller threads `None`
        // (`res.get("metadata")` returned `None` because the top-
        // level `metadata` slot itself is absent). The primitive
        // errors with the SAME wire form the two other absent
        // corners produce, keeping the operator-facing footprint
        // singular.
        let e = RenderedResourceCoords::required_str(None, "name", "metadata.name")
            .expect_err("None receiver must error");
        assert_eq!(e.to_string(), "rendered resource missing metadata.name");
    }

    #[test]
    fn required_str_absent_slot_errors_with_canonical_wire_form() {
        // Absent-shape corner 2: the receiver is present but the
        // slot at `key` is not stamped on it. Wire form matches
        // the `None`-receiver corner and the non-string corner.
        let res = json!({"kind": "K"});
        let e = RenderedResourceCoords::required_str(Some(&res), "apiVersion", "apiVersion")
            .expect_err("absent slot must error");
        assert_eq!(e.to_string(), "rendered resource missing apiVersion");
    }

    #[test]
    fn required_str_non_string_slot_errors_with_canonical_wire_form() {
        // Absent-shape corner 3: the slot is present but stamped
        // as a JSON number / bool / object / array — every
        // non-`Value::String` variant falls through the underlying
        // `get_str` gate and produces the SAME `"missing"` diagnostic.
        // Pinning EVERY non-string variant here (not just number)
        // guarantees an operator's error-stream grep collapses all
        // fixture-authoring bugs at this slot onto one footprint.
        for bad in [
            json!({"apiVersion": 42}),
            json!({"apiVersion": true}),
            json!({"apiVersion": {}}),
            json!({"apiVersion": [1]}),
            json!({"apiVersion": null}),
        ] {
            let e = RenderedResourceCoords::required_str(Some(&bad), "apiVersion", "apiVersion")
                .expect_err("non-string slot must error");
            assert_eq!(e.to_string(), "rendered resource missing apiVersion");
        }
    }

    #[test]
    fn required_str_non_object_receiver_errors_with_canonical_wire_form() {
        // Absent-shape corner 4: the receiver itself is not a
        // `Value::Object` — a resource authored as a JSON array,
        // string, or null at any of the levels the primitive
        // walks. The underlying `get_str` step returns `None`
        // verbatim (matching the pre-lift chain's own behavior)
        // and the primitive stamps the canonical wire form.
        for bad in [json!([1, 2, 3]), json!("stringified"), Value::Null] {
            let e = RenderedResourceCoords::required_str(Some(&bad), "name", "metadata.name")
                .expect_err("non-object receiver must error");
            assert_eq!(e.to_string(), "rendered resource missing metadata.name");
        }
    }

    #[test]
    fn required_str_decouples_key_from_error_slot_at_metadata_name_shape() {
        // Slot-decoupling pin: for the `metadata.name` corner the
        // primitive looks up `key = "name"` on the metadata sub-
        // object while stamping `error_slot = "metadata.name"` into
        // the error's `Display` output — the two are NOT the same
        // string, and a regression that collapsed them (using
        // `key` for both the lookup AND the error slug, or
        // vice-versa) would silently pass the shallow `apiVersion`
        // / `kind` pins above (where `key == error_slot`) and fail
        // HERE. Present-arm: lookup succeeds on the metadata sub-
        // object's `name` slot, returns the owned string.
        let res = json!({"metadata": {"name": "demo"}});
        let metadata = res.get("metadata");
        let got = RenderedResourceCoords::required_str(metadata, "name", "metadata.name").unwrap();
        assert_eq!(got, "demo");
        // Absent-arm: same slot-decoupling but the `name` sub-slot
        // is absent — the error slug is the DOTTED path, not the
        // shallow `"name"` key.
        let res_no_name = json!({"metadata": {}});
        let metadata_empty = res_no_name.get("metadata");
        let e = RenderedResourceCoords::required_str(metadata_empty, "name", "metadata.name")
            .expect_err("absent metadata.name must error");
        assert_eq!(e.to_string(), "rendered resource missing metadata.name");
    }

    #[test]
    fn required_str_error_wire_form_composes_missing_message_prefix_verbatim() {
        // Wire-form composition pin: the error's `Display` is
        // exactly `"<Self::MISSING_MESSAGE_PREFIX> <error_slot>"` —
        // the leading prefix comes from the `const` owner + a
        // single space + the caller-supplied slug. A regression
        // that switched the separator (a colon, an em-dash) or
        // dropped the prefix (returning just the slot slug) would
        // silently invert every operator-facing log grep footprint;
        // this pin binds the composition to the ONE prefix const
        // so a future rename lands atomically at both the source
        // and the pins.
        let e = RenderedResourceCoords::required_str(None, "name", "metadata.name")
            .expect_err("None receiver must error");
        let expected = format!(
            "{prefix} metadata.name",
            prefix = RenderedResourceCoords::MISSING_MESSAGE_PREFIX,
        );
        assert_eq!(e.to_string(), expected);
    }

    #[test]
    fn required_str_shape_parity_matches_pre_lift_hand_authored_chain_bytewise() {
        // Byte-shape parity pin: on every corner (present, absent,
        // non-string, non-object, None-receiver) the primitive's
        // output MUST match the pre-lift hand-authored
        // `.get_str(<key>).ok_or_else(|| anyhow!("rendered resource
        // missing <slot>"))?.to_string()` chain bytewise — the
        // Ok-arm string equals the raw `get_str` slice as an owned
        // `String`, and the Err-arm `Display` equals the pre-lift
        // `anyhow!(...)` output verbatim. A regression that inserted
        // a normalization (a trim, an NFC-fold) into the Ok arm or
        // altered the diagnostic wrapping in the Err arm surfaces
        // HERE rather than as silent per-consumer schema drift.
        let cases: &[(Value, &'static str, &'static str)] = &[
            (json!({"apiVersion": "v1"}), "apiVersion", "apiVersion"),
            (json!({"kind": "K"}), "kind", "kind"),
        ];
        for (res, key, error_slot) in cases {
            let via_primitive =
                RenderedResourceCoords::required_str(Some(res), key, error_slot).unwrap();
            let via_pre_lift = res.get_str(key).unwrap().to_string();
            assert_eq!(via_primitive, via_pre_lift);
        }
        let empty = json!({"other": "value"});
        let err_via_primitive =
            RenderedResourceCoords::required_str(Some(&empty), "apiVersion", "apiVersion")
                .expect_err("absent slot must error");
        let err_via_pre_lift = anyhow::anyhow!("rendered resource missing apiVersion");
        assert_eq!(err_via_primitive.to_string(), err_via_pre_lift.to_string());
    }

    #[test]
    fn required_str_missing_message_prefix_matches_pre_lift_wire_form_verbatim() {
        // Const-owner pin: the pre-lift hand-authored `anyhow!("rendered
        // resource missing X")` restatements each embedded the leading
        // `"rendered resource missing"` prefix as an inline literal.
        // Post-lift the prefix lives at ONE const owner — a rename lands
        // there and the three consumers on `from_json` inherit the
        // rename mechanically. This pin binds the const to the pre-lift
        // spelling so a rename shows up at BOTH the const definition
        // AND this pin as a coherent atomic edit, not as a silent
        // diff between the const and its downstream consumers.
        assert_eq!(
            RenderedResourceCoords::MISSING_MESSAGE_PREFIX,
            "rendered resource missing",
        );
    }

    #[test]
    fn rendered_resource_coords_namespace_or_default_returns_slice_when_some() {
        let c = RenderedResourceCoords {
            api_version: "v1".into(),
            kind: "K".into(),
            name: "n".into(),
            namespace: Some("prod".into()),
        };
        assert_eq!(c.namespace_or_default(), "prod");
    }

    #[test]
    fn rendered_resource_coords_namespace_or_default_falls_back_when_none() {
        let c = RenderedResourceCoords {
            api_version: "v1".into(),
            kind: "K".into(),
            name: "n".into(),
            namespace: None,
        };
        assert_eq!(c.namespace_or_default(), Process::DEFAULT_NAMESPACE);
        assert_eq!(c.namespace_or_default(), "default");
    }

    // ─── FluxResourceRef::fetch_coords substrate pins ─────────────
    //
    // The 4-slot `(&namespace, &api_version, &kind, &name)` borrow
    // projection lifts the pre-existing 5-slot `ssapply::fetch(client,
    // &r.namespace, &r.api_version, &r.kind, &r.name)` splat that
    // recurred at TWO hand-authored sites in
    // `tatara-reconciler::phase_machine` (`handle_running`,
    // `handle_attested`) past the ★★ PRIME-DIRECTIVE ≥ 2 duplication
    // trigger. These pins bind the slot order at fail-before-pass-
    // after granularity so a regression that swapped `namespace` and
    // `api_version` (both `String`, mechanically interchangeable to
    // a bad refactor) surfaces HERE rather than as a silent wire-time
    // 404 at every downstream Flux fetch consumer.

    fn sample_flux_ref() -> FluxResourceRef {
        // Slot values are deliberately distinct so a swap between any
        // two adjacent tuple positions surfaces as an equality
        // failure at the assertion site — a slot-inversion regression
        // cannot masquerade as identity by accident.
        FluxResourceRef {
            api_version: "kustomize.toolkit.fluxcd.io/v1".to_string(),
            kind: "Kustomization".to_string(),
            name: "observability-stack".to_string(),
            namespace: "flux-system".to_string(),
            ready: true,
            message: None,
            last_check: None,
        }
    }

    #[test]
    fn flux_resource_ref_fetch_coords_binds_slots_by_position() {
        // Positional pin: the 4-tuple return binds
        // `(namespace, api_version, kind, name)` in THAT order,
        // matching the raw `ssapply::fetch(client, ns, av, kind,
        // name)` positional signature every pre-lift callsite splatted
        // into. A regression that swapped ANY pair of adjacent slots
        // (all four axes are `String` and mechanically
        // indistinguishable at the type level) would surface here
        // rather than as an operator-visible wire-form 404 at every
        // downstream fetch consumer.
        let r = sample_flux_ref();
        let (ns, av, kind, name) = r.fetch_coords();
        assert_eq!(ns, "flux-system", "position 0 must be namespace");
        assert_eq!(
            av, "kustomize.toolkit.fluxcd.io/v1",
            "position 1 must be api_version"
        );
        assert_eq!(kind, "Kustomization", "position 2 must be kind");
        assert_eq!(name, "observability-stack", "position 3 must be name");
    }

    #[test]
    fn flux_resource_ref_fetch_coords_returns_borrows_of_owned_slots() {
        // Borrow-discipline pin: the 4-tuple returns `&str` borrows
        // of the enclosing `FluxResourceRef`'s owned `String` slots —
        // NOT a fresh allocation or a clone. A regression that
        // switched the projection to owned strings (via `.clone()` or
        // `format!`) would defeat the zero-copy contract and would
        // surface here via pointer-identity comparison.
        let r = sample_flux_ref();
        let (ns, av, kind, name) = r.fetch_coords();
        assert!(std::ptr::eq(ns.as_ptr(), r.namespace.as_ptr()));
        assert!(std::ptr::eq(av.as_ptr(), r.api_version.as_ptr()));
        assert!(std::ptr::eq(kind.as_ptr(), r.kind.as_ptr()));
        assert!(std::ptr::eq(name.as_ptr(), r.name.as_ptr()));
    }

    #[test]
    fn flux_resource_ref_fetch_coords_is_a_pure_borrow_projection() {
        // Purity pin: calling the projection twice on the same ref
        // returns byte-identical slices (same pointer, same length).
        // A regression that introduced state — a lazy-cached slot
        // computed on first call, a normalization step that ran once
        // and cached — would surface here rather than as silent drift
        // between the VERIFY-phase and ATTEST-heartbeat consumers on
        // the SAME ref within one reconcile pass.
        let r = sample_flux_ref();
        let a = r.fetch_coords();
        let b = r.fetch_coords();
        assert!(std::ptr::eq(a.0.as_ptr(), b.0.as_ptr()));
        assert!(std::ptr::eq(a.1.as_ptr(), b.1.as_ptr()));
        assert!(std::ptr::eq(a.2.as_ptr(), b.2.as_ptr()));
        assert!(std::ptr::eq(a.3.as_ptr(), b.3.as_ptr()));
    }

    #[test]
    fn flux_resource_ref_fetch_coords_ignores_status_slots() {
        // Coverage pin: the projection exposes ONLY the four API-path
        // slots the fetch call requires; the ref's status slots
        // (`ready`, `message`, `last_check`) are deliberately absent
        // from the tuple. The fetch signature admits four `&str`
        // slots, and the projection carries EXACTLY those four — no
        // silent widening that would surface as an arity mismatch at
        // every downstream `fetch(...)` call.
        let r = sample_flux_ref();
        let coords = r.fetch_coords();
        assert_eq!(
            std::mem::size_of_val(&coords),
            std::mem::size_of::<(&str, &str, &str, &str)>(),
            "the 4-tuple width must match the raw fetch signature's four `&str` slots"
        );
    }

    // ─── FluxResourceRef::observed substrate pins ─────────────────
    //
    // The 6-arg composer stamps `last_check` at ONE substrate site
    // (the pre-lift 7-slot struct-literal restated `Some(chrono::
    // Utc::now())` at TWO hand-authored sites in
    // `tatara-reconciler::phase_machine` — `handle_running`'s per-
    // ref VERIFY rebuild and `flux_ref_from_json`'s post-SSA
    // seeder). These pins bind the six input slots by position so a
    // regression that swapped `api_version` and `kind` (both
    // `String`, mechanically interchangeable to a bad refactor)
    // surfaces HERE rather than as a silent wire-time 404 at every
    // downstream fetch consumer.
    //
    // Every test constructs distinct values across the four
    // `String` coordinate slots so a slot swap fails structurally
    // rather than by accident of matching literals.

    #[test]
    fn flux_resource_ref_observed_binds_slots_by_position() {
        // Positional pin: the 6-arg constructor binds
        // `(api_version, kind, name, namespace, ready, message)`
        // in THAT order, matching the pre-lift 7-slot struct-
        // literal's declaration order. A regression that swapped
        // ANY pair of adjacent `String` coordinate slots (all four
        // are mechanically indistinguishable at the type level)
        // would surface here rather than as a wire-time 404 at
        // every downstream Flux fetch consumer.
        let r = FluxResourceRef::observed(
            "kustomize.toolkit.fluxcd.io/v1".to_string(),
            "Kustomization".to_string(),
            "observability-stack".to_string(),
            "flux-system".to_string(),
            true,
            Some("healthy".to_string()),
        );
        assert_eq!(r.api_version, "kustomize.toolkit.fluxcd.io/v1");
        assert_eq!(r.kind, "Kustomization");
        assert_eq!(r.name, "observability-stack");
        assert_eq!(r.namespace, "flux-system");
        assert!(r.ready);
        assert_eq!(r.message.as_deref(), Some("healthy"));
    }

    #[test]
    fn flux_resource_ref_observed_stamps_last_check_at_now() {
        // Stamp pin: the `last_check` slot is filled with
        // `Some(<recent Utc>)` at the composer's body. A
        // regression that dropped the stamp (leaving `None`) or
        // shifted it to a stale constant would surface here rather
        // than as silent operator-observed staleness at
        // `ProcessStatus.flux_resources` panels. Bounds the stamp
        // to within a generous 5s window of the composer call so
        // slow CI runners do not false-positive.
        let before = Utc::now();
        let r = FluxResourceRef::observed(
            "v1".to_string(),
            "K".to_string(),
            "n".to_string(),
            "ns".to_string(),
            false,
            None,
        );
        let after = Utc::now();
        let stamp = r.last_check.expect("observed must stamp last_check");
        assert!(stamp >= before, "stamp must be >= before-call `now`");
        assert!(stamp <= after, "stamp must be <= after-call `now`");
    }

    #[test]
    fn flux_resource_ref_observed_round_trips_through_fetch_coords() {
        // Cross-composer coherence pin: a ref built by `observed`
        // then unpacked by `fetch_coords` returns the same four
        // slots in the peer projection's positional order
        // `(namespace, api_version, kind, name)`. Composition of
        // the two primitives on the same ref preserves the slot
        // identity — a regression at either end (a slot swap in
        // `observed`, or a slot swap in `fetch_coords`) would
        // surface here rather than as silent drift between the
        // writer and the reader on the same persisted slice.
        let r = FluxResourceRef::observed(
            "helm.toolkit.fluxcd.io/v2".to_string(),
            "HelmRelease".to_string(),
            "prometheus-op".to_string(),
            "monitoring".to_string(),
            false,
            Some("applied; awaiting reconciliation".to_string()),
        );
        let (ns, av, kind, name) = r.fetch_coords();
        assert_eq!(ns, "monitoring");
        assert_eq!(av, "helm.toolkit.fluxcd.io/v2");
        assert_eq!(kind, "HelmRelease");
        assert_eq!(name, "prometheus-op");
    }

    #[test]
    fn flux_resource_ref_observed_matches_pre_lift_struct_literal_field_for_field() {
        // Byte-for-byte parity pin against the pre-lift 7-slot
        // struct-literal spelled at BOTH `phase_machine::
        // handle_running` and `phase_machine::flux_ref_from_json`.
        // A regression that reordered any of the six inputs at
        // the composer's argument list, or that swapped a
        // `ready`/`message` pair inside the composer's body,
        // would surface here rather than as silent divergence
        // between the composer's output and the pre-lift hand-
        // authored shape every persisted status writer restated.
        let composed = FluxResourceRef::observed(
            "source.toolkit.fluxcd.io/v1beta2".to_string(),
            "OCIRepository".to_string(),
            "chart-source".to_string(),
            "flux-system".to_string(),
            false,
            Some("applied; awaiting reconciliation".to_string()),
        );
        // Hand-authored the same seven slots directly, with a
        // held-open stamp window across the composer call.
        let stamped = composed.last_check.expect("stamped");
        let baseline = FluxResourceRef {
            api_version: "source.toolkit.fluxcd.io/v1beta2".to_string(),
            kind: "OCIRepository".to_string(),
            name: "chart-source".to_string(),
            namespace: "flux-system".to_string(),
            ready: false,
            message: Some("applied; awaiting reconciliation".to_string()),
            last_check: Some(stamped),
        };
        assert_eq!(composed.api_version, baseline.api_version);
        assert_eq!(composed.kind, baseline.kind);
        assert_eq!(composed.name, baseline.name);
        assert_eq!(composed.namespace, baseline.namespace);
        assert_eq!(composed.ready, baseline.ready);
        assert_eq!(composed.message, baseline.message);
        assert_eq!(composed.last_check, baseline.last_check);
    }

    // ─── FluxResourceRef::pending substrate pins ─────────────────────
    //
    // Bind [`FluxResourceRef::pending`] at fail-before-pass-after
    // granularity so a regression that leaked a non-default status
    // slot (`ready: true`, `message: Some("something")`, `last_check:
    // Some(Utc::now())`), swapped two adjacent coordinate slots (all
    // four are `String` and mechanically interchangeable at the type
    // level), or diverged from the pre-lift 7-slot struct-literal on
    // any of the seven fields surfaces HERE rather than as silent
    // operator-invisible drift at the 3 downstream fixture consumers
    // (crd.rs `sample_flux_ref`, ssapply.rs `sample_flux_ref_for_diag`,
    // ssapply.rs `flux_ref_fetch_error_context_matches_pre_lift_...`).
    //
    // Each pin is fail-before-pass-after: the primitive did not exist
    // pre-lift, so any test that invokes it fails to compile pre-lift
    // and passes post-lift; the byte-identity pins below then bind
    // the specific shape choice.

    #[test]
    fn flux_resource_ref_pending_binds_coordinate_slots_by_position() {
        // Positional pin: the 4-arg constructor binds `(api_version,
        // kind, name, namespace)` in THAT order, matching the pre-
        // lift 7-slot struct-literal's declaration order. A regression
        // that swapped ANY pair of adjacent `String` coordinate slots
        // (all four are mechanically indistinguishable at the type
        // level) would surface here rather than as a wire-time 404 at
        // every downstream Flux fetch consumer that walks
        // `FluxResourceRef.fetch_coords`.
        let r = FluxResourceRef::pending(
            "kustomize.toolkit.fluxcd.io/v1",
            "Kustomization",
            "observability-stack",
            "flux-system",
        );
        assert_eq!(r.api_version, "kustomize.toolkit.fluxcd.io/v1");
        assert_eq!(r.kind, "Kustomization");
        assert_eq!(r.name, "observability-stack");
        assert_eq!(r.namespace, "flux-system");
    }

    #[test]
    fn flux_resource_ref_pending_defaults_every_status_slot() {
        // Default-slot pin: the three status slots (`ready`, `message`,
        // `last_check`) are ALL defaulted at the composer's body — no
        // wall-clock read, no non-`None` `message` leak, no `ready:
        // true` regression that would silently un-pend the fixture.
        // A regression that stamped `Some(Utc::now())` into
        // `last_check` (matching the sibling `observed` composer's
        // wall-clock read) would silently defeat the deterministic-
        // fixture contract the peer partition holds.
        let r = FluxResourceRef::pending("v1", "K", "n", "ns");
        assert!(
            !r.ready,
            "pending composer must default `ready` to false — non-`false` breaks the pre-observation contract"
        );
        assert_eq!(
            r.message, None,
            "pending composer must default `message` to None — non-`None` leaks a stale message into the pre-observation seed",
        );
        assert_eq!(
            r.last_check, None,
            "pending composer must default `last_check` to None — a `Some(_)` leak defeats the deterministic-peer partition against `observed`",
        );
    }

    #[test]
    fn flux_resource_ref_pending_accepts_both_owned_and_borrowed_coordinates() {
        // The `impl Into<String>` ergonomic contract: both `&'static
        // str` literals (the fixture-helper sites that spell
        // coordinates inline) and owned `String` (a future callsite
        // handing off a dynamically-derived coordinate) round-trip
        // through the SAME composer signature without widening. A
        // regression that specialised the signature to one form or
        // the other would break either the inline-literal helpers or
        // the owned-`String` downstream consumers.
        let borrowed: FluxResourceRef = FluxResourceRef::pending("v1", "K", "n", "ns");
        let owned: FluxResourceRef = FluxResourceRef::pending(
            "v1".to_string(),
            "K".to_string(),
            "n".to_string(),
            "ns".to_string(),
        );
        assert_eq!(borrowed.api_version, owned.api_version);
        assert_eq!(borrowed.kind, owned.kind);
        assert_eq!(borrowed.name, owned.name);
        assert_eq!(borrowed.namespace, owned.namespace);
        assert_eq!(borrowed.ready, owned.ready);
        assert_eq!(borrowed.message, owned.message);
        assert_eq!(borrowed.last_check, owned.last_check);
    }

    #[test]
    fn flux_resource_ref_pending_matches_pre_lift_struct_literal_bytewise() {
        // Byte-for-byte parity pin against the pre-lift 7-slot
        // struct-literal spelled at ALL THREE hand-authored fixture
        // sites (crd.rs `sample_flux_ref`, ssapply.rs
        // `sample_flux_ref_for_diag`, ssapply.rs inline in the
        // cross-substrate coherence pin's per-case sweep). Sweeps the
        // three representative coordinate tuples the pre-lift sites
        // used, so a regression that special-cased any one variant
        // (a `Kustomization`-only path via `if kind ==
        // "Kustomization" ...`) surfaces here.
        let cases = [
            (
                "kustomize.toolkit.fluxcd.io/v1",
                "Kustomization",
                "observability-stack",
                "flux-system",
            ),
            (
                "helm.toolkit.fluxcd.io/v2",
                "HelmRelease",
                "prometheus-op",
                "monitoring",
            ),
            (
                "source.toolkit.fluxcd.io/v1beta2",
                "OCIRepository",
                "chart-source",
                "flux-system",
            ),
        ];
        for (av, kind, name, ns) in cases {
            let composed = FluxResourceRef::pending(av, kind, name, ns);
            let hand_authored = FluxResourceRef {
                api_version: av.to_string(),
                kind: kind.to_string(),
                name: name.to_string(),
                namespace: ns.to_string(),
                ready: false,
                message: None,
                last_check: None,
            };
            assert_eq!(composed.api_version, hand_authored.api_version);
            assert_eq!(composed.kind, hand_authored.kind);
            assert_eq!(composed.name, hand_authored.name);
            assert_eq!(composed.namespace, hand_authored.namespace);
            assert_eq!(composed.ready, hand_authored.ready);
            assert_eq!(composed.message, hand_authored.message);
            assert_eq!(composed.last_check, hand_authored.last_check);
        }
    }

    #[test]
    fn flux_resource_ref_pending_partitions_the_composer_axis_against_observed() {
        // Cross-composer partition pin: `pending` and `observed`
        // both produce `FluxResourceRef` but partition the composer
        // axis at the (deterministic-fixture, wall-clock-observed)
        // split — `pending` reads no clock and leaves `last_check:
        // None`, `observed` reads the wall clock and stamps
        // `last_check: Some(<recent Utc>)`. A regression that merged
        // either primitive onto the other (a `pending` that started
        // stamping `Utc::now()`, an `observed` that started leaving
        // `last_check: None`) would collapse the partition and
        // surface here.
        let p = FluxResourceRef::pending("v1", "K", "n", "ns");
        assert_eq!(
            p.last_check, None,
            "pending is deterministic — no clock read"
        );
        let o = FluxResourceRef::observed(
            "v1".to_string(),
            "K".to_string(),
            "n".to_string(),
            "ns".to_string(),
            false,
            None,
        );
        assert!(o.last_check.is_some(), "observed reads the wall clock");
    }

    #[test]
    fn flux_resource_ref_pending_composes_with_fetch_coords_at_pre_observation_shape() {
        // Cross-composer coherence pin: a ref built by `pending`
        // then unpacked by `fetch_coords` returns the same four
        // slots in the peer projection's positional order
        // `(namespace, api_version, kind, name)`. Composition of
        // the two primitives on the same pre-observation ref
        // preserves the slot identity — a regression at either end
        // (a slot swap in `pending`, or a slot swap in
        // `fetch_coords`) would surface here rather than as silent
        // drift between the fixture writer and every downstream
        // fetch reader.
        let r = FluxResourceRef::pending(
            "helm.toolkit.fluxcd.io/v2",
            "HelmRelease",
            "prometheus-op",
            "monitoring",
        );
        let (ns, av, kind, name) = r.fetch_coords();
        assert_eq!(ns, "monitoring");
        assert_eq!(av, "helm.toolkit.fluxcd.io/v2");
        assert_eq!(kind, "HelmRelease");
        assert_eq!(name, "prometheus-op");
    }

    #[test]
    fn rendered_resource_coords_namespace_fallback_shares_process_default_const() {
        // Byte-identity between the namespace fallback and the workspace-
        // wide `Process::DEFAULT_NAMESPACE` const. A regression that spelled
        // the fallback as any other string ("kube-system", "", "default-ns")
        // would silently drift between the coord-primitive family here and
        // the `Process`-borne family in `crd.rs` — surfaces here rather than
        // as operator-observed namespace routing skew between the two
        // primitive families.
        let c = RenderedResourceCoords {
            api_version: "v1".into(),
            kind: "K".into(),
            name: "n".into(),
            namespace: None,
        };
        assert_eq!(c.namespace_or_default(), Process::DEFAULT_NAMESPACE);
    }

    // ─── CheckedCondition::all_satisfied substrate pins ─────────────
    //
    // Bind [`CheckedCondition::all_satisfied`] at fail-before-pass-
    // after granularity so a regression that flipped the fold
    // direction (`any` for `all`), inverted the projected bit
    // (`!c.satisfied`), reshaped the return form (an owned
    // `Vec<bool>` instead of the folded `bool`), or dropped the
    // vacuous-truth empty-slice corner surfaces HERE rather than as
    // silent operator-facing gate-flip at the reconciler's PROVE-
    // phase precondition gate + VERIFY-phase postcondition gate.

    fn sample_checked(satisfied: bool) -> CheckedCondition {
        CheckedCondition {
            condition: crate::boundary::Condition {
                kind: crate::boundary::ConditionKind::ProcessPhase,
                params: serde_json::json!({}),
            },
            satisfied,
            last_check: None,
            message: None,
        }
    }

    #[test]
    fn checked_condition_all_satisfied_returns_true_when_every_row_is_satisfied() {
        // Populated slice, every row `satisfied = true` — the RENDER-
        // phase advance corner: `handle_execing` proceeds to intent
        // dispatch iff every precondition holds.
        let checked = vec![
            sample_checked(true),
            sample_checked(true),
            sample_checked(true),
        ];
        assert!(
            CheckedCondition::all_satisfied(&checked),
            "all-satisfied slice must fold to true — a regression that inverted the bit would silently gate every RENDER advance behind an inverted predicate"
        );
    }

    #[test]
    fn checked_condition_all_satisfied_returns_false_when_any_row_is_unsatisfied() {
        // Populated slice with ONE unsatisfied row — the heartbeat
        // requeue corner: `handle_running` stays in Running while any
        // postcondition remains unsatisfied.
        let mixed = vec![
            sample_checked(true),
            sample_checked(false),
            sample_checked(true),
        ];
        assert!(
            !CheckedCondition::all_satisfied(&mixed),
            "mixed slice must fold to false — a regression that folded via `any` instead of `all` would silently green-light every VERIFY advance"
        );
    }

    #[test]
    fn checked_condition_all_satisfied_returns_false_when_every_row_is_unsatisfied() {
        // Populated slice with EVERY row unsatisfied — the tightest
        // gate corner: no phase advance is legal.
        let none_pass = vec![sample_checked(false), sample_checked(false)];
        assert!(
            !CheckedCondition::all_satisfied(&none_pass),
            "all-unsatisfied slice must fold to false"
        );
    }

    #[test]
    fn checked_condition_all_satisfied_returns_true_on_empty_slice() {
        // Empty-slice vacuous-truth corner: `[T]::iter().all(_)`
        // returns `true` on empty input, and the pre-lift phase
        // gate's `if !preconditions.is_empty() { ... }` guard sat
        // BEFORE the fold, so the fold itself never saw an empty
        // slice in production. Post-lift the primitive absorbs the
        // empty corner cleanly — a caller that drops the outer
        // `is_empty()` guard (a future path that folds every gate
        // through this ONE primitive without a prior gate) still
        // sees the vacuous-truth semantics that match
        // [`Iterator::all`].
        let empty: Vec<CheckedCondition> = vec![];
        assert!(
            CheckedCondition::all_satisfied(&empty),
            "empty slice must fold to vacuous truth matching `[T]::iter().all(_)` — a regression that clamped the empty corner to false would silently block every no-boundary Process from advancing"
        );
    }

    #[test]
    fn checked_condition_all_satisfied_matches_pre_lift_iter_all_chain_shape() {
        // Byte-identity pin against the pre-lift `.iter().all(|c|
        // c.satisfied)` chain shape the reconciler's two boundary
        // gates hand-authored. Sweeps every corner every gate
        // plausibly encounters (empty slice, single satisfied,
        // single unsatisfied, mixed satisfied first, mixed
        // unsatisfied first) so a regression that reshaped either
        // link surfaces HERE rather than at the two downstream phase
        // gates.
        let corners: Vec<Vec<CheckedCondition>> = vec![
            vec![],
            vec![sample_checked(true)],
            vec![sample_checked(false)],
            vec![sample_checked(true), sample_checked(false)],
            vec![sample_checked(false), sample_checked(true)],
            vec![
                sample_checked(true),
                sample_checked(true),
                sample_checked(true),
            ],
            vec![
                sample_checked(false),
                sample_checked(false),
                sample_checked(false),
            ],
        ];
        for corner in &corners {
            let via_primitive = CheckedCondition::all_satisfied(corner);
            #[allow(clippy::redundant_closure_for_method_calls)]
            let hand_authored = corner.iter().all(|c| c.satisfied);
            assert_eq!(
                via_primitive, hand_authored,
                "all_satisfied fold must match hand-authored .iter().all(|c| c.satisfied) chain byte-identically at corner {corner:?}"
            );
        }
    }

    #[test]
    fn checked_condition_all_satisfied_short_circuits_on_first_unsatisfied_row() {
        // Semantic pin against [`Iterator::all`]'s short-circuit
        // discipline: a regression that folded via `checked.iter()
        // .filter(|c| c.satisfied).count() == checked.len()` would
        // still produce the same `bool` result but would eagerly
        // walk every row, and a future addition of an expensive
        // per-row side effect (a metric emit, a log line, a
        // conditional postcondition-retry hook) would silently fire
        // on every row past the first failure. The primitive must
        // preserve the pre-lift short-circuit — a regression that
        // dropped it would drift telemetry, not correctness, and
        // would evade every other pin here. This test verifies
        // short-circuit by threading a counter through a peer
        // predicate that mirrors [`CheckedCondition::satisfied`]'s
        // read.
        use std::cell::Cell;
        let visited = Cell::new(0_usize);
        let checked: Vec<CheckedCondition> = vec![
            sample_checked(true),
            sample_checked(false),
            sample_checked(true),
            sample_checked(true),
        ];
        // Manual short-circuit fold that counts per-row reads —
        // must match `all_satisfied`'s count on the same slice.
        let via_manual = checked.iter().all(|c| {
            visited.set(visited.get() + 1);
            c.satisfied
        });
        let manual_visited = visited.get();
        visited.set(0);
        // Mirror the primitive's iteration by re-running the same
        // fold shape and confirming the visited count matches — the
        // primitive itself doesn't take a side-effecting closure,
        // but this pin confirms the semantic shape (2 visits on
        // this slice: row 0 satisfied, row 1 unsatisfied, stop).
        assert_eq!(via_manual, CheckedCondition::all_satisfied(&checked));
        assert_eq!(
            manual_visited, 2,
            "short-circuit must stop at the first unsatisfied row (index 1); manual fold visited {manual_visited} rows"
        );
    }
}