af_workflow/

contract.rs

//! Stable contracts for durable workflow execution.
//!
//! These types describe immutable definitions, accepted source facts and
//! external effects. Products provide capabilities; the workflow kernel owns
//! lifecycle, fencing and persistence semantics.

use af_context::{InstanceId, RunId, SubjectId, TenantId};
use std::collections::{BTreeMap, BTreeSet};

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

/// Named workflow whose revisions are immutable.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkflowDefinition {
    /// Stable identifier of this record.
    pub id: String,
    /// Display name.
    pub name: String,
}

/// One durable instance pinned to a revision and execution profile for its lifetime.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowInstance {
    /// Stable identifier of this record.
    pub id: String,
    /// Tenant that owns this record.
    pub tenant_id: TenantId,
    /// Subject (user or service principal) acting on or owning this record.
    pub subject_id: SubjectId,
    /// Workflow definition this record belongs to.
    pub definition_id: String,
    /// Monotonic revision number.
    pub revision: u64,
    /// Execution profile the instance pins.
    pub execution_profile_id: String,
    /// Pinned execution profile revision.
    pub execution_profile_revision: u64,
    /// Lifecycle policy applied on database time.
    pub lifecycle: LifecyclePolicy,
    /// Current lifecycle status.
    pub status: String,
    /// CAS version of the authoritative state.
    pub state_version: i64,
    /// Sequence of the last authoritative event.
    pub event_sequence: i64,
    /// Instance control epoch; pause/resume bump it and stale workers fail.
    pub control_epoch: i64,
}

/// Subscription of an instance to a source event type, with a JSON-containment predicate.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TriggerBinding {
    /// Stable identifier of this record.
    pub id: String,
    /// Monotonic revision number.
    pub revision: u64,
    /// Source the binding subscribes to.
    pub source: String,
    /// Stable machine-readable event type.
    pub event_type: String,
    /// Workflow instance this record refers to.
    pub instance_id: InstanceId,
    /// Target branch; omitted legacy requests resolve only for a single-branch spec.
    #[serde(default)]
    pub branch_id: Option<af_context::BranchId>,
    /// JSON object the event payload must contain (`@>`) to create a delivery.
    pub predicate: Value,
    /// How out-of-order source sequences are handled.
    pub ordering: OrderingPolicy,
    /// Earliest time the binding or instance is active.
    pub starts_at: Option<DateTime<Utc>>,
    /// When the record stops being valid.
    pub expires_at: Option<DateTime<Utc>>,
    /// How long a missing source sequence may block before it is reported as a gap.
    #[serde(default = "default_gap_wait_ms")]
    pub gap_wait_ms: u64,
    /// Maximum buffered out-of-order deliveries before the source is blocked.
    #[serde(default = "default_gap_limit")]
    pub gap_limit: u32,
}

impl TriggerBinding {
    /// Reject blank identifiers, out-of-range gap settings and inverted windows.
    pub fn validate(&self) -> Result<(), ContractError> {
        for (name, value) in [
            ("trigger binding id", self.id.as_str()),
            ("trigger source", self.source.as_str()),
            ("trigger event_type", self.event_type.as_str()),
            ("trigger instance_id", self.instance_id.as_str()),
        ] {
            required(name, value)?;
        }
        if self.gap_wait_ms == 0 || self.gap_wait_ms > 3_600_000 {
            return Err(ContractError::Invalid(
                "trigger gap_wait_ms must be between 1 and 3600000".into(),
            ));
        }
        if self.gap_limit == 0 || self.gap_limit > 10_000 {
            return Err(ContractError::Invalid(
                "trigger gap_limit must be between 1 and 10000".into(),
            ));
        }
        if !self.predicate.is_object() {
            return Err(ContractError::Invalid(
                "trigger predicate must be a JSON object".into(),
            ));
        }
        if matches!((self.starts_at, self.expires_at), (Some(start), Some(end)) if end <= start) {
            return Err(ContractError::Invalid(
                "trigger expires_at must be after starts_at".into(),
            ));
        }
        Ok(())
    }
}

const fn default_gap_wait_ms() -> u64 {
    30_000
}

const fn default_gap_limit() -> u32 {
    100
}

/// Authoritative, append-only fact recorded by a committed transition.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowEvent {
    /// Workflow instance this record refers to.
    pub instance_id: InstanceId,
    /// Position in the instance's event log.
    pub sequence: i64,
    /// Stable machine-readable event type.
    pub event_type: String,
    /// Structured payload.
    pub payload: Value,
    /// Content hash that makes the referenced artifact immutable.
    pub content_digest: String,
    /// When the event happened.
    pub occurred_at: DateTime<Utc>,
}

/// Where an execution profile is allowed to act.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ExecutionMode {
    /// In-memory dry run; no external effects.
    Simulation,
    /// Historical replay.
    Backtest,
    /// Live data, simulated effects.
    Paper,
    /// Real external effects.
    Live,
}

/// Durability the deployment must provide for an execution profile.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DurabilityGrade {
    /// Up to five minutes RPO for ordinary work.
    Standard,
    /// Committed intents synchronously preserved before dispatch; required for funds actions.
    FundsGrade,
}

/// Immutable execution profile: mode, durability grade and provider bindings an instance pins.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ExecutionProfileRevision {
    /// Stable identifier of this record.
    pub id: String,
    /// Monotonic revision number.
    pub revision: u64,
    /// Content hash that makes the referenced artifact immutable.
    pub content_digest: String,
    /// Execution mode this record was produced under.
    pub mode: ExecutionMode,
    /// Durability the deployment guarantees for this profile.
    pub durability_grade: DurabilityGrade,
    /// Provider that feeds triggers.
    pub trigger_provider: String,
    /// Provider that feeds market or reference data.
    pub data_provider: String,
    /// Clock source; `database` is the only kernel-supported value.
    pub clock_model: String,
    /// Provider that dispatches actions.
    pub action_provider: String,
    /// Model bindings by role.
    #[serde(default)]
    pub models: BTreeMap<String, Value>,
    /// Environment settings the product interprets.
    #[serde(default)]
    pub environment: Value,
    /// Product policy bundle applied to permissions and guards.
    #[serde(default)]
    pub policy_bundle: Value,
    /// Named external connections by role.
    #[serde(default)]
    pub connection_bindings: BTreeMap<String, String>,
}

impl ExecutionProfileRevision {
    /// Reject blank identifiers and provider names.
    pub fn validate(&self) -> Result<(), ContractError> {
        for (name, value) in [
            ("execution profile id", self.id.as_str()),
            (
                "execution profile content_digest",
                self.content_digest.as_str(),
            ),
            ("trigger_provider", self.trigger_provider.as_str()),
            ("data_provider", self.data_provider.as_str()),
            ("clock_model", self.clock_model.as_str()),
            ("action_provider", self.action_provider.as_str()),
        ] {
            required(name, value)?;
        }
        Ok(())
    }
}

/// Exact capability identity: id, contract version and content digest.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct CapabilityPin {
    /// Stable identifier of this record.
    pub id: String,
    /// Contract version of the capability.
    pub contract_version: String,
    /// Content hash that makes the referenced artifact immutable.
    pub content_digest: String,
}

