camel_core/lifecycle/adapters/step_compilers/

mod.rs

//! StepCompiler registry pattern — extract from step_resolution.rs
//!
//! Each compiler group is responsible for matching specific `BuilderStep` variants.
//! The registry dispatches each step to compilers in registration order; the first
//! compiler that returns `Matched` wins. Compilers that don't handle a variant
//! return `NotHandled(step)` to pass it to the next compiler.

use std::sync::Arc;

use camel_api::{
    BodyType, BoxProcessor, CamelError, FunctionInvoker, ProducerContext, SpanKindHint,
    StepLifecycle,
};
use camel_component_api::{ComponentContext, RuntimeObservability};
use camel_endpoint::parse_uri;

use crate::intercept::InterceptRules;
use crate::lifecycle::adapters::route_controller::SharedLanguageRegistry;
use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;
use crate::lifecycle::application::route_definition::BuilderStep;
use crate::{CacheRegistry, ClaimCheckRegistry, IdempotentRegistry};
use camel_bean::BeanRegistry;

mod control_flow;
mod core;
mod endpoints;
mod error_handling;
mod routing;
mod splitting;
mod transforms;

/// A compiled pipeline step.
///
/// `Process` is the normal case: a boxed processor plus its optional body
/// contract. `Stop` (added in Task 3b) is the Stop EIP marker — `run_steps`
/// recognises it and produces `PipelineOutcome::Stopped` without invoking a
/// Tower service. `Segment` (added in Task 3) wraps an `OutcomeSegment` for
/// structural EIPs with outcome-aware sub-pipelines.
///
/// **Boundary:** `CompiledStep` is the compile-time representation. At runtime,
/// `run_steps` consumes a `Vec<CompiledStep>` (Stop variants included) and
/// produces a `PipelineOutcome`; the wrapping `Service<Exchange>` impl
/// translates `PipelineOutcome` back to `Result<Exchange, CamelError>`. See
/// ADR-0024.
#[derive(Debug, Clone)]
pub enum CompiledStep {
    Process {
        processor: BoxProcessor,
        body_contract: Option<BodyType>,
        /// Lifecycle handle for this processor, if it is stateful.
        /// `None` for stateless processors (the common case).
        lifecycle: Option<Arc<dyn StepLifecycle>>,
        /// Stable span name (e.g. `to:direct`, `split`), stamped by
        /// `StepCompilerRegistry::compile_step` from
        /// `BuilderStep::span_label`. `None` for anonymous steps — spans
        /// fall back to the positional `step-{index}` name. Read by
        /// `compose_traced_pipeline` / `segment_span` in route_compiler.rs.
        label: Option<Arc<str>>,
        /// Span kind for this step's step span, stamped by
        /// `StepCompilerRegistry::compile_step` from
        /// `BuilderStep::span_kind_hint`. `Internal` for steps built
        /// outside the registry. Read by `compose_traced_pipeline` in
        /// route_compiler.rs.
        kind_hint: SpanKindHint,
    },
    /// Stop EIP marker. `run_steps` produces `PipelineOutcome::Stopped(ex)`
    /// without invoking a Tower service. Replaces `StopService` (Task 7).
    Stop,
    /// Outcome-aware structural EIP segment. `run_steps` invokes
    /// `segment.run(ex)` and matches on the returned `PipelineOutcome`.
    /// See ADR-0025.
    Segment {
        segment: camel_api::OutcomeSegment,
        body_contract: Option<BodyType>,
        /// Lifecycle handles from children nested inside this segment.
        /// `Option<Vec<...>>` (not `Option<Arc<...>>`) because multiple stateful
        /// children (e.g. Idempotent+Resequencer inside Filter) each
        /// register independently.
        lifecycle: Option<Vec<Arc<dyn StepLifecycle>>>,
        /// Stable span name, same contract as [`CompiledStep::Process`]'s
        /// `label` field.
        label: Option<Arc<str>>,
    },
}

impl CompiledStep {
    /// Overwrite the span label of a `Process`/`Segment` step; no-op on `Stop`
    /// (the marker produces no step span, so it carries no label).
    pub(crate) fn set_label(&mut self, label: Option<String>) {
        let label = label.map(Arc::from);
        match self {
            CompiledStep::Process { label: slot, .. }
            | CompiledStep::Segment { label: slot, .. } => {
                *slot = label;
            }
            CompiledStep::Stop => {}
        }
    }

    /// Overwrite the span kind hint of a `Process` step; no-op on `Stop` and
    /// `Segment` (neither produces an individual step span, so neither
    /// carries a kind hint).
    pub(crate) fn set_kind_hint(&mut self, hint: SpanKindHint) {
        match self {
            CompiledStep::Process {
                kind_hint: slot, ..
            } => *slot = hint,
            CompiledStep::Stop | CompiledStep::Segment { .. } => {}
        }
    }
}

