aion_core/

workloop.rs

//! Workloop domain vocabulary: arming, invariants, tolerance, health, alarms,
//! and the hatch dedupe identity.
//!
//! A workloop is the perpetual-work document kind: its completion is an
//! incident, its header declares invariants with tolerances, and its cadence is
//! an engine-side dead-man switch. This module carries the pure vocabulary the
//! engine, stores, and surfaces share; the engine-side services live in the
//! `aion` crate.

use std::collections::BTreeMap;
use std::time::Duration;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::WorkflowId;

/// Search attribute name that records a run's document kind on the listing
/// surface.
///
/// Workloop runs project a distinct kind so list renderers and filters can
/// separate loops from workflows WITHOUT any change to
/// [`crate::WorkflowStatus`] — the kind is an additive attribute, folded from
/// recorded history exactly like `aion.display_name`. Absent = an ordinary
/// workflow.
pub const WORKFLOW_KIND_ATTRIBUTE: &str = "aion.kind";

/// The [`WORKFLOW_KIND_ATTRIBUTE`] value recorded for workloop runs.
pub const WORKLOOP_KIND: &str = "workloop";

/// The run's document kind, projected from an already-folded search-attribute
/// map (the same map [`crate::search_attributes_from_events`] produces).
///
/// Returns `None` for ordinary workflows — histories that never recorded a
/// kind attribute.
#[must_use]
pub fn workflow_kind_from_attributes<S: std::hash::BuildHasher>(
    attributes: &std::collections::HashMap<String, crate::SearchAttributeValue, S>,
) -> Option<String> {
    match attributes.get(WORKFLOW_KIND_ATTRIBUTE) {
        Some(crate::SearchAttributeValue::String(kind)) => Some(kind.clone()),
        _ => None,
    }
}

/// The run's document kind, projected from recorded history.
///
/// Folds every [`crate::Event::SearchAttributesUpdated`] (last write wins) and
/// reads the [`WORKFLOW_KIND_ATTRIBUTE`], mirroring [`crate::display_name`].
#[must_use]
pub fn workflow_kind(events: &[crate::Event]) -> Option<String> {
    workflow_kind_from_attributes(&crate::search_attributes_from_events(events))
}

/// Cause carried on the ONE alarm path (design brief R4.2).
///
/// From the invariant's view a missed window and a failed sample are the same
/// event — *the invariant is not confirmed held* — so cause is a FIELD on the
/// alarm, never a separate alarm channel. The set is closed and additive:
/// trigger selectors (Leg 3) arm on named causes as ALLOWLISTS (R5.2a), so an
/// unforeseen cause fails safe by not firing anything.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[serde(rename_all = "kebab-case")]
pub enum AlarmCause {
    /// The loop ran and produced unconfirmed/red samples past tolerance.
    SampleRed,
    /// The loop missed its declared cadence window(s) past tolerance — the
    /// engine-side dead-man switch (R4.3), which also covers the hung
    /// iteration (R3.3a): an iteration that produces no terminal by its next
    /// window is a missed window, never waited on.
    WindowMissed,
    /// The engine positively knows the loop cannot run: its run is terminal
    /// without a declared retirement, or its history is gone. Triggers must
    /// never arm remediation on this cause (R5.3) — it means watching stopped.
    LoopDead,
    /// Duration-form tolerance expired with no evidence either way: no sample
    /// arrived and no window exists to miss (a signal-armed loop gone silent —
    /// the R2.4a silent-death family). Named per the brief's own read-time
    /// vocabulary ("unconfirmed-unknown", R2.5a).
    UnconfirmedUnknown,
}

/// Health verdict of one sample against one invariant.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
pub enum HealthStatus {
    /// A route declared as confirming the invariant was taken (R3.3).
    Confirmed,
    /// The iteration closed or failed without confirming the invariant — an
    /// unhealthy sample. Every invariant is sampled on the same tick (R2.2),
    /// so a closing iteration yields a sample per invariant.
    Unconfirmed,
}

/// One health sample recorded against a loop invariant (R3.3).
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct HealthSample {
    /// Invariant the sample lands against.
    pub invariant: String,
    /// Sample verdict.
    pub status: HealthStatus,
    /// Cadence window the sample belongs to; `None` on a signal-only loop,
    /// which has no windows.
    pub window_seq: Option<u64>,
}

/// A raised invariant alarm — the payload of
/// [`crate::Event::InvariantUnconfirmed`], carried as one value so the health
/// engine, the Recorder, and the trigger layer all speak the same record.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct InvariantAlarm {
    /// Invariant that is not confirmed held.
    pub invariant: String,
    /// Why confirmation is missing (R4.2's cause field).
    pub cause: AlarmCause,
    /// Cadence window at which tolerance was exceeded; `None` on a
    /// signal-only loop.
    pub window_seq: Option<u64>,
    /// When the invariant was last confirmed, if ever (completeness claim).
    pub last_confirmed_at: Option<DateTime<Utc>>,
    /// Consecutive unconfirmed samples/windows observed at alarm time.
    pub consecutive_unconfirmed: u64,
}