/// Immutable, digest-locked revision of a workflow spec with its pinned capabilities.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowRevision {
    /// Workflow definition this record belongs to.
    pub definition_id: String,
    /// Monotonic revision number.
    pub revision: u64,
    /// Content hash that makes the referenced artifact immutable.
    pub content_digest: String,
    /// Kernel ABI the revision was validated against.
    pub kernel_abi_version: String,
    /// Digest over the pinned capabilities.
    pub dependency_set_digest: String,
    /// Versions of generic expressions used by the spec.
    #[serde(default)]
    pub expression_versions: BTreeMap<String, String>,
    /// Capabilities the spec may dispatch.
    #[serde(default)]
    pub capabilities: Vec<CapabilityPin>,
    /// Where the spec came from (template, draft command, author).
    #[serde(default)]
    pub template_provenance: Value,
    /// The validated spec.
    pub spec: crate::Spec,
}

impl WorkflowRevision {
    /// Check this contract's invariants; returns the first violation.
    pub fn validate(&self) -> Result<(), ContractError> {
        required("definition_id", &self.definition_id)?;
        required("content_digest", &self.content_digest)?;
        required("kernel_abi_version", &self.kernel_abi_version)?;
        required("dependency_set_digest", &self.dependency_set_digest)?;
        self.spec
            .validate_structure()
            .map_err(|error| ContractError::Invalid(error.to_string()))?;
        let unique = self
            .capabilities
            .iter()
            .map(|pin| &pin.id)
            .collect::<BTreeSet<_>>();
        if unique.len() != self.capabilities.len() {
            return Err(ContractError::Invalid(
                "capability pins must be unique by id within one revision".into(),
            ));
        }
        for pin in &self.capabilities {
            required("capability id", &pin.id)?;
            required("capability contract_version", &pin.contract_version)?;
            required("capability content_digest", &pin.content_digest)?;
        }
        Ok(())
    }
}

/// Role a capability plays in a graph.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CapabilityKind {
    /// Produces events.
    Trigger,
    /// Pure or read-only transform.
    Expression,
    /// Authorization, freshness, reservation or policy check dominating an action.
    Guard,
    /// External effect dispatched through an intent.
    Action,
    /// Terminal consumer.
    Sink,
}

/// Side-effect class of a capability.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Effect {
    /// No I/O.
    Pure,
    /// Reads external state.
    Read,
    /// Writes kernel-owned state only.
    InternalWrite,
    /// Writes external state; requires authorization dominance.
    ExternalWrite,
    /// Moves value; requires authorization, freshness, reservation and funds-grade durability.
    Funds,
}

/// Whether repeating a dispatch is safe.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum IdempotencyMode {
    /// Not idempotent; never retried automatically.
    None,
    /// The provider deduplicates by idempotency key; safe to retry.
    Native,
    /// Must reconcile the previous attempt before dispatching again.
    ReconcileBeforeRetry,
    /// Retry only through an explicit operator action.
    NeverAutomaticRetry,
}

/// Deployment state of a capability provider.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CapabilityLifecycle {
    /// Registered but not yet serving.
    Installed,
    /// Serving new and existing work.
    Active,
    /// Serving pinned work only; new pins rejected.
    Deprecated,
    /// Not serving; pinned work waits.
    Disabled,
    /// Temporarily unreachable.
    Unavailable,
    /// Blocks every action immediately.
    EmergencyRevoked,
}

/// Automatic retry policy of a capability.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RetryPolicy {
    /// Total dispatch attempts allowed.
    pub max_attempts: u32,
    /// Per-attempt deadline.
    pub timeout_ms: u64,
    /// Backoff after the first failure.
    pub initial_backoff_ms: u64,
    /// Cap on exponential backoff.
    pub max_backoff_ms: u64,
}

impl RetryPolicy {
    /// Exponential backoff for the 1-based `attempt`, capped at `max_backoff_ms`
    /// with up to 25% deterministic jitter derived from `jitter_seed`.
    pub fn backoff_ms(&self, attempt: u32, jitter_seed: u64) -> u64 {
        let factor = 1_u64
            .checked_shl(attempt.saturating_sub(1).min(20))
            .unwrap_or(u64::MAX);
        let base = self
            .initial_backoff_ms
            .saturating_mul(factor)
            .min(self.max_backoff_ms);
        let jitter_ceiling = (base / 4).max(1);
        base.saturating_add(jitter_seed % jitter_ceiling)
            .min(self.max_backoff_ms)
    }
}

/// Immutable contract of a trigger, expression, guard, action or sink.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CapabilityManifest {
    /// Stable identifier of this record.
    pub id: String,
    /// Contract version of the capability.
    pub contract_version: String,
    /// Content hash that makes the referenced artifact immutable.
    pub content_digest: String,
    /// Discriminator naming the variant of this record.
    pub kind: CapabilityKind,
    /// JSON Schema for node configuration exposed to authoring clients.
    #[serde(default = "object_schema")]
    pub config_schema: Value,
    /// JSON Schema for provider input.
    pub input_schema: Value,
    /// JSON Schema for provider output.
    pub output_schema: Value,
    /// Side-effect class of the action.
    pub effect: Effect,
    /// Whether equal inputs always produce equal outputs.
    pub deterministic: bool,
    /// Retry safety class.
    pub idempotency_mode: IdempotencyMode,
    /// Retry policy.
    pub retry: RetryPolicy,
    /// Permissions granted or required.
    #[serde(default)]
    pub permissions: BTreeSet<String>,
    /// Guards that must dominate the action.
    #[serde(default)]
    pub required_guards: BTreeSet<GuardKind>,
    /// Which inputs may be LLM-tainted.
    #[serde(default)]
    pub taint_rules: Value,
    /// Declared cost for budgeting.
    #[serde(default)]
    pub resource_cost: Value,
    /// Usable in simulation.
    pub supports_simulation: bool,
    /// Usable in backtest replay.
    pub supports_replay: bool,
    /// Usable in paper mode.
    pub supports_paper: bool,
    /// Usable live.
    pub supports_live: bool,
    /// Clock guarantees the provider needs.
    #[serde(default)]
    pub clock_requirements: Value,
    /// Data freshness the provider needs.
    #[serde(default)]
    pub data_requirements: Value,
    /// Whether the provider can observe a dispatched action's outcome.
    pub supports_reconciliation: bool,
    /// Deployment state.
    pub lifecycle: CapabilityLifecycle,
}

fn object_schema() -> Value {
    serde_json::json!({"type":"object"})
}

impl CapabilityManifest {
    /// Manifest for an action capability with conservative defaults (one attempt, 30 s timeout).
    pub fn action(
        id: impl Into<String>,
        contract_version: impl Into<String>,
        content_digest: impl Into<String>,
        effect: Effect,
        idempotency_mode: IdempotencyMode,
        supports_reconciliation: bool,
    ) -> Self {
        let required_guards = match effect {
            Effect::Funds => BTreeSet::from([
                GuardKind::Authorization,
                GuardKind::Freshness,
                GuardKind::Reservation,
            ]),
            Effect::ExternalWrite => BTreeSet::from([GuardKind::Authorization]),
            _ => BTreeSet::new(),
        };
        Self {
            id: id.into(),
            contract_version: contract_version.into(),
            content_digest: content_digest.into(),
            kind: CapabilityKind::Action,
            config_schema: object_schema(),
            input_schema: serde_json::json!({"type": "object"}),
            output_schema: serde_json::json!({"type": "object"}),
            effect,
            deterministic: false,
            idempotency_mode,
            retry: RetryPolicy {
                max_attempts: 1,
                timeout_ms: 30_000,
                initial_backoff_ms: 100,
                max_backoff_ms: 5_000,
            },
            permissions: BTreeSet::new(),
            required_guards,
            taint_rules: Value::Null,
            resource_cost: Value::Null,
            supports_simulation: true,
            supports_replay: false,
            supports_paper: true,
            supports_live: true,
            clock_requirements: Value::Null,
            data_requirements: Value::Null,
            supports_reconciliation,
            lifecycle: CapabilityLifecycle::Active,
        }
    }