/// Result from a compiler: either it handled the step, or it did not recognize
/// the variant and returns the step for the next compiler.
///
/// `Result` is intentionally outside this enum (returned by `StepCompiler::compile`)
/// so that compiler arms can use the `?` operator for propagated errors instead of
/// wrapping every `Err` in `Matched(Err(e))`.
pub(crate) enum CompileOutcome {
    Matched(CompiledStep),
    NotHandled(BuilderStep),
}

/// A compiler that can handle one or more `BuilderStep` variants.
///
/// The `compile` method receives ownership of the step. If the compiler recognizes
/// the variant it returns `Ok(CompileOutcome::Matched(...))`. Otherwise it returns
/// `Ok(CompileOutcome::NotHandled(step))` to pass the step to the next compiler.
/// Compilation errors are returned as `Err(CamelError)`.
pub(crate) trait StepCompiler: Send + Sync {
    fn compile(
        &self,
        step: BuilderStep,
        step_index: usize,
        ctx: &CompilationContext,
        registry: &StepCompilerRegistry,
    ) -> Result<CompileOutcome, CamelError>;
}

/// Shared context passed to every compiler invocation.
pub(crate) struct CompilationContext<'a> {
    pub producer_ctx: &'a ProducerContext,
    pub rt: Arc<dyn RuntimeObservability>,
    pub languages: &'a SharedLanguageRegistry,
    pub beans: &'a Arc<std::sync::Mutex<BeanRegistry>>,
    pub function_invoker: Option<Arc<dyn FunctionInvoker>>,
    pub component_ctx: Arc<dyn ComponentContext>,
    pub route_id: Option<&'a str>,
    pub staging_mode: &'a FunctionStagingMode,
    /// Idempotent repository registry. Used by the `IdempotentConsumer`
    /// compiler arm to resolve repository names into `Arc<dyn IdempotentRepository>`.
    pub idempotent_repositories: &'a IdempotentRegistry,
    /// Claim check repository registry. Used by the `ClaimCheck` compiler arm
    /// to resolve repository names into `Arc<dyn ClaimCheckRepository>`.
    pub claim_check_repositories: &'a ClaimCheckRegistry,
    /// Cache repository registry. Used by the `Cache`, `CacheInvalidate`, and
    /// `CachePeekStale` compiler arms to resolve repository names into
    /// `Arc<dyn CacheRepository>`.
    pub cache_repositories: &'a CacheRegistry,
    /// Route send-point interception rules captured at compile time.
    /// Consulted by send-step compilation (Task 4/5).
    pub intercept: InterceptRules,
}

impl<'a> CompilationContext<'a> {
    /// Recursively compile child steps. Used by compilers that have sub-pipelines
    /// (Filter, Choice, Split, Loop, etc.).
    pub fn compile_children(
        &self,
        steps: Vec<BuilderStep>,
        registry: &StepCompilerRegistry,
    ) -> Result<Vec<CompiledStep>, CamelError> {
        registry.compile_steps(steps, self)
    }

    /// Recursively compile child steps and map them into outcome-aware segments.
    ///
    /// Each `CompiledStep` variant is converted to a `Box<dyn OutcomePipeline>`:
    /// - `Process` → `BoxProcessorSegment`, optionally wrapped in `BodyCoercingSegment`
    /// - `Stop` → `StopSegment` (produces `PipelineOutcome::Stopped(ex)`)
    /// - `Segment` → its inner `OutcomeSegment` (which now implements OutcomePipeline)
    ///
    /// This replaces the 22-line duplicated closure in Filter/DeclarativeFilter
    /// (and will prevent 14+ more duplicates in T9–T16).
    #[allow(clippy::type_complexity)]
    pub fn compile_children_segments(
        &self,
        steps: Vec<BuilderStep>,
        registry: &StepCompilerRegistry,
    ) -> Result<
        (
            Vec<Box<dyn camel_api::OutcomePipeline>>,
            Vec<Arc<dyn camel_api::StepLifecycle>>,
        ),
        CamelError,
    > {
        let pairs = self.compile_children(steps, registry)?;
        let mut lifecycle_handles: Vec<Arc<dyn camel_api::StepLifecycle>> = Vec::new();
        let segments: Vec<Box<dyn camel_api::OutcomePipeline>> = pairs
            .into_iter()
            .map(|c| match c {
                CompiledStep::Process {
                    processor,
                    body_contract,
                    lifecycle,
                    label: _,
                    kind_hint: _,
                } => {
                    if let Some(lc) = lifecycle {
                        lifecycle_handles.push(lc);
                    }
                    let inner: Box<dyn camel_api::OutcomePipeline> = Box::new(
                        crate::lifecycle::adapters::route_compiler::BoxProcessorSegment::new(
                            processor,
                        ),
                    );
                    match body_contract {
                        Some(contract) => Box::new(
                            crate::lifecycle::adapters::route_compiler::BodyCoercingSegment::new(
                                inner, contract,
                            ),
                        ),
                        None => inner,
                    }
                }
                CompiledStep::Stop => {
                    Box::new(crate::lifecycle::adapters::route_compiler::StopSegment)
                        as Box<dyn camel_api::OutcomePipeline>
                }
                CompiledStep::Segment {
                    segment,
                    body_contract: _,
                    lifecycle,
                    label: _,
                } => {
                    if let Some(lcs) = lifecycle {
                        lifecycle_handles.extend(lcs);
                    }
                    Box::new(segment)
                }
            })
            .collect();
        Ok((segments, lifecycle_handles))
    }
}