/// Errors refusing an invalid workloop declaration at the engine boundary.
///
/// The estate rule is NO ASSUMED DEFAULTS: tolerance, cadence, and retention
/// are declarations, and an absent declaration is refused — here as well as at
/// the AWL checker, because the engine validates at its own boundary.
#[derive(thiserror::Error, Clone, Debug, PartialEq, Eq)]
pub enum WorkloopSpecError {
    /// A tolerance declared neither the count form nor the duration form.
    #[error("tolerance must declare at least one form (N windows or unconfirmed-for duration)")]
    ToleranceUndeclared,
    /// A duration-form tolerance declared a zero duration.
    #[error("tolerance duration must be greater than zero")]
    ToleranceZeroDuration,
    /// A cadence arming declared a zero period.
    #[error("cadence period must be greater than zero")]
    ZeroCadencePeriod,
    /// A signal arming declared no signals.
    #[error("signal arming must name at least one signal")]
    NoSignals,
    /// A signal name was empty.
    #[error("signal names must be non-empty")]
    EmptySignalName,
    /// An invariant on a signal-only loop declared only the count form
    /// (the R2.4a silent-death cross-rule).
    #[error(
        "invariant `{invariant}` on a signal-only workloop must declare the duration-form \
         tolerance: with no windows to miss, a count of samples can never alarm on total silence"
    )]
    SignalOnlyNeedsDurationTolerance {
        /// Invariant missing the duration form.
        invariant: String,
    },
    /// An invariant name was empty.
    #[error("invariant names must be non-empty")]
    EmptyInvariantName,
    /// Two invariants shared a name.
    #[error("invariant `{invariant}` is declared more than once")]
    DuplicateInvariant {
        /// The duplicated invariant name.
        invariant: String,
    },
    /// An invariant declared no current-state record type.
    #[error("invariant `{invariant}` must declare the type of its current-state record")]
    MissingRecordType {
        /// Invariant missing the record type.
        invariant: String,
    },
    /// An invariant declared no confirming route.
    #[error(
        "invariant `{invariant}` must declare at least one confirming route; an invariant \
         nothing can confirm alarms unconditionally"
    )]
    NoConfirmingRoutes {
        /// Invariant with no confirming route.
        invariant: String,
    },
    /// A confirming route name was empty.
    #[error("confirming route names on invariant `{invariant}` must be non-empty")]
    EmptyConfirmingRoute {
        /// Invariant carrying the empty route name.
        invariant: String,
    },
    /// The workloop declared no invariants.
    #[error(
        "a workloop must declare at least one invariant; a loop with no invariants has no \
         health surface and rebuilds the silent-death family"
    )]
    NoInvariants,
    /// The retention window was zero.
    #[error("retention window must be greater than zero")]
    ZeroRetention,
    /// The retention window could not be expressed as a calendar duration, so
    /// no cutoff instant could ever be derived from it.
    #[error(
        "retention window of {seconds}s cannot be expressed as a calendar duration, so no \
         retention cutoff can be derived from it; declare a window the clock can subtract"
    )]
    UnrepresentableRetention {
        /// The declared window, in whole seconds.
        seconds: u64,
    },
    /// A carry field name was empty.
    #[error("carry field names must be non-empty")]
    EmptyCarryField,
    /// A generation-1 start payload was not JSON at all.
    #[error("a workloop start payload carrying declared carry fields must be a JSON document")]
    CarrySeedTargetNotJson,
    /// A generation-1 start payload was JSON but not an object.
    #[error(
        "a workloop start payload must be a JSON object so declared carry fields can be seeded \
         into it; seeding a scalar or an array has nowhere to put a named field"
    )]
    CarrySeedTargetNotAnObject,
    /// A hatch identity part was empty.
    #[error("hatch identity parts (namespace, workflow type, key) must be non-empty")]
    EmptyHatchIdentityPart,
    /// A hatch identity part contained a NUL byte, which the derivation
    /// reserves as its separator.
    #[error("hatch identity parts must not contain NUL bytes")]
    HatchIdentityNulByte,
}

/// Declared tolerance for one invariant (R2.3): *tolerates N consecutive
/// unhealthy samples* and/or *unconfirmed for duration D*.
///
/// There is NO default tolerance — the constructors are the only way to build
/// one, and each requires an explicit declaration. Deserialization re-validates
/// through the same constructors, so a stored record cannot smuggle an
/// undeclared tolerance back in.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(try_from = "ToleranceSpecWire", into = "ToleranceSpecWire")]
pub struct ToleranceSpec {
    consecutive_windows: Option<u64>,
    unconfirmed_for: Option<Duration>,
}

impl ToleranceSpec {
    /// Count form: tolerates `windows` consecutive unhealthy samples; the
    /// (`windows` + 1)-th consecutive unhealthy sample exceeds tolerance.
    /// Zero is a legitimate DECLARED value (alarm on the first unhealthy
    /// sample) — what is refused is absence, never zero.
    #[must_use]
    pub const fn count(windows: u64) -> Self {
        Self {
            consecutive_windows: Some(windows),
            unconfirmed_for: None,
        }
    }

