tatara_process/

pool.rs

//! `EphemeralPool` CRD — a population of warm, pre-attested ephemeral
//! Processes that get *allocated* to requestors (e.g., a GitHub PR
//! flow) on demand and *returned* (per a typed policy) when the
//! requestor releases them.
//!
//! Compounding move: the pool is a population manager **over the
//! existing Process algebra**, not a parallel runtime. A pool member
//! is just a `Process` with `Lifetime::Permanent` while in the free
//! list; allocation is "the operator (the pool reconciler) flips
//! that Process's lifetime slot to Ephemeral with the requestor's
//! TTL." Zero new compute primitive.
//!
//! Topology:
//!
//! ```text
//! EphemeralPool       (this CRD)
//!   ├── PoolSpec      (desired_size, template (EphemeralSpec), return_policy, selector)
//!   ├── PoolStatus    (phase, free / allocated / spawning / returning counts, members)
//!   └── owns N Processes via ownerReferences (one per pool slot)
//!
//! EphemeralAllocation (see allocation.rs)
//!   ├── AllocationSpec (pool_ref, requestor, requested_at, lifetime override)
//!   └── AllocationStatus (phase, assigned_process_ref, allocated_at, expires_at)
//! ```

use chrono::{DateTime, Utc};
use kube::CustomResource;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

use crate::ephemeral::EphemeralSpec;

/// `EphemeralPool` CRD spec — typed pool of warm Processes.
///
/// ```yaml
/// apiVersion: tatara.pleme.io/v1alpha1
/// kind: EphemeralPool
/// metadata:
///   name: attest-pool
///   namespace: ephemeral-pools
/// spec:
///   desiredSize: 3
///   minSize: 1
///   maxSize: 5
///   returnPolicy: Reset
///   selector:
///     repos: ["pleme-io/demo-*"]
///     branches: ["main", "release-*"]
///     prLabels: ["needs-ephemeral"]
///   template:
///     aplicacao:
///       chartRef: "oci://ghcr.io/pleme-io/charts/lareira-demo-app"
///       version: "0.5.5"
///       profile: "all-in-one"
///       …
///     ttl: "2h"
///     teardown: OnAttested
///     postconditions: [ … ]
/// ```
#[derive(CustomResource, Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[kube(
    group = "tatara.pleme.io",
    version = "v1alpha1",
    kind = "EphemeralPool",
    plural = "ephemeralpools",
    shortname = "epool",
    namespaced,
    status = "PoolStatus",
    printcolumn = r#"{"name":"Desired","type":"integer","jsonPath":".spec.desiredSize"}"#,
    printcolumn = r#"{"name":"Ready","type":"integer","jsonPath":".status.readyCount"}"#,
    printcolumn = r#"{"name":"Allocated","type":"integer","jsonPath":".status.allocatedCount"}"#,
    printcolumn = r#"{"name":"Phase","type":"string","jsonPath":".status.phase"}"#,
    printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"#
)]
#[serde(rename_all = "camelCase")]
pub struct PoolSpec {
    /// Target number of warm Processes the pool maintains in `Free`
    /// state (sum of Free + Spawning targets `desired_size`).
    pub desired_size: u32,

    /// Hard floor on the free count. The reconciler refuses to scale
    /// below this even on cost-pressure signals. Default = 0.
    #[serde(default)]
    pub min_size: u32,

    /// Hard ceiling on total pool members (free + allocated + spawning).
    /// `0` = no cap. Default = 0.
    #[serde(default)]
    pub max_size: u32,

    /// What to do when an allocation releases.
    #[serde(default)]
    pub return_policy: ReturnPolicy,

    /// Routing selector — which allocation requests this pool serves.
    /// The reconciler matches incoming `EphemeralAllocation` CRs
    /// against this selector (most-specific wins across pools sharing
    /// a namespace).
    #[serde(default)]
    pub selector: PoolSelector,

    /// Template for each pool member — a typed `EphemeralSpec` that
    /// the reconciler lowers to `ProcessSpec` and instantiates.
    /// While in the free list each member's lifetime is overridden
    /// to `Permanent`; allocation flips it back to `Ephemeral` with
    /// the requestor's TTL.
    pub template: EphemeralSpec,

    /// How long a pool member may sit in `Free` before the reconciler
    /// recycles it (humantime). Defends against drift / stale state.
    /// Default `"24h"`.
    #[serde(default = "default_free_ttl")]
    pub free_ttl: String,

    /// Max time the reconciler allows a single allocation to hold a
    /// member before forcibly returning it (humantime). Hard cap
    /// independent of the allocation's own TTL. Default `"4h"`.
    #[serde(default = "default_max_allocation_ttl")]
    pub max_allocation_ttl: String,

    /// **R5 desired-count loop** — when set non-zero, the pool
    /// reconciler maintains exactly this many *healthy* (Running or
    /// Attested) Processes regardless of allocation pressure. Drives
    /// the "always seeking stability" property: failed members are
    /// replaced per `replacement_policy`. `0` keeps the legacy
    /// allocation-driven sizing (desired = floor of free + allocated).
    ///
    /// Operator usage: `desired: 5` means "always have 5 of these
    /// running"; failures auto-replace.
    #[serde(default)]
    pub desired: u32,

    /// **R5** — what the pool reconciler does when a member reaches
    /// `Failed` phase.
    #[serde(default)]
    pub replacement_policy: ReplacementPolicy,

    /// **R5** — when true, exactly one healthy member of the pool
    /// holds the unprefixed-form DNS hostnames declared in
    /// `template.routing` at any moment. The claim arbiter (see
    /// `tatara-reconciler::claim`) transfers atomically when the
    /// holder fails.
    #[serde(default)]
    pub stable_name_claim: bool,
}

impl EphemeralPool {
    /// Borrow-form metadata-projection primitive on the `metadata.name`
    /// axis of `EphemeralPool`: returns the K8s object name slice with
    /// the missing-name corner collapsed to the load-bearing empty-string
    /// sentinel — the ONE-liner collapse of the paired
    /// `self.metadata.name.as_deref().unwrap_or("")` incantation every
    /// pool-side consumer restated by hand pre-lift.
    ///
    /// Pre-lift the `.metadata.name.as_deref().unwrap_or("")` chain
    /// was hand-authored at TWO sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication threshold in `tatara-pool-reconciler`, both keyed
    /// by the pool's own name slot:
    /// * `router::pool_name` — the tie-break comparator inside
    ///   `best_match`; a deterministic lexicographic-min-name arbiter
    ///   across two pool candidates whose specificity scores tie.
    /// * `controller_allocation::reconcile_inner` — the `HashMap<
    ///   pool-name, Vec<PoolMember>>` lookup closure fed into
    ///   `decide_allocation_reconcile`; keys the "which pool members
    ///   back this allocation candidate?" projection at every
    ///   allocation-reconcile pass.
    ///
    /// Both sites walked the SAME `.as_deref().unwrap_or("")` chain
    /// and both wanted the `&str` form the primitive returns — as a
    /// borrow suitable for lexicographic `str::cmp` in the tie-break
    /// AND for the `HashMap<String, _>::get(&str)` lookup. Post-lift
    /// each caller reaches for `pool.name_or_empty()` and the produced
    /// slice feeds the same downstream comparator / lookup unchanged.
    ///
    /// The empty-string fallback is the SAME sentinel the sibling
    /// borrow-form primitive [`crate::crd::Process::uid_or_empty`]
    /// returns AND the SAME sentinel the owned-form sibling
    /// [`crate::crd::Process::owned_name_or_empty`] returns on the
    /// `metadata.name` axis of the sister CRD — the three primitives
    /// partition the (borrow-form × owned-form) × (uid × name) corner
    /// of the metadata-slot family on identical fallback semantics
    /// (empty string means "the slot is unset"), so a consumer that
    /// switches between the CRD surfaces based on downstream keying
    /// requirements never sees a different missing-slot spelling as
    /// a side effect.
    ///
    /// Return-form axis: `&str` mirrors the borrow-first discipline
    /// of the peer metadata primitives on `Process`
    /// ([`crate::crd::Process::namespace_or_default`],
    /// [`crate::crd::Process::name_or_placeholder`],
    /// [`crate::crd::Process::uid_or_empty`]). The one missing-slot
    /// corner the chain swallowed pre-lift (missing `metadata.name`)
    /// collapses to the empty-string sentinel so `str::is_empty` /
    /// `HashMap::get` on an unnamed pool behaves identically to what
    /// the pre-lift `.as_deref().unwrap_or("")` chain produced.
    ///
    /// A future normalization step (a name-canonicalization pass, a
    /// case-fold key builder, a per-cluster prefix stripper for
    /// cross-cluster pool-name aliasing) lands at ONE substrate
    /// method here and both downstream consumers pick up the upgrade
    /// mechanically — no per-callsite hand-edit at `pool_name` /
    /// `reconcile_inner`.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the `.metadata.name.as_deref().unwrap_or("")` chain recurred
    /// at two hand-authored sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication trigger, and is lifted to ONE owner here).
    /// THEORY.md §II.1 invariant 5 (composition preserves proofs —
    /// the pins bind the missing-name corner + the empty-string
    /// sentinel byte-shape + the borrow-form `&str` lifetime + the
    /// byte-identical parity with the pre-lift chain + the fallback-
    /// value coherence with `Process::uid_or_empty` /
    /// `Process::owned_name_or_empty` on the metadata-slot × empty-
    /// sentinel axis, so a regression that drifted any surface at
    /// `tests::name_or_empty_*` here rather than as silent operator-
    /// facing skew between the router tie-break and the allocation
    /// member-lookup on the SAME pool candidate).
    pub fn name_or_empty(&self) -> &str {
        self.metadata.name.as_deref().unwrap_or("")
    }

    /// Owned-form metadata-projection primitive on the `metadata.name`
    /// axis of `EphemeralPool`: returns an owned `String` copy of the K8s
    /// object name with the missing-name corner collapsed to the load-
    /// bearing empty-string sentinel — the ONE-liner collapse of the
    /// paired `self.metadata.name.clone().unwrap_or_default()` incantation
    /// every pool-side consumer restated by hand pre-lift.
    ///
    /// Pre-lift the `.metadata.name.clone().unwrap_or_default()` chain
    /// was hand-authored at TWO sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication threshold in `tatara-pool-reconciler`, both keyed by
    /// the pool's own name slot in an `owned String` context:
    /// * `controller_allocation::reconcile_inner` — the
    ///   `HashMap<String, Vec<PoolMember>>` key seed inside a
    ///   `pools.iter().map(|p| ...).collect()` fanout; the map key is
    ///   the owned `String` form because the produced `HashMap<String, _>`
    ///   outlives the pool-list borrow that generated it and the
    ///   downstream `pool_members.get(pool.name_or_empty())` closure
    ///   consumes it as `&str`.
    /// * `allocation_decide::AllocationConvergenceCtx::observe` — the
    ///   `AllocationRef::name` slot seed stamped on the matched-pool
    ///   handle; the struct literal is `AllocationRef { name: String,
    ///   namespace: String }` and the produced value is threaded through
    ///   the `Decision::decide` transition rule downstream.
    ///
    /// Both sites walked the SAME `.clone().unwrap_or_default()` chain
    /// and both wanted the `String` form the primitive returns — as the
    /// owned key of a `HashMap<String, _>` and as the `String` slot of
    /// an `AllocationRef` struct literal. Post-lift each callsite reads
    /// `pool.owned_name_or_empty()` and the produced value feeds the
    /// same downstream key / struct-literal slot unchanged.
    ///
    /// The empty-string fallback is the SAME sentinel the sibling
    /// borrow-form primitive [`Self::name_or_empty`] returns AND the
    /// SAME sentinel the sibling owned-form primitive
    /// [`crate::crd::Process::owned_name_or_empty`] returns on the
    /// `metadata.name` axis of the sister CRD — the three primitives
    /// partition the (borrow-form × owned-form) corner of the metadata-
    /// name family across BOTH tatara-process CRDs on identical missing-
    /// slot semantics (empty string means "the slot is unset"), so a
    /// consumer that switches between the CRD surfaces based on
    /// downstream ownership requirements never sees a different
    /// missing-slot spelling as a side effect.
    ///
    /// Peer to [`Self::name_or_empty`] on the (return-form × ownership)
    /// axis pair — closes the corner the pool-side family previously
    /// left open:
    ///
    /// * borrow + empty sentinel → [`Self::name_or_empty`] (router tie-
    ///   break comparator, `HashMap<String, _>::get(&str)` lookup —
    ///   consumers whose downstream keys by `&str` and allocates
    ///   nothing);
    /// * owned + empty sentinel → **this method** (HashMap-key seed in
    ///   an outliving-borrow context, `AllocationRef::name` struct-
    ///   literal slot — consumers whose downstream requires the owned
    ///   `String` form because the produced value outlives the source-
    ///   pool borrow).
    ///
    /// A future normalization step (a name-canonicalization pass, a
    /// case-fold key builder, a per-cluster prefix stripper for cross-
    /// cluster pool-name aliasing) lands at ONE substrate method here
    /// and both downstream consumers pick up the upgrade mechanically —
    /// no per-callsite hand-edit at `reconcile_inner` /
    /// `AllocationConvergenceCtx::observe`.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the `.metadata.name.clone().unwrap_or_default()` chain recurred
    /// at two hand-authored sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication trigger, and is lifted to ONE owner here).
    /// THEORY.md §II.1 invariant 5 (composition preserves proofs —
    /// the pins bind the missing-name corner + the empty-string
    /// sentinel byte-shape + the owned-form `String` return type + the
    /// byte-identical parity with the pre-lift chain + the fallback-
    /// value coherence with [`Self::name_or_empty`] +
    /// [`crate::crd::Process::owned_name_or_empty`] on the metadata-
    /// slot × empty-sentinel axis, so a regression that drifted any
    /// surface at `tests::owned_name_or_empty_*` here rather than as
    /// silent operator-facing skew between the pool-members lookup key
    /// and the AllocationRef seed on the SAME pool candidate).
    pub fn owned_name_or_empty(&self) -> String {
        self.metadata.name.clone().unwrap_or_default()
    }