/// Registry of step compilers. Steps are dispatched to compilers in registration
/// order. The first matching compiler handles the step.
pub(crate) struct StepCompilerRegistry {
    compilers: Vec<Box<dyn StepCompiler>>,
}

impl StepCompilerRegistry {
    pub fn new() -> Self {
        Self {
            compilers: Vec::new(),
        }
    }

    pub fn register(&mut self, compiler: Box<dyn StepCompiler>) {
        self.compilers.push(compiler);
    }

    /// Try each compiler in order. The first to return `Matched` wins.
    /// If all return `NotHandled`, returns `Ok(None)`. Compilation errors
    /// short-circuit and return `Err(CamelError)`.
    pub fn compile_step(
        &self,
        step: BuilderStep,
        step_index: usize,
        ctx: &CompilationContext,
    ) -> Result<Option<CompiledStep>, CamelError> {
        // Capture the span label and kind hint BEFORE the dispatch loop
        // moves the step.
        let label = step.span_label();
        let kind_hint = step.span_kind_hint();
        let mut step = step;
        for compiler in &self.compilers {
            match compiler.compile(step, step_index, ctx, self)? {
                CompileOutcome::Matched(mut s) => {
                    s.set_label(label);
                    s.set_kind_hint(kind_hint);
                    return Ok(Some(s));
                }
                CompileOutcome::NotHandled(s) => step = s,
            }
        }
        Ok(None)
    }

    /// Compile all steps in a vector.
    pub fn compile_steps(
        &self,
        steps: Vec<BuilderStep>,
        ctx: &CompilationContext,
    ) -> Result<Vec<CompiledStep>, CamelError> {
        let mut out = Vec::with_capacity(steps.len());
        for (i, step) in steps.into_iter().enumerate() {
            match self.compile_step(step, i, ctx)? {
                Some(c) => out.push(c),
                None => {
                    return Err(CamelError::RouteError(
                        "no compiler registered for step variant".into(),
                    ));
                }
            }
        }
        Ok(out)
    }
}

/// A resolved send target: the pieces the un-intercepted `To` path produces
/// for one URI.
pub(super) struct ResolvedSend {
    pub producer: BoxProcessor,
    pub body_contract: Option<BodyType>,
    pub lifecycle: Option<Arc<dyn StepLifecycle>>,
}

/// Resolve a send URI into producer/body-contract/lifecycle, exactly as the
/// un-intercepted `To` path does. Canonical resolution path for send steps;
/// [`resolve_producer_with_lifecycle`] and [`resolve_producer`] are thin
/// wrappers over this.
pub(super) fn resolve_send(
    ctx: &CompilationContext,
    uri: &str,
) -> Result<ResolvedSend, CamelError> {
    let parsed = parse_uri(uri)?;
    let component = ctx
        .component_ctx
        .resolve_component(&parsed.scheme)
        .ok_or_else(|| CamelError::ComponentNotFound(parsed.scheme.clone()))?;
    let endpoint = component.create_endpoint(uri, ctx.component_ctx.as_ref())?;
    let body_contract = endpoint.body_contract();
    let producer = endpoint.create_producer(Arc::clone(&ctx.rt), ctx.producer_ctx)?;
    // Capture the endpoint's lifecycle handle so the route controller can
    // start/shut it down in route order (ADR-0022). Default is `None` for
    // stateless endpoints — see `Endpoint::lifecycle` in camel-component-api.
    let lifecycle: Option<Arc<dyn StepLifecycle>> = endpoint.lifecycle();
    Ok(ResolvedSend {
        producer,
        body_contract,
        lifecycle,
    })
}

/// Parse a URI and create a producer, also returning the endpoint's
/// [`StepLifecycle`] handle (if any). Thin wrapper over [`resolve_send`] that
/// discards the body contract. Use directly when the lifecycle is needed
/// (e.g. `WireTap` propagation).
pub(crate) fn resolve_producer_with_lifecycle(
    ctx: &CompilationContext,
    uri: &str,
) -> Result<(BoxProcessor, Option<Arc<dyn StepLifecycle>>), CamelError> {
    let resolved = resolve_send(ctx, uri)?;
    Ok((resolved.producer, resolved.lifecycle))
}

/// Parse a URI and create a producer, reusing `component_ctx`, `rt`, and `producer_ctx`
/// from the compilation context.
///
/// Thin wrapper over [`resolve_producer_with_lifecycle`] that discards the
/// endpoint's [`StepLifecycle`] handle.
pub(crate) fn resolve_producer(
    ctx: &CompilationContext,
    uri: &str,
) -> Result<BoxProcessor, CamelError> {
    Ok(resolve_producer_with_lifecycle(ctx, uri)?.0)
}