    /// Check this contract's invariants; returns the first violation.
    pub fn validate(&self) -> Result<(), ContractError> {
        required("capability id", &self.id)?;
        required("contract_version", &self.contract_version)?;
        required("content_digest", &self.content_digest)?;
        if !self.config_schema.is_object()
            || jsonschema::validator_for(&self.config_schema).is_err()
        {
            return Err(ContractError::Invalid(
                "capability config_schema must be a valid JSON Schema object".into(),
            ));
        }
        if self.retry.max_attempts == 0
            || self.retry.timeout_ms == 0
            || self.retry.initial_backoff_ms == 0
            || self.retry.max_backoff_ms < self.retry.initial_backoff_ms
        {
            return Err(ContractError::Invalid(
                "retry attempts, timeout and backoff bounds are invalid".into(),
            ));
        }
        if self.effect == Effect::Funds {
            if self.kind != CapabilityKind::Action {
                return Err(ContractError::Invalid(
                    "funds effect is only valid for action capabilities".into(),
                ));
            }
            if !self.supports_reconciliation && self.idempotency_mode != IdempotencyMode::Native {
                return Err(ContractError::Invalid(
                    "funds actions require native idempotency or reconciliation".into(),
                ));
            }
            for guard in [
                GuardKind::Authorization,
                GuardKind::Freshness,
                GuardKind::Reservation,
            ] {
                if !self.required_guards.contains(&guard) {
                    return Err(ContractError::Invalid(format!(
                        "funds action must require {guard:?} guard"
                    )));
                }
            }
        }
        Ok(())
    }

    /// Whether new intents may pin this provider (`installed` or `active`).
    pub fn can_start_new_work(&self) -> bool {
        matches!(
            self.lifecycle,
            CapabilityLifecycle::Installed
                | CapabilityLifecycle::Active
                | CapabilityLifecycle::Deprecated
        )
    }
}

/// Role a guard plays on an action's dominating path.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum GuardKind {
    /// Caller may perform the effect.
    Authorization,
    /// Inputs are recent enough.
    Freshness,
    /// Resources are reserved and fenced.
    Reservation,
    /// Product policy allows it.
    Policy,
}

/// How a binding treats source sequence order.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OrderingPolicy {
    /// Reject a gap; the next accepted sequence must be previous + 1.
    StrictSequence,
    /// Order does not matter.
    Commutative,
    /// Only the latest event matters.
    LatestStateReconcile,
    /// Reject anything older than the last accepted sequence.
    RejectUnordered,
}

/// Immutable source event as delivered by a trigger adapter.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TriggerEnvelope {
    /// Source-unique event id.
    pub event_id: String,
    /// Stable machine-readable event type.
    pub event_type: String,
    /// Trigger adapter that delivered the event.
    pub source: String,
    /// Projection schema version.
    pub schema_version: String,
    /// Tenant that owns this record.
    pub tenant_id: TenantId,
    /// Subject (user or service principal) acting on or owning this record.
    pub subject_id: SubjectId,
    /// Aggregate the event belongs to (for ordering).
    pub aggregate_id: String,
    /// Per-aggregate sequence from the source.
    pub source_sequence: Option<i64>,
    /// Version of the observed object, if any.
    pub observed_version: Option<String>,
    /// When the event happened.
    pub occurred_at: DateTime<Utc>,
    /// When the receipt arrived.
    pub received_at: DateTime<Utc>,
    /// Source watermark up to which events are complete.
    pub watermark: Option<DateTime<Utc>>,
    /// Key that correlates related events.
    pub correlation_key: String,
    /// Key that deduplicates redeliveries.
    pub dedup_key: String,
    /// Source cursor after this event.
    pub cursor: Option<String>,
    /// Structured payload.
    pub payload: Value,
    /// Distributed tracing context.
    #[serde(default)]
    pub trace_context: Value,
}

impl TriggerEnvelope {
    /// Check this contract's invariants; returns the first violation.
    pub fn validate(&self) -> Result<(), ContractError> {
        for (name, value) in [
            ("event_id", self.event_id.as_str()),
            ("event_type", self.event_type.as_str()),
            ("source", self.source.as_str()),
            ("schema_version", self.schema_version.as_str()),
            ("tenant_id", self.tenant_id.as_str()),
            ("aggregate_id", self.aggregate_id.as_str()),
            ("correlation_key", self.correlation_key.as_str()),
            ("dedup_key", self.dedup_key.as_str()),
        ] {
            required(name, value)?;
        }
        if self
            .cursor
            .as_ref()
            .is_some_and(|cursor| cursor.is_empty() || cursor.len() > 4096)
        {
            return Err(ContractError::Invalid(
                "source cursor requires 1..4096 bytes".into(),
            ));
        }
        if self.occurred_at > self.received_at + chrono::Duration::minutes(5) {
            return Err(ContractError::Invalid(
                "occurred_at is implausibly ahead of received_at".into(),
            ));
        }
        Ok(())
    }
}

/// Cursor to resume an admitted source adapter after restart.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SourceAdapterState {
    /// Exact source.
    pub source: af_context::WorkflowSourceId,
    /// Opaque provider cursor.
    pub cursor: String,
    /// Event whose transaction last advanced the cursor; absent for an initial seed.
    pub source_event_id: Option<String>,
}
/// Receipt returned only after source fact, binding snapshot, deliveries and cursor commit.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SourceIngestionReceipt {
    /// Durable event identity safe for adapter ACK.
    pub event_id: String,
    /// True for an identical committed retry.
    pub duplicate: bool,
    /// Newly inserted deliveries; duplicate retries report zero.
    pub deliveries: u64,
    /// Current committed resume cursor, including one advanced after this event.
    pub committed_cursor: Option<String>,
}

/// When an instance completes on its own.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CompletionPolicy {
    /// Only an operator stops it.
    ExplicitStop,
    /// After the first consumed trigger or timer.
    FirstTrigger,
    /// After the first evaluation that matched.
    FirstMatch,
    /// After the first action reaches a terminal state.
    FirstActionTerminal,
    /// After the first successful evaluation.
    FirstSuccess,
    /// After this many matched evaluations.
    AfterMatchedEvaluations(u64),
    /// After this many successful evaluations.
    AfterSuccessfulRuns(u64),
}

/// What happens when a lifecycle deadline passes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ExpiryPolicy {
    /// Stop new work; let dispatched actions finish.
    Drain,
    /// Stop and request cancellation.
    Cancel,
}