    /// Duration form: alarms once the invariant has gone unconfirmed for
    /// `unconfirmed_for` — the only form evaluable engine-side with zero
    /// samples (R2.4a).
    ///
    /// # Errors
    ///
    /// Refuses a zero duration ([`WorkloopSpecError::ToleranceZeroDuration`]).
    pub const fn duration(unconfirmed_for: Duration) -> Result<Self, WorkloopSpecError> {
        if unconfirmed_for.is_zero() {
            return Err(WorkloopSpecError::ToleranceZeroDuration);
        }
        Ok(Self {
            consecutive_windows: None,
            unconfirmed_for: Some(unconfirmed_for),
        })
    }

    /// Both forms together: the count form may be declared in addition to the
    /// duration form, never instead (R2.4a).
    ///
    /// # Errors
    ///
    /// Refuses a zero duration ([`WorkloopSpecError::ToleranceZeroDuration`]).
    pub const fn both(windows: u64, unconfirmed_for: Duration) -> Result<Self, WorkloopSpecError> {
        if unconfirmed_for.is_zero() {
            return Err(WorkloopSpecError::ToleranceZeroDuration);
        }
        Ok(Self {
            consecutive_windows: Some(windows),
            unconfirmed_for: Some(unconfirmed_for),
        })
    }

    /// The declared count form, when present.
    #[must_use]
    pub const fn consecutive_windows(&self) -> Option<u64> {
        self.consecutive_windows
    }

    /// The declared duration form, when present.
    #[must_use]
    pub const fn unconfirmed_for(&self) -> Option<Duration> {
        self.unconfirmed_for
    }
}

/// Serde wire shape for [`ToleranceSpec`]; decoding re-runs the declaration
/// checks so both-absent can never be represented.
#[derive(Serialize, Deserialize, Clone, Debug)]
struct ToleranceSpecWire {
    consecutive_windows: Option<u64>,
    unconfirmed_for: Option<Duration>,
}

impl TryFrom<ToleranceSpecWire> for ToleranceSpec {
    type Error = WorkloopSpecError;

    fn try_from(wire: ToleranceSpecWire) -> Result<Self, Self::Error> {
        match (wire.consecutive_windows, wire.unconfirmed_for) {
            (None, None) => Err(WorkloopSpecError::ToleranceUndeclared),
            (Some(windows), None) => Ok(Self::count(windows)),
            (None, Some(duration)) => Self::duration(duration),
            (Some(windows), Some(duration)) => Self::both(windows, duration),
        }
    }
}

impl From<ToleranceSpec> for ToleranceSpecWire {
    fn from(spec: ToleranceSpec) -> Self {
        Self {
            consecutive_windows: spec.consecutive_windows,
            unconfirmed_for: spec.unconfirmed_for,
        }
    }
}

/// How a workloop is armed (R2.4): `every <duration>` (cadence) and/or
/// `on <signal>` (triggered), one construct family. A loop with neither is a
/// declaration error, refused by the constructors.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(try_from = "WorkloopArmingWire", into = "WorkloopArmingWire")]
pub struct WorkloopArming {
    every: Option<Duration>,
    signals: Vec<String>,
}

impl WorkloopArming {
    /// Cadence-armed loop: the engine fires every `period` (R4.3).
    ///
    /// # Errors
    ///
    /// Refuses a zero period ([`WorkloopSpecError::ZeroCadencePeriod`]).
    pub fn every(period: Duration) -> Result<Self, WorkloopSpecError> {
        if period.is_zero() {
            return Err(WorkloopSpecError::ZeroCadencePeriod);
        }
        Ok(Self {
            every: Some(period),
            signals: Vec::new(),
        })
    }

    /// Cadence-armed loop that additionally fires on the named signals.
    ///
    /// # Errors
    ///
    /// Refuses a zero period or an empty signal name.
    pub fn every_with_signals(
        period: Duration,
        signals: Vec<String>,
    ) -> Result<Self, WorkloopSpecError> {
        if period.is_zero() {
            return Err(WorkloopSpecError::ZeroCadencePeriod);
        }
        validate_signals(&signals, false)?;
        Ok(Self {
            every: Some(period),
            signals,
        })
    }

    /// Signal-only loop: fires on signal arrival, with no cadence window for
    /// the engine to miss. The R2.4a cross-rule applies: every invariant on a
    /// signal-only loop must declare the duration-form tolerance (enforced by
    /// [`WorkloopSpec::new`]).
    ///
    /// # Errors
    ///
    /// Refuses an empty signal list or an empty signal name.
    pub fn signal_only(signals: Vec<String>) -> Result<Self, WorkloopSpecError> {
        validate_signals(&signals, true)?;
        Ok(Self {
            every: None,
            signals,
        })
    }

    /// The declared cadence period, when the loop is cadence-armed.
    #[must_use]
    pub const fn cadence_period(&self) -> Option<Duration> {
        self.every
    }

    /// The declared triggering signals (empty for a pure-cadence loop).
    #[must_use]
    pub fn signals(&self) -> &[String] {
        &self.signals
    }

    /// Whether this loop has no cadence window (signal-only arming).
    #[must_use]
    pub const fn is_signal_only(&self) -> bool {
        self.every.is_none()
    }
}

fn validate_signals(signals: &[String], require_nonempty: bool) -> Result<(), WorkloopSpecError> {
    if require_nonempty && signals.is_empty() {
        return Err(WorkloopSpecError::NoSignals);
    }
    if signals.iter().any(String::is_empty) {
        return Err(WorkloopSpecError::EmptySignalName);
    }
    Ok(())
}