/// Pack a lifecycle Vec into `None` when empty, `Some` when non-empty.
/// Preserves the invariant that `Some` always implies ≥1 handle.
pub(super) fn pack_lifecycles(
    lifecycles: Vec<Arc<dyn StepLifecycle>>,
) -> Option<Vec<Arc<dyn StepLifecycle>>> {
    if lifecycles.is_empty() {
        None
    } else {
        Some(lifecycles)
    }
}

/// Build the full registry with all compiler groups.
pub(crate) fn build_registry() -> StepCompilerRegistry {
    let mut reg = StepCompilerRegistry::new();
    reg.register(Box::new(core::CoreCompiler));
    reg.register(Box::new(endpoints::EndpointsCompiler));
    reg.register(Box::new(transforms::TransformsCompiler));
    reg.register(Box::new(routing::RoutingCompiler));
    reg.register(Box::new(control_flow::ControlFlowCompiler));
    reg.register(Box::new(splitting::SplittingCompiler));
    reg.register(Box::new(error_handling::ErrorHandlingCompiler));
    reg
}

#[cfg(test)]
mod segment_tests {
    use super::*;
    use camel_api::{Exchange, OutcomePipeline, PipelineOutcome};
    use std::future::Future;
    use std::pin::Pin;

    #[derive(Clone)]
    struct EchoSegment;