/// How a schedule fires.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ScheduleCadence {
    /// Cron expression evaluated in the schedule's timezone.
    Cron {
        /// Six- or seven-field cron expression.
        expression: String,
    },
    /// Fixed period measured from the scheduled time.
    FixedRate {
        /// Period in milliseconds.
        milliseconds: u64,
    },
    /// Fixed delay measured from the previous completion.
    FixedDelay {
        /// Delay in milliseconds.
        milliseconds: u64,
    },
}

/// How missed schedule windows are handled.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CatchUpPolicy {
    /// Run only the most recent due occurrence.
    Skip,
    /// Run once, listing the missed windows it covers.
    CatchUpOnce,
    /// Run every missed occurrence oldest first, at most `limit` per transition.
    CatchUpAll {
        /// Maximum occurrences per transition.
        limit: u32,
    },
}

/// Cadence, timezone and catch-up behavior of a schedule.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SchedulePolicy {
    /// Cron, fixed-rate or fixed-delay.
    pub cadence: ScheduleCadence,
    /// IANA timezone for cron evaluation and DST.
    pub timezone: String,
    /// Missed-window behavior.
    pub catch_up: CatchUpPolicy,
}

impl SchedulePolicy {
    /// Check this contract's invariants; returns the first violation.
    pub fn validate(&self) -> Result<(), ContractError> {
        self.timezone.parse::<chrono_tz::Tz>().map_err(|_| {
            ContractError::Invalid(format!("unknown IANA timezone '{}'", self.timezone))
        })?;
        match &self.cadence {
            ScheduleCadence::Cron { expression } => {
                expression.parse::<cron::Schedule>().map_err(|error| {
                    ContractError::Invalid(format!("invalid cron '{expression}': {error}"))
                })?;
            }
            ScheduleCadence::FixedRate { milliseconds }
            | ScheduleCadence::FixedDelay { milliseconds }
                if *milliseconds == 0 =>
            {
                return Err(ContractError::Invalid(
                    "schedule interval must be positive".into(),
                ));
            }
            _ => {}
        }
        if matches!(self.catch_up, CatchUpPolicy::CatchUpAll { limit: 0 }) {
            return Err(ContractError::Invalid(
                "catch_up_all limit must be positive".into(),
            ));
        }
        Ok(())
    }
}

/// Start, completion, timeout and expiry rules evaluated on database time.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct LifecyclePolicy {
    /// Earliest time the binding or instance is active.
    pub starts_at: Option<DateTime<Utc>>,
    /// When the record stops being valid.
    pub expires_at: Option<DateTime<Utc>>,
    /// When the instance completes.
    pub completion: CompletionPolicy,
    /// Terminalize after this long without a consumed event.
    pub event_idle_timeout_ms: Option<u64>,
    /// Terminalize after this long without a committed transition.
    pub progress_timeout_ms: Option<u64>,
    /// Drain or cancel on expiry.
    pub on_expiry: ExpiryPolicy,
    /// Hard stop for draining.
    pub drain_deadline: Option<DateTime<Utc>>,
}

impl LifecyclePolicy {
    /// One-shot lifecycle: completes on the first success, drains on expiry.
    pub fn run_once() -> Self {
        Self {
            starts_at: None,
            expires_at: None,
            completion: CompletionPolicy::FirstSuccess,
            event_idle_timeout_ms: None,
            progress_timeout_ms: None,
            on_expiry: ExpiryPolicy::Drain,
            drain_deadline: None,
        }
    }

    /// Check this contract's invariants; returns the first violation.
    pub fn validate(&self) -> Result<(), ContractError> {
        if self
            .starts_at
            .zip(self.expires_at)
            .is_some_and(|(a, b)| a >= b)
        {
            return Err(ContractError::Invalid(
                "lifecycle starts_at must be before expires_at".into(),
            ));
        }
        if self.drain_deadline.is_some() && self.on_expiry != ExpiryPolicy::Drain {
            return Err(ContractError::Invalid(
                "drain_deadline requires on_expiry=drain".into(),
            ));
        }
        if self
            .expires_at
            .zip(self.drain_deadline)
            .is_some_and(|(expires, drain)| drain <= expires)
        {
            return Err(ContractError::Invalid(
                "lifecycle drain_deadline must be after expires_at".into(),
            ));
        }
        if self.event_idle_timeout_ms == Some(0) || self.progress_timeout_ms == Some(0) {
            return Err(ContractError::Invalid(
                "lifecycle timeouts must be positive".into(),
            ));
        }
        if matches!(
            self.completion,
            CompletionPolicy::AfterMatchedEvaluations(0) | CompletionPolicy::AfterSuccessfulRuns(0)
        ) {
            return Err(ContractError::Invalid(
                "lifecycle completion count must be positive".into(),
            ));
        }
        Ok(())
    }
}

/// How a step ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StepOutcomeKind {
    /// Terminal success.
    Succeeded,
    /// Filtered out.
    Skipped,
    /// Parked on a wake condition.
    Waiting,
    /// Errored.
    Failed,
    /// Cancelled.
    Cancelled,
}

/// Recorded outcome of one step.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct StepOutcome {
    /// Discriminator naming the variant of this record.
    pub kind: StepOutcomeKind,
    /// Stable code for material outcomes.
    pub reason_code: Option<String>,
    /// What resumes a waiting step.
    #[serde(default)]
    pub wake_condition: Value,
}

/// Lifecycle of an action intent; `dispatch_committed` is the point of no return.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ActionState {
    /// Created by a transition; not yet authorized.
    Prepared,
    /// Waiting for a human confirmation.
    AwaitingConfirmation,
    /// Guards passed; may be dispatched.
    Authorized,
    /// Effect committed; cancellation can no longer claim it was not performed.
    DispatchCommitted,
    /// Provider reports progress.
    Executing,
    /// Terminal success.
    Succeeded,
    /// Terminal rejection.
    Rejected,
    /// Failed; waits for its backoff, then dispatches again.
    Retryable,
    /// Outcome unknown; waits for reconciliation.
    Unknown,
    /// Outcome established by reconciliation.
    Reconciled,
}

impl ActionState {
    /// Whether the state machine allows `self -> next`.
    pub fn can_transition_to(self, next: Self) -> bool {
        use ActionState::*;
        matches!(
            (self, next),
            (Prepared, AwaitingConfirmation | Authorized | Rejected)
                | (AwaitingConfirmation, Authorized | Rejected)
                | (Authorized, DispatchCommitted | Rejected)
                | (
                    DispatchCommitted,
                    Executing | Succeeded | Rejected | Retryable | Unknown
                )
                | (Executing, Succeeded | Rejected | Retryable | Unknown)
                | (Retryable, DispatchCommitted | Reconciled)
                | (Unknown, Reconciled)
                | (Succeeded | Rejected, Reconciled)
        )
    }

    /// Whether the effect may already have happened.
    pub fn dispatch_committed(self) -> bool {
        matches!(
            self,
            Self::DispatchCommitted
                | Self::Executing
                | Self::Succeeded
                | Self::Rejected
                | Self::Retryable
                | Self::Unknown
                | Self::Reconciled
        )
    }
}

/// Control epochs an intent was created under; any bump makes it stale.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ControlEpochs {
    /// Tenant control epoch.
    pub tenant: i64,
    /// Resource control epoch.
    pub resource: i64,
    /// Instance control epoch.
    pub instance: i64,
}