/// Serde wire shape for [`WorkloopArming`]; decoding re-runs the declaration
/// checks so an unarmed loop can never be represented.
#[derive(Serialize, Deserialize, Clone, Debug)]
struct WorkloopArmingWire {
    every: Option<Duration>,
    signals: Vec<String>,
}

impl TryFrom<WorkloopArmingWire> for WorkloopArming {
    type Error = WorkloopSpecError;

    fn try_from(wire: WorkloopArmingWire) -> Result<Self, Self::Error> {
        match wire.every {
            Some(period) if wire.signals.is_empty() => Self::every(period),
            Some(period) => Self::every_with_signals(period, wire.signals),
            None => Self::signal_only(wire.signals),
        }
    }
}

impl From<WorkloopArming> for WorkloopArmingWire {
    fn from(arming: WorkloopArming) -> Self {
        Self {
            every: arming.every,
            signals: arming.signals,
        }
    }
}

/// One declared invariant (R2.1): a name, the type of its current-state
/// record, its tolerance, and the routes that confirm it (R3.3).
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct InvariantSpec {
    /// Invariant name, unique within the loop.
    pub name: String,
    /// Declared type of the invariant's current-state record (R7): the AWL
    /// type name the surface checked the record payload against. The engine is
    /// type-erased and carries the name as provenance, never as a schema.
    pub record_type: String,
    /// Declared tolerance — never defaulted (R2.3).
    pub tolerance: ToleranceSpec,
    /// Routes whose taking confirms this invariant (R3.3).
    pub confirms: Vec<String>,
}

impl InvariantSpec {
    fn validate(&self) -> Result<(), WorkloopSpecError> {
        if self.name.is_empty() {
            return Err(WorkloopSpecError::EmptyInvariantName);
        }
        if self.record_type.is_empty() {
            return Err(WorkloopSpecError::MissingRecordType {
                invariant: self.name.clone(),
            });
        }
        if self.confirms.is_empty() {
            return Err(WorkloopSpecError::NoConfirmingRoutes {
                invariant: self.name.clone(),
            });
        }
        if self.confirms.iter().any(String::is_empty) {
            return Err(WorkloopSpecError::EmptyConfirmingRoute {
                invariant: self.name.clone(),
            });
        }
        Ok(())
    }
}

/// The carry a workloop threads from one generation to the next, and the
/// DEFAULTS that seed generation 1 (R-carry).
///
/// # 🔴 WHY DEFAULTS ARE A CONTRACT AND NOT A CONVENIENCE
///
/// A workloop's iteration body reads its carry fields unconditionally — the
/// generated input codec requires them. Every generation after the first gets
/// them from the previous iteration's `route start` payload. Generation 1 has
/// no previous iteration, so unless something seeds those fields the very
/// first run of a loop fails to decode its own input: a start payload that
/// passes schema admission and is then unreadable by the workflow it was
/// admitted for.
///
/// Seeding is therefore the engine's job at the START, not the author's job
/// at every call site — an operator starting a loop cannot be expected to
/// know which fields the compiled codec will demand.
///
/// # 🔴 CALLER VALUES WIN
///
/// The merge is default-filling, never overwriting: a field the caller
/// supplied keeps the caller's value. A default that clobbered an explicit
/// start value would make the declaration silently override the operator.
#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
#[serde(try_from = "CarryContractWire", into = "CarryContractWire")]
pub struct CarryContract {
    defaults: BTreeMap<String, serde_json::Value>,
}

impl CarryContract {
    /// A loop that declares no carry: nothing to seed.
    #[must_use]
    pub fn none() -> Self {
        Self::default()
    }

    /// Builds a carry contract from declared field defaults.
    ///
    /// # Errors
    ///
    /// Refuses an empty field name ([`WorkloopSpecError::EmptyCarryField`]).
    pub fn new(defaults: BTreeMap<String, serde_json::Value>) -> Result<Self, WorkloopSpecError> {
        if defaults.keys().any(String::is_empty) {
            return Err(WorkloopSpecError::EmptyCarryField);
        }
        Ok(Self { defaults })
    }

    /// The declared field defaults.
    #[must_use]
    pub const fn defaults(&self) -> &BTreeMap<String, serde_json::Value> {
        &self.defaults
    }

    /// Whether this loop declares any carry at all.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.defaults.is_empty()
    }

    /// Merge the declared defaults into a generation-1 start payload.
    ///
    /// Returns the payload unchanged when nothing is declared. Caller-supplied
    /// fields are preserved; only ABSENT fields are filled.
    ///
    /// # Errors
    ///
    /// Refuses a start payload that is not a JSON object
    /// ([`WorkloopSpecError::CarrySeedTargetNotAnObject`]) — there is nowhere
    /// to put a named field in a scalar or an array, and silently dropping the
    /// seed would reproduce the very decode failure seeding exists to prevent.
    pub fn seed(&self, input: &crate::Payload) -> Result<crate::Payload, WorkloopSpecError> {
        if self.defaults.is_empty() {
            return Ok(input.clone());
        }
        let mut document: serde_json::Value = serde_json::from_slice(input.bytes())
            .map_err(|_| WorkloopSpecError::CarrySeedTargetNotJson)?;
        let object = document
            .as_object_mut()
            .ok_or(WorkloopSpecError::CarrySeedTargetNotAnObject)?;
        for (field, default) in &self.defaults {
            if !object.contains_key(field) {
                object.insert(field.clone(), default.clone());
            }
        }
        let bytes =
            serde_json::to_vec(&document).map_err(|_| WorkloopSpecError::CarrySeedTargetNotJson)?;
        Ok(crate::Payload::new(crate::ContentType::Json, bytes))
    }
}