    /// Copy-form metadata-projection primitive on the deletion-tombstone
    /// axis of `EphemeralPool`: returns `true` iff the K8s API server
    /// has stamped a `metadata.deletionTimestamp` on this pool (the
    /// moment the object entered the "being deleted" corner of its
    /// lifecycle, after which further mutating writes are refused and
    /// finalizers are drained before the object is actually removed) —
    /// the ONE-liner collapse of the paired
    /// `self.metadata.deletion_timestamp.is_some()` incantation every
    /// pool-side consumer restated by hand pre-lift.
    ///
    /// Pre-lift the `.metadata.deletion_timestamp.is_some()` chain was
    /// hand-authored at TWO sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication threshold in `tatara-pool-reconciler`, both
    /// projecting the SAME tombstone-presence predicate on an
    /// `EphemeralPool` value:
    /// * `pool_decide::decide_pool_reconcile` — the pure decision
    ///   function's deletion-preempt gate that forces
    ///   [`PoolDecision::Drain`] as soon as the API server stamps
    ///   the tombstone, before the (desired vs actual) supply-arithmetic
    ///   branches get a chance to run. Wired at the very top of the
    ///   decision so a draining pool never spawns / reaps / expires
    ///   through the normal replenishment arithmetic while the
    ///   deletion is in flight.
    /// * `controller_pool::pool_phase_from_members` — the observed-
    ///   phase composer's tombstone-first arm that returns
    ///   [`PoolPhase::Draining`] regardless of the supply / demand
    ///   arithmetic that would otherwise pick `Ready` / `Scaling` /
    ///   `Degraded`. Keeps the reported phase honest during the
    ///   finalizer drain so operators reading `kubectl get
    ///   ephemeralpools` see the tombstone-present state as
    ///   `Draining`, not as a stale `Ready`.
    ///
    /// Both sites walked the SAME `.metadata.deletion_timestamp
    /// .is_some()` chain and both wanted the `bool` form the primitive
    /// returns — the `decide_pool_reconcile` site to gate the
    /// `→ Drain` short-circuit and the `pool_phase_from_members` site
    /// to gate the `→ Draining` short-circuit. Post-lift each callsite
    /// reads `pool.is_being_deleted()` and the produced `bool` feeds
    /// the same downstream short-circuit unchanged.
    ///
    /// Sibling to [`crate::crd::Process::is_being_deleted`] on the
    /// deletion-tombstone axis of the sister CRD — the two primitives
    /// now partition the tombstone-presence probe across BOTH
    /// tatara-process CRDs on identical missing-slot semantics
    /// (present timestamp means "the API server has begun deletion"),
    /// so an operator or reconciler that switches between the CRD
    /// surfaces never sees a different tombstone-detection spelling
    /// as a side effect.
    ///
    /// Return-form axis: `bool` matches the copy-form discipline of
    /// the sibling [`crate::crd::Process::is_being_deleted`] and of
    /// the pool-side [`crate::phase::ProcessPhase::is_alive`] +
    /// [`Self::name_or_empty`]-family primitives — the underlying
    /// slot is a wire-format `Option<Time>` that carries only
    /// presence information at this axis (the RFC-3339 timestamp
    /// payload itself is not what the two consumers read; both only
    /// probe presence to detect the tombstone-stamped state).
    /// Returning the raw `Option<&Time>` would push the `.is_some()`
    /// probe back to every callsite, restating the pre-lift chain
    /// one link shorter without collapsing the primitive.
    ///
    /// Peer to [`Self::name_or_empty`] and [`Self::owned_name_or_empty`]
    /// on the metadata-projection axis for `EphemeralPool`; this method
    /// opens the presence-probe corner for the tombstone slot. Future
    /// metadata-presence projections on the pool CRD (an
    /// `is_being_finalized` projection on
    /// `metadata.finalizers.is_empty()`'s negation, a `has_owner`
    /// projection on `metadata.owner_references.is_empty()`'s
    /// negation) land as peer methods on this same axis.
    ///
    /// A future normalization step (a per-tombstone staleness gate
    /// that returns `false` for a tombstone older than the reconciler's
    /// grace-period budget, a canonicalization pass that treats a
    /// tombstone from a paused controller as absent, a cross-cluster
    /// tombstone-observation clock skew guard) lands at ONE substrate
    /// method here and both downstream consumers pick up the upgrade
    /// mechanically — no per-callsite hand-edit at
    /// `decide_pool_reconcile` / `pool_phase_from_members`.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the `.metadata.deletion_timestamp.is_some()` chain recurred at
    /// two hand-authored sites past the ★★ PRIME-DIRECTIVE ≥ 2
    /// duplication trigger, and is lifted to ONE owner here).
    /// THEORY.md §II.1 invariant 5 (composition preserves proofs —
    /// the pins bind the missing-tombstone corner + the present-
    /// tombstone corner + the copy-form `bool` return + the byte-
    /// identical parity with the pre-lift `.is_some()` chain + the
    /// cross-CRD coherence with `crate::crd::Process::is_being_deleted`
    /// on the tombstone axis, so a regression that drifted any surface
    /// at `tests::is_being_deleted_*` rather than as silent operator-
    /// facing skew between the pool-reconciler's `→ Drain` decision
    /// and the observed-phase composer's `→ Draining` report on the
    /// SAME `EphemeralPool` within one reconcile pass).
    pub fn is_being_deleted(&self) -> bool {
        self.metadata.deletion_timestamp.is_some()
    }

    /// Owned-form metadata-projection primitive on the `metadata.namespace`
    /// axis of `EphemeralPool`: returns an owned `String` copy of the K8s
    /// namespace with the missing-namespace corner collapsed to the load-
    /// bearing empty-string sentinel — the ONE-liner collapse of the
    /// paired `self.metadata.namespace.clone().unwrap_or_default()`
    /// incantation every pool-side consumer restated by hand pre-lift.
    ///
    /// Pre-lift the `.metadata.namespace.clone().unwrap_or_default()`
    /// chain was hand-authored at TWO sites past the ★★ PRIME-DIRECTIVE
    /// ≥ 2 duplication threshold, both stamping the `AllocationRef
    /// { namespace: String, .. }` slot inside an owned-`String` context:
    /// * `tatara-pool-reconciler::allocation_decide::AllocationConvergenceCtx
    ///   ::observe` — the matched-pool seed's `AllocationRef.namespace`
    ///   slot, right beside the peer [`Self::owned_name_or_empty`] call
    ///   that owns the paired name half. This is the exact site the
    ///   pre-existing peer-primitive doc-comment forecast (`"a future
    ///   run may lift owned_namespace_or_empty as the sibling axis
    ///   peer"`).
    /// * `crate::pool::tests::allocation_ref_new_composes_with_owned_name_or_empty_pool_projection`
    ///   — the composition pin that seeded an `AllocationRef` from the
    ///   same paired-primitive-half construction the production consumer
    ///   in `allocation_decide::observe` performs. Post-lift the pin
    ///   composes two peer primitives (`owned_name_or_empty` +
    ///   `owned_namespace_or_empty`) rather than one primitive plus the
    ///   pre-lift chain, sharpening it from a mixed-form composition
    ///   check into a paired-primitive-family composition check.
    ///
    /// Both sites walked the SAME `.clone().unwrap_or_default()` chain
    /// and both wanted the `String` form the primitive returns — as the
    /// `String` slot of an `AllocationRef` struct literal built through
    /// [`crate::pool::AllocationRef::new`]. Post-lift each callsite reads
    /// `pool.owned_namespace_or_empty()` and the produced value feeds
    /// the same downstream `AllocationRef` slot unchanged.
    ///
    /// The empty-string fallback is the SAME sentinel the sibling owned-
    /// form primitive [`Self::owned_name_or_empty`] returns on the
    /// `metadata.name` axis of the same CRD — the two primitives now
    /// partition the (owned `String` × `metadata.<slot>`) corner of the
    /// pool CRD's metadata family across BOTH object-coordinate slots
    /// on identical missing-slot semantics (empty string means "the
    /// slot is unset"), so the [`crate::pool::AllocationRef::new`]
    /// composer sees a coherent owned-empty pair regardless of which
    /// slot is absent on the source pool. Coherent with the workspace-
    /// wide owned-empty sentinel that the peer primitives
    /// [`crate::crd::Process::uid_or_empty`],
    /// [`crate::crd::Process::owned_name_or_empty`],
    /// [`Self::name_or_empty`], and [`Self::owned_name_or_empty`]
    /// already share on the metadata-slot × empty-sentinel axis.
    ///
    /// Peer to [`Self::owned_name_or_empty`] on the
    /// (`metadata.name` × `metadata.namespace`) axis of the owned-form
    /// projection family — closes the corner the pool-side family
    /// previously left open:
    ///
    /// * owned + name + empty sentinel → [`Self::owned_name_or_empty`]
    ///   (`AllocationRef.name` seed, `HashMap<String, _>` key seed);
    /// * owned + namespace + empty sentinel → **this method**
    ///   (`AllocationRef.namespace` seed — the paired half the same
    ///   `AllocationRef::new(name, namespace)` constructor consumes);
    /// * copy + deletion + tombstone probe → [`Self::is_being_deleted`]
    ///   (the presence-probe corner of the same metadata axis, already
    ///   opened).
    ///
    /// A future normalization step (a namespace-canonicalization pass,
    /// a case-fold key builder, a per-cluster prefix stripper, or the
    /// canonical-namespace default lift that would substitute
    /// [`crate::crd::Process::DEFAULT_NAMESPACE`] on the missing-slot
    /// corner rather than the empty-string sentinel) lands at ONE
    /// substrate method here and both downstream consumers pick up the
    /// upgrade mechanically — no per-callsite hand-edit at
    /// `AllocationConvergenceCtx::observe` / the composition pin.
    ///
    /// The empty-string fallback (rather than
    /// [`crate::crd::Process::DEFAULT_NAMESPACE`]) is DELIBERATELY
    /// pinned: the sole downstream consumer
    /// (`AllocationConvergenceCtx::observe`'s matched-pool seed) feeds
    /// the produced value into `AllocationRef.namespace`, which is then
    /// matched byte-identically against `spec.pool_ref.namespace` at
    /// [`crate::pool::allocation_decide::resolve_pool`]-style comparators.
    /// A silent substitution of `"default"` at this primitive would
    /// alias every namespace-absent pool to the `"default"` bucket at
    /// the matcher, hiding the missing-slot corner from an operator
    /// who explicitly authored an allocation against a namespace-
    /// unset pool. The load-bearing empty-string sentinel keeps the
    /// pre-lift `.clone().unwrap_or_default()` shape verbatim so the
    /// downstream matcher's byte-comparison stays honest.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the `.metadata.namespace.clone().unwrap_or_default()` chain
    /// recurred at two hand-authored sites past the ★★ PRIME-DIRECTIVE
    /// ≥ 2 duplication trigger, and is lifted to ONE owner here).
    /// THEORY.md §II.1 invariant 5 (composition preserves proofs —
    /// the pins bind the missing-namespace corner + the empty-string
    /// sentinel byte-shape + the owned-form `String` return type + the
    /// byte-identical parity with the pre-lift chain + the fallback-
    /// value coherence with [`Self::owned_name_or_empty`] on the
    /// paired-slot axis, so a regression that drifted any surface at
    /// `tests::owned_namespace_or_empty_*` rather than as silent
    /// operator-facing skew between the paired name / namespace halves
    /// of the SAME `AllocationRef` seed).
    pub fn owned_namespace_or_empty(&self) -> String {
        self.metadata.namespace.clone().unwrap_or_default()
    }
}

/// What the pool reconciler does when a member reaches `Failed`.
///
/// Sibling closed-set lifts on the same `tatara-process` axis:
/// [`crate::compliance::VerificationPhase::ALL`],
/// [`crate::signal::SighupStrategy::ALL`],
/// [`crate::spec::MustReachPhase::ALL`],
/// [`crate::intent::WorkloadKind::ALL`],
/// [`crate::export::ReportFormat::ALL`],
/// [`crate::encapsulates::EncapsulationMode::ALL`],
/// [`crate::export::ExportTrigger::ALL`],
/// [`crate::lifetime::TeardownPolicy::ALL`],
/// [`crate::boundary::ConditionKind::ALL`],
/// [`crate::lifetime::LifetimeKind::ALL`],
/// [`crate::intent::IntentKind::ALL`],
/// [`crate::phase::ProcessPhase::ALL`],
/// [`crate::signal::ProcessSignal::ALL`].
#[derive(
    Clone,
    Copy,
    Debug,
    Default,
    Serialize,
    Deserialize,
    JsonSchema,
    PartialEq,
    Eq,
    Hash,
    tatara_closed_set::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum ReplacementPolicy {
    /// **Default** — Failed member is reaped + replaced immediately
    /// (pool stays at `desired` count). Most production-like.
    #[default]
    ReplaceImmediate,
    /// Failed member stays for inspection; pool runs short until the
    /// operator manually reaps it. Useful for debugging.
    HoldFailed,
    /// Failed member triggers pool-wide pause: `desired` is
    /// effectively 0 until the operator manually resumes via a
    /// pool-status patch. Used for "halt on any failure" workflows.
    PausePool,
}

impl ReplacementPolicy {
    /// The closed set of replacement policies — single source of truth
    /// that drives the `as_str` / Display / `FromStr` triad and the
    /// `replaces_failed` / `pauses_on_failure` predicate pair. Adding a
    /// fourth variant lands at one `ALL` entry + one `as_str` arm + one
    /// predicate arm per projection — exhaustively checked by the
    /// compiler (the `[Self; 3]` array literal forces the arity) and by
    /// the predicate-pair injectivity test below (a new variant must
    /// land in its own (replaces_failed, pauses_on_failure) bucket or
    /// the author has to extend the consumer dispatch in
    /// `tatara-pool-reconciler::desired::PoolConvergence::decide`).
    pub const ALL: [Self; 3] = [Self::ReplaceImmediate, Self::HoldFailed, Self::PausePool];

    /// Canonical PascalCase wire-format projection — matches the serde
    /// `rename_all = "PascalCase"` output verbatim AND the CRD `enum:`
    /// enumeration the pool reconciler stamps on the
    /// `ephemeralpools.tatara.pleme.io` schema. Pinned by
    /// `replacement_policy_as_str_matches_serde` so a variant rename
    /// can't drift between the typed surface, the CRD enum, the YAML
    /// wire format AND the operator-facing diagnostic (the
    /// `desired.rs` Pause reason composes `policy={policy}` via
    /// Display, not a hard-coded `"PausePool"` literal that would
    /// silently rot).
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::ReplaceImmediate => "ReplaceImmediate",
            Self::HoldFailed => "HoldFailed",
            Self::PausePool => "PausePool",
        }
    }

    /// Should the pool auto-spawn a replacement for a Failed member?
    /// Closed-set match (not `matches!`) so a future variant triggers
    /// the compiler's exhaustiveness check at this site rather than
    /// silently defaulting to `false`. Paired with
    /// `pauses_on_failure` they form the two-axis projection
    /// consumers in `tatara-pool-reconciler::desired::PoolConvergence`
    /// pattern-match against — `replaces_failed` true ⇒ emit
    /// `ReapFailed` per failure; `pauses_on_failure` true with any
    /// failure ⇒ emit `Pause` and short-circuit. The pair is
    /// `(true, false) | (false, false) | (false, true)` — pinned
    /// injective by `replacement_policy_predicate_pair_is_injective`.
    pub const fn replaces_failed(self) -> bool {
        match self {
            Self::ReplaceImmediate => true,
            Self::HoldFailed | Self::PausePool => false,
        }
    }

    /// Should reaching Failed on any member pause the whole pool?
    /// See `replaces_failed` for the closed-match rationale + the
    /// predicate-pair contract.
    pub const fn pauses_on_failure(self) -> bool {
        match self {
            Self::PausePool => true,
            Self::ReplaceImmediate | Self::HoldFailed => false,
        }
    }
}

// `impl FromStr for ReplacementPolicy` + `impl tatara_lisp::ClosedSet for
// ReplacementPolicy` + `impl fmt::Display for ReplacementPolicy` are
// generated by `#[derive(tatara_closed_set::DeriveClosedSet)]` on the enum
// declaration above. `label` delegates to the inherent
// `ReplacementPolicy::as_str` via `#[closed_set(via = "as_str")]` so the
// PascalCase wire-format projection stays load-bearing (matches the
// serde `rename_all = "PascalCase"` output AND the
// `tatara-pool-reconciler::desired::PoolConvergence` Pause reason
// emission verbatim) while generic `T: ClosedSet` consumers reach the
// STABLE workspace-wide name (`label`); Display delegates to the same
// inherent projection via `#[closed_set(display)]` so the
// `Pause` reason emitter's `policy={policy}` composition stays
// pinned on the closed-set algebra rather than on a hand-rolled
// `fmt::Display` block per implementor.

// `pub struct UnknownReplacementPolicy(pub String)` is generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` + `#[closed_set(generate_unknown)]`
// on the enum declaration above. The auto-derived label
// `"replacement policy"` matches the prior hand-rolled
// `#[error("unknown replacement policy: {0}")]` verbatim. Symmetric to
// [`UnknownMemberState`], [`UnknownPoolPhase`], [`UnknownReturnPolicy`],
// [`crate::export::UnknownReportFormat`],
// [`crate::export::UnknownChannelKind`],
// [`crate::export::UnknownExportTrigger`],
// [`crate::lifetime::UnknownTeardownPolicy`],
// [`crate::boundary::UnknownConditionKind`], and
// [`crate::phase::UnknownPhase`].