    impl OutcomePipeline for EchoSegment {
        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
            Box::new(EchoSegment)
        }
        fn run<'a>(
            &'a mut self,
            exchange: Exchange,
        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
            Box::pin(async move { PipelineOutcome::Completed(exchange) })
        }
    }

    #[test]
    fn compiled_step_segment_clone_compiles() {
        let seg = camel_api::OutcomeSegment::new(Box::new(EchoSegment));
        let step = CompiledStep::Segment {
            segment: seg,
            body_contract: None,
            lifecycle: None,
            label: None,
        };
        let _cloned = step.clone();
        if let CompiledStep::Segment { .. } = _cloned {
            // ok
        } else {
            panic!("clone should preserve variant");
        }
    }

    #[test]
    fn compiled_step_segment_debug_renders() {
        let seg = camel_api::OutcomeSegment::new(Box::new(EchoSegment));
        let step = CompiledStep::Segment {
            segment: seg,
            body_contract: None,
            lifecycle: None,
            label: None,
        };
        let s = format!("{:?}", step);
        assert!(
            s.contains("Segment"),
            "debug should mention Segment variant: {s}"
        );
    }

    #[test]
    fn outcome_segment_satisfies_clone_send_static() {
        fn assert_traits<T: Clone + Send + 'static>() {}
        assert_traits::<camel_api::OutcomeSegment>();
    }

    #[tokio::test]
    #[allow(clippy::arc_with_non_send_sync)]
    async fn outcome_segment_survives_arcswap_swap() {
        use arc_swap::ArcSwap;
        use camel_api::{Exchange, Message, OutcomePipeline, PipelineOutcome};
        use std::sync::Arc;

        #[derive(Clone)]
        struct EchoSegment;
        impl OutcomePipeline for EchoSegment {
            fn clone_box(&self) -> Box<dyn OutcomePipeline> {
                Box::new(EchoSegment)
            }
            fn run<'a>(
                &'a mut self,
                ex: Exchange,
            ) -> std::pin::Pin<Box<dyn std::future::Future<Output = PipelineOutcome> + Send + 'a>>
            {
                Box::pin(async move { PipelineOutcome::Completed(ex) })
            }
        }

        let seg = camel_api::OutcomeSegment::new(Box::new(EchoSegment));
        let slot: ArcSwap<Option<camel_api::OutcomeSegment>> = ArcSwap::from_pointee(None);
        slot.store(Arc::new(Some(seg.clone())));
        slot.store(Arc::new(Some(seg)));

        let mut borrowed = slot.load().as_ref().clone().unwrap();
        let outcome = borrowed.run(Exchange::new(Message::new("ping"))).await;
        assert!(matches!(outcome, PipelineOutcome::Completed(_)));
    }

    /// Test lifecycle handle used by compile_children_segments_bubbles_child_lifecycle.
    #[derive(Debug)]
    struct TestLifecycle;

    #[async_trait::async_trait]
    impl camel_api::StepLifecycle for TestLifecycle {
        fn name(&self) -> &'static str {
            "test-lifecycle"
        }
        async fn shutdown(
            &self,
            _reason: camel_api::StepShutdownReason,
        ) -> Result<(), camel_api::CamelError> {
            Ok(())
        }
    }

    /// Custom compiler that injects a lifecycle handle into every
    /// `BuilderStep::Processor` it compiles.
    struct LifecycleInjectorCompiler {
        handle: Arc<dyn camel_api::StepLifecycle>,
    }

    impl StepCompiler for LifecycleInjectorCompiler {
        fn compile(
            &self,
            step: BuilderStep,
            _step_index: usize,
            _ctx: &CompilationContext,
            _registry: &StepCompilerRegistry,
        ) -> Result<CompileOutcome, CamelError> {
            match step {
                BuilderStep::Processor(op) => Ok(CompileOutcome::Matched(CompiledStep::Process {
                    processor: op.0,
                    body_contract: None,
                    lifecycle: Some(self.handle.clone()),
                    label: None,
                    kind_hint: SpanKindHint::Internal,
                })),
                other => Ok(CompileOutcome::NotHandled(other)),
            }
        }
    }

    #[tokio::test]
    async fn compile_children_segments_bubbles_child_lifecycle() {
        use std::collections::HashMap;
        use std::sync::Mutex;

        use camel_api::{
            BoxProcessor, BoxProcessorExt, FilterPredicate, OpaqueProcessor, StepLifecycle,
        };
        use camel_bean::BeanRegistry;
        use camel_component_api::{
            ComponentContext, NoOpComponentContext, RuntimeObservability,
            test_support::NoopRuntimeObservability,
        };

        use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;

        let handle: Arc<dyn StepLifecycle> = Arc::new(TestLifecycle);

        // Register lifecycle injector + real control-flow compiler so
        // compile_children_segments runs through a structural EIP path.
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(LifecycleInjectorCompiler {
            handle: handle.clone(),
        }));
        reg.register(Box::new(super::control_flow::ControlFlowCompiler));

        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = CompilationContext {
            producer_ctx: &pc,
            rt,
            languages: &languages,
            beans: &beans,
            function_invoker: None,
            component_ctx,
            route_id: None,
            staging_mode: &staging,
            idempotent_repositories: &idempotent_repositories,
            claim_check_repositories: &claim_check_repositories,
            cache_repositories: &cache_repositories,
            intercept: InterceptRules::default(),
        };

        // Compile a Filter with a child Processor step.
        let filter_step = BuilderStep::Filter {
            predicate: FilterPredicate::new(|_| true),
            steps: vec![BuilderStep::Processor(OpaqueProcessor(
                BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) })),
            ))],
        };

        let result = reg.compile_step(filter_step, 0, &ctx);
        let compiled = result
            .expect("compilation should succeed")
            .expect("should match");

        match compiled {
            CompiledStep::Segment {
                lifecycle,
                body_contract,
                ..
            } => {
                assert_eq!(body_contract, None, "body_contract should be None");
                let handles = lifecycle.expect("Segment should have lifecycle handles");
                assert_eq!(handles.len(), 1, "expected 1 lifecycle handle");
                assert_eq!(handles[0].name(), "test-lifecycle", "handle name mismatch");
            }
            other => panic!("Expected CompiledStep::Segment, got {other:?}"),
        }
    }

    /// A lifecycle handle with a configurable name for multi-handle tests.
    #[derive(Debug)]
    struct NamedLifecycle(&'static str);

    #[async_trait::async_trait]
    impl camel_api::StepLifecycle for NamedLifecycle {
        fn name(&self) -> &'static str {
            self.0
        }
        async fn shutdown(
            &self,
            _reason: camel_api::StepShutdownReason,
        ) -> Result<(), camel_api::CamelError> {
            Ok(())
        }
    }

    /// Test A: Multiple stateful children in one Segment → Vec length 2.
    #[tokio::test]
    async fn compile_children_segments_multiple_stateful_children() {
        use std::collections::HashMap;
        use std::sync::Mutex;

        use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;
        use camel_api::{
            BoxProcessor, BoxProcessorExt, FilterPredicate, OpaqueProcessor, StepLifecycle,
        };
        use camel_bean::BeanRegistry;
        use camel_component_api::{
            ComponentContext, NoOpComponentContext, RuntimeObservability,
            test_support::NoopRuntimeObservability,
        };

        let handle: Arc<dyn StepLifecycle> = Arc::new(NamedLifecycle("multi"));

        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(LifecycleInjectorCompiler {
            handle: handle.clone(),
        }));
        reg.register(Box::new(super::control_flow::ControlFlowCompiler));

        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = CompilationContext {
            producer_ctx: &pc,
            rt,
            languages: &languages,
            beans: &beans,
            function_invoker: None,
            component_ctx,
            route_id: None,
            staging_mode: &staging,
            idempotent_repositories: &idempotent_repositories,
            claim_check_repositories: &claim_check_repositories,
            cache_repositories: &cache_repositories,
            intercept: InterceptRules::default(),
        };

        // Filter with TWO child Processors → both get the same lifecycle handle.
        let filter_step = BuilderStep::Filter {
            predicate: FilterPredicate::new(|_| true),
            steps: vec![
                BuilderStep::Processor(OpaqueProcessor(BoxProcessor::from_fn(|ex| {
                    Box::pin(async move { Ok(ex) })
                }))),
                BuilderStep::Processor(OpaqueProcessor(BoxProcessor::from_fn(|ex| {
                    Box::pin(async move { Ok(ex) })
                }))),
            ],
        };

        let result = reg.compile_step(filter_step, 0, &ctx);
        let compiled = result
            .expect("compilation should succeed")
            .expect("should match");

        match compiled {
            CompiledStep::Segment { lifecycle, .. } => {
                let handles = lifecycle.expect("Segment should have lifecycle handles");
                assert_eq!(
                    handles.len(),
                    2,
                    "expected 2 lifecycle handles for 2 children"
                );
                for h in &handles {
                    assert_eq!(h.name(), "multi", "all handles should be 'multi'");
                }
            }
            other => panic!("Expected CompiledStep::Segment, got {other:?}"),
        }
    }

    /// Test B: Multi-branch accumulation across Choice when-clauses.
    #[tokio::test]
    async fn compile_children_segments_multi_branch_accumulation() {
        use std::collections::HashMap;
        use std::sync::Mutex;

        use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;
        use crate::lifecycle::application::route_definition::WhenStep;
        use camel_api::{
            BoxProcessor, BoxProcessorExt, FilterPredicate, OpaqueProcessor, StepLifecycle,
        };
        use camel_bean::BeanRegistry;
        use camel_component_api::{
            ComponentContext, NoOpComponentContext, RuntimeObservability,
            test_support::NoopRuntimeObservability,
        };

        let handle: Arc<dyn StepLifecycle> = Arc::new(NamedLifecycle("branch"));

        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(LifecycleInjectorCompiler {
            handle: handle.clone(),
        }));
        reg.register(Box::new(super::control_flow::ControlFlowCompiler));

        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = CompilationContext {
            producer_ctx: &pc,
            rt,
            languages: &languages,
            beans: &beans,
            function_invoker: None,
            component_ctx,
            route_id: None,
            staging_mode: &staging,
            idempotent_repositories: &idempotent_repositories,
            claim_check_repositories: &claim_check_repositories,
            cache_repositories: &cache_repositories,
            intercept: InterceptRules::default(),
        };

        // Choice with 2 when branches, each containing 1 stateful child.
        let choice_step = BuilderStep::Choice {
            whens: vec![
                WhenStep {
                    predicate: FilterPredicate::new(|_| true),
                    steps: vec![BuilderStep::Processor(OpaqueProcessor(
                        BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) })),
                    ))],
                },
                WhenStep {
                    predicate: FilterPredicate::new(|_| false),
                    steps: vec![BuilderStep::Processor(OpaqueProcessor(
                        BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) })),
                    ))],
                },
            ],
            otherwise: None,
        };

        let result = reg.compile_step(choice_step, 0, &ctx);
        let compiled = result
            .expect("compilation should succeed")
            .expect("should match");

        match compiled {
            CompiledStep::Segment { lifecycle, .. } => {
                let handles = lifecycle.expect("Segment should have lifecycle handles");
                assert_eq!(
                    handles.len(),
                    2,
                    "expected 2 lifecycle handles from 2 branches"
                );
                for h in &handles {
                    assert_eq!(h.name(), "branch", "all handles should be 'branch'");
                }
            }
            other => panic!("Expected CompiledStep::Segment, got {other:?}"),
        }
    }

    /// Test C: Nested Segment-in-Segment flattening — outer Segment contains
    /// innermost lifecycle handle from a grandchild Processor.
    #[tokio::test]
    async fn compile_children_segments_nested_segment_flattening() {
        use std::collections::HashMap;
        use std::sync::Mutex;

        use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;
        use camel_api::{
            BoxProcessor, BoxProcessorExt, FilterPredicate, OpaqueProcessor, StepLifecycle,
        };
        use camel_bean::BeanRegistry;
        use camel_component_api::{
            ComponentContext, NoOpComponentContext, RuntimeObservability,
            test_support::NoopRuntimeObservability,
        };

        let inner_handle: Arc<dyn StepLifecycle> = Arc::new(NamedLifecycle("deep"));

        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(LifecycleInjectorCompiler {
            handle: inner_handle.clone(),
        }));
        reg.register(Box::new(super::control_flow::ControlFlowCompiler));

        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = CompilationContext {
            producer_ctx: &pc,
            rt,
            languages: &languages,
            beans: &beans,
            function_invoker: None,
            component_ctx,
            route_id: None,
            staging_mode: &staging,
            idempotent_repositories: &idempotent_repositories,
            claim_check_repositories: &claim_check_repositories,
            cache_repositories: &cache_repositories,
            intercept: InterceptRules::default(),
        };

        // Outer Filter containing an inner Filter that has a stateful Processor.
        // The outer Segment's lifecycle should contain the innermost handle
        // (proves recursive flattening through compile_children_segments).
        let inner_filter = BuilderStep::Filter {
            predicate: FilterPredicate::new(|_| true),
            steps: vec![BuilderStep::Processor(OpaqueProcessor(
                BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) })),
            ))],
        };

        let outer_filter = BuilderStep::Filter {
            predicate: FilterPredicate::new(|_| true),
            steps: vec![inner_filter],
        };

        let result = reg.compile_step(outer_filter, 0, &ctx);
        let compiled = result
            .expect("compilation should succeed")
            .expect("should match");

        match compiled {
            CompiledStep::Segment { lifecycle, .. } => {
                let handles = lifecycle.expect("outer Segment should have lifecycle handles");
                assert_eq!(handles.len(), 1, "expected 1 innermost lifecycle handle");
                assert_eq!(
                    handles[0].name(),
                    "deep",
                    "handle should be from innermost child"
                );
            }
            other => panic!("Expected CompiledStep::Segment, got {other:?}"),
        }
    }
}