/// Serde wire shape for [`CarryContract`]; decoding re-runs the field checks.
#[derive(Serialize, Deserialize, Clone, Debug)]
struct CarryContractWire {
    defaults: BTreeMap<String, serde_json::Value>,
}

impl TryFrom<CarryContractWire> for CarryContract {
    type Error = WorkloopSpecError;

    fn try_from(wire: CarryContractWire) -> Result<Self, Self::Error> {
        Self::new(wire.defaults)
    }
}

impl From<CarryContract> for CarryContractWire {
    fn from(contract: CarryContract) -> Self {
        Self {
            defaults: contract.defaults,
        }
    }
}

/// The declared shape of one workloop, as the engine takes it: arming,
/// invariants, and the retention window — all REQUIRED parameters, refused
/// when absent or degenerate (the estate rule: no assumed defaults).
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(try_from = "WorkloopSpecWire", into = "WorkloopSpecWire")]
pub struct WorkloopSpec {
    arming: WorkloopArming,
    invariants: Vec<InvariantSpec>,
    retention: Duration,
    carry: CarryContract,
}

impl WorkloopSpec {
    /// Builds a validated workloop spec.
    ///
    /// # Errors
    ///
    /// Refuses: no invariants; duplicate/empty invariant names; a missing
    /// record type or confirming route; a zero retention window; and — the
    /// R2.4a cross-rule — any invariant on a signal-only loop whose tolerance
    /// lacks the duration form.
    pub fn new(
        arming: WorkloopArming,
        invariants: Vec<InvariantSpec>,
        retention: Duration,
    ) -> Result<Self, WorkloopSpecError> {
        Self::with_carry(arming, invariants, retention, CarryContract::none())
    }

    /// [`WorkloopSpec::new`] for a loop that DECLARES carry fields, with the
    /// defaults that seed generation 1.
    ///
    /// # Errors
    ///
    /// As [`WorkloopSpec::new`], plus the carry contract's own refusals.
    pub fn with_carry(
        arming: WorkloopArming,
        invariants: Vec<InvariantSpec>,
        retention: Duration,
        carry: CarryContract,
    ) -> Result<Self, WorkloopSpecError> {
        if invariants.is_empty() {
            return Err(WorkloopSpecError::NoInvariants);
        }
        if retention.is_zero() {
            return Err(WorkloopSpecError::ZeroRetention);
        }
        // 🔴 A RETENTION WINDOW THE CLOCK CANNOT SUBTRACT IS REFUSED HERE.
        //
        // The close path derives its prune cutoff as `now - retention`. A
        // window that cannot be converted to a calendar duration has no
        // cutoff, and the only two things the close could then do are refuse
        // every park or invent a fallback — and the obvious fallback (zero)
        // points the WRONG WAY: it makes the cutoff `now` and prunes every
        // prior generation, the exact inverse of a very long retention. The
        // declaration boundary is the place to refuse it, so no loop can be
        // registered whose retention can never be applied.
        if chrono::Duration::from_std(retention).is_err() {
            return Err(WorkloopSpecError::UnrepresentableRetention {
                seconds: retention.as_secs(),
            });
        }
        let mut seen = std::collections::HashSet::new();
        for invariant in &invariants {
            invariant.validate()?;
            if !seen.insert(invariant.name.clone()) {
                return Err(WorkloopSpecError::DuplicateInvariant {
                    invariant: invariant.name.clone(),
                });
            }
            if arming.is_signal_only() && invariant.tolerance.unconfirmed_for().is_none() {
                return Err(WorkloopSpecError::SignalOnlyNeedsDurationTolerance {
                    invariant: invariant.name.clone(),
                });
            }
        }
        Ok(Self {
            arming,
            invariants,
            retention,
            carry,
        })
    }

    /// The loop's declared arming.
    #[must_use]
    pub const fn arming(&self) -> &WorkloopArming {
        &self.arming
    }

    /// The loop's declared invariants.
    #[must_use]
    pub fn invariants(&self) -> &[InvariantSpec] {
        &self.invariants
    }

    /// The loop's declared carry contract.
    #[must_use]
    pub const fn carry(&self) -> &CarryContract {
        &self.carry
    }

    /// The declared retention window for prior invariant-record generations
    /// (R8.1). Exactly one current record per invariant survives indefinitely;
    /// prior generations older than this window are pruned.
    #[must_use]
    pub const fn retention(&self) -> Duration {
        self.retention
    }
}