fn default_free_ttl() -> String {
    "24h".to_string()
}
fn default_max_allocation_ttl() -> String {
    "4h".to_string()
}

/// `EphemeralPool.status` — observed pool population state.
#[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PoolStatus {
    /// Pool lifecycle phase.
    #[serde(default)]
    pub phase: PoolPhase,

    /// When the pool entered the current phase.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub phase_since: Option<DateTime<Utc>>,

    /// Number of members currently in `Free` state (ready for allocation).
    #[serde(default)]
    pub ready_count: u32,

    /// Number of members currently `Allocated`.
    #[serde(default)]
    pub allocated_count: u32,

    /// Number of members currently `Spawning` (not yet Attested).
    #[serde(default)]
    pub spawning_count: u32,

    /// Number of members currently `Returning` (reset or replace
    /// in progress).
    #[serde(default)]
    pub returning_count: u32,

    /// Member ledger — one entry per pool slot.
    #[serde(default)]
    pub members: Vec<PoolMember>,

    /// Operator-visible message (e.g., "scaled down to floor").
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,

    /// Standard Kubernetes Conditions.
    #[serde(default)]
    pub conditions: Vec<PoolCondition>,
}

/// One pool slot's state.
#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PoolMember {
    /// `metadata.name` of the backing Process.
    pub process_name: String,
    /// Pool member's current slot state.
    pub state: MemberState,
    /// When the member entered the current state.
    pub entered_state_at: DateTime<Utc>,
    /// If allocated: the AllocationRef holding this slot.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub allocation_ref: Option<AllocationRef>,
}

impl PoolStatus {
    /// Substrate constructor for the observed [`PoolStatus`] seed:
    /// composes the `(phase, phase_since, ready/allocated/spawning
    /// /returning counts, members, message, conditions)` 9-slot record
    /// every pool-reconciler status-patch site restated by hand pre-
    /// lift. The four counters ride a SINGLE closed-set-driven fold
    /// over the members list (one pass rather than four independent
    /// filter-and-count passes); the `message` + `conditions` slots
    /// stay at their invariant `None` / `vec![]` defaults every pre-
    /// lift caller stamped verbatim, and `phase_since` is derived from
    /// the caller-supplied `now` timestamp so the constructor stays
    /// clock-injectable rather than implicitly reading wall time.
    ///
    /// Pre-lift the 11-line
    /// ```rust,ignore
    /// PoolStatus {
    ///     phase,
    ///     phase_since: Some(Utc::now()),
    ///     ready_count: count_state(&members, MemberState::Free),
    ///     allocated_count: count_state(&members, MemberState::Allocated),
    ///     spawning_count: count_state(&members, MemberState::Spawning),
    ///     returning_count: count_state(&members, MemberState::Returning),
    ///     members: members.clone(),
    ///     message: None,
    ///     conditions: vec![],
    /// }
    /// ```
    /// incantation was hand-authored at TWO sites past the ★★ PRIME-
    /// DIRECTIVE ≥ 2 duplication threshold in
    /// `tatara-pool-reconciler::controller_pool::reconcile_inner`,
    /// both restating the same 4-slot count fanout + defaults:
    /// * The `desired > 0` path — status patch after the
    ///   convergence-action loop when the operator drives the pool
    ///   through the R11 desired-count invariant.
    /// * The legacy allocation-driven path (`desired == 0`) — status
    ///   patch after the [`crate::pool::PoolDecision`] apply loop.
    ///
    /// Both sites walked the SAME 4-slot count fanout on the SAME
    /// four `MemberState` variants (Free/Allocated/Spawning/Returning)
    /// and stamped the SAME defaults (`message: None`, `conditions:
    /// vec![]`), even though the four counters walked the members list
    /// four independent times pre-lift when a single pass suffices.
    /// Post-lift both callers write
    /// `PoolStatus::observed(phase, members, Utc::now())` and share
    /// ONE substrate owner; a future counter slot (e.g., a
    /// `warming_count` for a `MemberState::Warming` variant between
    /// Spawning and Free) plugs into the fold at ONE match arm and
    /// both status-patch sites inherit the new slot mechanically.
    ///
    /// The `Failed` variant is deliberately absent from the fold — no
    /// `PoolStatus` slot counts failed members (they surface via
    /// `pool_phase_from_members`'s `PoolPhase::Degraded` transition
    /// instead), and the closed-set match on
    /// [`MemberState`] pins that a future variant which SHOULD count
    /// toward one of the four buckets triggers the compiler's
    /// exhaustiveness check at this fold rather than silently sinking
    /// into `Failed`'s no-op arm.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the 11-line status-seed incantation recurred at two hand-
    /// authored sites past the ★★ PRIME-DIRECTIVE ≥ 2 duplication
    /// trigger, and is lifted to ONE owner here). THEORY.md §II.1
    /// invariant 5 (composition preserves proofs — the pins bind the
    /// 4-slot count fanout + the closed-set exhaustiveness on
    /// `MemberState` + the invariant defaults, so a regression that
    /// dropped a counter slot or swapped a variant surfaces at
    /// `tests::pool_status_observed_*` rather than as silent operator-
    /// facing skew between the two status-patch sites on the SAME
    /// pool).
    #[must_use]
    pub fn observed(phase: PoolPhase, members: Vec<PoolMember>, now: DateTime<Utc>) -> Self {
        let (ready_count, allocated_count, spawning_count, returning_count) =
            PoolMember::state_count_fanout(&members);
        Self {
            phase,
            phase_since: Some(now),
            ready_count,
            allocated_count,
            spawning_count,
            returning_count,
            members,
            message: None,
            conditions: vec![],
        }
    }
}

impl PoolMember {
    /// Substrate primitive: single-pass closed-set fold over a
    /// `[PoolMember]` slice producing the `(ready, allocated,
    /// spawning, returning)` 4-tuple every `PoolStatus` seed stamps at
    /// its four counter slots. The `Failed` arm is a no-op (no
    /// `PoolStatus` counter tracks failed members — they surface via
    /// [`PoolPhase::Degraded`] instead), pinned by the closed-set
    /// match so a future variant that SHOULD count toward one of the
    /// four buckets triggers the compiler's exhaustiveness check here
    /// rather than silently falling through.
    ///
    /// Consumed by [`PoolStatus::observed`]. A caller that needs a
    /// single per-variant count outside the status-seed fanout should
    /// keep spelling `members.iter().filter(...).count()` rather than
    /// walking this 4-tuple — the fanout is shaped for the
    /// `PoolStatus` fill, not for arbitrary per-variant queries.
    #[must_use]
    pub fn state_count_fanout(members: &[Self]) -> (u32, u32, u32, u32) {
        let mut ready = 0u32;
        let mut allocated = 0u32;
        let mut spawning = 0u32;
        let mut returning = 0u32;
        for m in members {
            match m.state {
                MemberState::Free => ready += 1,
                MemberState::Allocated => allocated += 1,
                MemberState::Spawning => spawning += 1,
                MemberState::Returning => returning += 1,
                MemberState::Failed => {}
            }
        }
        (ready, allocated, spawning, returning)
    }
}

/// Light reference to an `EphemeralAllocation`.
#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AllocationRef {
    pub name: String,
    pub namespace: String,
}

impl AllocationRef {
    /// Substrate constructor for [`AllocationRef`]: composes the
    /// `(name, namespace)` pair through ONE `impl Into<String>`-gated
    /// entry point — the ONE-liner collapse of the paired
    /// `AllocationRef { name: n.into(), namespace: ns.into() }`
    /// struct-literal incantation every downstream consumer restated
    /// by hand pre-lift.
    ///
    /// Pre-lift the `AllocationRef { name, namespace }` struct-literal
    /// was hand-authored at FOUR production sites past the ★★ PRIME-
    /// DIRECTIVE ≥ 2 duplication threshold across the workspace, all
    /// composing an owned `(name: String, namespace: String)` pair
    /// under one of two roles:
    /// * `tatara-pool-reconciler::controller_allocation::reconcile_inner`
    ///   Bind path — the `assignedProcess` status slot's ref, pairing
    ///   the just-bound member Process name with the allocation's
    ///   containing namespace.
    /// * `tatara-pool-reconciler::controller_allocation::reconcile_inner`
    ///   Release path — the same `assignedProcess` slot shape, stamped
    ///   at the release-side status patch alongside the (unchanged)
    ///   `boundPool` ref.
    /// * `tatara-pool-reconciler::allocation_decide::AllocationConvergenceCtx::observe`
    ///   pool-matched handle — the `matched_pool` slot's ref, pairing
    ///   [`EphemeralPool::owned_name_or_empty`] with the pool's
    ///   containing namespace.
    /// * `tatara-github-watcher::allocation_factory::allocation_from_pr`
    ///   — the `pool_ref` slot on the `AllocationSpec` emitted from a
    ///   PullRequestEvent, pairing the operator-configured pool name
    ///   with the watcher's target namespace.
    ///
    /// All FOUR sites walked the SAME two-field struct-literal shape
    /// — an owned name half, an owned namespace half — differing only
    /// in provenance. Post-lift each callsite reads
    /// `AllocationRef::new(name, ns)` and the produced value feeds the
    /// same downstream slot (`assignedProcess` / `bound_pool` /
    /// `matched_pool` / `spec.pool_ref`) unchanged. The `impl Into<String>`
    /// signature accepts every provenance the pre-lift sites carried —
    /// owned `String` (the reconciler's owned-form projections), `&str`
    /// (the factory's `n.to_string()` / `namespace.to_string()`
    /// borrow-to-owned promotions), `Cow<str>`, and every other
    /// `Into<String>` implementor — so no callsite has to change its
    /// upstream provenance to route through the primitive.
    ///
    /// Return-form axis: owned [`AllocationRef`] — the wire-format
    /// shape [`crate::pool::AllocationRef`]'s serde `rename_all =
    /// "camelCase"` produces on both spec (`poolRef`) and status
    /// (`boundPool` / `assignedProcess`) slots. The primitive owns
    /// the axis-order `(name, namespace)` — the same order the four
    /// consumers spelled — so a slot swap surfaces at the
    /// `allocation_ref_new_positional_axis_order` pin below rather
    /// than as silent `<namespace>/<name>` inversion downstream.
    ///
    /// Peer to the sibling substrate primitives already opened on the
    /// pool-side (name, namespace) axis pair:
    /// [`EphemeralPool::name_or_empty`] (borrow-form name),
    /// [`EphemeralPool::owned_name_or_empty`] (owned-form name); this
    /// constructor is the composer that folds the owned-form projections
    /// into the wire-format ref shape.
    ///
    /// A future refactor of [`AllocationRef`]'s field set (a
    /// `resource_kind: String` field for cross-CRD refs, an
    /// `api_version: String` field for FQN references, a
    /// canonicalization pass over the namespace half, a non-empty-name
    /// gate) lands at ONE substrate constructor site here and every
    /// downstream consumer inherits the upgrade mechanically — no per-
    /// callsite hand-edit at the FOUR reconciler + factory sites.
    ///
    /// Theory anchor: THEORY.md §VI.1 (generation over composition —
    /// the `AllocationRef { name, namespace }` struct-literal shape
    /// recurred at four hand-authored sites past the ★★ PRIME-
    /// DIRECTIVE ≥ 2 duplication trigger, and is lifted to ONE owner
    /// here). THEORY.md §II.1 invariant 5 (composition preserves
    /// proofs — the pins bind the positional axis-order + the
    /// `Into<String>` provenance closure + byte-identical parity with
    /// the pre-lift struct-literal + `PartialEq` coherence with the
    /// hand-authored form, so a regression that reshaped any surface
    /// at `tests::allocation_ref_new_*` rather than as silent
    /// operator-facing skew between the assignedProcess / bound_pool
    /// / matched_pool / spec.pool_ref slots on the SAME allocation).
    #[must_use]
    pub fn new(name: impl Into<String>, namespace: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            namespace: namespace.into(),
        }
    }
}

/// Per-slot state in the pool's free list.
///
/// Sibling closed-sets on the `EphemeralPool` axis: [`ReplacementPolicy::ALL`]
/// (the on-failure policy that the pool reconciler dispatches against
/// the [`Self::is_failed`] projection), [`ReturnPolicy::ALL`] (the
/// release-time disposition that transitions an [`Self::Allocated`]
/// member into [`Self::Returning`] before it either re-enters
/// [`Self::Free`] or gets [`Self::Spawning`]'d as a fresh slot).
#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    Hash,
    Serialize,
    Deserialize,
    JsonSchema,
    tatara_closed_set::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum MemberState {
    /// Pool reconciler is creating/converging the backing Process.
    Spawning,
    /// Process is `Attested`; ready for allocation.
    Free,
    /// Held by an `EphemeralAllocation`.
    Allocated,
    /// Return policy is being applied (Reset → reset Job; Replace →
    /// Process is being torn down and recreated).
    Returning,
    /// Permanent failure — the member needs operator attention.
    Failed,
}

impl MemberState {
    /// The closed set of member states — single source of truth that
    /// drives the `as_str` / Display / `FromStr` triad AND the
    /// `is_failed` / `counts_toward_supply` predicate pair. Adding a
    /// sixth variant lands at one `ALL` entry + one `as_str` arm + one
    /// arm per predicate — exhaustively checked by the compiler (the
    /// `[Self; 5]` array literal forces the arity) and by the
    /// per-variant truth-table contract test (a new variant must
    /// declare its own `(is_failed, counts_toward_supply)` projection
    /// or the consumer dispatch in
    /// `tatara-pool-reconciler::controller_pool::pool_phase_from_members`
    /// and `tatara-pool-reconciler::pool_decide::decide_pool_reconcile`
    /// will silently bucket it into the wrong lifecycle column).
    pub const ALL: [Self; 5] = [
        Self::Spawning,
        Self::Free,
        Self::Allocated,
        Self::Returning,
        Self::Failed,
    ];

    /// Canonical PascalCase wire-format projection — matches the serde
    /// `rename_all = "PascalCase"` output verbatim AND the CRD `enum:`
    /// enumeration that `ephemeralpools.tatara.pleme.io` stamps on
    /// `status.members[].state`. Pinned by
    /// `member_state_as_str_matches_serde` so a variant rename can't
    /// drift between the typed surface, the CRD enum, the YAML wire
    /// format AND any future operator-facing diagnostic that composes
    /// `state={state}` via Display rather than a hard-coded literal
    /// that would silently rot.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Spawning => "Spawning",
            Self::Free => "Free",
            Self::Allocated => "Allocated",
            Self::Returning => "Returning",
            Self::Failed => "Failed",
        }
    }

    /// Is this member in a permanent-failure state — needs operator
    /// attention? Closed-set match (not `matches!`) so a future variant
    /// triggers the compiler's exhaustiveness check at this site rather
    /// than silently defaulting to `false`. Consumed by
    /// `tatara-pool-reconciler::pool_decide::decide_pool_reconcile` to
    /// gate the highest-priority `ReplaceMembers` decision branch — a
    /// future variant that should also trigger replacement (e.g.
    /// `MemberState::Quarantined`) flips this predicate at one site
    /// and inherits the priority-1 dispatch without touching the
    /// consumer match arm.
    pub const fn is_failed(self) -> bool {
        match self {
            Self::Failed => true,
            Self::Spawning | Self::Free | Self::Allocated | Self::Returning => false,
        }
    }

    /// Does this member contribute to the pool's *available supply*
    /// (current ready slots + slots coming online)? Closed-set match so
    /// a future variant triggers the compiler's exhaustiveness check.
    /// Consumed by
    /// `tatara-pool-reconciler::controller_pool::pool_phase_from_members`
    /// — the `(free + spawning)` supply calc collapses into one
    /// predicate-driven filter, so a future "warming-up" state
    /// (`MemberState::Warming` between Spawning and Free) plugs into
    /// the supply count at one site rather than three. Disjoint with
    /// `is_failed` — pinned by `member_state_failed_implies_no_supply`
    /// (a Failed member can never count toward supply; the pool
    /// reconciler would otherwise double-count failures as available
    /// capacity).
    pub const fn counts_toward_supply(self) -> bool {
        match self {
            Self::Free | Self::Spawning => true,
            Self::Allocated | Self::Returning | Self::Failed => false,
        }
    }
}

