camel_api/

runtime.rs

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use crate::CamelError;
use crate::declarative::LanguageExpressionDef;
use crate::splitter::StreamSplitConfig;

pub const CANONICAL_CONTRACT_NAME: &str = "canonical-v1";
pub const CANONICAL_CONTRACT_VERSION: u32 = 2;
pub const CANONICAL_CONTRACT_SUPPORTED_STEPS: &[&str] = &[
    "to",
    "log",
    "wire_tap",
    "script",
    "filter",
    "choice",
    "split",
    "aggregate",
    "stop",
    "delay",
    "cache",
    "cache_invalidate",
    "cache_peek_stale",
];
pub const CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS: &[&str] =
    &["script", "filter", "choice", "split"];
pub const CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS: &[&str] = &[
    "set_header",
    "set_property",
    "set_body",
    "multicast",
    "convert_body_to",
    "bean",
    "marshal",
    "unmarshal",
];
pub const CANONICAL_CONTRACT_RUST_ONLY_STEPS: &[&str] = &[
    "processor",
    "process",
    "process_fn",
    "map_body",
    "set_body_fn",
    "set_header_fn",
];

pub fn canonical_contract_supports_step(step: &str) -> bool {
    CANONICAL_CONTRACT_SUPPORTED_STEPS.contains(&step)
}