/// Serde wire shape for [`WorkloopSpec`]; decoding re-runs every declaration
/// check so a stored spec cannot drift out of the declared invariants.
///
/// # 🔴 EVERY FIELD IS REQUIRED, INCLUDING `carry`
///
/// `carry` carries no `#[serde(default)]`, and that absence is load-bearing.
/// A default would make an encoding that FAILED to carry the contract decode
/// as "this loop declared no carry" — indistinguishable from a loop that
/// really declared none, and silently reproducing the undecodable-generation-1
/// defect the carry contract exists to close: the seeded fields would vanish
/// and the successor generation's input would no longer satisfy the compiled
/// codec. A missing field must REFUSE the decode and name itself, so the
/// operator sees a spec that cannot be read rather than a loop that quietly
/// lost its declaration.
#[derive(Serialize, Deserialize, Clone, Debug)]
struct WorkloopSpecWire {
    arming: WorkloopArming,
    invariants: Vec<InvariantSpec>,
    retention: Duration,
    carry: CarryContract,
}

impl TryFrom<WorkloopSpecWire> for WorkloopSpec {
    type Error = WorkloopSpecError;

    fn try_from(wire: WorkloopSpecWire) -> Result<Self, Self::Error> {
        Self::with_carry(wire.arming, wire.invariants, wire.retention, wire.carry)
    }
}

impl From<WorkloopSpec> for WorkloopSpecWire {
    fn from(spec: WorkloopSpec) -> Self {
        Self {
            carry: spec.carry,
            arming: spec.arming,
            invariants: spec.invariants,
            retention: spec.retention,
        }
    }
}

/// Fixed UUID-v5 namespace for hatch dedupe identities (R13.1).
///
/// Never change this value: the derived workflow ids ARE the dedupe index —
/// re-minting the same (namespace, workflow type, key) must yield the same id
/// forever, across replays, retries, and releases.
const HATCH_IDENTITY_NAMESPACE: Uuid = Uuid::from_bytes([
    0xa1, 0x0f, 0x7a, 0x8e, 0x9d, 0x3c, 0x45, 0xf1, 0x8f, 0x2a, 0x4b, 0x6e, 0x1c, 0x9d, 0x2e, 0x73,
]);