// `impl FromStr for MemberState` + `impl tatara_lisp::ClosedSet for
// MemberState` + `impl fmt::Display for MemberState` are generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` on the enum declaration
// above. `label` delegates to the inherent `MemberState::as_str` via
// `#[closed_set(via = "as_str")]` so the
// `pool_phase_from_members` supply calc can keep keying on
// `counts_toward_supply` against the typed variant while a generic
// `T: ClosedSet` consumer reaches the STABLE workspace-wide name
// (`label`) without knowing this enum lives in `tatara-process::pool`;
// Display delegates to the same inherent projection via
// `#[closed_set(display)]` so the diagnostic emitter's
// `state={state}` composition stays pinned on the closed-set algebra.

// `pub struct UnknownMemberState(pub String)` is generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` + `#[closed_set(generate_unknown)]`
// on the enum declaration above. The auto-derived label `"member state"`
// matches the prior hand-rolled `#[error("unknown member state: {0}")]`
// verbatim. Symmetric to [`UnknownReplacementPolicy`],
// [`UnknownPoolPhase`], [`UnknownReturnPolicy`],
// [`crate::lifetime::UnknownTeardownPolicy`],
// [`crate::boundary::UnknownConditionKind`], and
// [`crate::phase::UnknownPhase`].

/// Pool lifecycle phase (observed across the whole pool population).
///
/// Sibling closed-set on the same `EphemeralPool` axis as
/// [`MemberState::ALL`] (the per-slot lifecycle this phase aggregates
/// over via [`MemberState::counts_toward_supply`]),
/// [`ReplacementPolicy::ALL`] (on-failure policy) and
/// [`ReturnPolicy::ALL`] (release-time disposition). Together with
/// `MemberState`, this closes the pool reconciler's
/// `(slot-state, pool-phase)` two-tier observation algebra on the
/// same closed-set discipline as the rest of `tatara-process`.
#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    Hash,
    Serialize,
    Deserialize,
    JsonSchema,
    tatara_closed_set::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum PoolPhase {
    /// Just admitted; no members yet.
    Initializing,
    /// `ready_count == desired_size`.
    Steady,
    /// `ready_count + spawning_count < desired_size` and reconciler
    /// is creating new members.
    ScalingUp,
    /// `ready_count > desired_size` and reconciler is reaping excess.
    ScalingDown,
    /// `min_size` constraint violated.
    Degraded,
    /// Pool is being deleted; reconciler is reaping all members.
    Draining,
}

impl Default for PoolPhase {
    fn default() -> Self {
        Self::Initializing
    }
}

impl PoolPhase {
    /// The closed set of pool phases — single source of truth that
    /// drives the `as_str` / Display / `FromStr` triad AND the
    /// `is_steady` / `is_terminal` predicate pair. Adding a seventh
    /// variant lands at one `ALL` entry + one `as_str` arm + one arm
    /// per predicate — exhaustively checked by the compiler (the
    /// `[Self; 6]` array literal forces the arity) AND by the
    /// per-variant truth-table contract test (a new variant must
    /// declare its own `(is_steady, is_terminal)` projection or any
    /// future status-aggregator surface — `feira pool list
    /// --healthy`, the operator-facing condition aggregator, the
    /// desired-loop heartbeat short-circuit — will silently bucket
    /// it into the wrong lifecycle column).
    pub const ALL: [Self; 6] = [
        Self::Initializing,
        Self::Steady,
        Self::ScalingUp,
        Self::ScalingDown,
        Self::Degraded,
        Self::Draining,
    ];

    /// Canonical PascalCase wire-format projection — matches the
    /// serde `rename_all = "PascalCase"` output verbatim AND the CRD
    /// `enum:` enumeration that `ephemeralpools.tatara.pleme.io`
    /// stamps on `status.phase`. Pinned by
    /// `pool_phase_as_str_matches_serde` so a variant rename can't
    /// drift between the typed surface, the CRD enum, the YAML wire
    /// format AND any future operator-facing diagnostic that
    /// composes `phase={phase}` via Display rather than a hard-coded
    /// literal that would silently rot. Display + FromStr triad
    /// over `ALL` mirrors `MemberState` / `ReplacementPolicy` /
    /// `ReturnPolicy` / `AllocationPhase` / `TeardownPolicy` /
    /// `ConditionKind` / `ProcessPhase` / `ProcessSignal`.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Initializing => "Initializing",
            Self::Steady => "Steady",
            Self::ScalingUp => "ScalingUp",
            Self::ScalingDown => "ScalingDown",
            Self::Degraded => "Degraded",
            Self::Draining => "Draining",
        }
    }

    /// Is the pool fully converged — supply matches desired, no
    /// reconciler-driven population change pending? Closed-set match
    /// (not `matches!`) so a future variant triggers the compiler's
    /// exhaustiveness check at this site rather than silently
    /// defaulting to `false`. Paired with `is_terminal` they form
    /// the two-axis projection that future status aggregators
    /// (operator-facing fleet health, `feira pool list --healthy`,
    /// the SSE filter "show non-steady pools") dispatch against —
    /// `is_steady && !is_terminal` ⇒ converged (goal state);
    /// `!is_steady && is_terminal` ⇒ being deleted (no future
    /// spawn); `!is_steady && !is_terminal` ⇒ transient
    /// (Initializing | ScalingUp | ScalingDown | Degraded — pool
    /// is in motion toward desired). The impossible bucket
    /// `(true, true)` — a draining pool that's somehow also steady
    /// — is pinned empty by `pool_phase_steady_excludes_terminal`.
    pub const fn is_steady(self) -> bool {
        match self {
            Self::Steady => true,
            Self::Initializing
            | Self::ScalingUp
            | Self::ScalingDown
            | Self::Degraded
            | Self::Draining => false,
        }
    }

    /// Is the pool in its absorbing exit state — deletion-stamped,
    /// reconciler is reaping every member, no spawn will ever
    /// happen again? Closed-set match so a future variant triggers
    /// the compiler's exhaustiveness check. See `is_steady` for the
    /// predicate-pair contract + bucket definitions.
    pub const fn is_terminal(self) -> bool {
        match self {
            Self::Draining => true,
            Self::Initializing
            | Self::Steady
            | Self::ScalingUp
            | Self::ScalingDown
            | Self::Degraded => false,
        }
    }
}

// `impl FromStr for PoolPhase` + `impl tatara_lisp::ClosedSet for PoolPhase`
// + `impl fmt::Display for PoolPhase` are generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` on the enum declaration above.
// `label` delegates to the inherent `PoolPhase::as_str` via
// `#[closed_set(via = "as_str")]` so the operator-facing
// `phase={phase}` Display composition keeps reading the same canonical
// PascalCase projection while a generic `T: ClosedSet` consumer (a
// status-aggregator filter, the `feira pool list --healthy` predicate, a
// future SSE event router) can walk every variant without knowing the
// closed set lives in `tatara-process::pool`; Display delegates to the
// same inherent projection via `#[closed_set(display)]` so the
// `phase={phase}` composition stays pinned on the closed-set algebra
// rather than a hand-rolled `fmt::Display` block.

// `pub struct UnknownPoolPhase(pub String)` is generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` + `#[closed_set(generate_unknown)]`
// on the enum declaration above. The auto-derived label `"pool phase"`
// matches the prior hand-rolled `#[error("unknown pool phase: {0}")]`
// verbatim. Symmetric to [`UnknownMemberState`],
// [`UnknownReplacementPolicy`], [`UnknownReturnPolicy`],
// [`crate::lifetime::UnknownTeardownPolicy`],
// [`crate::boundary::UnknownConditionKind`], and
// [`crate::phase::UnknownPhase`].

/// Standard K8s Condition shape (kept local so tatara-process doesn't
/// depend on k8s_openapi types in its public schema).
#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PoolCondition {
    pub type_: String,
    pub status: String,
    pub reason: String,
    pub message: String,
    pub last_transition_time: DateTime<Utc>,
}

/// What the pool does when an allocation releases a member.
///
/// Sibling closed-set on the `EphemeralPool` axis:
/// [`ReplacementPolicy::ALL`]. Sibling closed-sets on the
/// `tatara-process` algebra: [`crate::lifetime::TeardownPolicy::ALL`]
/// (the *release*-time counterpart for non-pooled ephemeral envs),
/// [`crate::boundary::ConditionKind::ALL`],
/// [`crate::lifetime::LifetimeKind::ALL`],
/// [`crate::intent::IntentKind::ALL`],
/// [`crate::phase::ProcessPhase::ALL`],
/// [`crate::signal::ProcessSignal::ALL`].
#[derive(
    Clone,
    Copy,
    Debug,
    Hash,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    JsonSchema,
    Default,
    tatara_closed_set::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum ReturnPolicy {
    /// Tear down the Process + create a fresh one. Safe but slow
    /// (1-2 min spin-up before the slot is Free again).
    #[default]
    Replace,
    /// Keep the Process running; run a typed `:reset` Job that wipes
    /// state (DB drop, secrets rotate). Fast (~5-10s) but depends on
    /// the reset Job being correct for the workload. API-authoritative
    /// systems are natural fits because the control API owns all state.
    Reset,
    /// Keep the Process indefinitely after release (debugging aid;
    /// operator must `feira pool reap NAME` to clean up). Useful for
    /// post-mortem of a flaky test.
    Keep,
}

impl ReturnPolicy {
    /// The closed set of return policies — single source of truth that
    /// drives the `as_str` / Display / `FromStr` triad and the
    /// `keeps_process` / `runs_reset_job` predicate pair. Adding a
    /// fourth variant lands at one `ALL` entry + one `as_str` arm +
    /// one arm per predicate — exhaustively checked by the compiler
    /// (the `[Self; 3]` array literal forces the arity) and by the
    /// predicate-pair injectivity test (a new variant must land in
    /// its own (keeps_process, runs_reset_job) bucket or the author
    /// has to extend the consumer dispatch in
    /// `tatara-pool-reconciler::return_policy::plan_return`).
    pub const ALL: [Self; 3] = [Self::Replace, Self::Reset, Self::Keep];

    /// Canonical PascalCase wire-format projection — matches the
    /// serde `rename_all = "PascalCase"` output verbatim AND the CRD
    /// `enum:` enumeration the pool reconciler stamps on the
    /// `ephemeralpools.tatara.pleme.io` schema. Pinned by
    /// `return_policy_as_str_matches_serde` so a variant rename can't
    /// drift between the typed surface, the CRD enum, the YAML wire
    /// format AND any future operator-facing diagnostic that composes
    /// `policy={policy}` via Display rather than a hard-coded literal.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Replace => "Replace",
            Self::Reset => "Reset",
            Self::Keep => "Keep",
        }
    }

    /// Does the pool keep the backing Process alive across release?
    /// Closed-set match (not `matches!`) so a future variant triggers
    /// the compiler's exhaustiveness check at this site rather than
    /// silently defaulting to `false`. Paired with `runs_reset_job`
    /// they form the two-axis projection that the consumer in
    /// `tatara-pool-reconciler::return_policy::plan_return` matches
    /// against — `keeps_process` false ⇒ `DeleteAndRespawn`;
    /// `keeps_process && runs_reset_job` ⇒ `ResetThenFree`;
    /// `keeps_process && !runs_reset_job` ⇒ `KeepForInspection`. The
    /// pair is `(false, false) | (true, true) | (true, false)` —
    /// pinned injective by
    /// `return_policy_predicate_pair_is_injective`.
    pub const fn keeps_process(self) -> bool {
        match self {
            Self::Replace => false,
            Self::Reset | Self::Keep => true,
        }
    }

    /// Does the policy run a typed `:reset` Job to wipe state in
    /// place? See `keeps_process` for the closed-match rationale +
    /// the predicate-pair contract.
    pub const fn runs_reset_job(self) -> bool {
        match self {
            Self::Reset => true,
            Self::Replace | Self::Keep => false,
        }
    }
}

// `impl FromStr for ReturnPolicy` + `impl tatara_lisp::ClosedSet for
// ReturnPolicy` + `impl fmt::Display for ReturnPolicy` are generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` on the enum declaration
// above. `label` delegates to the inherent `ReturnPolicy::as_str` via
// `#[closed_set(via = "as_str")]` so the
// `tatara-pool-reconciler::return_policy::plan_return` dispatch keeps
// reading the canonical PascalCase projection that matches the CRD
// `enum:` literal verbatim, while a generic `T: ClosedSet` consumer
// plugs in without knowing the enum lives in `tatara-process::pool`;
// Display delegates to the same inherent projection via
// `#[closed_set(display)]` so the `policy={policy}` diagnostic
// composition stays pinned on the closed-set algebra.

// `pub struct UnknownReturnPolicy(pub String)` is generated by
// `#[derive(tatara_closed_set::DeriveClosedSet)]` + `#[closed_set(generate_unknown)]`
// on the enum declaration above. The auto-derived label `"return policy"`
// matches the prior hand-rolled `#[error("unknown return policy: {0}")]`
// verbatim. Symmetric to [`UnknownReplacementPolicy`],
// [`UnknownMemberState`], [`UnknownPoolPhase`],
// [`crate::lifetime::UnknownTeardownPolicy`],
// [`crate::boundary::UnknownConditionKind`], and
// [`crate::phase::UnknownPhase`].

/// Routing selector — matches an `EphemeralAllocation`'s requestor
/// against pool-eligibility predicates.
#[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PoolSelector {
    /// Glob-matched against `EphemeralAllocation.spec.requestor.repo`.
    /// Empty = match every repo.
    #[serde(default)]
    pub repos: Vec<String>,

    /// Glob-matched against `EphemeralAllocation.spec.requestor.branch`.
    /// Empty = match every branch.
    #[serde(default)]
    pub branches: Vec<String>,

    /// PR labels (all-must-match, AND semantics). Empty = no label
    /// requirement.
    #[serde(default)]
    pub pr_labels: Vec<String>,

    /// Allocation `kind` strings this pool can serve (e.g., "github-pr",
    /// "manual", "ci-run"). Empty = any kind.
    #[serde(default)]
    pub kinds: Vec<String>,
}

impl PoolSelector {
    /// Does this selector match the given allocation routing key?
    /// Pure: no side effects.
    pub fn matches(&self, key: &MatchKey<'_>) -> bool {
        glob_any(&self.repos, key.repo)
            && glob_any(&self.branches, key.branch)
            && labels_subset(&self.pr_labels, key.pr_labels)
            && kind_any(&self.kinds, key.kind)
    }

    /// Specificity score — higher = more specific. Used by the
    /// reconciler to break ties between selectors that all match.
    pub fn specificity(&self) -> u32 {
        let mut score = 0;
        if !self.repos.is_empty() {
            score += 8;
        }
        if !self.branches.is_empty() {
            score += 4;
        }
        score += (self.pr_labels.len() as u32) * 2;
        if !self.kinds.is_empty() {
            score += 1;
        }
        score
    }
}

/// Allocation routing key — what the reconciler matches against pool selectors.
#[derive(Clone, Copy, Debug)]
pub struct MatchKey<'a> {
    pub repo: &'a str,
    pub branch: &'a str,
    pub pr_labels: &'a [String],
    pub kind: &'a str,
}