/// Scope of a control command.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ControlScope {
    /// Whole tenant.
    Tenant,
    /// One external resource.
    Resource,
    /// One instance.
    Instance,
}

/// Operating mode set by a control command.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ControlMode {
    /// Normal operation.
    Running,
    /// No new evaluations or dispatches.
    Paused,
    /// Terminal stop.
    Stopped,
    /// Immediate stop that also revokes providers.
    EmergencyStopped,
}

/// Operator command that bumps a control epoch.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ControlCommand {
    /// Scope kind.
    pub scope: ControlScope,
    /// Scope identity.
    pub scope_id: String,
    /// Execution mode this record was produced under.
    pub mode: ControlMode,
    /// Operator issuing the command.
    pub operator_subject_id: SubjectId,
    /// Human-readable reason.
    pub reason: String,
    /// When an override lapses.
    pub override_expires_at: Option<DateTime<Utc>>,
}

impl ControlCommand {
    /// Check this contract's invariants; returns the first violation.
    pub fn validate(&self) -> Result<(), ContractError> {
        required("control scope_id", &self.scope_id)?;
        required("control operator_subject_id", &self.operator_subject_id)?;
        required("control reason", &self.reason)?;
        Ok(())
    }
}

/// Reference to a product-owned fenced reservation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ResourceReservationRef {
    /// Reservation identity.
    pub reservation_id: String,
    /// Fence the reservation was taken under.
    pub fencing_token: i64,
}

/// Durable intent to perform one external effect, created inside a fenced transition.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ActionIntent {
    /// Stable identifier of this record.
    pub id: String,
    /// Tenant that owns this record.
    pub tenant_id: TenantId,
    /// Workflow instance this record refers to.
    pub instance_id: InstanceId,
    /// Run this record belongs to.
    pub run_id: RunId,
    /// Pinned capability (id, contract version, content digest).
    pub capability: CapabilityPin,
    /// Caller-supplied key that makes repeated submissions return the first result.
    pub idempotency_key: String,
    /// Current lifecycle state.
    pub state: ActionState,
    /// Structured input handed to the provider.
    pub input: Value,
    /// Side-effect class of the action.
    pub effect: Effect,
    /// Idempotency class that decides whether automatic retry is safe.
    pub retry_class: IdempotencyMode,
    /// Control epochs at creation.
    pub control_epochs: ControlEpochs,
    /// Resource the effect targets; empty when unscoped.
    pub resource_scope_id: String,
    /// Instance lease version at creation.
    pub lease_epoch: i64,
    /// Bumped on every claim; fences claim owners.
    pub action_epoch: i64,
    /// Latest time by which the work must finish.
    pub deadline: Option<DateTime<Utc>>,
    /// Fenced reservation for funds actions.
    pub reservation: Option<ResourceReservationRef>,
    /// When the record was created (database time).
    pub created_at: DateTime<Utc>,
}

impl ActionIntent {
    /// Check this contract's invariants; returns the first violation.
    pub fn validate(&self) -> Result<(), ContractError> {
        required("action id", &self.id)?;
        required("action idempotency_key", &self.idempotency_key)?;
        if self.effect == Effect::Funds && self.reservation.is_none() {
            return Err(ContractError::Invalid(
                "funds action requires a resource reservation".into(),
            ));
        }
        if self.effect == Effect::Funds && self.resource_scope_id.trim().is_empty() {
            return Err(ContractError::Invalid(
                "funds action requires a resource scope".into(),
            ));
        }
        if self.effect == Effect::Funds && self.retry_class == IdempotencyMode::None {
            return Err(ContractError::Invalid(
                "funds action requires an explicit retry class".into(),
            ));
        }
        if self
            .deadline
            .is_some_and(|deadline| deadline <= self.created_at)
        {
            return Err(ContractError::Invalid(
                "action deadline must be after creation".into(),
            ));
        }
        Ok(())
    }

    /// Validate an intent at the only creation boundary. Later states are
    /// reached exclusively through the fenced store transition.
    pub fn validate_prepared(&self) -> Result<(), ContractError> {
        self.validate()?;
        if self.state != ActionState::Prepared {
            return Err(ContractError::Invalid(
                "new action intent must start in prepared state".into(),
            ));
        }
        Ok(())
    }
}

/// Provider's answer to a dispatch.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ActionReceipt {
    /// Stable identifier of this record.
    pub id: String,
    /// Intent this refers to.
    pub action_intent_id: String,
    /// Provider version that produced it.
    pub provider_version: String,
    /// When the receipt arrived.
    pub received_at: DateTime<Utc>,
    /// State the provider reports.
    pub outcome: ActionState,
    /// Structured payload.
    pub payload: Value,
    /// Digest of the raw provider response; deduplicates receipts.
    pub raw_receipt_digest: String,
}

/// Observed state of a dispatched action.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ActionObservation {
    /// Stable identifier of this record.
    pub id: String,
    /// Intent this refers to.
    pub action_intent_id: String,
    /// Provider version that produced it.
    pub provider_version: String,
    /// When the state was observed.
    pub observed_at: DateTime<Utc>,
    /// Current lifecycle state.
    pub state: String,
    /// External resource created or affected.
    pub resource_ref: Option<Value>,
    /// Digest of the raw provider response; deduplicates receipts.
    pub raw_receipt_digest: String,
    /// Whether this observation ends the action.
    pub terminal: bool,
    /// A reconciliation provider sets this only after proving that repeating
    /// the same idempotent action is safe.
    #[serde(default)]
    pub retry_authorized: bool,
}

/// A missing source-sequence range that blocks ordered delivery for an aggregate.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SourceGap {
    /// Tenant that owns this record.
    pub tenant_id: TenantId,
    /// Source the gap belongs to.
    pub source: String,
    /// Aggregate whose sequence has the hole.
    pub aggregate_id: String,
    /// First missing sequence (inclusive).
    pub missing_from: i64,
    /// Last missing sequence (inclusive).
    pub missing_to: i64,
    /// When waiting for the gap expires and delivery blocks.
    pub deadline: DateTime<Utc>,
    /// Gap status (`waiting`, `blocked`, `unblocked`).
    pub status: String,
}

/// Scope of a resource reservation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReservationScope {
    /// Whole tenant.
    Tenant,
    /// One external resource.
    Resource,
}

/// Lifecycle of a reservation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReservationState {
    /// Held.
    Reserved,
    /// Used by a committed dispatch.
    Consumed,
    /// Given back.
    Released,
    /// Lapsed unused.
    Expired,
}

/// Mirror of a product-owned fenced reservation.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ResourceReservation {
    /// Stable identifier of this record.
    pub id: String,
    /// Tenant that owns this record.
    pub tenant_id: TenantId,
    /// Scope kind.
    pub scope: ReservationScope,
    /// Scope identity.
    pub scope_id: String,
    /// Kind of resource reserved.
    pub resource_kind: String,
    /// Reserved amount as a decimal string.
    pub amount: String,
    /// Product policy version that granted it.
    pub policy_version: String,
    /// When the record stops being valid.
    pub expires_at: DateTime<Utc>,
    /// Fence the reservation was taken under.
    pub fencing_token: i64,
    /// Provider-side reservation id.
    pub provider_reservation_id: Option<String>,
    /// Current lifecycle state.
    pub state: ReservationState,
}