/// Derives the deterministic workflow id for a hatch (R13.1 mandatory dedupe).
///
/// Identity = (namespace + target workflow type + key), defined ONCE and
/// identically for workflow and workloop documents, so the two kinds can never
/// drift on what "same hatch" means. The id is a UUID-v5 over NUL-separated
/// parts: an iteration retry or replay re-mints the SAME identity, so the same
/// observed subject hatches ONE workflow, never two — the double-firing poller
/// is unrepresentable, not documented.
///
/// # Errors
///
/// Refuses empty parts and parts containing the reserved NUL separator.
pub fn hatch_workflow_id(
    namespace: &str,
    workflow_type: &str,
    key: &str,
) -> Result<WorkflowId, WorkloopSpecError> {
    for part in [namespace, workflow_type, key] {
        if part.is_empty() {
            return Err(WorkloopSpecError::EmptyHatchIdentityPart);
        }
        if part.contains('\0') {
            return Err(WorkloopSpecError::HatchIdentityNulByte);
        }
    }
    let name = format!("{namespace}\0{workflow_type}\0{key}");
    Ok(WorkflowId::new(Uuid::new_v5(
        &HATCH_IDENTITY_NAMESPACE,
        name.as_bytes(),
    )))
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::time::Duration;

    use super::{
        AlarmCause, HealthSample, HealthStatus, InvariantSpec, ToleranceSpec, WorkloopArming,
        WorkloopSpec, WorkloopSpecError, hatch_workflow_id, workflow_kind_from_attributes,
    };
    use crate::SearchAttributeValue;

    fn invariant(name: &str, tolerance: ToleranceSpec) -> InvariantSpec {
        InvariantSpec {
            name: String::from(name),
            record_type: String::from("ServeState"),
            tolerance,
            confirms: vec![String::from("sweep")],
        }
    }

    fn cadence_arming() -> Result<WorkloopArming, WorkloopSpecError> {
        WorkloopArming::every(Duration::from_secs(900))
    }

    #[test]
    fn tolerance_requires_a_declared_form() -> Result<(), Box<dyn std::error::Error>> {
        let undeclared = serde_json::json!({
            "consecutive_windows": null,
            "unconfirmed_for": null,
        });
        let error = serde_json::from_value::<ToleranceSpec>(undeclared)
            .err()
            .ok_or("both-absent tolerance must refuse to decode")?;
        assert!(error.to_string().contains("at least one form"));
        Ok(())
    }

    #[test]
    fn tolerance_zero_count_is_a_legitimate_declared_value()
    -> Result<(), Box<dyn std::error::Error>> {
        let zero = ToleranceSpec::count(0);
        assert_eq!(zero.consecutive_windows(), Some(0));
        let json = serde_json::to_string(&zero)?;
        assert_eq!(serde_json::from_str::<ToleranceSpec>(&json)?, zero);
        Ok(())
    }

    #[test]
    fn tolerance_zero_duration_is_refused() {
        assert_eq!(
            ToleranceSpec::duration(Duration::ZERO),
            Err(WorkloopSpecError::ToleranceZeroDuration)
        );
        assert_eq!(
            ToleranceSpec::both(3, Duration::ZERO),
            Err(WorkloopSpecError::ToleranceZeroDuration)
        );
    }

    #[test]
    fn arming_refuses_zero_period_and_empty_signals() {
        assert_eq!(
            WorkloopArming::every(Duration::ZERO),
            Err(WorkloopSpecError::ZeroCadencePeriod)
        );
        assert_eq!(
            WorkloopArming::signal_only(Vec::new()),
            Err(WorkloopSpecError::NoSignals)
        );
        assert_eq!(
            WorkloopArming::signal_only(vec![String::new()]),
            Err(WorkloopSpecError::EmptySignalName)
        );
    }

    #[test]
    fn arming_round_trips_and_revalidates_on_decode() -> Result<(), Box<dyn std::error::Error>> {
        let arming = WorkloopArming::every_with_signals(
            Duration::from_secs(1500),
            vec![String::from("drain")],
        )?;
        let json = serde_json::to_string(&arming)?;
        assert_eq!(serde_json::from_str::<WorkloopArming>(&json)?, arming);

        let unarmed = serde_json::json!({ "every": null, "signals": [] });
        assert!(serde_json::from_value::<WorkloopArming>(unarmed).is_err());
        Ok(())
    }

    #[test]
    fn spec_requires_invariants_and_retention() -> Result<(), Box<dyn std::error::Error>> {
        assert_eq!(
            WorkloopSpec::new(cadence_arming()?, Vec::new(), Duration::from_secs(1)),
            Err(WorkloopSpecError::NoInvariants)
        );
        assert_eq!(
            WorkloopSpec::new(
                cadence_arming()?,
                vec![invariant("serving", ToleranceSpec::count(3))],
                Duration::ZERO,
            ),
            Err(WorkloopSpecError::ZeroRetention)
        );
        Ok(())
    }

    #[test]
    fn spec_refuses_duplicate_and_degenerate_invariants() -> Result<(), Box<dyn std::error::Error>>
    {
        let duplicate = WorkloopSpec::new(
            cadence_arming()?,
            vec![
                invariant("serving", ToleranceSpec::count(3)),
                invariant("serving", ToleranceSpec::count(1)),
            ],
            Duration::from_secs(86_400),
        );
        assert_eq!(
            duplicate,
            Err(WorkloopSpecError::DuplicateInvariant {
                invariant: String::from("serving")
            })
        );

        let mut nameless = invariant("serving", ToleranceSpec::count(3));
        nameless.name = String::new();
        assert_eq!(
            WorkloopSpec::new(cadence_arming()?, vec![nameless], Duration::from_secs(1)),
            Err(WorkloopSpecError::EmptyInvariantName)
        );

        let mut untyped = invariant("serving", ToleranceSpec::count(3));
        untyped.record_type = String::new();
        assert_eq!(
            WorkloopSpec::new(cadence_arming()?, vec![untyped], Duration::from_secs(1)),
            Err(WorkloopSpecError::MissingRecordType {
                invariant: String::from("serving")
            })
        );

        let mut unconfirmable = invariant("serving", ToleranceSpec::count(3));
        unconfirmable.confirms = Vec::new();
        assert_eq!(
            WorkloopSpec::new(
                cadence_arming()?,
                vec![unconfirmable],
                Duration::from_secs(1)
            ),
            Err(WorkloopSpecError::NoConfirmingRoutes {
                invariant: String::from("serving")
            })
        );
        Ok(())
    }

    #[test]
    fn signal_only_loop_requires_duration_form_tolerance() -> Result<(), Box<dyn std::error::Error>>
    {
        let arming = WorkloopArming::signal_only(vec![String::from("task_ready")])?;

        // Count-only tolerance on a signal-only loop: the R2.4a refusal.
        assert_eq!(
            WorkloopSpec::new(
                arming.clone(),
                vec![invariant("serving", ToleranceSpec::count(3))],
                Duration::from_secs(86_400),
            ),
            Err(WorkloopSpecError::SignalOnlyNeedsDurationTolerance {
                invariant: String::from("serving")
            })
        );

        // Duration form (alone or with count) is accepted: the fixture pair.
        let duration_form = ToleranceSpec::duration(Duration::from_secs(2700))?;
        WorkloopSpec::new(
            arming.clone(),
            vec![invariant("serving", duration_form)],
            Duration::from_secs(86_400),
        )?;
        let both_forms = ToleranceSpec::both(3, Duration::from_secs(2700))?;
        WorkloopSpec::new(
            arming,
            vec![invariant("serving", both_forms)],
            Duration::from_secs(86_400),
        )?;
        Ok(())
    }

    #[test]
    fn spec_round_trips_and_revalidates_on_decode() -> Result<(), Box<dyn std::error::Error>> {
        let spec = WorkloopSpec::new(
            cadence_arming()?,
            vec![invariant(
                "serving",
                ToleranceSpec::both(3, Duration::from_secs(2700))?,
            )],
            Duration::from_secs(14 * 86_400),
        )?;
        let json = serde_json::to_string(&spec)?;
        assert_eq!(serde_json::from_str::<WorkloopSpec>(&json)?, spec);
        Ok(())
    }

    /// 🔴 A SPEC THAT DID NOT CARRY ITS CARRY CONTRACT MUST REFUSE TO DECODE.
    ///
    /// The failure this guards is silent, not loud: with a `#[serde(default)]`
    /// on the field, an encoding that lost `carry` decodes as a loop that
    /// DECLARED no carry. Generation 1's seeding then fills nothing, the start
    /// payload is admitted, and the compiled codec cannot decode it — the
    /// undecodable-generation-1 defect, reproduced by an absence rather than a
    /// declaration.
    ///
    /// The control below is what makes the refusal mean something: the same
    /// object WITH the field decodes, so a blanket "this shape never decodes"
    /// cannot satisfy this test.
    #[test]
    fn a_spec_encoding_missing_its_carry_contract_refuses_to_decode()
    -> Result<(), Box<dyn std::error::Error>> {
        let spec = WorkloopSpec::new(
            cadence_arming()?,
            vec![invariant("serving", ToleranceSpec::count(3))],
            Duration::from_secs(14 * 86_400),
        )?;
        let mut encoded = serde_json::to_value(&spec)?;
        let object = encoded
            .as_object_mut()
            .ok_or("a workloop spec must encode as a JSON object")?;
        assert!(
            object.contains_key("carry"),
            "fixture control: the encoding must carry the field this test removes, or \
             removing it proves nothing: {object:?}"
        );

        // The control: unmodified, it decodes.
        assert_eq!(
            serde_json::from_value::<WorkloopSpec>(encoded.clone())?,
            spec
        );

        let object = encoded
            .as_object_mut()
            .ok_or("a workloop spec must encode as a JSON object")?;
        object.remove("carry");
        let refusal = serde_json::from_value::<WorkloopSpec>(encoded)
            .err()
            .ok_or("a spec encoding with no carry contract must not decode")?;
        assert!(
            refusal.to_string().contains("carry"),
            "the refusal must NAME the missing field so an operator knows what is absent: \
             {refusal}"
        );
        Ok(())
    }

    #[test]
    fn alarm_causes_serialize_as_kebab_case_vocabulary() -> Result<(), serde_json::Error> {
        for (cause, wire) in [
            (AlarmCause::SampleRed, "\"sample-red\""),
            (AlarmCause::WindowMissed, "\"window-missed\""),
            (AlarmCause::LoopDead, "\"loop-dead\""),
            (AlarmCause::UnconfirmedUnknown, "\"unconfirmed-unknown\""),
        ] {
            assert_eq!(serde_json::to_string(&cause)?, wire);
            assert_eq!(serde_json::from_str::<AlarmCause>(wire)?, cause);
        }
        Ok(())
    }

    #[test]
    fn health_samples_round_trip_through_json() -> Result<(), serde_json::Error> {
        for sample in [
            HealthSample {
                invariant: String::from("serving"),
                status: HealthStatus::Confirmed,
                window_seq: Some(41),
            },
            HealthSample {
                invariant: String::from("serving"),
                status: HealthStatus::Unconfirmed,
                window_seq: None,
            },
        ] {
            let json = serde_json::to_string(&sample)?;
            assert_eq!(serde_json::from_str::<HealthSample>(&json)?, sample);
        }
        Ok(())
    }

    #[test]
    fn hatch_identity_is_deterministic_and_discriminating() -> Result<(), Box<dyn std::error::Error>>
    {
        let first = hatch_workflow_id("default", "process_task", "task-42")?;
        let again = hatch_workflow_id("default", "process_task", "task-42")?;
        assert_eq!(first, again);

        // Every part discriminates.
        assert_ne!(
            first,
            hatch_workflow_id("other", "process_task", "task-42")?
        );
        assert_ne!(
            first,
            hatch_workflow_id("default", "other_task", "task-42")?
        );
        assert_ne!(
            first,
            hatch_workflow_id("default", "process_task", "task-43")?
        );

        // Concatenation ambiguity is broken by the separator.
        assert_ne!(
            hatch_workflow_id("a", "bc", "d")?,
            hatch_workflow_id("ab", "c", "d")?
        );
        Ok(())
    }

    #[test]
    fn hatch_identity_refuses_empty_and_nul_parts() {
        assert_eq!(
            hatch_workflow_id("", "process_task", "task-42"),
            Err(WorkloopSpecError::EmptyHatchIdentityPart)
        );
        assert_eq!(
            hatch_workflow_id("default", "", "task-42"),
            Err(WorkloopSpecError::EmptyHatchIdentityPart)
        );
        assert_eq!(
            hatch_workflow_id("default", "process_task", ""),
            Err(WorkloopSpecError::EmptyHatchIdentityPart)
        );
        assert_eq!(
            hatch_workflow_id("default", "process\0task", "task-42"),
            Err(WorkloopSpecError::HatchIdentityNulByte)
        );
    }

    #[test]
    fn workflow_kind_projects_from_attributes() {
        let mut attributes = HashMap::new();
        assert_eq!(workflow_kind_from_attributes(&attributes), None);
        attributes.insert(
            String::from(super::WORKFLOW_KIND_ATTRIBUTE),
            SearchAttributeValue::String(String::from(super::WORKLOOP_KIND)),
        );
        assert_eq!(
            workflow_kind_from_attributes(&attributes),
            Some(String::from("workloop"))
        );
        attributes.insert(
            String::from(super::WORKFLOW_KIND_ATTRIBUTE),
            SearchAttributeValue::Int(7),
        );
        assert_eq!(workflow_kind_from_attributes(&attributes), None);
    }
}