pub fn canonical_contract_rejection_reason(step: &str) -> Option<&'static str> {
    if CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&step) {
        return Some(
            "declared out-of-scope for canonical v2; use declarative route compilation path outside CQRS canonical commands",
        );
    }

    if CANONICAL_CONTRACT_RUST_ONLY_STEPS.contains(&step) {
        return Some("rust-only programmable step; not representable in canonical v2 contract");
    }

    if canonical_contract_supports_step(step)
        && CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS.contains(&step)
    {
        return Some(
            "supported only as declarative/serializable expression form; closure/processor variants are outside canonical v2",
        );
    }

    None
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub struct CanonicalRouteSpec {
    /// Stable minimal route representation for runtime command registration.
    ///
    /// Scope notes:
    /// - This is intentionally a partial model and does not mirror every `BuilderStep`.
    /// - Version 2 adds: auto_startup, startup_order, concurrency.
    /// - Still excluded: error_handler, unit_of_work. These are set to defaults
    ///   when compiling from canonical.
    /// - Round-trip (YAML → Canonical → YAML) loses these fields.
    /// - Advanced EIPs continue to use the existing RouteDefinition/BuilderStep path.
    pub route_id: String,
    pub from: String,
    pub steps: Vec<CanonicalStepSpec>,
    pub circuit_breaker: Option<CanonicalCircuitBreakerSpec>,
    pub auto_startup: Option<bool>,
    pub startup_order: Option<i32>,
    pub concurrency: Option<CanonicalConcurrencySpec>,
    pub version: u32,
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(tag = "step", content = "config", rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
#[non_exhaustive]
pub enum CanonicalStepSpec {
    To {
        uri: String,
    },
    Log {
        message: String,
    },
    WireTap {
        uri: String,
    },
    Script {
        expression: LanguageExpressionDef,
    },
    Filter {
        predicate: LanguageExpressionDef,
        steps: Vec<CanonicalStepSpec>,
    },
    Choice {
        whens: Vec<CanonicalWhenSpec>,
        otherwise: Option<Vec<CanonicalStepSpec>>,
    },
    Split {
        expression: CanonicalSplitExpressionSpec,
        aggregation: CanonicalSplitAggregationSpec,
        parallel: bool,
        parallel_limit: Option<usize>,
        /// Threshold above which split fragments start new traces (0 = off).
        /// Absent means the runtime default applies.
        trace_item_threshold: Option<usize>,
        stop_on_exception: bool,
        steps: Vec<CanonicalStepSpec>,
    },
    Aggregate(CanonicalAggregateSpec),
    Stop,
    Delay {
        #[ts(type = "number")]
        delay_ms: u64,
        dynamic_header: Option<String>,
    },
    Cache {
        repository: Option<String>,
        key: String,
        ttl: Option<String>,
        max_entry_bytes: Option<usize>,
        /// Coalesce concurrent misses on the same key into a single `on_miss`
        /// run. Absent/null means `false`.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        coalesce_misses: Option<bool>,
        on_miss: Vec<CanonicalStepSpec>,
    },
    CacheInvalidate {
        repository: Option<String>,
        /// Exact key to invalidate (simple-language expression).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        key: Option<String>,
        /// Namespace prefix to invalidate (simple-language expression).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        key_prefix: Option<String>,
    },
    CacheClear {
        repository: Option<String>,
    },
    CacheStats {
        repository: Option<String>,
    },
    CachePeekStale {
        repository: Option<String>,
        key: String,
        /// On-miss policy: `"stop"` (default) or `"continue"`. Absent/null means `"stop"`.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        on_miss: Option<String>,
    },
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub struct CanonicalWhenSpec {
    pub predicate: LanguageExpressionDef,
    pub steps: Vec<CanonicalStepSpec>,
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
#[non_exhaustive]
pub enum CanonicalSplitExpressionSpec {
    BodyLines,
    BodyJsonArray,
    Language(LanguageExpressionDef),
    Stream(StreamSplitConfig),
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
#[non_exhaustive]
pub enum CanonicalSplitAggregationSpec {
    LastWins,
    CollectAll,
    Original,
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
#[non_exhaustive]
pub enum CanonicalAggregateStrategySpec {
    CollectAll,
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub struct CanonicalAggregateSpec {
    pub header: String,
    pub completion_size: Option<usize>,
    #[ts(type = "number")]
    pub completion_timeout_ms: Option<u64>,
    pub correlation_key: Option<String>,
    pub force_completion_on_stop: Option<bool>,
    pub discard_on_timeout: Option<bool>,
    pub strategy: CanonicalAggregateStrategySpec,
    pub max_buckets: Option<usize>,
    /// Per-bucket accumulation bound (audit 2026-08-31, F6-2).
    /// Absent = builder default (10_000).
    #[serde(default)]
    pub max_bucket_size: Option<usize>,
    #[ts(type = "number")]
    pub bucket_ttl_ms: Option<u64>,
    /// Language expression predicate; completes the bucket when it evaluates
    /// true against the incoming exchange (runtime-resolved via the language
    /// registry). Mirrors `CorrelationStrategy::Expression`.
    #[serde(default)]
    pub completion_predicate: Option<LanguageExpressionDef>,
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub struct CanonicalCircuitBreakerSpec {
    pub failure_threshold: u32,
    #[ts(type = "number")]
    pub open_duration_ms: u64,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub fallback: Vec<CanonicalStepSpec>,
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(tag = "mode", rename_all = "snake_case")]
#[non_exhaustive]
pub enum CanonicalConcurrencySpec {
    Sequential,
    Concurrent { max: usize },
}

impl CanonicalRouteSpec {
    pub fn new(route_id: impl Into<String>, from: impl Into<String>) -> Self {
        Self {
            route_id: route_id.into(),
            from: from.into(),
            steps: Vec::new(),
            circuit_breaker: None,
            auto_startup: None,
            startup_order: None,
            concurrency: None,
            version: CANONICAL_CONTRACT_VERSION,
        }
    }

    pub fn with_auto_startup(mut self, auto: bool) -> Self {
        self.auto_startup = Some(auto);
        self
    }

    pub fn with_startup_order(mut self, order: i32) -> Self {
        self.startup_order = Some(order);
        self
    }

    pub fn with_concurrency(mut self, concurrency: CanonicalConcurrencySpec) -> Self {
        self.concurrency = Some(concurrency);
        self
    }

    pub fn validate_contract(&self) -> Result<(), CamelError> {
        if self.route_id.trim().is_empty() {
            return Err(CamelError::RouteError(
                "canonical contract violation: route_id cannot be empty".to_string(),
            ));
        }
        if self.from.trim().is_empty() {
            return Err(CamelError::RouteError(
                "canonical contract violation: from cannot be empty".to_string(),
            ));
        }
        if self.version == 0 || self.version > CANONICAL_CONTRACT_VERSION {
            return Err(CamelError::RouteError(format!(
                "canonical contract violation: expected version {}, got {}",
                CANONICAL_CONTRACT_VERSION, self.version
            )));
        }
        validate_steps(&self.steps)?;
        if let Some(cb) = &self.circuit_breaker {
            if cb.failure_threshold == 0 {
                return Err(CamelError::RouteError(
                    "canonical contract violation: circuit_breaker.failure_threshold must be > 0"
                        .to_string(),
                ));
            }
            if cb.open_duration_ms == 0 {
                return Err(CamelError::RouteError(
                    "canonical contract violation: circuit_breaker.open_duration_ms must be > 0"
                        .to_string(),
                ));
            }
            validate_steps(&cb.fallback)?;
        }
        if let Some(CanonicalConcurrencySpec::Concurrent { max: 0 }) = &self.concurrency {
            return Err(CamelError::RouteError(
                "canonical contract violation: concurrency max must be > 0".to_string(),
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct CanonicalFieldLoss {
    pub field: &'static str,
    pub reason: String,
    pub target_version: u32,
}

#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize)]
pub struct CanonicalLossReport {
    pub dropped_fields: Vec<CanonicalFieldLoss>,
}

impl CanonicalLossReport {
    pub fn from_field(field: &'static str, reason: &str, target_version: u32) -> Self {
        Self {
            dropped_fields: vec![CanonicalFieldLoss {
                field,
                reason: reason.to_string(),
                target_version,
            }],
        }
    }

    pub fn is_empty(&self) -> bool {
        self.dropped_fields.is_empty()
    }
}

fn validate_steps(steps: &[CanonicalStepSpec]) -> Result<(), CamelError> {
    for step in steps {
        match step {
            CanonicalStepSpec::To { uri } | CanonicalStepSpec::WireTap { uri } => {
                if uri.trim().is_empty() {
                    return Err(CamelError::RouteError(
                        "canonical contract violation: endpoint uri cannot be empty".to_string(),
                    ));
                }
            }
            CanonicalStepSpec::Filter { steps, .. } => validate_steps(steps)?,
            CanonicalStepSpec::Choice { whens, otherwise } => {
                for when in whens {
                    validate_steps(&when.steps)?;
                }
                if let Some(otherwise) = otherwise {
                    validate_steps(otherwise)?;
                }
            }
            CanonicalStepSpec::Split {
                parallel_limit,
                steps,
                ..
            } => {
                if let Some(limit) = parallel_limit
                    && *limit == 0
                {
                    return Err(CamelError::RouteError(
                        "canonical contract violation: split.parallel_limit must be > 0"
                            .to_string(),
                    ));
                }
                validate_steps(steps)?;
            }
            CanonicalStepSpec::Aggregate(config) => {
                let header_present = !config.header.trim().is_empty();
                let key_present = config
                    .correlation_key
                    .as_deref()
                    .is_some_and(|k| !k.trim().is_empty());
                if !key_present && config.correlation_key.is_some() {
                    return Err(CamelError::RouteError(
                        "canonical contract violation: aggregate.correlation_key cannot be empty"
                            .to_string(),
                    ));
                }
                if !header_present && !key_present {
                    return Err(CamelError::RouteError(
                        "canonical contract violation: aggregate requires a correlation source: header or correlation_key"
                            .to_string(),
                    ));
                }
                if let Some(size) = config.completion_size
                    && size == 0
                {
                    return Err(CamelError::RouteError(
                        "canonical contract violation: aggregate.completion_size must be > 0"
                            .to_string(),
                    ));
                }
            }
            CanonicalStepSpec::Cache { on_miss, .. } => {
                validate_steps(on_miss)?;
            }
            CanonicalStepSpec::Log { .. }
            | CanonicalStepSpec::Script { .. }
            | CanonicalStepSpec::Stop
            | CanonicalStepSpec::Delay { .. }
            | CanonicalStepSpec::CacheInvalidate { .. }
            | CanonicalStepSpec::CacheClear { .. }
            | CanonicalStepSpec::CacheStats { .. }
            | CanonicalStepSpec::CachePeekStale { .. } => {}
        }
    }
    Ok(())
}

#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum RuntimeCommand {
    RegisterRoute {
        spec: CanonicalRouteSpec,
        command_id: String,
        causation_id: Option<String>,
    },
    StartRoute {
        route_id: String,
        command_id: String,
        causation_id: Option<String>,
    },
    StopRoute {
        route_id: String,
        command_id: String,
        causation_id: Option<String>,
    },
    SuspendRoute {
        route_id: String,
        command_id: String,
        causation_id: Option<String>,
    },
    ResumeRoute {
        route_id: String,
        command_id: String,
        causation_id: Option<String>,
    },
    ReloadRoute {
        route_id: String,
        command_id: String,
        causation_id: Option<String>,
    },
    /// Internal lifecycle command emitted by runtime adapters when a route crashes at runtime.
    ///
    /// This keeps aggregate/projection state aligned with controller-observed failures.
    FailRoute {
        route_id: String,
        error: String,
        command_id: String,
        causation_id: Option<String>,
    },
    RemoveRoute {
        route_id: String,
        command_id: String,
        causation_id: Option<String>,
    },
    ReloadTlsCerts {
        scheme: String,
        host: String,
        port: u16,
        command_id: String,
        causation_id: Option<String>,
    },
    /// Reload template sources for a route (infrastructure command).
    ///
    /// Intercepted in `RuntimeBus::execute` BEFORE journal recovery + dedup,
    /// exactly like `ReloadTlsCerts`: it is idempotent, NOT journaled, and does
    /// not mutate `RouteStatus`. Dispatches to
    /// `TemplateReloadRegistry::reload_route`.
    ReloadTemplates {
        route_id: String,
        command_id: String,
        causation_id: Option<String>,
    },
}

impl RuntimeCommand {
    pub fn command_id(&self) -> &str {
        match self {
            RuntimeCommand::RegisterRoute { command_id, .. }
            | RuntimeCommand::StartRoute { command_id, .. }
            | RuntimeCommand::StopRoute { command_id, .. }
            | RuntimeCommand::SuspendRoute { command_id, .. }
            | RuntimeCommand::ResumeRoute { command_id, .. }
            | RuntimeCommand::ReloadRoute { command_id, .. }
            | RuntimeCommand::FailRoute { command_id, .. }
            | RuntimeCommand::RemoveRoute { command_id, .. }
            | RuntimeCommand::ReloadTlsCerts { command_id, .. }
            | RuntimeCommand::ReloadTemplates { command_id, .. } => command_id,
        }
    }

    pub fn causation_id(&self) -> Option<&str> {
        match self {
            RuntimeCommand::RegisterRoute { causation_id, .. }
            | RuntimeCommand::StartRoute { causation_id, .. }
            | RuntimeCommand::StopRoute { causation_id, .. }
            | RuntimeCommand::SuspendRoute { causation_id, .. }
            | RuntimeCommand::ResumeRoute { causation_id, .. }
            | RuntimeCommand::ReloadRoute { causation_id, .. }
            | RuntimeCommand::FailRoute { causation_id, .. }
            | RuntimeCommand::RemoveRoute { causation_id, .. }
            | RuntimeCommand::ReloadTlsCerts { causation_id, .. }
            | RuntimeCommand::ReloadTemplates { causation_id, .. } => causation_id.as_deref(),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum RuntimeCommandResult {
    Accepted,
    Duplicate {
        command_id: String,
    },
    RouteRegistered {
        route_id: String,
    },
    RouteStateChanged {
        route_id: String,
        status: String,
    },
    TlsCertsReloaded {
        scheme: String,
        host: String,
        port: u16,
    },
    TemplatesReloaded {
        route_id: String,
    },
}

#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum RuntimeQuery {
    GetRouteStatus {
        route_id: String,
    },
    /// **Note:** This variant is intercepted by `RuntimeBus::ask` *before* reaching
    /// `execute_query`. Do not handle it in `execute_query` — it has no access to
    /// the in-flight counter. See `runtime_bus.rs` for the intercept.
    InFlightCount {
        route_id: String,
    },
    ListRoutes,
}

#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum RuntimeQueryResult {
    InFlightCount { route_id: String, count: u64 },
    RouteNotFound { route_id: String },
    RouteStatus { route_id: String, status: String },
    Routes { route_ids: Vec<String> },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub enum RuntimeEvent {
    RouteRegistered { route_id: String },
    RouteStartRequested { route_id: String },
    RouteStarted { route_id: String },
    RouteFailed { route_id: String, error: String },
    RouteStopped { route_id: String },
    RouteSuspended { route_id: String },
    RouteResumed { route_id: String },
    RouteReloaded { route_id: String },
    RouteRemoved { route_id: String },
}

#[async_trait]
pub trait RuntimeCommandBus: Send + Sync {
    async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError>;
}

#[async_trait]
pub trait RuntimeQueryBus: Send + Sync {
    async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError>;
}

pub trait RuntimeHandle: RuntimeCommandBus + RuntimeQueryBus {}

impl<T> RuntimeHandle for T where T: RuntimeCommandBus + RuntimeQueryBus {}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use futures::executor::block_on;

    struct NoopRuntime;

    #[async_trait]
    impl RuntimeCommandBus for NoopRuntime {
        async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError> {
            Ok(match cmd {
                RuntimeCommand::RegisterRoute { spec, .. } => {
                    RuntimeCommandResult::RouteRegistered {
                        route_id: spec.route_id,
                    }
                }
                RuntimeCommand::StartRoute { route_id, .. }
                | RuntimeCommand::StopRoute { route_id, .. }
                | RuntimeCommand::SuspendRoute { route_id, .. }
                | RuntimeCommand::ResumeRoute { route_id, .. }
                | RuntimeCommand::ReloadRoute { route_id, .. }
                | RuntimeCommand::FailRoute { route_id, .. }
                | RuntimeCommand::RemoveRoute { route_id, .. } => {
                    RuntimeCommandResult::RouteStateChanged {
                        route_id,
                        status: "ok".to_string(),
                    }
                }
                RuntimeCommand::ReloadTlsCerts {
                    scheme, host, port, ..
                } => RuntimeCommandResult::TlsCertsReloaded { scheme, host, port },
                RuntimeCommand::ReloadTemplates { route_id, .. } => {
                    RuntimeCommandResult::TemplatesReloaded { route_id }
                }
            })
        }
    }

    #[async_trait]
    impl RuntimeQueryBus for NoopRuntime {
        async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError> {
            Ok(match query {
                RuntimeQuery::GetRouteStatus { route_id } => RuntimeQueryResult::RouteStatus {
                    route_id,
                    status: "Started".to_string(),
                },
                RuntimeQuery::InFlightCount { route_id } => {
                    RuntimeQueryResult::InFlightCount { route_id, count: 0 }
                }
                RuntimeQuery::ListRoutes => RuntimeQueryResult::Routes {
                    route_ids: vec!["r1".to_string()],
                },
            })
        }
    }

    #[test]
    fn command_and_query_ids_are_exposed() {
        let cmd = RuntimeCommand::StartRoute {
            route_id: "r1".into(),
            command_id: "c1".into(),
            causation_id: None,
        };
        assert_eq!(cmd.command_id(), "c1");
    }

    #[test]
    fn canonical_spec_requires_route_id_and_from() {
        let spec = CanonicalRouteSpec::new("r1", "timer:tick");
        assert_eq!(spec.route_id, "r1");
        assert_eq!(spec.from, "timer:tick");
        assert_eq!(spec.version, CANONICAL_CONTRACT_VERSION);
        assert!(spec.steps.is_empty());
        assert!(spec.circuit_breaker.is_none());
    }

    #[test]
    fn canonical_contract_rejects_invalid_version() {
        let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
        spec.version = 3;
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("expected version"));
    }

    #[test]
    fn canonical_contract_declares_subset_scope() {
        assert!(canonical_contract_supports_step("to"));
        assert!(canonical_contract_supports_step("split"));
        assert!(!canonical_contract_supports_step("set_header"));
        assert!(!canonical_contract_supports_step("set_property"));

        assert!(CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS.contains(&"split"));
        assert!(CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&"set_header"));
        assert!(CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&"set_property"));
        assert!(CANONICAL_CONTRACT_RUST_ONLY_STEPS.contains(&"processor"));
    }

    #[test]
    fn canonical_contract_rejection_reason_is_explicit() {
        let set_header_reason = canonical_contract_rejection_reason("set_header")
            .expect("set_header should have explicit reason");
        assert!(set_header_reason.contains("out-of-scope"));

        let set_property_reason = canonical_contract_rejection_reason("set_property")
            .expect("set_property should have explicit reason");
        assert!(set_property_reason.contains("out-of-scope"));

        let processor_reason = canonical_contract_rejection_reason("processor")
            .expect("processor should be rust-only");
        assert!(processor_reason.contains("rust-only"));

        let split_reason = canonical_contract_rejection_reason("split")
            .expect("split should require declarative form");
        assert!(split_reason.contains("declarative"));
    }

    #[test]
    fn command_causation_id_is_exposed() {
        let cmd = RuntimeCommand::StopRoute {
            route_id: "r1".into(),
            command_id: "c2".into(),
            causation_id: Some("c1".into()),
        };
        assert_eq!(cmd.command_id(), "c2");
        assert_eq!(cmd.causation_id(), Some("c1"));
    }

    #[test]
    fn canonical_contract_rejects_empty_route_id_and_from() {
        let spec = CanonicalRouteSpec::new("   ", "timer:tick");
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("route_id cannot be empty"));

        let spec = CanonicalRouteSpec::new("r1", "  ");
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("from cannot be empty"));
    }

    #[test]
    fn canonical_contract_rejects_invalid_nested_steps() {
        let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
        spec.steps = vec![CanonicalStepSpec::Split {
            expression: CanonicalSplitExpressionSpec::BodyLines,
            aggregation: CanonicalSplitAggregationSpec::CollectAll,
            parallel: true,
            parallel_limit: Some(0),
            trace_item_threshold: None,
            stop_on_exception: false,
            steps: vec![CanonicalStepSpec::To {
                uri: "log:ok".to_string(),
            }],
        }];
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("split.parallel_limit must be > 0"));

        spec.steps = vec![CanonicalStepSpec::To {
            uri: "   ".to_string(),
        }];
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("endpoint uri cannot be empty"));
    }

    #[test]
    fn canonical_contract_rejects_invalid_aggregate_and_circuit_breaker() {
        let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
        spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
            header: " ".to_string(),
            completion_size: Some(1),
            completion_timeout_ms: None,
            correlation_key: None,
            force_completion_on_stop: None,
            discard_on_timeout: None,
            strategy: CanonicalAggregateStrategySpec::CollectAll,
            max_buckets: None,
            max_bucket_size: None,
            bucket_ttl_ms: None,
            completion_predicate: None,
        })];
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("correlation source"));

        spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
            header: "k".to_string(),
            completion_size: Some(0),
            completion_timeout_ms: None,
            correlation_key: None,
            force_completion_on_stop: None,
            discard_on_timeout: None,
            strategy: CanonicalAggregateStrategySpec::CollectAll,
            max_buckets: None,
            max_bucket_size: None,
            bucket_ttl_ms: None,
            completion_predicate: None,
        })];
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("aggregate.completion_size must be > 0"));

        spec.steps = vec![];
        spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
            failure_threshold: 0,
            open_duration_ms: 10,
            fallback: vec![],
        });
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("failure_threshold must be > 0"));

        spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
            failure_threshold: 1,
            open_duration_ms: 0,
            fallback: vec![],
        });
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("open_duration_ms must be > 0"));
    }

    #[test]
    fn contract_accepts_expression_only_aggregate() {
        let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
        spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
            header: String::new(),
            completion_size: None,
            completion_timeout_ms: None,
            correlation_key: Some("${header.orderId}".to_string()),
            force_completion_on_stop: None,
            discard_on_timeout: None,
            strategy: CanonicalAggregateStrategySpec::CollectAll,
            max_buckets: None,
            max_bucket_size: None,
            bucket_ttl_ms: None,
            completion_predicate: None,
        })];
        assert!(spec.validate_contract().is_ok());
    }

    #[test]
    fn contract_rejects_aggregate_missing_both_sources() {
        let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
        spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
            header: String::new(),
            completion_size: None,
            completion_timeout_ms: None,
            correlation_key: None,
            force_completion_on_stop: None,
            discard_on_timeout: None,
            strategy: CanonicalAggregateStrategySpec::CollectAll,
            max_buckets: None,
            max_bucket_size: None,
            bucket_ttl_ms: None,
            completion_predicate: None,
        })];
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("correlation source"));
    }

    #[test]
    fn contract_rejects_empty_correlation_key() {
        let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
        spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
            header: "region".to_string(),
            completion_size: None,
            completion_timeout_ms: None,
            correlation_key: Some(String::new()),
            force_completion_on_stop: None,
            discard_on_timeout: None,
            strategy: CanonicalAggregateStrategySpec::CollectAll,
            max_buckets: None,
            max_bucket_size: None,
            bucket_ttl_ms: None,
            completion_predicate: None,
        })];
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("correlation_key cannot be empty"));
    }

    #[test]
    fn canonical_contract_rejects_invalid_fallback_step() {
        let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
        spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
            failure_threshold: 1,
            open_duration_ms: 10,
            fallback: vec![CanonicalStepSpec::To {
                uri: "   ".to_string(),
            }],
        });
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("endpoint uri cannot be empty"));
    }

    #[test]
    fn canonical_contract_rejection_reason_none_for_regular_steps() {
        assert!(canonical_contract_rejection_reason("to").is_none());
        assert!(canonical_contract_rejection_reason("unknown-step").is_none());
    }

    #[test]
    fn command_helpers_cover_all_variants() {
        let spec = CanonicalRouteSpec::new("r1", "timer:tick");
        let cmds = [
            RuntimeCommand::RegisterRoute {
                spec,
                command_id: "c1".into(),
                causation_id: Some("root".into()),
            },
            RuntimeCommand::StartRoute {
                route_id: "r1".into(),
                command_id: "c2".into(),
                causation_id: None,
            },
            RuntimeCommand::StopRoute {
                route_id: "r1".into(),
                command_id: "c3".into(),
                causation_id: None,
            },
            RuntimeCommand::SuspendRoute {
                route_id: "r1".into(),
                command_id: "c4".into(),
                causation_id: None,
            },
            RuntimeCommand::ResumeRoute {
                route_id: "r1".into(),
                command_id: "c5".into(),
                causation_id: None,
            },
            RuntimeCommand::ReloadRoute {
                route_id: "r1".into(),
                command_id: "c6".into(),
                causation_id: None,
            },
            RuntimeCommand::FailRoute {
                route_id: "r1".into(),
                error: "boom".into(),
                command_id: "c7".into(),
                causation_id: None,
            },
            RuntimeCommand::RemoveRoute {
                route_id: "r1".into(),
                command_id: "c8".into(),
                causation_id: None,
            },
            RuntimeCommand::ReloadTlsCerts {
                scheme: "https".into(),
                host: "example.com".into(),
                port: 8443,
                command_id: "c9".into(),
                causation_id: None,
            },
        ];

        let ids: Vec<&str> = cmds.iter().map(RuntimeCommand::command_id).collect();
        assert_eq!(
            ids,
            vec!["c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9"]
        );
        assert_eq!(cmds[0].causation_id(), Some("root"));
        assert_eq!(cmds[1].causation_id(), None);
    }

    #[test]
    fn canonical_route_spec_serde_roundtrip() {
        let mut spec = CanonicalRouteSpec::new("test-route", "timer:tick?period=1000");
        spec.steps.push(CanonicalStepSpec::Log {
            message: "Hello".into(),
        });
        spec.steps.push(CanonicalStepSpec::To {
            uri: "log:info".into(),
        });
        spec.steps.push(CanonicalStepSpec::Stop);

        let json = serde_json::to_string(&spec).unwrap();
        let deserialized: CanonicalRouteSpec = serde_json::from_str(&json).unwrap();
        assert_eq!(spec, deserialized);
    }

    #[test]
    fn canonical_step_spec_serde_variants() {
        let steps = vec![
            CanonicalStepSpec::To {
                uri: "direct:a".into(),
            },
            CanonicalStepSpec::Log {
                message: "msg".into(),
            },
            CanonicalStepSpec::WireTap {
                uri: "direct:audit".into(),
            },
            CanonicalStepSpec::Stop,
            CanonicalStepSpec::Delay {
                delay_ms: 100,
                dynamic_header: None,
            },
        ];
        let json = serde_json::to_string_pretty(&steps).unwrap();
        let back: Vec<CanonicalStepSpec> = serde_json::from_str(&json).unwrap();
        assert_eq!(steps, back);
    }

    #[test]
    fn canonical_cache_peek_stale_on_miss_round_trip() {
        let some = CanonicalStepSpec::CachePeekStale {
            repository: None,
            key: "k".into(),
            on_miss: Some("continue".into()),
        };
        let json = serde_json::to_string(&some).unwrap();
        assert!(json.contains("continue"));
        let back: CanonicalStepSpec = serde_json::from_str(&json).unwrap();
        assert_eq!(some, back);

        let none = CanonicalStepSpec::CachePeekStale {
            repository: None,
            key: "k".into(),
            on_miss: None,
        };
        let json = serde_json::to_string(&none).unwrap();
        assert!(
            !json.contains("on_miss"),
            "on_miss: None must omit the key from JSON, got: {json}"
        );
        let back: CanonicalStepSpec = serde_json::from_str(&json).unwrap();
        assert_eq!(none, back);
    }

    #[test]
    fn canonical_cache_invalidate_prefix_round_trip() {
        let prefixed = CanonicalStepSpec::CacheInvalidate {
            repository: Some("persistent".into()),
            key: None,
            key_prefix: Some("ns:".into()),
        };
        let json = serde_json::to_string(&prefixed).unwrap();
        assert!(json.contains("key_prefix"), "must emit key_prefix: {json}");
        let back: CanonicalStepSpec = serde_json::from_str(&json).unwrap();
        assert_eq!(prefixed, back);

        // Legacy wire form carrying only `key` still deserializes with
        // `key_prefix: None` (no key-iteration field).
        let legacy = r#"{"step":"cache_invalidate","config":{"repository":null,"key":"k"}}"#;
        let parsed: CanonicalStepSpec = serde_json::from_str(legacy).unwrap();
        assert_eq!(
            parsed,
            CanonicalStepSpec::CacheInvalidate {
                repository: None,
                key: Some("k".into()),
                key_prefix: None,
            }
        );
    }

    #[test]
    fn canonical_cache_clear_stats_round_trip() {
        let clear = CanonicalStepSpec::CacheClear {
            repository: Some("persistent".into()),
        };
        let json = serde_json::to_string(&clear).unwrap();
        assert!(json.contains("cache_clear"));
        let back: CanonicalStepSpec = serde_json::from_str(&json).unwrap();
        assert_eq!(clear, back);

        let stats = CanonicalStepSpec::CacheStats { repository: None };
        let json = serde_json::to_string(&stats).unwrap();
        assert!(json.contains("cache_stats"));
        let back: CanonicalStepSpec = serde_json::from_str(&json).unwrap();
        assert_eq!(stats, back);
    }

    #[test]
    fn canonical_circuit_breaker_fallback_roundtrip() {
        let mut spec = CanonicalRouteSpec::new("cb-fallback", "direct:start");
        spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
            failure_threshold: 1,
            open_duration_ms: 60000,
            fallback: vec![CanonicalStepSpec::CachePeekStale {
                repository: Some("persistent".into()),
                key: "tile-xyz".into(),
                on_miss: None,
            }],
        });

        let json = serde_json::to_string(&spec).unwrap();
        let back: CanonicalRouteSpec = serde_json::from_str(&json).unwrap();
        assert_eq!(spec, back);
        assert_eq!(back.circuit_breaker.as_ref().unwrap().fallback.len(), 1);

        // A spec serialized without the fallback key deserializes with empty fallback.
        let without = CanonicalCircuitBreakerSpec {
            failure_threshold: 1,
            open_duration_ms: 60000,
            fallback: vec![],
        };
        let json = serde_json::to_string(&without).unwrap();
        assert!(
            !json.contains("fallback"),
            "empty fallback must omit the key from JSON, got: {json}"
        );
        let back: CanonicalCircuitBreakerSpec = serde_json::from_str(&json).unwrap();
        assert!(back.fallback.is_empty());
    }

    #[test]
    fn canonical_route_spec_json_schema_generates() {
        let schema = schemars::schema_for!(CanonicalRouteSpec);
        let json = serde_json::to_string(&schema).unwrap();
        assert!(json.contains("CanonicalRouteSpec"));
        assert!(json.contains("route_id"));
    }

    #[test]
    fn canonical_json_schema_has_no_function_step() {
        let schema = schemars::schema_for!(CanonicalRouteSpec);
        let json = serde_json::to_string(&schema).unwrap();
        assert!(
            !json.contains("\"function\""),
            "canonical JSON schema must not contain 'function' step"
        );
    }

    #[test]
    fn canonical_contract_does_not_support_function() {
        assert!(
            !canonical_contract_supports_step("function"),
            "function must not be in CANONICAL_CONTRACT_SUPPORTED_STEPS"
        );
    }

    #[test]
    fn runtime_command_result_all_variants_are_distinct() {
        let accepted = RuntimeCommandResult::Accepted;
        let dup = RuntimeCommandResult::Duplicate {
            command_id: "c1".into(),
        };
        let registered = RuntimeCommandResult::RouteRegistered {
            route_id: "r1".into(),
        };
        let changed = RuntimeCommandResult::RouteStateChanged {
            route_id: "r1".into(),
            status: "Started".into(),
        };

        assert_ne!(accepted, dup);
        assert_ne!(dup, registered);
        assert_ne!(registered, changed);

        let dup2 = RuntimeCommandResult::Duplicate {
            command_id: "c1".into(),
        };
        assert_eq!(dup, dup2);
    }

    #[test]
    fn runtime_event_serialization_round_trip() {
        let event = RuntimeEvent::RouteFailed {
            route_id: "route-a".to_string(),
            error: "boom".to_string(),
        };
        let json = serde_json::to_string(&event).unwrap();
        let back: RuntimeEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(event, back);
    }

    #[test]
    fn noop_runtime_execute_and_ask_return_expected_shapes() {
        let rt = NoopRuntime;
        let cmd = RuntimeCommand::RegisterRoute {
            spec: CanonicalRouteSpec::new("r2", "timer:tick"),
            command_id: "c1".into(),
            causation_id: None,
        };
        let cmd_result = block_on(rt.execute(cmd)).unwrap();
        assert_eq!(
            cmd_result,
            RuntimeCommandResult::RouteRegistered {
                route_id: "r2".into()
            }
        );

        let query_result = block_on(rt.ask(RuntimeQuery::GetRouteStatus {
            route_id: "r2".into(),
        }))
        .unwrap();
        assert_eq!(
            query_result,
            RuntimeQueryResult::RouteStatus {
                route_id: "r2".into(),
                status: "Started".into()
            }
        );
    }

    #[test]
    fn canonical_contract_name_and_version_constants_match() {
        assert_eq!(CANONICAL_CONTRACT_NAME, "canonical-v1");
        assert_eq!(CANONICAL_CONTRACT_VERSION, 2);
    }

    #[test]
    fn canonical_concurrency_spec_rejects_zero_max() {
        let spec = CanonicalRouteSpec::new("r1", "timer:tick")
            .with_concurrency(CanonicalConcurrencySpec::Concurrent { max: 0 });
        let err = spec.validate_contract().unwrap_err().to_string();
        assert!(err.contains("concurrency max must be > 0"), "{err}");
    }

    #[test]
    fn canonical_v2_round_trip() {
        let spec = CanonicalRouteSpec::new("r1", "timer:tick")
            .with_auto_startup(false)
            .with_startup_order(42)
            .with_concurrency(CanonicalConcurrencySpec::Concurrent { max: 8 });
        spec.validate_contract().unwrap();
    }

    #[test]
    fn canonical_v2_version_is_2() {
        assert_eq!(CANONICAL_CONTRACT_VERSION, 2);
    }

    #[test]
    fn canonical_loss_report_builder() {
        let report =
            CanonicalLossReport::from_field("error_handler", "not supported by canonical path", 2);
        assert_eq!(report.dropped_fields.len(), 1);
        assert_eq!(report.dropped_fields[0].field, "error_handler");
    }

    #[test]
    fn canonical_v1_json_deserializes_in_v2() {
        let json = r#"{"route_id":"r1","from":"timer:tick","steps":[],"version":1}"#;
        let spec: CanonicalRouteSpec = serde_json::from_str(json).unwrap();
        assert_eq!(spec.route_id, "r1");
        assert!(spec.auto_startup.is_none());
        assert!(spec.startup_order.is_none());
        assert!(spec.concurrency.is_none());
        // CRITICAL: v1 specs must pass validation in v2 runtime (backward compat)
        spec.validate_contract().unwrap();
    }
}