/// A value with provenance and freshness metadata.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ObservedValue<T> {
    /// The value carried by this record.
    pub value: T,
    /// Where the value was observed.
    pub source: String,
    /// When the state was observed.
    pub observed_at: DateTime<Utc>,
    /// When the receipt arrived.
    pub received_at: DateTime<Utc>,
    /// Version of the source.
    pub source_version: String,
    /// Quality label from the source.
    pub quality: String,
    /// Digest of the value.
    pub digest: String,
}

impl<T> ObservedValue<T> {
    /// Whether the value is fresh enough for `max_age` at `now`.
    pub fn is_accepted(
        &self,
        now: DateTime<Utc>,
        max_age: chrono::Duration,
        accepted_quality: &BTreeSet<String>,
    ) -> bool {
        self.observed_at <= now
            && now - self.observed_at <= max_age
            && accepted_quality.contains(&self.quality)
    }
}

/// Everything a decision depended on, for replay and audit.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct DecisionSnapshot {
    /// Digests of the inputs.
    pub input_digests: Vec<String>,
    /// Config version.
    pub instance_config_version: String,
    /// Product policy version that granted it.
    pub policy_version: String,
    /// Capabilities in force.
    pub capability_versions: Vec<CapabilityPin>,
    /// Execution profile revision in force.
    pub execution_profile_revision_id: String,
    /// Context values used.
    #[serde(default)]
    pub context_snapshot: Value,
    /// Policy values used.
    #[serde(default)]
    pub policy_snapshot: Value,
    /// Which Agent artifacts influenced the decision.
    #[serde(default)]
    pub agent_provenance: Value,
}

/// Who supplied a parameter value.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ParameterProvenance {
    /// Explicitly supplied by the user.
    UserSupplied,
    /// Product default.
    ProductDefault,
    /// Inferred by an Agent.
    AgentInferred,
    /// Computed from other parameters.
    Derived,
}

/// A parameter with its provenance.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ParameterValue {
    /// The value carried by this record.
    pub value: Value,
    /// Who supplied it.
    pub provenance: ParameterProvenance,
}

/// Declared parameter and its live-mode requirements.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ParameterSpec {
    /// Display name.
    pub name: String,
    /// Must be present.
    pub required: bool,
    /// Must be user-supplied in live mode.
    pub required_explicit_for_live: bool,
}

/// Check supplied parameters against their specs for `mode`.
pub fn validate_parameters(
    mode: ExecutionMode,
    specs: &[ParameterSpec],
    values: &BTreeMap<String, ParameterValue>,
) -> Result<(), MissingRequirements> {
    let missing = specs
        .iter()
        .filter(|spec| match values.get(&spec.name) {
            None => {
                spec.required || (mode == ExecutionMode::Live && spec.required_explicit_for_live)
            }
            Some(value) => {
                mode == ExecutionMode::Live
                    && spec.required_explicit_for_live
                    && value.provenance != ParameterProvenance::UserSupplied
            }
        })
        .map(|spec| spec.name.clone())
        .collect::<Vec<_>>();
    if missing.is_empty() {
        Ok(())
    } else {
        Err(MissingRequirements {
            parameters: missing,
        })
    }
}

/// Parameters that block a live run.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
#[error("missing explicit workflow requirements: {parameters:?}")]
pub struct MissingRequirements {
    /// Missing or non-explicit parameter names.
    pub parameters: Vec<String>,
}

/// Budget reserved atomically before a root operation starts children.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RootOperationBudget {
    /// Root operation identity.
    pub root_operation_id: String,
    /// Maximum nesting depth.
    pub max_depth: u32,
    /// Maximum descendants.
    pub descendant_limit: u32,
    /// Maximum runs.
    pub run_limit: u32,
    /// Token budget.
    pub token_budget: u64,
    /// Cost budget in micro-units.
    pub cost_budget_micros: u64,
    /// Maximum actions.
    pub action_budget: u32,
    /// Latest time by which the work must finish.
    pub deadline: DateTime<Utc>,
    /// Permissions children may not exceed.
    #[serde(default)]
    pub permission_ceiling: BTreeSet<String>,
}

/// Persisted Agent decision reused after recovery instead of re-asking the model.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct DecisionArtifact {
    /// Stable identifier of this record.
    pub id: String,
    /// Root operation identity.
    pub root_operation_id: String,
    /// Content hash that makes the referenced artifact immutable.
    pub content_digest: String,
    /// Model identifier as registered in the model registry.
    pub model: String,
    /// Digest of the prompt.
    pub prompt_digest: String,
    /// Model output.
    pub output: Value,
    /// When the record was created (database time).
    pub created_at: DateTime<Utc>,
}

/// Operator-facing diagnostic attached to an instance.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct DiagnosticRecord {
    /// Stable diagnostic code.
    pub code: String,
    /// Human-readable message.
    pub message: String,
    /// When it was recorded.
    pub at: DateTime<Utc>,
    /// Structured detail.
    #[serde(default)]
    pub detail: Value,
}

/// Rebuildable read model of an instance; never authorizes or schedules work.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ExecutionProjection {
    /// Projection schema version.
    pub schema_version: String,
    /// Authoritative event sequence the projection was built from.
    pub source_sequence: i64,
    /// Steps in progress.
    pub current_steps: Vec<String>,
    /// Steps completed.
    pub completed_steps: Vec<String>,
    /// What the instance waits for.
    pub waiting_on: Option<Value>,
    /// Most recent decision.
    pub last_decision: Option<Value>,
    /// Steps that may run next.
    pub next_possible_steps: Vec<String>,
    /// Next scheduled evaluation.
    pub next_trigger_at: Option<DateTime<Utc>>,
    /// Actions prepared but not dispatched.
    pub planned_actions: Vec<String>,
    /// Most recent diagnostic.
    pub latest_diagnostic: Option<DiagnosticRecord>,
    /// Product-defined progress.
    pub progress: Value,
    /// When the record stops being valid.
    pub expires_at: Option<DateTime<Utc>>,
}

/// A contract value violates its invariants.
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum ContractError {
    /// Invalid workflow contract.
    #[error("invalid workflow contract: {0}")]
    Invalid(String),
}