#[cfg(test)]
mod dispatch_tests {
    use super::*;
    use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;
    use camel_api::{BoxProcessor, BoxProcessorExt};
    use camel_bean::BeanRegistry;
    use camel_component_api::{
        ComponentContext, NoOpComponentContext, RuntimeObservability,
        test_support::NoopRuntimeObservability,
    };
    use std::collections::HashMap;
    use std::sync::Mutex;

    /// Compiler that handles `BuilderStep::To` → `CompiledStep::Stop`.
    struct ToStopCompiler;

    impl StepCompiler for ToStopCompiler {
        fn compile(
            &self,
            step: BuilderStep,
            _step_index: usize,
            _ctx: &CompilationContext,
            _registry: &StepCompilerRegistry,
        ) -> Result<CompileOutcome, CamelError> {
            match step {
                BuilderStep::To(_) => Ok(CompileOutcome::Matched(CompiledStep::Stop)),
                other => Ok(CompileOutcome::NotHandled(other)),
            }
        }
    }

    /// Compiler that handles `BuilderStep::To` → `CompiledStep::Process`.
    struct ToProcessCompiler;

    impl StepCompiler for ToProcessCompiler {
        fn compile(
            &self,
            step: BuilderStep,
            _step_index: usize,
            _ctx: &CompilationContext,
            _registry: &StepCompilerRegistry,
        ) -> Result<CompileOutcome, CamelError> {
            match step {
                BuilderStep::To(_) => Ok(CompileOutcome::Matched(CompiledStep::Process {
                    processor: BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) })),
                    body_contract: None,
                    lifecycle: None,
                    label: None,
                    kind_hint: SpanKindHint::Internal,
                })),
                other => Ok(CompileOutcome::NotHandled(other)),
            }
        }
    }

    /// Minimal outcome pipeline for building `Segment` fixtures in tests.
    #[derive(Clone)]
    struct NoopPipeline;

    impl camel_api::OutcomePipeline for NoopPipeline {
        fn clone_box(&self) -> Box<dyn camel_api::OutcomePipeline> {
            Box::new(NoopPipeline)
        }

        fn run<'a>(
            &'a mut self,
            exchange: camel_api::Exchange,
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = camel_api::PipelineOutcome> + Send + 'a>,
        > {
            Box::pin(async move { camel_api::PipelineOutcome::Completed(exchange) })
        }
    }

    /// Compiler that passes all steps through (never handles any variant).
    /// Returns `NotHandled(step)` for every input, proving by-value pass-through.
    struct PassThroughCompiler;

    impl StepCompiler for PassThroughCompiler {
        fn compile(
            &self,
            step: BuilderStep,
            _step_index: usize,
            _ctx: &CompilationContext,
            _registry: &StepCompilerRegistry,
        ) -> Result<CompileOutcome, CamelError> {
            Ok(CompileOutcome::NotHandled(step))
        }
    }

    /// Shared context builder — avoids repeating the 15-line setup in every test.
    #[allow(clippy::too_many_arguments)]
    fn ctx<'a>(
        pc: &'a ProducerContext,
        rt: Arc<dyn RuntimeObservability>,
        languages: &'a SharedLanguageRegistry,
        beans: &'a Arc<Mutex<BeanRegistry>>,
        component_ctx: Arc<dyn ComponentContext>,
        staging: &'a FunctionStagingMode,
        idempotent_repositories: &'a crate::IdempotentRegistry,
        claim_check_repositories: &'a crate::ClaimCheckRegistry,
        cache_repositories: &'a crate::CacheRegistry,
    ) -> CompilationContext<'a> {
        CompilationContext {
            producer_ctx: pc,
            rt,
            languages,
            beans,
            function_invoker: None,
            component_ctx,
            route_id: None,
            staging_mode: staging,
            idempotent_repositories,
            claim_check_repositories,
            cache_repositories,
            intercept: InterceptRules::default(),
        }
    }

    /// Test that the dispatcher respects registration order: the first matching
    /// compiler wins even when a later compiler also matches.
    #[test]
    fn compile_step_preserves_registry_order() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let context = ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        // Register ToStopCompiler FIRST, ToProcessCompiler SECOND.
        // Both match BuilderStep::To. The first-registered should win.
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(ToStopCompiler));
        reg.register(Box::new(ToProcessCompiler));

        let result = reg
            .compile_step(BuilderStep::To("test".into()), 0, &context)
            .expect("compilation should succeed")
            .expect("should match");

        assert!(
            matches!(result, CompiledStep::Stop),
            "expected ToStopCompiler (first registered) to win, got {result:?}"
        );
    }

    /// Task 1.2 (span-name-enrichment): `set_label` is a no-op on `Stop`.
    #[test]
    fn set_label_noop_on_stop() {
        let mut step = CompiledStep::Stop;
        step.set_label(Some("x".into()));
        assert!(
            matches!(step, CompiledStep::Stop),
            "set_label must not change a Stop step, got {step:?}"
        );
    }

    /// Task 1.2 (span-name-enrichment): `compile_step` stamps the outcome with
    /// the step's `span_label` (here `to:direct` from `BuilderStep::To`).
    #[test]
    fn compile_step_stamps_label() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let context = ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(ToProcessCompiler));

        let result = reg
            .compile_step(BuilderStep::To("direct:x".into()), 0, &context)
            .expect("compilation should succeed")
            .expect("should match");

        match result {
            CompiledStep::Process { label, .. } => {
                assert_eq!(label.as_deref(), Some("to:direct"));
            }
            other => panic!("expected Process, got {other:?}"),
        }
    }

    /// Task 1.2 (span-kind-hint): `set_kind_hint` is a no-op on `Stop` and
    /// `Segment` — only `Process` carries a kind hint, and a Segment's label
    /// must survive the call untouched.
    #[test]
    fn set_kind_hint_noop_on_stop_and_segment() {
        use camel_api::SpanKindHint;

        // Stop: the marker produces no step span, so it carries no kind.
        let mut step = CompiledStep::Stop;
        step.set_kind_hint(SpanKindHint::Client);
        assert!(
            matches!(step, CompiledStep::Stop),
            "set_kind_hint must not change a Stop step, got {step:?}"
        );

        // Segment: no kind_hint field; its label is unchanged.
        let mut step = CompiledStep::Segment {
            segment: camel_api::OutcomeSegment::new(Box::new(NoopPipeline)),
            body_contract: None,
            lifecycle: None,
            label: Some("seg".into()),
        };
        step.set_kind_hint(SpanKindHint::Client);
        match step {
            CompiledStep::Segment { label, .. } => {
                assert_eq!(label.as_deref(), Some("seg"));
            }
            other => panic!("expected Segment to stay a Segment, got {other:?}"),
        }
    }

    /// Task 1.2 (span-kind-hint): `compile_step` stamps the outcome with the
    /// step's `span_kind_hint` (here `kafka` → Producer) alongside the label.
    #[test]
    fn compile_step_stamps_kind_hint() {
        use camel_api::SpanKindHint;

        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let context = ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(ToProcessCompiler));

        let result = reg
            .compile_step(BuilderStep::To("kafka:orders".into()), 0, &context)
            .expect("compilation should succeed")
            .expect("should match");

        match result {
            CompiledStep::Process {
                kind_hint, label, ..
            } => {
                assert_eq!(kind_hint, SpanKindHint::Producer);
                assert_eq!(label.as_deref(), Some("to:kafka"));
            }
            other => panic!("expected Process, got {other:?}"),
        }
    }

    /// Test that when no compiler handles a variant, the dispatcher returns
    /// `Ok(None)` rather than an error.
    #[test]
    fn compile_step_unhandled_returns_ok_none() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let context = ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        // Register a compiler that only handles BuilderStep::To.
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(ToStopCompiler));

        // Send a BuilderStep::Log variant — ToStopCompiler does not match it.
        let result = reg.compile_step(
            BuilderStep::Log {
                level: camel_processor::LogLevel::Info,
                message: "unhandled".into(),
            },
            0,
            &context,
        );

        assert!(
            matches!(result, Ok(None)),
            "expected Ok(None), got {result:?}"
        );
    }

    /// Test that `NotHandled(step)` passes the step by-value to the next
    /// compiler. Compiler N returns `NotHandled(step)` for a variant that
    /// compiler N+1 handles — proving the step is not dropped or replaced.
    #[test]
    fn compile_step_nothandled_passes_step_intact_to_next_compiler() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let context = ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        // Register PassThroughCompiler FIRST (never handles anything),
        // ToStopCompiler SECOND (handles BuilderStep::To).
        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(PassThroughCompiler));
        reg.register(Box::new(ToStopCompiler));

        let result = reg
            .compile_step(BuilderStep::To("passthrough".into()), 0, &context)
            .expect("compilation should succeed")
            .expect("should match");

        assert!(
            matches!(result, CompiledStep::Stop),
            "expected ToStopCompiler (N+1) to win after pass-through, got {result:?}"
        );
    }
}