fn glob_any(patterns: &[String], value: &str) -> bool {
    if patterns.is_empty() {
        return true;
    }
    patterns.iter().any(|p| glob_match(p, value))
}

fn kind_any(kinds: &[String], value: &str) -> bool {
    if kinds.is_empty() {
        return true;
    }
    kinds.iter().any(|k| k == value)
}

fn labels_subset(required: &[String], present: &[String]) -> bool {
    required.iter().all(|r| present.iter().any(|p| p == r))
}

/// Minimal glob: supports trailing `*` only (e.g., `"pleme-io/*"`,
/// `"release-*"`). Sufficient for repo/branch routing. Empty pattern
/// matches anything.
fn glob_match(pattern: &str, value: &str) -> bool {
    if pattern.is_empty() {
        return true;
    }
    if let Some(prefix) = pattern.strip_suffix('*') {
        value.starts_with(prefix)
    } else {
        pattern == value
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    // The closed-set tests below call `T::from_str(bad)` via the
    // derive-generated `FromStr` impls — bring the trait into scope at
    // the test module so the lib body doesn't carry an otherwise-unused
    // `use std::str::FromStr;` at the file head.
    use std::str::FromStr;

    #[test]
    fn glob_trailing_star_matches_prefix() {
        assert!(glob_match("pleme-io/*", "pleme-io/demo-app"));
        assert!(!glob_match("pleme-io/*", "drzln/dotfiles"));
        assert!(glob_match("release-*", "release-2026-05"));
        assert!(!glob_match("release-*", "main"));
        assert!(glob_match("main", "main"));
        assert!(!glob_match("main", "develop"));
    }

    #[test]
    fn empty_selector_matches_anything() {
        let s = PoolSelector::default();
        assert!(s.matches(&MatchKey {
            repo: "any/repo",
            branch: "any-branch",
            pr_labels: &[],
            kind: "any",
        }));
    }

    #[test]
    fn repo_glob_filters_match_key() {
        let s = PoolSelector {
            repos: vec!["pleme-io/demo-*".into()],
            ..Default::default()
        };
        assert!(s.matches(&MatchKey {
            repo: "pleme-io/demo-app",
            branch: "x",
            pr_labels: &[],
            kind: "y",
        }));
        assert!(!s.matches(&MatchKey {
            repo: "pleme-io/other-repo",
            branch: "x",
            pr_labels: &[],
            kind: "y",
        }));
    }

    #[test]
    fn pr_labels_require_all() {
        let s = PoolSelector {
            pr_labels: vec!["needs-ephemeral".into(), "integration".into()],
            ..Default::default()
        };
        // Both labels present → match.
        assert!(s.matches(&MatchKey {
            repo: "x",
            branch: "y",
            pr_labels: &[
                "needs-ephemeral".into(),
                "integration".into(),
                "extra".into()
            ],
            kind: "z",
        }));
        // One label missing → no match.
        assert!(!s.matches(&MatchKey {
            repo: "x",
            branch: "y",
            pr_labels: &["needs-ephemeral".into()],
            kind: "z",
        }));
    }

    #[test]
    fn specificity_ranks_more_constrained_higher() {
        let general = PoolSelector::default();
        let specific = PoolSelector {
            repos: vec!["pleme-io/*".into()],
            branches: vec!["main".into()],
            pr_labels: vec!["needs-ephemeral".into()],
            kinds: vec!["github-pr".into()],
        };
        assert!(specific.specificity() > general.specificity());
    }

    #[test]
    fn return_policy_defaults_to_replace() {
        assert_eq!(ReturnPolicy::default(), ReturnPolicy::Replace);
    }

    #[test]
    fn pool_phase_defaults_to_initializing() {
        assert_eq!(PoolPhase::default(), PoolPhase::Initializing);
    }

    // ── closed-set algebra contracts for ReplacementPolicy
    //    (ALL × as_str × FromStr × predicate-pair) ────────────────────

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

    /// CANONICAL-KEY CONTRACT: `as_str` matches serde's PascalCase
    /// output verbatim for every variant. A future variant rename (or
    /// an `as_str` arm typo) lands here at one site, instead of
    /// drifting between the typed surface, the CRD enum, and the
    /// YAML wire format.
    #[test]
    fn replacement_policy_as_str_matches_serde() {
        crate::tagged_union::assert_label_matches_serde_serialization::<ReplacementPolicy>();
    }

    /// The Display impl IS `as_str` — pinning this lets future callers
    /// reach for either projection without drift. The operator-facing
    /// "policy={policy}" diagnostic in `tatara-pool-reconciler::desired`
    /// composes through Display rather than through a hard-coded
    /// variant string.
    #[test]
    fn replacement_policy_display_matches_as_str() {
        crate::tagged_union::assert_display_matches_label::<ReplacementPolicy>();
    }

    /// `FromStr` rejects strings that aren't in the canonical
    /// projection — lowercased / typo / cross-axis-leaked — and the
    /// error echoes the input verbatim so the operator-facing
    /// diagnostic carries the offending value, not a normalized form.
    /// The empty-input arm is pinned by
    /// [`replacement_policy_is_well_formed_closed_set`] via the
    /// `tatara_lisp::ClosedSet` testkit; the cases here pin the
    /// verbatim-echo contract on the [`UnknownReplacementPolicy`]
    /// newtype, which the trait's `make_unknown` can't see.
    #[test]
    fn unknown_replacement_policy_errors() {
        for bad in [
            "replaceimmediate",
            "PAUSEPOOL",
            "Replace-Immediate",
            "hold_failed",
            "Pause",
            "Reset",
        ] {
            let err = ReplacementPolicy::from_str(bad).unwrap_err();
            assert_eq!(err.0, bad, "error payload should echo input verbatim");
        }
    }

    /// TRUTH-TABLE CONTRACT: the predicate pair agrees with the
    /// documented per-variant on-failure behavior.
    #[test]
    fn replacement_policy_predicate_truth_tables() {
        assert!(ReplacementPolicy::ReplaceImmediate.replaces_failed());
        assert!(!ReplacementPolicy::ReplaceImmediate.pauses_on_failure());

        assert!(!ReplacementPolicy::HoldFailed.replaces_failed());
        assert!(!ReplacementPolicy::HoldFailed.pauses_on_failure());

        assert!(!ReplacementPolicy::PausePool.replaces_failed());
        assert!(ReplacementPolicy::PausePool.pauses_on_failure());
    }

    /// DISJOINTNESS CONTRACT: no variant returns true from BOTH
    /// predicates simultaneously — the two on-failure actions
    /// (reap-each-failed vs pause-whole-pool) are mutually exclusive.
    /// A future `ReplacementPolicy::PauseAndReap` that returned true
    /// from both would FAIL here, forcing the author to either pick
    /// one bucket or extend the consumer dispatch site in
    /// `tatara-pool-reconciler::desired::PoolConvergence::decide`
    /// deliberately rather than silently double-firing both branches.
    #[test]
    fn replacement_policy_predicates_are_disjoint() {
        for policy in ReplacementPolicy::ALL {
            assert!(
                !(policy.replaces_failed() && policy.pauses_on_failure()),
                "{policy:?} returns true from both replaces_failed and pauses_on_failure",
            );
        }
    }

    /// INJECTIVITY CONTRACT: the pair `(replaces_failed,
    /// pauses_on_failure)` is injective across `ALL`. Each variant
    /// projects to its own `(bool, bool)` bucket: `(true, false)` =
    /// reap; `(false, false)` = hold; `(false, true)` = pause. Pairing
    /// this with the disjointness contract above forces a future
    /// variant to land in a fresh `(replaces_failed,
    /// pauses_on_failure)` bucket — or the author extends the consumer
    /// dispatch in `tatara-pool-reconciler::desired::PoolConvergence`
    /// to recognize the new projection bucket.
    #[test]
    fn replacement_policy_predicate_pair_is_injective() {
        let projections: Vec<(bool, bool)> = ReplacementPolicy::ALL
            .into_iter()
            .map(|p| (p.replaces_failed(), p.pauses_on_failure()))
            .collect();
        let unique: std::collections::HashSet<_> = projections.iter().copied().collect();
        assert_eq!(
            projections.len(),
            unique.len(),
            "predicate pair projection is not injective: {projections:?}",
        );
    }

    /// DEFAULT-AGREEMENT CONTRACT: `ReplacementPolicy::default()`
    /// returns the variant tagged `#[default]` in the enum, AND that
    /// variant reaps (the production-safe behavior). A future #[default]
    /// rename without flipping the predicates fails here.
    #[test]
    fn replacement_policy_default_replaces_failed() {
        let d = ReplacementPolicy::default();
        assert_eq!(d, ReplacementPolicy::ReplaceImmediate);
        assert!(d.replaces_failed());
        assert!(!d.pauses_on_failure());
    }

    #[test]
    fn kinds_filter_to_known_set() {
        let s = PoolSelector {
            kinds: vec!["github-pr".into(), "manual".into()],
            ..Default::default()
        };
        assert!(s.matches(&MatchKey {
            repo: "x",
            branch: "y",
            pr_labels: &[],
            kind: "github-pr",
        }));
        assert!(!s.matches(&MatchKey {
            repo: "x",
            branch: "y",
            pr_labels: &[],
            kind: "scheduled",
        }));
    }

    // ── closed-set algebra contracts for ReturnPolicy
    //    (ALL × as_str × FromStr × predicate-pair) ────────────────────

    /// Structural well-formedness of [`ReturnPolicy`] as a
    /// [`tatara_lisp::ClosedSet`] implementor — testkit lift
    /// symmetric to [`replacement_policy_is_well_formed_closed_set`]
    /// above.
    #[test]
    fn return_policy_is_well_formed_closed_set() {
        tatara_closed_set::assert_closed_set_well_formed::<ReturnPolicy>();
    }

    /// CANONICAL-KEY CONTRACT: `as_str` matches serde's PascalCase
    /// output verbatim for every variant. A future variant rename (or
    /// an `as_str` arm typo) lands here at one site, instead of
    /// drifting between the typed surface, the CRD enum, and the
    /// YAML wire format.
    #[test]
    fn return_policy_as_str_matches_serde() {
        crate::tagged_union::assert_label_matches_serde_serialization::<ReturnPolicy>();
    }

    /// The Display impl IS `as_str` — pinning this lets future callers
    /// reach for either projection without drift, mirroring the
    /// `ReplacementPolicy` discipline.
    #[test]
    fn return_policy_display_matches_as_str() {
        crate::tagged_union::assert_display_matches_label::<ReturnPolicy>();
    }

    /// `FromStr` rejects strings that aren't in the canonical
    /// projection — lowercased / typo / cross-axis-leaked — and the
    /// error echoes the input verbatim so the operator-facing
    /// diagnostic carries the offending value, not a normalized form.
    /// The empty-input arm is pinned by
    /// [`return_policy_is_well_formed_closed_set`] via the
    /// `tatara_lisp::ClosedSet` testkit.
    #[test]
    fn unknown_return_policy_errors() {
        for bad in [
            "replace",
            "RESET",
            "Re-place",
            "keep_for_inspection",
            "DeleteAndRespawn",
            "ReplaceImmediate",
        ] {
            let err = ReturnPolicy::from_str(bad).unwrap_err();
            assert_eq!(err.0, bad, "error payload should echo input verbatim");
        }
    }

    /// TRUTH-TABLE CONTRACT: the predicate pair agrees with the
    /// documented per-variant on-release behavior.
    #[test]
    fn return_policy_predicate_truth_tables() {
        assert!(!ReturnPolicy::Replace.keeps_process());
        assert!(!ReturnPolicy::Replace.runs_reset_job());

        assert!(ReturnPolicy::Reset.keeps_process());
        assert!(ReturnPolicy::Reset.runs_reset_job());

        assert!(ReturnPolicy::Keep.keeps_process());
        assert!(!ReturnPolicy::Keep.runs_reset_job());
    }

    /// IMPLICATION CONTRACT: `runs_reset_job` implies `keeps_process`.
    /// You cannot run a typed `:reset` Job against a Process you've
    /// just deleted; the impossible bucket `(false, true)` must stay
    /// empty. A future variant returning true from `runs_reset_job`
    /// while returning false from `keeps_process` fails here, which
    /// forces the author to either flip `keeps_process` to true or
    /// extend the consumer dispatch site in
    /// `tatara-pool-reconciler::return_policy::plan_return`
    /// deliberately rather than letting an impossible state slip in.
    #[test]
    fn return_policy_reset_implies_keeps_process() {
        for policy in ReturnPolicy::ALL {
            if policy.runs_reset_job() {
                assert!(
                    policy.keeps_process(),
                    "{policy:?} runs a reset job but does not keep the process",
                );
            }
        }
    }

    /// INJECTIVITY CONTRACT: the pair `(keeps_process, runs_reset_job)`
    /// is injective across `ALL`. Each variant projects to its own
    /// `(bool, bool)` bucket: `(false, false)` = delete + respawn;
    /// `(true, true)` = reset-in-place; `(true, false)` = keep for
    /// inspection. Pairing this with the implication contract above
    /// forces a future variant to land in a fresh
    /// `(keeps_process, runs_reset_job)` bucket — or the author
    /// extends the consumer dispatch in
    /// `tatara-pool-reconciler::return_policy::plan_return` to
    /// recognize the new projection bucket.
    #[test]
    fn return_policy_predicate_pair_is_injective() {
        let projections: Vec<(bool, bool)> = ReturnPolicy::ALL
            .into_iter()
            .map(|p| (p.keeps_process(), p.runs_reset_job()))
            .collect();
        let unique: std::collections::HashSet<_> = projections.iter().copied().collect();
        assert_eq!(
            projections.len(),
            unique.len(),
            "predicate pair projection is not injective: {projections:?}",
        );
    }

    /// DEFAULT-AGREEMENT CONTRACT: `ReturnPolicy::default()` returns
    /// the variant tagged `#[default]` in the enum, AND that variant
    /// is the safe "tear down + respawn" behavior — neither keeps the
    /// process nor runs a reset Job. A future `#[default]` rename
    /// without flipping the predicates fails here.
    #[test]
    fn return_policy_default_is_replace_and_neither_predicate_fires() {
        let d = ReturnPolicy::default();
        assert_eq!(d, ReturnPolicy::Replace);
        assert!(!d.keeps_process());
        assert!(!d.runs_reset_job());
    }

    // ── closed-set algebra contracts for MemberState
    //    (ALL × as_str × FromStr × predicate pair) ────────────────────

    /// Structural well-formedness of [`MemberState`] as a
    /// [`tatara_lisp::ClosedSet`] implementor — testkit lift
    /// symmetric to [`replacement_policy_is_well_formed_closed_set`]
    /// and [`return_policy_is_well_formed_closed_set`] above.
    #[test]
    fn member_state_is_well_formed_closed_set() {
        tatara_closed_set::assert_closed_set_well_formed::<MemberState>();
    }

    /// CANONICAL-KEY CONTRACT: `as_str` matches serde's PascalCase
    /// output verbatim for every variant. A future variant rename (or
    /// an `as_str` arm typo) lands here at one site, instead of
    /// drifting between the typed surface, the CRD enum, and the YAML
    /// wire format the pool reconciler stamps on
    /// `status.members[].state`.
    #[test]
    fn member_state_as_str_matches_serde() {
        crate::tagged_union::assert_label_matches_serde_serialization::<MemberState>();
    }

    /// The Display impl IS `as_str` — pinning this lets future callers
    /// reach for either projection without drift. Any operator-facing
    /// "state={state}" diagnostic that composes through Display
    /// inherits the canonical wire-format string automatically.
    #[test]
    fn member_state_display_matches_as_str() {
        crate::tagged_union::assert_display_matches_label::<MemberState>();
    }

    /// `FromStr` rejects strings that aren't in the canonical
    /// projection — lowercased / typo / cross-axis-leaked — and
    /// the error echoes the input verbatim so the operator-facing
    /// diagnostic carries the offending value, not a normalized form.
    /// The empty-input arm is pinned by
    /// [`member_state_is_well_formed_closed_set`] via the
    /// `tatara_lisp::ClosedSet` testkit. The cross-axis leak cases
    /// pin the closed-set REJECTION contract that the trait can't see:
    /// `"ReplaceImmediate"`, `"Reset"`, and `"Attested"` are valid
    /// labels for sibling enums (`ReplacementPolicy`, `ReturnPolicy`,
    /// `ProcessPhase`) but MUST reject here, because the codomains
    /// are disjoint.
    #[test]
    fn unknown_member_state_errors() {
        for bad in [
            "free",
            "SPAWNING",
            "Free-State",
            "allocated_now",
            "ReplaceImmediate", // ReplacementPolicy-axis leak
            "Reset",            // ReturnPolicy-axis leak
            "Attested",         // ProcessPhase-axis leak
        ] {
            let err = MemberState::from_str(bad).unwrap_err();
            assert_eq!(err.0, bad, "error payload should echo input verbatim");
        }
    }

    /// TRUTH-TABLE CONTRACT: the predicate pair agrees with the
    /// documented per-variant lifecycle role. The pool reconciler's
    /// `pool_phase_from_members` supply calc collapses
    /// `count_state(Free) + count_state(Spawning)` into one
    /// `counts_toward_supply` filter; this table pins the per-variant
    /// projection that consumer depends on.
    #[test]
    fn member_state_predicate_truth_tables() {
        assert!(!MemberState::Spawning.is_failed());
        assert!(MemberState::Spawning.counts_toward_supply());

        assert!(!MemberState::Free.is_failed());
        assert!(MemberState::Free.counts_toward_supply());

        assert!(!MemberState::Allocated.is_failed());
        assert!(!MemberState::Allocated.counts_toward_supply());

        assert!(!MemberState::Returning.is_failed());
        assert!(!MemberState::Returning.counts_toward_supply());

        assert!(MemberState::Failed.is_failed());
        assert!(!MemberState::Failed.counts_toward_supply());
    }

    /// DISJOINTNESS CONTRACT: no variant returns true from BOTH
    /// `is_failed` and `counts_toward_supply` simultaneously — a
    /// failed member can never be counted as available capacity. A
    /// future variant that returned true from both would FAIL here,
    /// forcing the author to either drop it from supply, or extend
    /// the consumer's bucketing in
    /// `tatara-pool-reconciler::controller_pool::pool_phase_from_members`
    /// deliberately rather than silently inflating the pool's supply
    /// count with failed slots.
    #[test]
    fn member_state_failed_implies_no_supply() {
        for state in MemberState::ALL {
            assert!(
                !(state.is_failed() && state.counts_toward_supply()),
                "{state:?} returns true from both is_failed and counts_toward_supply — \
                 a failed member can never be counted as available pool capacity",
            );
        }
    }

    /// COVERAGE CONTRACT: every variant lands somewhere — either
    /// in supply, or as a failed slot, or as an in-use bucket
    /// (`Allocated | Returning`). A future variant that returns
    /// `false` from `counts_toward_supply` AND `false` from
    /// `is_failed` is fine *iff* it represents an in-use slot; this
    /// test pins the existing variants in their declared buckets so
    /// the consumer-side dispatch in
    /// `tatara-pool-reconciler::pool_decide::decide_pool_reconcile`
    /// stays grounded.
    #[test]
    fn member_state_buckets_cover_every_variant() {
        let mut supply = 0u32;
        let mut failed = 0u32;
        let mut in_use = 0u32;
        for state in MemberState::ALL {
            match (state.is_failed(), state.counts_toward_supply()) {
                (true, false) => failed += 1,
                (false, true) => supply += 1,
                (false, false) => in_use += 1,
                (true, true) => panic!("disjointness already pins this empty for {state:?}"),
            }
        }
        assert_eq!(supply, 2, "supply bucket: Free + Spawning");
        assert_eq!(failed, 1, "failed bucket: Failed");
        assert_eq!(in_use, 2, "in-use bucket: Allocated + Returning");
        assert_eq!(supply + failed + in_use, MemberState::ALL.len() as u32);
    }

    // ── closed-set algebra contracts for PoolPhase
    //    (ALL × as_str × FromStr × predicate pair) ────────────────────

    /// Structural well-formedness of [`PoolPhase`] as a
    /// [`tatara_lisp::ClosedSet`] implementor — testkit lift
    /// symmetric to [`member_state_is_well_formed_closed_set`] above.
    #[test]
    fn pool_phase_is_well_formed_closed_set() {
        tatara_closed_set::assert_closed_set_well_formed::<PoolPhase>();
    }

    /// CANONICAL-KEY CONTRACT: `as_str` matches serde's PascalCase
    /// output verbatim for every variant. A future variant rename (or
    /// an `as_str` arm typo) lands here at one site, instead of
    /// drifting between the typed surface, the CRD enum, and the YAML
    /// wire format the pool reconciler stamps on `status.phase`.
    #[test]
    fn pool_phase_as_str_matches_serde() {
        crate::tagged_union::assert_label_matches_serde_serialization::<PoolPhase>();
    }

    /// The Display impl IS `as_str` — pinning this lets future callers
    /// reach for either projection without drift. Any operator-facing
    /// "phase={phase}" diagnostic that composes through Display
    /// inherits the canonical wire-format string automatically.
    #[test]
    fn pool_phase_display_matches_as_str() {
        crate::tagged_union::assert_display_matches_label::<PoolPhase>();
    }

    /// `FromStr` rejects strings that aren't in the canonical
    /// projection — lowercased / typo / cross-axis-leaked — and
    /// the error echoes the input verbatim so the operator-facing
    /// diagnostic carries the offending value, not a normalized form.
    /// The empty-input arm is pinned by
    /// [`pool_phase_is_well_formed_closed_set`] via the
    /// `tatara_lisp::ClosedSet` testkit. The cross-axis leak cases
    /// (`"Free"`, `"Replace"`, `"Attested"`, `"HoldFailed"`) pin the
    /// closed-set REJECTION contract that the trait can't see — those
    /// are valid sibling-axis labels but MUST reject here.
    #[test]
    fn unknown_pool_phase_errors() {
        for bad in [
            "steady",
            "SCALINGUP",
            "Scaling-Up",
            "scaling_down",
            "Free",       // MemberState-axis leak
            "Replace",    // ReturnPolicy-axis leak
            "Attested",   // ProcessPhase-axis leak
            "HoldFailed", // ReplacementPolicy-axis leak
        ] {
            let err = PoolPhase::from_str(bad).unwrap_err();
            assert_eq!(err.0, bad, "error payload should echo input verbatim");
        }
    }

    /// TRUTH-TABLE CONTRACT: the predicate pair agrees with the
    /// documented per-variant lifecycle role. Pinning this table at
    /// one site means any future status-aggregator surface
    /// (`feira pool list --healthy`, the SSE filter, the desired-loop
    /// heartbeat short-circuit) reads the same projection that the
    /// reconciler writes.
    #[test]
    fn pool_phase_predicate_truth_tables() {
        assert!(!PoolPhase::Initializing.is_steady());
        assert!(!PoolPhase::Initializing.is_terminal());

        assert!(PoolPhase::Steady.is_steady());
        assert!(!PoolPhase::Steady.is_terminal());

        assert!(!PoolPhase::ScalingUp.is_steady());
        assert!(!PoolPhase::ScalingUp.is_terminal());

        assert!(!PoolPhase::ScalingDown.is_steady());
        assert!(!PoolPhase::ScalingDown.is_terminal());

        assert!(!PoolPhase::Degraded.is_steady());
        assert!(!PoolPhase::Degraded.is_terminal());

        assert!(!PoolPhase::Draining.is_steady());
        assert!(PoolPhase::Draining.is_terminal());
    }

    /// DISJOINTNESS CONTRACT: no variant returns true from BOTH
    /// `is_steady` and `is_terminal` simultaneously — a draining pool
    /// is by definition transitioning OUT, not the goal converged
    /// state. A future variant that returned true from both would
    /// FAIL here, forcing the author to either pick one bucket or
    /// extend the consumer dispatch sites (status aggregators,
    /// heartbeat short-circuit) deliberately rather than silently
    /// double-firing both branches.
    #[test]
    fn pool_phase_steady_excludes_terminal() {
        for phase in PoolPhase::ALL {
            assert!(
                !(phase.is_steady() && phase.is_terminal()),
                "{phase:?} returns true from both is_steady and is_terminal — \
                 a draining pool is by definition not the converged goal state",
            );
        }
    }

    /// COVERAGE CONTRACT: every variant lands somewhere — either the
    /// converged goal (`Steady`), the absorbing exit (`Draining`),
    /// or the transient bucket (`Initializing | ScalingUp |
    /// ScalingDown | Degraded` — pool is in motion toward desired).
    /// A future variant that returns `false` from BOTH predicates is
    /// fine *iff* it represents an in-motion state; this test pins
    /// the existing variants in their declared buckets so the
    /// projection consumers stay grounded.
    #[test]
    fn pool_phase_buckets_cover_every_variant() {
        let mut converged = 0u32;
        let mut terminal = 0u32;
        let mut transient = 0u32;
        for phase in PoolPhase::ALL {
            match (phase.is_steady(), phase.is_terminal()) {
                (true, false) => converged += 1,
                (false, true) => terminal += 1,
                (false, false) => transient += 1,
                (true, true) => panic!("disjointness already pins this empty for {phase:?}"),
            }
        }
        assert_eq!(converged, 1, "converged bucket: Steady");
        assert_eq!(terminal, 1, "terminal bucket: Draining");
        assert_eq!(
            transient, 4,
            "transient bucket: Initializing + ScalingUp + ScalingDown + Degraded"
        );
        assert_eq!(
            converged + terminal + transient,
            PoolPhase::ALL.len() as u32
        );
    }

    /// DEFAULT-AGREEMENT CONTRACT: `PoolPhase::default()` returns the
    /// variant a freshly-admitted pool should land in — `Initializing`
    /// — AND that variant is neither steady (no members yet) nor
    /// terminal (not deletion-stamped). A future `Default` rename
    /// without flipping the predicates fails here.
    #[test]
    fn pool_phase_default_is_initializing_in_transient_bucket() {
        let d = PoolPhase::default();
        assert_eq!(d, PoolPhase::Initializing);
        assert!(!d.is_steady());
        assert!(!d.is_terminal());
    }

    // ─────────────────────────────────────────────────────────────────
    // `EphemeralPool::name_or_empty` — borrow-form metadata-projection
    // primitive on the `metadata.name` axis. Pins the missing-slot
    // corner, the populated-slot corner, the pre-lift chain-shape
    // parity, and the pure-projection discipline that the two
    // `tatara-pool-reconciler` consumers routed onto the primitive
    // depend on. See the primitive's doc-comment for the full
    // migration rationale.
    // ─────────────────────────────────────────────────────────────────

    fn empty_template() -> EphemeralSpec {
        EphemeralSpec {
            aplicacao: crate::intent::AplicacaoIntent {
                chart_ref: "oci://x".into(),
                version: "1".into(),
                profile: String::new(),
                values_overlay: serde_json::Value::Null,
                release_name: None,
                target_namespace: None,
                install_timeout: None,
            },
            ttl: "1h".into(),
            teardown: crate::lifetime::TeardownPolicy::Always,
            max_concurrent: 0,
            postconditions: vec![],
            preconditions: vec![],
            verify_timeout: None,
            classification: None,
            parent: None,
            exports: vec![],
            routing: None,
        }
    }

    fn pool_spec() -> PoolSpec {
        PoolSpec {
            desired_size: 1,
            min_size: 0,
            max_size: 0,
            return_policy: ReturnPolicy::Replace,
            selector: PoolSelector::default(),
            template: empty_template(),
            free_ttl: "24h".into(),
            max_allocation_ttl: "4h".into(),
            desired: 0,
            replacement_policy: ReplacementPolicy::default(),
            stable_name_claim: false,
        }
    }

    fn pool_named(name: &str) -> EphemeralPool {
        EphemeralPool::new(name, pool_spec())
    }

    fn pool_unnamed() -> EphemeralPool {
        let mut p = EphemeralPool::new("scratch", pool_spec());
        p.metadata.name = None;
        p
    }

    #[test]
    fn name_or_empty_returns_empty_string_when_metadata_name_is_none() {
        let p = pool_unnamed();
        assert!(p.metadata.name.is_none(), "fixture invariant");
        assert_eq!(p.name_or_empty(), "");
    }

    #[test]
    fn name_or_empty_returns_populated_slot_verbatim() {
        let p = pool_named("attest-pool");
        assert_eq!(p.name_or_empty(), "attest-pool");
    }

    #[test]
    fn name_or_empty_returns_empty_string_when_slot_is_explicitly_empty_string() {
        // Corner between `None` (missing slot) and `Some(String::new())`
        // (populated slot containing the empty string): the primitive
        // MUST fold both to the same `""` byte-shape so a downstream
        // `HashMap<String,_>::get(name)` / `str::cmp` sees ONE
        // "unnamed pool" bucket regardless of which shape the K8s API
        // server materialized. This is byte-identical to what the
        // pre-lift `.as_deref().unwrap_or("")` chain produced.
        let mut p = pool_named("scratch");
        p.metadata.name = Some(String::new());
        assert_eq!(p.name_or_empty(), "");
    }

    #[test]
    fn name_or_empty_is_a_pure_projection() {
        // Consecutive calls return byte-identical slices — no cached
        // state, no mutation on the `EphemeralPool` between calls.
        // Guards against a future refactor that plants a cache field
        // and drifts one caller from another silently.
        let p = pool_named("router-pool");
        assert_eq!(p.name_or_empty(), p.name_or_empty());
        assert_eq!(p.name_or_empty(), "router-pool");
        assert_eq!(p.name_or_empty(), "router-pool");
    }

    #[test]
    fn name_or_empty_matches_pre_lift_chain_verbatim() {
        // Byte-identical parity with the two hand-authored
        // `.metadata.name.as_deref().unwrap_or("")` chains the
        // primitive replaces in `tatara-pool-reconciler::router` and
        // `tatara-pool-reconciler::controller_allocation`. Runs across
        // the FULL corner set of the metadata.name slot: absent,
        // present-with-value, present-with-empty-string.
        let cases: [(Option<String>, &str); 3] = [
            (None, ""),
            (Some("attest-pool".into()), "attest-pool"),
            (Some(String::new()), ""),
        ];
        for (slot, expected) in cases {
            let mut p = pool_named("scratch");
            p.metadata.name = slot.clone();
            let pre_lift = p.metadata.name.as_deref().unwrap_or("");
            assert_eq!(pre_lift, expected, "pre-lift chain sanity");
            assert_eq!(p.name_or_empty(), pre_lift);
            assert_eq!(p.name_or_empty(), expected);
        }
    }

    #[test]
    fn name_or_empty_borrows_from_metadata_name_slot() {
        // The returned `&str` is tied to the `EphemeralPool`'s
        // lifetime — the caller can compare / hash / index without
        // allocating. This is the load-bearing property that lets
        // the `HashMap<String, _>::get(pool.name_or_empty())` closure
        // in `controller_allocation::reconcile_inner` skip cloning.
        let p = pool_named("attest-pool");
        let s: &str = p.name_or_empty();
        assert_eq!(s.as_ptr(), p.metadata.name.as_deref().unwrap().as_ptr());
    }

    // ─── EphemeralPool::owned_name_or_empty substrate pins ────────────
    //
    // The owned-form peer of the borrow-form `name_or_empty` primitive
    // above. Sibling to the sister-CRD primitive
    // `crate::crd::Process::owned_name_or_empty` (owned + empty sentinel
    // on `Process::metadata.name`) — the four primitives now partition
    // the (borrow × owned) × (name × uid) corner of the metadata-slot
    // family on identical missing-slot semantics across BOTH tatara-
    // process CRDs (`Process::uid_or_empty` + `Process::owned_name_or_empty`
    // + `EphemeralPool::name_or_empty` + this method). Fail-before-pass-
    // after granularity: `owned_name_or_empty` did not exist on the pool
    // CRD pre-lift; the compiler cannot resolve the name until the impl
    // block above is in place, so a rollback of the primitive breaks
    // this whole module.
    #[test]
    fn owned_name_or_empty_returns_empty_string_when_metadata_name_is_none() {
        let p = pool_unnamed();
        assert!(p.metadata.name.is_none(), "fixture invariant");
        assert_eq!(p.owned_name_or_empty(), String::new());
    }

    #[test]
    fn owned_name_or_empty_returns_owned_string_when_slot_is_populated() {
        let p = pool_named("attest-pool");
        assert_eq!(p.owned_name_or_empty(), "attest-pool");
    }

    #[test]
    fn owned_name_or_empty_returns_empty_string_when_slot_is_explicitly_empty_string() {
        // Corner between `None` (missing slot) and `Some(String::new())`
        // (populated slot containing the empty string): the primitive
        // MUST fold both to the same `""` byte-shape so a downstream
        // `HashMap<String,_>::get(name)` sees ONE "unnamed pool" bucket
        // regardless of which shape the K8s API server materialized.
        // Byte-identical to what the pre-lift `.clone().unwrap_or_default()`
        // chain produced.
        let mut p = pool_named("scratch");
        p.metadata.name = Some(String::new());
        assert_eq!(p.owned_name_or_empty(), String::new());
        assert!(p.owned_name_or_empty().is_empty());
    }

    #[test]
    fn owned_name_or_empty_is_a_pure_projection() {
        // Consecutive calls return byte-identical Strings — no cached
        // state, no mutation on the `EphemeralPool` between calls.
        // Guards against a future refactor that plants a cache field
        // and drifts one caller from another silently.
        let p = pool_named("router-pool");
        assert_eq!(p.owned_name_or_empty(), p.owned_name_or_empty());
        assert_eq!(p.owned_name_or_empty(), "router-pool");
        assert_eq!(p.owned_name_or_empty(), "router-pool");
    }

    #[test]
    fn owned_name_or_empty_matches_pre_lift_chain_verbatim() {
        // Byte-identical parity with the two hand-authored
        // `.metadata.name.clone().unwrap_or_default()` chains the
        // primitive replaces in `tatara-pool-reconciler::
        // controller_allocation::reconcile_inner` (HashMap key seed)
        // and `tatara-pool-reconciler::allocation_decide::
        // AllocationConvergenceCtx::observe` (AllocationRef.name slot
        // seed). Runs across the FULL corner set of the metadata.name
        // slot: absent, present-with-value, present-with-empty-string.
        // A regression that inserted a normalization step at the
        // primitive the pre-lift chain does NOT apply — or vice versa —
        // surfaces here rather than as silent drift between the two
        // owned-form callsites and the ONE substrate owner they now
        // route through.
        let cases: [(Option<String>, &str); 3] = [
            (None, ""),
            (Some("attest-pool".into()), "attest-pool"),
            (Some(String::new()), ""),
        ];
        for (slot, expected) in cases {
            let mut p = pool_named("scratch");
            p.metadata.name = slot.clone();
            let pre_lift = p.metadata.name.clone().unwrap_or_default();
            assert_eq!(pre_lift.as_str(), expected, "pre-lift chain sanity");
            assert_eq!(p.owned_name_or_empty(), pre_lift);
            assert_eq!(p.owned_name_or_empty().as_str(), expected);
        }
    }

    #[test]
    fn owned_name_or_empty_matches_borrow_form_peer_on_populated_slot() {
        // Cross-primitive coherence pin at the sibling corner: when the
        // slot is present, the borrow-form (`name_or_empty`) and owned-
        // form (`owned_name_or_empty`) primitives return the SAME byte
        // sequence and differ only in ownership. A regression that
        // skewed one form's fallback would surface here rather than as
        // silent drift between the router tie-break comparator and the
        // AllocationRef seed on the SAME pool.
        let p = pool_named("attest-pool");
        assert_eq!(p.name_or_empty(), p.owned_name_or_empty().as_str());
    }

    #[test]
    fn owned_name_or_empty_matches_borrow_form_peer_on_missing_slot() {
        // Sibling corner of the coherence pin above: when the slot is
        // absent (or explicitly empty), BOTH primitives fold to the
        // same empty-string byte-shape. The load-bearing property is
        // that a caller who switches between the two return-forms
        // based on downstream ownership requirements never sees a
        // different missing-slot spelling as a side effect.
        let p = pool_unnamed();
        assert_eq!(p.name_or_empty(), p.owned_name_or_empty().as_str());
        assert_eq!(p.name_or_empty(), "");
        assert_eq!(p.owned_name_or_empty(), String::new());
    }

    // ─── EphemeralPool::is_being_deleted substrate pins ───────────────
    //
    // Pins the copy-form metadata-projection primitive on the deletion-
    // tombstone axis of the pool CRD. Peer to the borrow-form + owned-
    // form metadata-fallback family (`name_or_empty`,
    // `owned_name_or_empty`); this one opens the presence-probe corner
    // for the tombstone slot. Sibling to the sister-CRD primitive
    // `crate::crd::Process::is_being_deleted` — the two primitives
    // now partition the tombstone-presence probe across BOTH tatara-
    // process CRDs on identical missing-slot semantics. Fail-before-
    // pass-after granularity: `is_being_deleted` did not exist on the
    // pool CRD pre-lift; the compiler cannot resolve the name until
    // the impl block above is in place, so a rollback of the primitive
    // breaks this whole module.

    fn tombstoned_pool() -> EphemeralPool {
        let mut p = pool_named("attest-pool");
        p.metadata.namespace = Some("ephemeral-pools".into());
        p.metadata.deletion_timestamp = Some(k8s_openapi::apimachinery::pkg::apis::meta::v1::Time(
            Utc::now(),
        ));
        p
    }

    #[test]
    fn is_being_deleted_returns_false_when_deletion_timestamp_is_absent() {
        // Missing-tombstone corner pin: the primitive collapses the
        // no-tombstone case to `false` so the `→ Drain` short-circuit
        // at `decide_pool_reconcile` is NOT taken and the observed-
        // phase composer at `pool_phase_from_members` proceeds to its
        // normal (free / spawning / allocated) arithmetic branches
        // instead of short-circuiting to `PoolPhase::Draining`.
        // Matches the pre-lift `.is_some()` chain's `false` byte-
        // identically at every consumer's downstream gate.
        let mut p = pool_named("attest-pool");
        p.metadata.deletion_timestamp = None;
        assert!(!p.is_being_deleted());
    }

    #[test]
    fn is_being_deleted_returns_true_when_deletion_timestamp_is_present() {
        // Present-tombstone corner pin: the primitive returns `true`
        // on any populated `metadata.deletionTimestamp` slot regardless
        // of the timestamp payload — the two consumers only read the
        // tombstone's PRESENCE, never its RFC-3339 timestamp value.
        // A regression that gated the `true` return on the timestamp
        // being non-epoch, or parsed the timestamp before returning,
        // would surface here rather than as silent skew at the
        // `→ Drain` decision or the `→ Draining` phase report on the
        // SAME `EphemeralPool`.
        let p = tombstoned_pool();
        assert!(p.is_being_deleted());
    }

    #[test]
    fn is_being_deleted_is_a_pure_projection() {
        // Purity pin: two consecutive calls return byte-identical
        // `bool` values (no lazy materialization, no interior
        // mutation of `self`). Peer to the sibling
        // `name_or_empty_is_a_pure_projection` +
        // `owned_name_or_empty_is_a_pure_projection` pins in this
        // module and to `is_being_deleted_is_a_pure_projection` on
        // the sister-CRD `Process`; all four bind the pure-projection
        // discipline on the ONE substrate accessor per metadata slot.
        let p = tombstoned_pool();
        let a = p.is_being_deleted();
        let b = p.is_being_deleted();
        assert_eq!(a, b);
        assert!(a);
    }

    #[test]
    fn is_being_deleted_matches_pre_lift_pool_reconciler_chain_shape() {
        // Parity pin: sweeps the two corners every pre-lift consumer
        // plausibly encountered (missing tombstone, present tombstone)
        // and compares the substrate call against a hand-authored pre-
        // lift chain byte-identically. A regression that reshaped
        // either corner would surface here rather than as silent
        // operator-facing skew between the pool-reconciler's `→ Drain`
        // decision and the observed-phase composer's `→ Draining`
        // report on the SAME `EphemeralPool` within one reconcile
        // pass.
        fn pre_lift(p: &EphemeralPool) -> bool {
            p.metadata.deletion_timestamp.is_some()
        }
        // Missing slot.
        let mut p = pool_named("attest-pool");
        p.metadata.deletion_timestamp = None;
        assert_eq!(p.is_being_deleted(), pre_lift(&p));
        // Populated slot.
        let p = tombstoned_pool();
        assert_eq!(p.is_being_deleted(), pre_lift(&p));
    }

    #[test]
    fn is_being_deleted_composes_with_pool_phase_draining_at_reconcile_preempt() {
        // Call-site-shape pin: the `pool_phase_from_members`
        // deletion-preempt returns `PoolPhase::Draining` as soon as
        // `pool.is_being_deleted()` holds, regardless of the (free +
        // spawning) supply arithmetic that would otherwise pick
        // `Ready` / `Scaling` / `Degraded`. The `→ Drain` decision at
        // `decide_pool_reconcile` composes with the same probe on the
        // same tombstone-presence slot. A regression that broadened
        // the tombstone probe implicitly (returning `false` on a
        // present but zero-timestamp) or narrowed it (requiring an
        // additional `.finalizers.is_empty()` conjunct that the two
        // consumers never spelled) would surface here rather than as
        // silent operator-facing skew between the pool reconciler's
        // decision and the observed-phase composer on the SAME
        // `EphemeralPool` within one reconcile pass.
        let alive = pool_named("attest-pool");
        assert!(!alive.is_being_deleted());
        let dying = tombstoned_pool();
        assert!(dying.is_being_deleted());
    }

    // ─── EphemeralPool::owned_namespace_or_empty substrate pins ───────
    //
    // The owned-form peer of the `owned_name_or_empty` primitive on the
    // sibling `metadata.namespace` axis — the paired half of the
    // `AllocationRef { name, namespace }` struct literal both
    // `AllocationConvergenceCtx::observe` and the composition pin
    // consume through the SAME `AllocationRef::new(name, namespace)`
    // constructor. Fail-before-pass-after granularity:
    // `owned_namespace_or_empty` did not exist on the pool CRD pre-
    // lift; the compiler cannot resolve the name until the impl block
    // above is in place, so a rollback of the primitive breaks this
    // whole module.
    #[test]
    fn owned_namespace_or_empty_returns_empty_string_when_metadata_namespace_is_none() {
        // Missing-slot corner pin: the primitive collapses the no-
        // namespace case to the load-bearing empty-string sentinel so
        // the downstream `AllocationRef.namespace` slot carries `""`
        // rather than a defaulted `"default"` string. See the doc-
        // comment's DELIBERATE-EMPTY-SENTINEL rationale for why the
        // fallback matches `.clone().unwrap_or_default()` byte-for-
        // byte rather than substituting `Process::DEFAULT_NAMESPACE`
        // at the primitive.
        let mut p = pool_named("attest-pool");
        p.metadata.namespace = None;
        assert!(p.metadata.namespace.is_none(), "fixture invariant");
        assert_eq!(p.owned_namespace_or_empty(), String::new());
    }

    #[test]
    fn owned_namespace_or_empty_returns_owned_string_when_slot_is_populated() {
        let mut p = pool_named("attest-pool");
        p.metadata.namespace = Some("ephemeral-pools".into());
        assert_eq!(p.owned_namespace_or_empty(), "ephemeral-pools");
    }

    #[test]
    fn owned_namespace_or_empty_returns_empty_string_when_slot_is_explicitly_empty_string() {
        // Corner between `None` (missing slot) and `Some(String::new())`
        // (populated slot containing the empty string): the primitive
        // MUST fold both to the same `""` byte-shape so a downstream
        // `AllocationRef.namespace ==` comparator at
        // `resolve_pool` sees ONE "unset namespace" bucket regardless
        // of which shape the K8s API server materialized. Byte-
        // identical to what the pre-lift `.clone().unwrap_or_default()`
        // chain produced.
        let mut p = pool_named("attest-pool");
        p.metadata.namespace = Some(String::new());
        assert_eq!(p.owned_namespace_or_empty(), String::new());
        assert!(p.owned_namespace_or_empty().is_empty());
    }

    #[test]
    fn owned_namespace_or_empty_is_a_pure_projection() {
        // Consecutive calls return byte-identical Strings — no cached
        // state, no mutation on the `EphemeralPool` between calls.
        // Peer to the sibling `owned_name_or_empty_is_a_pure_projection`
        // pin in this module and to `is_being_deleted_is_a_pure_projection`
        // on the same CRD; all three bind the pure-projection
        // discipline on the ONE substrate accessor per metadata slot.
        let mut p = pool_named("attest-pool");
        p.metadata.namespace = Some("ephemeral-pools".into());
        assert_eq!(p.owned_namespace_or_empty(), p.owned_namespace_or_empty());
        assert_eq!(p.owned_namespace_or_empty(), "ephemeral-pools");
        assert_eq!(p.owned_namespace_or_empty(), "ephemeral-pools");
    }

    #[test]
    fn owned_namespace_or_empty_matches_pre_lift_chain_verbatim() {
        // Byte-identical parity with the two hand-authored
        // `.metadata.namespace.clone().unwrap_or_default()` chains
        // the primitive replaces in `tatara-pool-reconciler::
        // allocation_decide::AllocationConvergenceCtx::observe`
        // (matched-pool `AllocationRef.namespace` seed) and in the
        // sibling composition pin
        // `allocation_ref_new_composes_with_owned_name_or_empty_pool_projection`.
        // Runs across the FULL corner set of the metadata.namespace
        // slot: absent, present-with-value, present-with-empty-string.
        // A regression that inserted a normalization step at the
        // primitive the pre-lift chain does NOT apply — or vice versa —
        // surfaces here rather than as silent drift between the two
        // owned-form callsites and the ONE substrate owner they now
        // route through.
        let cases: [(Option<String>, &str); 3] = [
            (None, ""),
            (Some("ephemeral-pools".into()), "ephemeral-pools"),
            (Some(String::new()), ""),
        ];
        for (slot, expected) in cases {
            let mut p = pool_named("attest-pool");
            p.metadata.namespace = slot.clone();
            let pre_lift = p.metadata.namespace.clone().unwrap_or_default();
            assert_eq!(pre_lift.as_str(), expected, "pre-lift chain sanity");
            assert_eq!(p.owned_namespace_or_empty(), pre_lift);
            assert_eq!(p.owned_namespace_or_empty().as_str(), expected);
        }
    }

    #[test]
    fn owned_namespace_or_empty_composes_with_owned_name_or_empty_on_paired_slot_axis() {
        // Paired-axis coherence pin: the two owned-form primitives on
        // the pool CRD's `metadata.name` + `metadata.namespace` slots
        // share the SAME empty-string sentinel on the missing corner,
        // so a caller that composes both halves into an
        // `AllocationRef` (as `AllocationConvergenceCtx::observe`
        // does) never sees a mixed-fallback pair (one `""`, the
        // other `"default"`) as a side effect of one slot being
        // absent. A regression that skewed either primitive's
        // fallback would surface here rather than as silent operator-
        // facing skew between the paired halves of the SAME
        // `AllocationRef` seed.
        let mut p = pool_named("attest-pool");
        p.metadata.namespace = None;
        p.metadata.name = None;
        assert_eq!(p.owned_name_or_empty(), p.owned_namespace_or_empty());
        assert_eq!(p.owned_name_or_empty(), String::new());
        assert_eq!(p.owned_namespace_or_empty(), String::new());
    }

    #[test]
    fn owned_namespace_or_empty_does_not_default_to_process_default_namespace() {
        // Deliberate-empty-sentinel pin: the primitive's fallback is
        // `""`, NOT `crate::crd::Process::DEFAULT_NAMESPACE`. The
        // sole downstream consumer (`AllocationConvergenceCtx::observe`)
        // feeds the produced value into `AllocationRef.namespace`,
        // which is then matched byte-identically against
        // `spec.pool_ref.namespace` at `resolve_pool`. A silent
        // substitution of `"default"` at this primitive would alias
        // every namespace-absent pool to the `"default"` bucket at
        // the matcher, hiding the missing-slot corner from an
        // operator who explicitly authored an allocation against a
        // namespace-unset pool. Pinned so a future "helpful"
        // canonicalization step lands as a compiler-visible failure
        // here rather than as silent operator-facing skew at the
        // matched-pool seed.
        let mut p = pool_named("attest-pool");
        p.metadata.namespace = None;
        assert_ne!(
            p.owned_namespace_or_empty(),
            crate::crd::Process::DEFAULT_NAMESPACE
        );
        assert_eq!(p.owned_namespace_or_empty(), "");
    }

    // ─── AllocationRef::new substrate pins ────────────────────────────
    //
    // Pins the substrate constructor for [`AllocationRef`] — the
    // ONE-liner composer that lifts the paired
    // `AllocationRef { name, namespace }` struct-literal every
    // downstream consumer restated by hand pre-lift at FOUR production
    // sites (2 × controller_allocation.rs assignedProcess seeds, 1 ×
    // allocation_decide.rs pool_ref seed, 1 × allocation_factory.rs
    // pool_ref seed) onto ONE substrate owner on `AllocationRef`.
    // Fail-before-pass-after granularity: `AllocationRef::new` did not
    // exist pre-lift; the compiler cannot resolve the name until the
    // impl block above is in place, so a rollback of the primitive
    // breaks this whole module.

    #[test]
    fn allocation_ref_new_composes_owned_string_pair_verbatim() {
        // Happy-path pin: the constructor materializes an
        // `AllocationRef { name: <name>, namespace: <namespace> }`
        // byte-identical to the pre-lift struct literal every consumer
        // spelled. A regression that dropped either slot (e.g. an
        // erroneous `..Default::default()` on a shape that never had
        // a Default derive) surfaces here rather than as silent slot
        // loss downstream at the assignedProcess / bound_pool /
        // matched_pool / spec.pool_ref sinks.
        let r = AllocationRef::new(String::from("pr-42-demo"), String::from("ephemeral-pools"));
        assert_eq!(r.name, "pr-42-demo");
        assert_eq!(r.namespace, "ephemeral-pools");
    }

    #[test]
    fn allocation_ref_new_matches_pre_lift_struct_literal_verbatim() {
        // Byte-identical parity pin: the substrate constructor and the
        // hand-authored struct literal produce equal `AllocationRef`
        // values on every provenance the FOUR pre-lift sites carried
        // (owned `String` from an owned-form projection; `&str`
        // promoted through `.to_string()`). A regression that inserted
        // a normalization step at the primitive the pre-lift literal
        // does NOT apply — or vice versa — surfaces here rather than
        // as silent drift between the four consumers and the ONE
        // substrate owner they now route through.
        let owned_name = String::from("pr-42-demo");
        let owned_ns = String::from("ephemeral-pools");
        let lifted = AllocationRef::new(owned_name.clone(), owned_ns.clone());
        let pre_lift = AllocationRef {
            name: owned_name,
            namespace: owned_ns,
        };
        assert_eq!(lifted, pre_lift);
    }

    #[test]
    fn allocation_ref_new_accepts_str_provenance_via_into_string() {
        // `Into<String>` provenance-closure pin: the primitive accepts
        // every provenance the pre-lift sites carried. The
        // controller_allocation.rs assignedProcess seeds passed owned
        // `String` values (a moved `member_process_name` +
        // `ns.clone()`); the allocation_factory.rs pool_ref seed
        // passed `&str` (`n.to_string()` / `namespace.to_string()`).
        // Both provenances produce byte-identical output. A future
        // refactor of the constructor signature that demanded owned
        // `String` at author sites (dropping `impl Into<String>`)
        // would force `.to_string()` back at the FOUR call sites — the
        // pin fences that regression at ONE place.
        let from_str = AllocationRef::new("pr-42-demo", "ephemeral-pools");
        let from_string =
            AllocationRef::new(String::from("pr-42-demo"), String::from("ephemeral-pools"));
        assert_eq!(from_str, from_string);
        // Mixed provenance is also load-bearing: the allocation_decide.rs
        // matched_pool seed pairs an owned `String` (from
        // `EphemeralPool::owned_name_or_empty()`) with a hand-authored
        // `.clone().unwrap_or_default()` — also `String`. The
        // controller_allocation.rs paths pair a moved `String` name
        // with a `.clone()`-ed `ns: String`. Verify (owned, borrow)
        // and (borrow, owned) both compose to the same shape as
        // (owned, owned) / (borrow, borrow).
        let mixed_a = AllocationRef::new(String::from("pr-42-demo"), "ephemeral-pools");
        let mixed_b = AllocationRef::new("pr-42-demo", String::from("ephemeral-pools"));
        assert_eq!(from_str, mixed_a);
        assert_eq!(from_str, mixed_b);
    }

    #[test]
    fn allocation_ref_new_positional_axis_order_pinned_name_first_namespace_second() {
        // Axis-order pin: name is the FIRST positional argument;
        // namespace is the SECOND. Reversing the pair at the
        // constructor is the exact regression this pin fences — the
        // FOUR pre-lift sites all spelled `name` before `namespace`
        // (matching the struct definition's field order in
        // `pub struct AllocationRef { pub name, pub namespace }`)
        // and the wire-format serde output `{ "name": "...",
        // "namespace": "..." }` reflects that order. A slot swap at
        // the primitive would surface here rather than as silent
        // `<namespace>/<name>` inversion at every downstream
        // qualified-ref composer that reads `{ref.name}/{ref.namespace}`
        // as an audit-log key.
        let r = AllocationRef::new("alpha-name", "beta-namespace");
        assert_eq!(r.name, "alpha-name");
        assert_eq!(r.namespace, "beta-namespace");
        assert_ne!(r.name, "beta-namespace");
        assert_ne!(r.namespace, "alpha-name");
    }

    #[test]
    fn allocation_ref_new_preserves_empty_string_verbatim() {
        // Empty-string sentinel pin: the constructor is pure — it does
        // NOT canonicalize empty inputs (does NOT default an empty
        // namespace to `"default"`; does NOT reject an empty name).
        // Preserves the pre-lift shape the allocation_decide.rs
        // matched_pool seed relied on: when the pool's metadata.namespace
        // is absent, `.clone().unwrap_or_default()` yields the empty
        // string, and the AllocationRef's namespace slot carries that
        // empty string verbatim to the downstream `bound_pool` sink.
        // A future canonicalization pass (e.g. defaulting to
        // `Process::DEFAULT_NAMESPACE`) MUST land here, not at the
        // primitive body silently, so the pre-lift consumers' empty-
        // sentinel semantics are the visible contract of the new
        // constructor.
        let r = AllocationRef::new("", "");
        assert_eq!(r.name, "");
        assert_eq!(r.namespace, "");
        let mixed = AllocationRef::new("pr-42-demo", "");
        assert_eq!(mixed.name, "pr-42-demo");
        assert_eq!(mixed.namespace, "");
    }

    #[test]
    fn allocation_ref_new_composes_with_owned_name_or_empty_pool_projection() {
        // Composition pin: the constructor composes with the paired
        // substrate primitives [`EphemeralPool::owned_name_or_empty`]
        // + [`EphemeralPool::owned_namespace_or_empty`] at the
        // allocation_decide.rs pool_ref seed — the same primitive
        // family the pool CRD opened for both halves of the
        // `AllocationRef { name, namespace }` struct literal. The
        // composed pair carries an owned `String` name half (from
        // `pool.owned_name_or_empty()`) and an owned `String`
        // namespace half (from `pool.owned_namespace_or_empty()`) —
        // no pre-lift chain remains. A regression that broke the
        // primitive family's `impl Into<String>` acceptance of an
        // owned `String` return type would surface here rather than
        // as silent build failure at the pool-reconciler matched_pool
        // seed.
        let pool = pool_named("attest-pool");
        let r = AllocationRef::new(pool.owned_name_or_empty(), pool.owned_namespace_or_empty());
        assert_eq!(r.name, "attest-pool");
        assert_eq!(r.namespace, pool.owned_namespace_or_empty());
    }

    #[test]
    fn allocation_ref_new_returns_wire_format_serialization_verbatim() {
        // Wire-format pin: the constructor produces an
        // [`AllocationRef`] whose serde `rename_all = "camelCase"`
        // serialization is byte-identical to the pre-lift struct
        // literal's serialization. The `bound_pool` and
        // `assignedProcess` slots on `AllocationStatus` (and the
        // `poolRef` slot on `AllocationSpec`) all round-trip through
        // this shape — the pin fences a regression that added a
        // private field or a `#[serde(skip)]` accidentally.
        let r = AllocationRef::new("pr-42-demo", "ephemeral-pools");
        let yaml = serde_yaml::to_string(&r).expect("AllocationRef serializes to yaml");
        assert!(yaml.contains("name: pr-42-demo"), "{yaml}");
        assert!(yaml.contains("namespace: ephemeral-pools"), "{yaml}");
        let back: AllocationRef =
            serde_yaml::from_str(&yaml).expect("AllocationRef round-trips through yaml");
        assert_eq!(back, r);
    }

    fn member(state: MemberState) -> PoolMember {
        PoolMember {
            process_name: "m".into(),
            state,
            entered_state_at: DateTime::<Utc>::from_timestamp(0, 0).unwrap(),
            allocation_ref: None,
        }
    }

    #[test]
    fn state_count_fanout_returns_all_zeros_on_empty_slice() {
        // Zero-length pin: the empty-members corner produces a
        // 4-tuple of zero counters, matching the pre-lift
        // `count_state` fanout's four `.iter().filter(...).count()`
        // calls each returning 0 on an empty iterator.
        assert_eq!(PoolMember::state_count_fanout(&[]), (0, 0, 0, 0));
    }

    #[test]
    fn state_count_fanout_partitions_variants_into_correct_slots() {
        // Positional-axis pin: the returned 4-tuple's slot order
        // matches the four `PoolStatus` counter slots in declaration
        // order — `(ready, allocated, spawning, returning)`. A
        // regression that swapped two slots (e.g., `ready` ↔
        // `spawning`) surfaces here rather than as an operator-facing
        // scale-out oscillation at the pool reconciler.
        let members = vec![
            member(MemberState::Free),
            member(MemberState::Free),
            member(MemberState::Allocated),
            member(MemberState::Spawning),
            member(MemberState::Spawning),
            member(MemberState::Spawning),
            member(MemberState::Returning),
        ];
        assert_eq!(PoolMember::state_count_fanout(&members), (2, 1, 3, 1));
    }

    #[test]
    fn state_count_fanout_excludes_failed_from_every_counter() {
        // Closed-set pin: no `PoolStatus` slot counts `Failed` members
        // (they surface via `PoolPhase::Degraded` instead of a status
        // counter). This test fences a regression that let a `Failed`
        // member drift into one of the four counters and inflate the
        // operator-visible ready/allocated/spawning/returning fanout.
        let members = vec![
            member(MemberState::Failed),
            member(MemberState::Failed),
            member(MemberState::Failed),
        ];
        assert_eq!(PoolMember::state_count_fanout(&members), (0, 0, 0, 0));

        // Mixed with a Free member: the Free member is counted, the
        // Failed members are not.
        let mixed = vec![
            member(MemberState::Free),
            member(MemberState::Failed),
            member(MemberState::Failed),
        ];
        assert_eq!(PoolMember::state_count_fanout(&mixed), (1, 0, 0, 0));
    }

    #[test]
    fn state_count_fanout_matches_pre_lift_count_state_helper_verbatim() {
        // Parity pin: for every possible members list, the 4-tuple
        // returned by the substrate primitive matches the pre-lift
        // `count_state(&members, MemberState::<slot>)` fanout that
        // pool-reconciler restated at both status-patch sites. The
        // pre-lift helper was
        // ```rust,ignore
        // fn count_state(members: &[PoolMember], target: MemberState) -> u32 {
        //     members.iter().filter(|m| m.state == target).count() as u32
        // }
        // ```
        // — re-implemented inline here as an oracle.
        fn count_state(members: &[PoolMember], target: MemberState) -> u32 {
            members.iter().filter(|m| m.state == target).count() as u32
        }
        let members = vec![
            member(MemberState::Free),
            member(MemberState::Allocated),
            member(MemberState::Allocated),
            member(MemberState::Spawning),
            member(MemberState::Returning),
            member(MemberState::Returning),
            member(MemberState::Failed),
        ];
        let (ready, allocated, spawning, returning) = PoolMember::state_count_fanout(&members);
        assert_eq!(ready, count_state(&members, MemberState::Free));
        assert_eq!(allocated, count_state(&members, MemberState::Allocated));
        assert_eq!(spawning, count_state(&members, MemberState::Spawning));
        assert_eq!(returning, count_state(&members, MemberState::Returning));
    }

    #[test]
    fn pool_status_observed_composes_pre_lift_status_seed_verbatim() {
        // Composition pin: the substrate constructor produces a
        // `PoolStatus` structurally equal to the pre-lift 11-line
        // struct literal both pool-reconciler status-patch sites
        // stamped by hand. Any drift in the defaults (`message`,
        // `conditions`) or in the counter fanout surfaces here.
        let now = DateTime::<Utc>::from_timestamp(1_700_000_000, 0).unwrap();
        let members = vec![
            member(MemberState::Free),
            member(MemberState::Allocated),
            member(MemberState::Spawning),
            member(MemberState::Returning),
            member(MemberState::Failed),
        ];
        let member_count = members.len();
        let observed = PoolStatus::observed(PoolPhase::Steady, members, now);
        assert_eq!(observed.phase, PoolPhase::Steady);
        assert_eq!(observed.phase_since, Some(now));
        assert_eq!(observed.ready_count, 1);
        assert_eq!(observed.allocated_count, 1);
        assert_eq!(observed.spawning_count, 1);
        assert_eq!(observed.returning_count, 1);
        assert_eq!(observed.members.len(), member_count);
        assert!(observed.message.is_none());
        assert!(observed.conditions.is_empty());
    }

    #[test]
    fn pool_status_observed_moves_members_by_value_without_extra_clone() {
        // Ownership pin: the constructor consumes the members Vec by
        // value rather than borrowing + cloning internally. Both pre-
        // lift sites called `.clone()` on their `members` binding for
        // the struct-literal `members:` slot; the substrate lift keeps
        // the same one-clone bound at the caller (or a straight move
        // if the caller no longer needs the local `members` binding
        // after the seed) rather than accidentally cloning twice.
        let members = vec![member(MemberState::Free), member(MemberState::Spawning)];
        let now = DateTime::<Utc>::from_timestamp(0, 0).unwrap();
        let observed = PoolStatus::observed(PoolPhase::Steady, members, now);
        assert_eq!(observed.members.len(), 2);
    }
}