fn required(name: &str, value: &str) -> Result<(), ContractError> {
    if value.trim().is_empty() {
        Err(ContractError::Invalid(format!("{name} is required")))
    } else {
        Ok(())
    }
}

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

    fn empty_spec() -> crate::Spec {
        crate::Spec {
            spec_id: "test".into(),
            version: "1".into(),
            description: String::new(),
            aliases: vec![],
            instance_config_schema: None,
            display: BTreeMap::new(),
            branches: vec![],
        }
    }

    fn funds_manifest() -> CapabilityManifest {
        CapabilityManifest::action(
            "action.example",
            "1",
            "sha256:x",
            Effect::Funds,
            IdempotencyMode::ReconcileBeforeRetry,
            true,
        )
    }

    #[test]
    fn funds_capability_requires_reconciliation_and_all_guards() {
        assert!(funds_manifest().validate().is_ok());
        let mut invalid = funds_manifest();
        invalid.required_guards.remove(&GuardKind::Freshness);
        assert!(invalid.validate().is_err());
        invalid.required_guards.insert(GuardKind::Freshness);
        invalid.supports_reconciliation = false;
        assert!(invalid.validate().is_err());
    }

    #[test]
    fn action_state_never_skips_dispatch_commit() {
        assert!(ActionState::Authorized.can_transition_to(ActionState::DispatchCommitted));
        assert!(!ActionState::Authorized.can_transition_to(ActionState::Succeeded));
        assert!(ActionState::Unknown.can_transition_to(ActionState::Reconciled));
    }

    #[test]
    fn live_parameters_cannot_be_silently_inferred() {
        let specs = [ParameterSpec {
            name: "account".into(),
            required: true,
            required_explicit_for_live: true,
        }];
        let inferred = BTreeMap::from([(
            "account".into(),
            ParameterValue {
                value: Value::String("a".into()),
                provenance: ParameterProvenance::AgentInferred,
            },
        )]);
        assert_eq!(
            validate_parameters(ExecutionMode::Live, &specs, &inferred)
                .unwrap_err()
                .parameters,
            ["account"]
        );
        assert!(validate_parameters(ExecutionMode::Paper, &specs, &inferred).is_ok());
    }

    #[test]
    fn freshness_is_fail_closed() {
        let now = Utc::now();
        let observed = ObservedValue {
            value: 1,
            source: "source".into(),
            observed_at: now - chrono::Duration::seconds(2),
            received_at: now,
            source_version: "1".into(),
            quality: "good".into(),
            digest: "d".into(),
        };
        assert!(observed.is_accepted(
            now,
            chrono::Duration::seconds(3),
            &BTreeSet::from(["good".into()])
        ));
        assert!(!observed.is_accepted(
            now,
            chrono::Duration::seconds(1),
            &BTreeSet::from(["good".into()])
        ));
    }

    #[test]
    fn revision_and_manifest_contracts_fail_closed() {
        let mut revision = WorkflowRevision {
            definition_id: "definition".into(),
            revision: 1,
            content_digest: "digest".into(),
            kernel_abi_version: "1".into(),
            dependency_set_digest: "dependencies".into(),
            expression_versions: BTreeMap::new(),
            capabilities: vec![CapabilityPin {
                id: "action.example".into(),
                contract_version: "1".into(),
                content_digest: "capability-digest".into(),
            }],
            template_provenance: Value::Null,
            spec: empty_spec(),
        };
        assert!(revision.validate().is_ok());
        revision.capabilities.push(CapabilityPin {
            id: "action.example".into(),
            contract_version: "2".into(),
            content_digest: "capability-digest-v2".into(),
        });
        assert!(revision.validate().is_err());
        revision.capabilities.pop();
        revision.kernel_abi_version.clear();
        assert!(revision.validate().is_err());

        let external = CapabilityManifest::action(
            "action.notify",
            "1",
            "digest",
            Effect::ExternalWrite,
            IdempotencyMode::NeverAutomaticRetry,
            false,
        );
        assert_eq!(
            external.required_guards,
            BTreeSet::from([GuardKind::Authorization])
        );
        assert!(external.validate().is_ok());
        assert_eq!(external.config_schema, serde_json::json!({"type":"object"}));
        let mut invalid_schema = external.clone();
        invalid_schema.config_schema = serde_json::json!({"type":7});
        assert!(invalid_schema.validate().is_err());
        let mut invalid = external;
        invalid.retry.max_attempts = 0;
        assert!(invalid.validate().is_err());

        let mut wrong_kind = funds_manifest();
        wrong_kind.kind = CapabilityKind::Expression;
        assert!(wrong_kind.validate().is_err());
        wrong_kind.kind = CapabilityKind::Action;
        wrong_kind.idempotency_mode = IdempotencyMode::Native;
        wrong_kind.supports_reconciliation = false;
        assert!(wrong_kind.validate().is_ok());
        wrong_kind.lifecycle = CapabilityLifecycle::Disabled;
        assert!(!wrong_kind.can_start_new_work());
    }

    #[test]
    fn trigger_schedule_lifecycle_and_control_validate_boundaries() {
        let now = Utc::now();
        let mut trigger = TriggerEnvelope {
            event_id: "event".into(),
            event_type: "example".into(),
            source: "source".into(),
            schema_version: "1".into(),
            tenant_id: "tenant".parse().unwrap(),
            subject_id: "subject".parse().unwrap(),
            aggregate_id: "aggregate".into(),
            source_sequence: Some(1),
            observed_version: None,
            occurred_at: now,
            received_at: now,
            watermark: None,
            correlation_key: "key".into(),
            dedup_key: "dedup".into(),
            cursor: None,
            payload: Value::Null,
            trace_context: Value::Null,
        };
        assert!(trigger.validate().is_ok());
        trigger.occurred_at = now + chrono::Duration::minutes(6);
        assert!(trigger.validate().is_err());

        let valid_schedule = SchedulePolicy {
            cadence: ScheduleCadence::Cron {
                expression: "0 0 * * * *".into(),
            },
            timezone: "Asia/Shanghai".into(),
            catch_up: CatchUpPolicy::CatchUpOnce,
        };
        assert!(valid_schedule.validate().is_ok());
        for invalid in [
            SchedulePolicy {
                timezone: "Nowhere/Invalid".into(),
                ..valid_schedule.clone()
            },
            SchedulePolicy {
                cadence: ScheduleCadence::FixedRate { milliseconds: 0 },
                ..valid_schedule.clone()
            },
            SchedulePolicy {
                catch_up: CatchUpPolicy::CatchUpAll { limit: 0 },
                ..valid_schedule
            },
        ] {
            assert!(invalid.validate().is_err());
        }

        let mut lifecycle = LifecyclePolicy::run_once();
        assert!(lifecycle.validate().is_ok());
        lifecycle.starts_at = Some(now);
        lifecycle.expires_at = Some(now);
        assert!(lifecycle.validate().is_err());
        lifecycle.starts_at = None;
        lifecycle.expires_at = None;
        lifecycle.on_expiry = ExpiryPolicy::Cancel;
        lifecycle.drain_deadline = Some(now);
        assert!(lifecycle.validate().is_err());

        let mut command = ControlCommand {
            scope: ControlScope::Instance,
            scope_id: "instance".into(),
            mode: ControlMode::Paused,
            operator_subject_id: "operator".parse().unwrap(),
            reason: "maintenance".into(),
            override_expires_at: None,
        };
        assert!(command.validate().is_ok());
        command.reason.clear();
        assert!(command.validate().is_err());
    }

    #[test]
    fn funds_intent_requires_reservation_scope_and_retry_class() {
        let now = Utc::now();
        let mut intent = ActionIntent {
            id: "intent".into(),
            tenant_id: "tenant".parse().unwrap(),
            instance_id: "instance".parse().unwrap(),
            run_id: "run".parse().unwrap(),
            capability: CapabilityPin {
                id: "action.example".into(),
                contract_version: "1".into(),
                content_digest: "digest".into(),
            },
            idempotency_key: "idempotency".into(),
            state: ActionState::Prepared,
            input: Value::Null,
            effect: Effect::Funds,
            retry_class: IdempotencyMode::ReconcileBeforeRetry,
            control_epochs: ControlEpochs::default(),
            resource_scope_id: "resource".into(),
            lease_epoch: 1,
            action_epoch: 1,
            deadline: None,
            reservation: Some(ResourceReservationRef {
                reservation_id: "reservation".into(),
                fencing_token: 1,
            }),
            created_at: now,
        };
        assert!(intent.validate().is_ok());
        assert!(intent.validate_prepared().is_ok());
        intent.reservation = None;
        assert!(intent.validate().is_err());
        intent.reservation = Some(ResourceReservationRef {
            reservation_id: "reservation".into(),
            fencing_token: 1,
        });
        intent.resource_scope_id.clear();
        assert!(intent.validate().is_err());
        intent.resource_scope_id = "resource".into();
        intent.retry_class = IdempotencyMode::None;
        assert!(intent.validate().is_err());
        intent.retry_class = IdempotencyMode::ReconcileBeforeRetry;
        intent.state = ActionState::Authorized;
        assert!(intent.validate_prepared().is_err());
        assert!(!ActionState::Prepared.dispatch_committed());
        assert!(ActionState::Unknown.dispatch_committed());
    }

    fn trigger_binding() -> TriggerBinding {
        TriggerBinding {
            id: "binding".into(),
            revision: 1,
            source: "source".into(),
            event_type: "event".into(),
            instance_id: "instance".parse().unwrap(),
            branch_id: None,
            predicate: serde_json::json!({}),
            ordering: OrderingPolicy::Commutative,
            starts_at: None,
            expires_at: None,
            gap_wait_ms: default_gap_wait_ms(),
            gap_limit: default_gap_limit(),
        }
    }

    #[test]
    fn trigger_binding_rejects_blank_ids_bad_gaps_and_inverted_windows() {
        assert!(trigger_binding().validate().is_ok());
        let blank = TriggerBinding {
            source: "  ".into(),
            ..trigger_binding()
        };
        assert!(blank.validate().is_err());
        for gap_wait_ms in [0, 3_600_001] {
            let binding = TriggerBinding {
                gap_wait_ms,
                ..trigger_binding()
            };
            assert!(binding.validate().is_err(), "gap_wait_ms {gap_wait_ms}");
        }
        for gap_limit in [0, 10_001] {
            let binding = TriggerBinding {
                gap_limit,
                ..trigger_binding()
            };
            assert!(binding.validate().is_err(), "gap_limit {gap_limit}");
        }
        let array_predicate = TriggerBinding {
            predicate: serde_json::json!([1]),
            ..trigger_binding()
        };
        assert!(array_predicate.validate().is_err());
        let now = Utc::now();
        let inverted = TriggerBinding {
            starts_at: Some(now),
            expires_at: Some(now),
            ..trigger_binding()
        };
        assert!(inverted.validate().is_err());
        let ordered = TriggerBinding {
            starts_at: Some(now),
            expires_at: Some(now + chrono::Duration::seconds(1)),
            ..trigger_binding()
        };
        assert!(ordered.validate().is_ok());
        let defaults: TriggerBinding = serde_json::from_value(serde_json::json!({
            "id": "binding", "revision": 1, "source": "s", "event_type": "e",
            "instance_id": "i", "predicate": {}, "ordering": "commutative",
            "starts_at": null, "expires_at": null
        }))
        .unwrap();
        assert_eq!(defaults.gap_wait_ms, 30_000);
        assert_eq!(defaults.gap_limit, 100);
    }

    #[test]
    fn execution_profile_revision_requires_every_provider_name() {
        let profile = ExecutionProfileRevision {
            id: "profile".into(),
            revision: 1,
            content_digest: "sha256:profile".into(),
            mode: ExecutionMode::Paper,
            durability_grade: DurabilityGrade::Standard,
            trigger_provider: "triggers".into(),
            data_provider: "data".into(),
            clock_model: "database".into(),
            action_provider: "actions".into(),
            models: BTreeMap::new(),
            environment: Value::Null,
            policy_bundle: Value::Null,
            connection_bindings: BTreeMap::new(),
        };
        assert!(profile.validate().is_ok());
        let blank_provider = ExecutionProfileRevision {
            action_provider: String::new(),
            ..profile.clone()
        };
        assert!(blank_provider.validate().is_err());
        let blank_digest = ExecutionProfileRevision {
            content_digest: " ".into(),
            ..profile
        };
        assert!(blank_digest.validate().is_err());
    }

    #[test]
    fn retry_backoff_grows_exponentially_with_bounded_jitter_and_cap() {
        let policy = RetryPolicy {
            max_attempts: 5,
            timeout_ms: 1_000,
            initial_backoff_ms: 100,
            max_backoff_ms: 1_000,
        };
        assert_eq!(policy.backoff_ms(1, 0), 100);
        assert_eq!(policy.backoff_ms(2, 0), 200);
        assert_eq!(policy.backoff_ms(3, 0), 400);
        // Jitter is deterministic and never exceeds 25% of the base.
        assert_eq!(policy.backoff_ms(1, 24), 124);
        assert_eq!(policy.backoff_ms(1, 25), 100);
        // Jitter is taken modulo the ceiling (800 / 4 = 200 here).
        assert_eq!(policy.backoff_ms(4, 249), 849);
        // The cap holds for large attempts and for base + jitter.
        assert_eq!(policy.backoff_ms(5, 0), 1_000);
        assert_eq!(policy.backoff_ms(5, 249), 1_000);
        assert_eq!(policy.backoff_ms(40, u64::MAX), 1_000);
        // Attempt 0 behaves like the first attempt instead of underflowing.
        assert_eq!(policy.backoff_ms(0, 0), 100);
    }

    #[test]
    fn lifecycle_policy_rejects_inconsistent_windows_and_zero_counts() {
        let now = Utc::now();
        let later = now + chrono::Duration::hours(1);
        assert!(LifecyclePolicy::run_once().validate().is_ok());
        let inverted = LifecyclePolicy {
            starts_at: Some(later),
            expires_at: Some(now),
            ..LifecyclePolicy::run_once()
        };
        assert!(inverted.validate().is_err());
        let drain_without_policy = LifecyclePolicy {
            on_expiry: ExpiryPolicy::Cancel,
            drain_deadline: Some(later),
            ..LifecyclePolicy::run_once()
        };
        assert!(drain_without_policy.validate().is_err());
        let drain_before_expiry = LifecyclePolicy {
            expires_at: Some(later),
            drain_deadline: Some(now),
            ..LifecyclePolicy::run_once()
        };
        assert!(drain_before_expiry.validate().is_err());
        let zero_timeout = LifecyclePolicy {
            event_idle_timeout_ms: Some(0),
            ..LifecyclePolicy::run_once()
        };
        assert!(zero_timeout.validate().is_err());
        for completion in [
            CompletionPolicy::AfterMatchedEvaluations(0),
            CompletionPolicy::AfterSuccessfulRuns(0),
        ] {
            let zero_count = LifecyclePolicy {
                completion,
                ..LifecyclePolicy::run_once()
            };
            assert!(zero_count.validate().is_err());
        }
        let bounded = LifecyclePolicy {
            starts_at: Some(now),
            expires_at: Some(later),
            completion: CompletionPolicy::AfterSuccessfulRuns(2),
            event_idle_timeout_ms: Some(1_000),
            progress_timeout_ms: Some(1_000),
            on_expiry: ExpiryPolicy::Drain,
            drain_deadline: Some(later + chrono::Duration::minutes(1)),
        };
        assert!(bounded.validate().is_ok());
    }
}