camel_core/lifecycle/adapters/

route_compiler.rs

// adapters/route_compiler.rs
// Pipeline compilation functions: compose BuilderSteps into a Tower BoxProcessor.
// Tower types live here as this is the adapter layer responsible for
// translating declarative route definitions into executable pipelines.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use tokio_util::sync::CancellationToken;
use tower::Service;

use camel_api::metrics::MetricsCollector;
use camel_api::{
    BoxProcessor, CamelError, Exchange, IdentityProcessor, Message, NoOpMetrics,
    ORIGINAL_MESSAGE_EXTENSION, PipelineOutcome,
};

use camel_api::error_handler::{BoundaryKind, RetryOutcome, StepDisposition};
use camel_processor::{
    CircuitBreakerDecision, CircuitBreakerGate, RouteErrorHandler, invoke_processor,
};
use tracing::Instrument;

use crate::lifecycle::adapters::body_coercing::wrap_if_needed;
use crate::lifecycle::adapters::step_compilers::CompiledStep;
use crate::shared::observability::adapters::TracingProcessor;
use crate::shared::observability::domain::DetailLevel;

// Re-export outcome composition types so existing step_compiler import paths
// (`route_compiler::BoxProcessorSegment`, etc.) continue to work.
pub(crate) use super::outcome_composition::{
    BodyCoercingSegment, BoxProcessorSegment, StopSegment, compose_outcome_segment,
};

// Task-local cancel token — set by the pipeline task per-start, checked by
// `run_steps` between steps. Absent in direct tests (skip check).
//
// Design: per-start task-local, NOT compiled into the pipeline struct, to
// avoid the lifecycle bug where a compiled-in child token stays cancelled
// after stop→restart (the new start would inherit the cancelled state).
// ADR-0043.
tokio::task_local! {
    pub(crate) static CANCEL_TOKEN: CancellationToken;
}

/// Runtime context for metrics + route_id (B3). Cancel is via task-local (B1).
#[derive(Clone)]
pub struct PipelineRuntimeCtx {
    pub metrics: Arc<dyn MetricsCollector>,
    pub route_id: Arc<str>,
}

impl PipelineRuntimeCtx {
    /// Constructor for compile-time contexts where no MetricsCollector is available.
    /// The resulting pipeline will emit disposition counters to NoOpMetrics (no-op).
    /// Prefer constructing PipelineRuntimeCtx with real metrics at route startup.
    pub fn compile_time() -> Self {
        Self {
            metrics: Arc::new(NoOpMetrics),
            route_id: Arc::from(""),
        }
    }
}

/// Newtype around `Arc<[CompiledStep]>` that adds `Send + Sync`.
///
/// `CompiledStep` contains `BoxProcessor` (tower `BoxCloneService`) which
/// is `Send + !Sync` — the `!Sync` is an artifact of Tower's trait-object
/// bounds (`Box<dyn ... + Send>` lacks `+ Sync`), NOT because of interior
/// mutability. Verified: no `Rc`, `RefCell`, `Cell`, or `UnsafeCell` in
/// `OutcomePipeline`, `OutcomeSegment`, or `BoxProcessor`.
///
/// SAFETY: Concurrent access DOES occur — multiple Exchanges read
/// `&CompiledStep` from the same Arc simultaneously across tokio worker
/// threads. This is sound because `run_steps` only reads shared references
/// and calls `.clone()` to obtain owned copies before invoking. Read-only
/// `&T` + `clone()` of types free of `UnsafeCell` is sound across threads.
///
/// INVARIANT: If any `CompiledStep` variant ever introduces a type backed
/// by `UnsafeCell` (`Rc`, `RefCell`, `Cell`), this unsafe impl becomes
/// unsound UB. The compile-time guard below ensures `CompiledStep: Send`;
/// there is no mechanical guard for `Sync` — that requires manual review
/// (see `shared_snapshot_is_send_sync` test).
#[derive(Clone)]
struct SharedSnapshot(Arc<[CompiledStep]>);

// SAFETY: see struct doc above. `Send` is required so the future returned
// by `run_steps` (which owns the snapshot) is itself `Send` and compatible
// with `BoxCloneService`'s `Pin<Box<dyn Future + Send>>` return type.
unsafe impl Send for SharedSnapshot {}

// SAFETY: see struct doc above. `Sync` is required because the snapshot
// is held behind `Arc` and may be read concurrently by `&self` access on
// `SequentialPipeline`/`TracedPipeline` clones (e.g. `poll_ready` on one
// thread, `call` on another).
unsafe impl Sync for SharedSnapshot {}

// Compile-time guard: CompiledStep must remain Send.
// If this fails, SharedSnapshot's unsafe impl Send becomes unsound.
#[allow(dead_code)]
const _: () = {
    fn assert_send<T: Send>() {}
    fn _check() {
        assert_send::<CompiledStep>();
    }
};

/// Compose a list of CompiledSteps into a sub-pipeline (EIP internal).
///
/// Uses `into_tower_result()` so `PipelineOutcome::Stopped` maps to `Ok(ex)`.
/// Use [`compose_pipeline_with_handler`] for the top-level consumer-facing pipeline.
pub fn compose_pipeline(processors: Vec<CompiledStep>, ctx: PipelineRuntimeCtx) -> BoxProcessor {
    if processors.is_empty() {
        return BoxProcessor::new(IdentityProcessor);
    }
    BoxProcessor::new(SequentialPipeline {
        steps: SharedSnapshot(processors.into()),
        handler: None,
        ctx,
    })
}

/// Compose a list of CompiledSteps with an optional route error handler.
///
/// When a handler is present, step readiness errors are swallowed (poll_ready
/// returns Ready) and the handler's retry/recovery logic is invoked on step
/// failures. Otherwise, step readiness errors propagate immediately.
pub fn compose_pipeline_with_handler(
    processors: Vec<CompiledStep>,
    handler: Option<Arc<dyn RouteErrorHandler>>,
    ctx: PipelineRuntimeCtx,
) -> BoxProcessor {
    if processors.is_empty() {
        return BoxProcessor::new(IdentityProcessor);
    }
    BoxProcessor::new(SequentialPipeline {
        steps: SharedSnapshot(processors.into()),
        handler,
        ctx,
    })
}

/// Compose a list of CompiledSteps into a traced pipeline with Stop→Ok translation.
///
/// Each processor is wrapped with TracingProcessor to emit spans for observability.
/// When tracing is disabled, falls back to [`compose_pipeline_with_handler`] with zero overhead.
pub fn compose_traced_pipeline(
    processors: Vec<CompiledStep>,
    route_id: &str,
    trace_enabled: bool,
    detail_level: DetailLevel,
    metrics: Option<Arc<dyn MetricsCollector>>,
    handler: Option<Arc<dyn RouteErrorHandler>>,
    ctx: PipelineRuntimeCtx,
) -> BoxProcessor {
    if !trace_enabled {
        return compose_pipeline_with_handler(processors, handler, ctx);
    }

    if processors.is_empty() {
        return BoxProcessor::new(IdentityProcessor);
    }

    let wrapped: Vec<CompiledStep> = processors
        .into_iter()
        .enumerate()
        .map(|(idx, step)| {
            let (p, c, lc) = match step {
                CompiledStep::Process {
                    processor,
                    body_contract,
                    lifecycle,
                } => (processor, body_contract, lifecycle),
                CompiledStep::Stop => return CompiledStep::Stop,
                CompiledStep::Segment { .. } => return step,
            };
            let traced = BoxProcessor::new(TracingProcessor::new(
                p,
                route_id.to_string(),
                idx,
                detail_level.clone(),
                metrics.clone(),
            ));
            CompiledStep::Process {
                processor: traced,
                body_contract: c,
                lifecycle: lc,
            }
        })
        .collect();

    BoxProcessor::new(TracedPipeline {
        steps: SharedSnapshot(wrapped.into()),
        handler,
        ctx,
    })
}

/// Compose a list of `CompiledStep` items into a single pipeline with body coercion.
///
/// Each processor is optionally wrapped with `BodyCoercingProcessor` based on its
/// contract. Processors with `None` contract are passed through with zero overhead.
/// `CompiledStep::Stop` passes through without coercion.
pub fn compose_pipeline_with_contracts(
    processors: Vec<CompiledStep>,
    handler: Option<Arc<dyn RouteErrorHandler>>,
    ctx: PipelineRuntimeCtx,
) -> BoxProcessor {
    let wrapped: Vec<CompiledStep> = processors
        .into_iter()
        .map(|step| match step {
            CompiledStep::Process {
                processor,
                body_contract,
                lifecycle,
            } => {
                let coerced = wrap_if_needed(processor, body_contract);
                CompiledStep::Process {
                    processor: coerced,
                    body_contract: None,
                    lifecycle,
                }
            }
            CompiledStep::Stop => CompiledStep::Stop,
            CompiledStep::Segment { .. } => step,
        })
        .collect();
    compose_pipeline_with_handler(wrapped, handler, ctx)
}

/// Compose a list of `CompiledStep` items into a traced pipeline with body coercion.
///
/// Applies body coercion contracts first, then wraps with `TracingProcessor`.
/// When tracing is disabled, falls back to [`compose_pipeline_with_contracts`].
pub(crate) fn compose_traced_pipeline_with_contracts(
    processors: Vec<CompiledStep>,
    route_id: &str,
    trace_enabled: bool,
    detail_level: DetailLevel,
    metrics: Option<Arc<dyn MetricsCollector>>,
    handler: Option<Arc<dyn RouteErrorHandler>>,
    ctx: PipelineRuntimeCtx,
) -> BoxProcessor {
    if !trace_enabled {
        return compose_pipeline_with_contracts(processors, handler, ctx);
    }

    if processors.is_empty() {
        return BoxProcessor::new(IdentityProcessor);
    }

    let wrapped: Vec<CompiledStep> = processors
        .into_iter()
        .enumerate()
        .map(|(idx, step)| match step {
            CompiledStep::Process {
                processor,
                body_contract,
                lifecycle,
            } => {
                let coerced = wrap_if_needed(processor, body_contract);
                let traced = BoxProcessor::new(TracingProcessor::new(
                    coerced,
                    route_id.to_string(),
                    idx,
                    detail_level.clone(),
                    metrics.clone(),
                ));
                CompiledStep::Process {
                    processor: traced,
                    body_contract: None,
                    lifecycle,
                }
            }
            CompiledStep::Stop => CompiledStep::Stop,
            CompiledStep::Segment { .. } => step,
        })
        .collect();

    BoxProcessor::new(TracedPipeline {
        steps: SharedSnapshot(wrapped.into()),
        handler,
        ctx,
    })
}

/// A service that executes a sequence of CompiledSteps in order.
///
/// Uses `into_tower_result()` so `PipelineOutcome::Stopped(ex)` maps to
/// `Ok(ex)` — the Bug B fix that makes Stop indistinguishable from Completed
/// at the consumer boundary.
#[derive(Clone)]
struct SequentialPipeline {
    steps: SharedSnapshot,
    handler: Option<Arc<dyn RouteErrorHandler>>,
    ctx: PipelineRuntimeCtx,
}

impl Service<Exchange> for SequentialPipeline {
    type Response = Exchange;
    type Error = CamelError;
    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        match self.steps.0.first() {
            Some(CompiledStep::Process { processor, .. }) => {
                let mut proc = processor.clone();
                match proc.poll_ready(cx) {
                    Poll::Pending => Poll::Pending,
                    Poll::Ready(Err(_)) if self.handler.is_some() => Poll::Ready(Ok(())),
                    Poll::Ready(other) => Poll::Ready(other),
                }
            }
            Some(CompiledStep::Stop) => Poll::Ready(Ok(())),
            Some(CompiledStep::Segment { .. }) => Poll::Ready(Ok(())),
            None => Poll::Ready(Ok(())),
        }
    }

    // ADR-0024 reply-channel adapter: PipelineOutcome → Result<Exchange, CamelError>.
    // Completed(ex) and Stopped(ex) both map to Ok(ex); Failed(err) maps to Err.
    // Downstream consumers (RouteChannelService, ExchangeUoWLayer, HTTP/Kafka reply
    // finalisers) see Result<Exchange, CamelError> and treat Stop as success.
    fn call(&mut self, exchange: Exchange) -> Self::Future {
        // Cheap Arc::clone (refcount bump) on the SharedSnapshot newtype.
        // `SharedSnapshot: Send` so the future returned by `run_steps`
        // captures it directly without needing a Send-asserting wrapper.
        let steps = self.steps.clone();
        let handler = self.handler.clone();
        let ctx = self.ctx.clone();
        Box::pin(async move {
            run_steps(steps, exchange, handler, false, &ctx)
                .await
                .into_tower_result()
        })
    }
}

/// A traced service pipeline for wrapped CompiledSteps.
#[derive(Clone)]
struct TracedPipeline {
    steps: SharedSnapshot,
    handler: Option<Arc<dyn RouteErrorHandler>>,
    ctx: PipelineRuntimeCtx,
}

impl Service<Exchange> for TracedPipeline {
    type Response = Exchange;
    type Error = CamelError;
    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        match self.steps.0.first() {
            Some(CompiledStep::Process { processor, .. }) => {
                let mut proc = processor.clone();
                match proc.poll_ready(cx) {
                    Poll::Pending => Poll::Pending,
                    Poll::Ready(Err(_)) if self.handler.is_some() => Poll::Ready(Ok(())),
                    Poll::Ready(other) => Poll::Ready(other),
                }
            }
            Some(CompiledStep::Stop) => Poll::Ready(Ok(())),
            Some(CompiledStep::Segment { .. }) => Poll::Ready(Ok(())),
            None => Poll::Ready(Ok(())),
        }
    }

    // ADR-0024 reply-channel adapter (same as SequentialPipeline::call):
    // Completed(ex) and Stopped(ex) both map to Ok(ex). Bug B fix.
    fn call(&mut self, exchange: Exchange) -> Self::Future {
        let steps = self.steps.clone();
        let handler = self.handler.clone();
        let ctx = self.ctx.clone();
        Box::pin(async move {
            run_steps(steps, exchange, handler, true, &ctx)
                .await
                .into_tower_result()
        })
    }
}

/// Run a sequence of CompiledSteps with optional error recovery.
///
/// Each step is unified under [`OwnedRetryable`] — both Process and
/// Segment variants are treated uniformly via a stack-allocated enum
/// that dispatches to the existing `RetryableStep` impls on
/// `BoxProcessor` and `OutcomeSegment`. This eliminates the per-step
/// `Box::new(...) as Box<dyn RetryableStep>` heap allocation that the
/// pre-A2 implementation paid for every step of every Exchange (A2).
///
/// On failure:
/// 1. If a handler is present, `match_policy` selects a retry policy.
/// 2. `retry_step` attempts recovery; if exhausted, `handle_step` determines
///    the disposition:
///    - `Propagate` — return the error
///    - `Handled` — return the exchange early (success)
///    - `Continued` — clear the error and continue to the next step
/// 3. If no handler is present, the error is propagated directly.
///
/// CompiledStep::Stop short-circuits to `PipelineOutcome::Stopped(ex)` — the
/// handler is bypassed and no Tower service is invoked (ADR-0024 §3.5).
async fn run_steps(
    steps: SharedSnapshot,
    exchange: Exchange,
    handler: Option<Arc<dyn RouteErrorHandler>>,
    trace: bool,
    ctx: &PipelineRuntimeCtx,
) -> PipelineOutcome {
    use camel_api::error_handler::RetryableStep;
    let mut ex = exchange;
    // Index-based loop (not `for (i, step) in steps.0.iter().enumerate()`) so
    // the future stays `Send`: the iterator holds `&[CompiledStep]`, but
    // `CompiledStep: !Sync` (Tower `BoxCloneService` + `dyn OutcomePipeline`
    // trait objects), so `&[CompiledStep]: !Send`. The slice reference must
    // not span any `.await`. `&steps.0[i]` is consumed by the `match` scrutinee
    // and drops before the await — no borrow is live across the await point.
    let len = steps.0.len();
    for i in 0..len {
        // B1: cooperative cancellation between steps via task-local.
        // If the task-local is not set (direct test calls), skip the check.
        let cancelled = CANCEL_TOKEN.try_with(|t| t.is_cancelled()).unwrap_or(false);
        if cancelled {
            return PipelineOutcome::Failed(CamelError::ConsumerStopping);
        }
        // A2: dispatch to existing `RetryableStep` impls through a stack
        // enum instead of paying `Box::new(...) as Box<dyn RetryableStep>`
        // per step. `OwnedRetryable` is `enum { Processor, Segment }` with
        // discriminant-by-value layout — no extra heap alloc.
        let mut retryable: OwnedRetryable = match &steps.0[i] {
            CompiledStep::Stop => return PipelineOutcome::Stopped(ex),
            CompiledStep::Process { processor, .. } => OwnedRetryable::Processor(processor.clone()),
            CompiledStep::Segment { segment, .. } => OwnedRetryable::Segment(segment.clone()),
        };

        let original = handler.as_ref().map(|_| ex.clone());
        let outcome = if trace {
            invoke_with_span(&mut retryable, ex, i).await
        } else {
            retryable.invoke(ex).await
        };

        match outcome {
            PipelineOutcome::Completed(next) => {
                if camel_api::is_camel_stop(&next) {
                    return PipelineOutcome::Stopped(next);
                }
                ex = next;
            }
            PipelineOutcome::Stopped(stopped_ex) => {
                return PipelineOutcome::Stopped(stopped_ex);
            }
            PipelineOutcome::Failed(err) => {
                let (Some(handler), Some(original)) = (handler.as_ref(), original) else {
                    return PipelineOutcome::Failed(err);
                };
                let policy = handler.match_policy(&err);
                // `&mut retryable` auto-coerces from `&mut OwnedRetryable` to
                // `&mut dyn RetryableStep` via the trait impl on the enum.
                match handler
                    .retry_step(policy, &mut retryable, original, err)
                    .await
                {
                    RetryOutcome::Recovered(exchange) => {
                        ctx.metrics.record_counter(
                            "pipeline_disposition",
                            1.0,
                            &[("disposition", "recovered"), ("route_id", &ctx.route_id)],
                        );
                        ex = exchange;
                    }
                    RetryOutcome::Stopped(stopped_ex) => {
                        ctx.metrics.record_counter(
                            "pipeline_disposition",
                            1.0,
                            &[("disposition", "stopped"), ("route_id", &ctx.route_id)],
                        );
                        return PipelineOutcome::Stopped(stopped_ex);
                    }
                    RetryOutcome::Exhausted {
                        exchange,
                        error,
                        policy,
                    } => {
                        let disposition = if trace {
                            handler
                                .handle_step(policy, exchange, error)
                                .instrument(tracing::debug_span!("error_handler", step_index = i))
                                .await
                        } else {
                            handler.handle_step(policy, exchange, error).await
                        };
                        match disposition {
                            Ok(StepDisposition::Propagate(e)) => {
                                ctx.metrics.record_counter(
                                    "pipeline_disposition",
                                    1.0,
                                    &[("disposition", "propagated"), ("route_id", &ctx.route_id)],
                                );
                                return PipelineOutcome::Failed(e);
                            }
                            Ok(StepDisposition::Handled(done)) => {
                                ctx.metrics.record_counter(
                                    "pipeline_disposition",
                                    1.0,
                                    &[("disposition", "handled"), ("route_id", &ctx.route_id)],
                                );
                                return PipelineOutcome::Completed(done);
                            }
                            Ok(StepDisposition::Continued(next)) => {
                                ctx.metrics.record_counter(
                                    "pipeline_disposition",
                                    1.0,
                                    &[("disposition", "continued"), ("route_id", &ctx.route_id)],
                                );
                                ex = next;
                            }
                            Err(e) => {
                                ctx.metrics.record_counter(
                                    "pipeline_disposition",
                                    1.0,
                                    &[
                                        ("disposition", "handler_error"),
                                        ("route_id", &ctx.route_id),
                                    ],
                                );
                                return PipelineOutcome::Failed(e);
                            }
                            // Future StepDisposition variants fail the pipeline.
                            _ => {
                                return PipelineOutcome::Failed(CamelError::ProcessorError(
                                    "unknown step disposition".to_string(),
                                ));
                            }
                        }
                    }
                    // Future RetryOutcome variants fail the pipeline.
                    _ => {
                        return PipelineOutcome::Failed(CamelError::ProcessorError(
                            "unknown retry outcome".to_string(),
                        ));
                    }
                }
            }
        }
    }
    PipelineOutcome::Completed(ex)
}

/// Stack-allocated dispatcher that unifies `BoxProcessor` and
/// `OutcomeSegment` for the retry path without the heap allocation a
/// `Box<dyn RetryableStep>` would require. Sized by-value, dispatched
/// through a single trait method that fans out to the existing
/// `RetryableStep` impls on each variant.
///
/// A2: replaces `Box::new(processor.clone()) as Box<dyn RetryableStep>`
/// (and the equivalent for segments) with this enum, saving one heap
/// allocation per pipeline step per Exchange invocation.
enum OwnedRetryable {
    Processor(camel_api::BoxProcessor),
    Segment(camel_api::OutcomeSegment),
}

impl camel_api::error_handler::RetryableStep for OwnedRetryable {
    fn invoke<'a>(
        &'a mut self,
        exchange: Exchange,
    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
        match self {
            OwnedRetryable::Processor(p) => p.invoke(exchange),
            OwnedRetryable::Segment(s) => s.invoke(exchange),
        }
    }
}

async fn invoke_with_span(
    retryable: &mut dyn camel_api::error_handler::RetryableStep,
    exchange: Exchange,
    idx: usize,
) -> PipelineOutcome {
    retryable
        .invoke(exchange)
        .instrument(tracing::debug_span!("pipeline_step", index = idx))
        .await
}

/// Route channel with explicit security and circuit-breaker gates.
///
/// Gate order: Security → CB(before_call) → Pipeline → CB(after_result).
/// Errors from Security/CB gates go to `handler.handle_boundary`.
/// Errors from Pipeline go through the injected handler's retry/handle_step.
/// Pipeline Propagate returns Err — passed through to upstream.
#[derive(Clone)]
pub struct RouteChannelService {
    handler: Arc<dyn RouteErrorHandler>,
    security: Option<BoxProcessor>,
    cb_gate: Option<CircuitBreakerGate>,
    pipeline: BoxProcessor,
    /// When true, stash the original Message as `ORIGINAL_MESSAGE_EXTENSION`
    /// before any gate runs, so the error handler can restore it on failure.
    use_original_message: bool,
}

impl RouteChannelService {
    pub fn new(
        handler: Arc<dyn RouteErrorHandler>,
        security: Option<BoxProcessor>,
        cb_gate: Option<CircuitBreakerGate>,
        pipeline: BoxProcessor,
        use_original_message: bool,
    ) -> Self {
        Self {
            handler,
            security,
            cb_gate,
            pipeline,
            use_original_message,
        }
    }
}

impl Service<Exchange> for RouteChannelService {
    type Response = Exchange;
    type Error = CamelError;
    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), CamelError>> {
        // Swallow readiness errors from security gate — deferred to call()
        if let Some(ref mut sec) = self.security {
            match sec.clone().poll_ready(cx) {
                Poll::Pending => return Poll::Pending,
                Poll::Ready(Err(_)) | Poll::Ready(Ok(())) => {}
            }
        }
        // Pipeline readiness — swallow errors when handler present
        match self.pipeline.clone().poll_ready(cx) {
            Poll::Pending => return Poll::Pending,
            Poll::Ready(Err(_)) | Poll::Ready(Ok(())) => {}
        }
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, exchange: Exchange) -> Self::Future {
        let handler = self.handler.clone();
        let security = self.security.clone();
        let cb_gate = self.cb_gate.clone();
        let mut pipeline = self.pipeline.clone();
        let use_original_message = self.use_original_message;

        Box::pin(async move {
            let mut ex = exchange;

            // Stash original message for use_original_message support.
            // Done BEFORE any gate so the DLC can restore the pre-route message.
            // Only stashes when the flag is true to avoid perf regression on every Exchange.
            if use_original_message {
                let original: Arc<Message> = Arc::new(ex.input.clone());
                ex.set_extension(ORIGINAL_MESSAGE_EXTENSION, original);
            }

            // Gate 1: Security
            if let Some(mut sec) = security {
                let original = ex.clone();
                match invoke_processor(&mut sec, ex).await {
                    Ok(next) => ex = next,
                    Err(err) => {
                        return handler
                            .handle_boundary(BoundaryKind::Security, original, err)
                            .await;
                    }
                }
            }

            // Gate 2: CircuitBreaker — before_call
            if let Some(ref cb) = cb_gate {
                match cb.before_call() {
                    CircuitBreakerDecision::Allow => { /* proceed to pipeline */ }
                    CircuitBreakerDecision::Fallback(mut fb) => {
                        // Circuit open with fallback — call fallback.
                        // Fallback errors go through handle_boundary, not raw to upstream.
                        let original = ex.clone();
                        match invoke_processor(&mut fb, ex).await {
                            Ok(result) => return Ok(result),
                            Err(err) => {
                                return handler
                                    .handle_boundary(BoundaryKind::CircuitBreaker, original, err)
                                    .await;
                            }
                        }
                    }
                    CircuitBreakerDecision::Reject(err) => {
                        let original = ex.clone();
                        return handler
                            .handle_boundary(BoundaryKind::CircuitBreaker, original, err)
                            .await;
                    }
                }
            }

            // Pipeline (handler already injected for step errors)
            let result = invoke_processor(&mut pipeline, ex).await;

            // Gate 2: CircuitBreaker — after_result
            if let Some(ref cb) = cb_gate {
                cb.after_result(&result);
            }

            // Propagate from inner handler — pass through to upstream
            result
        })
    }
}

#[cfg(test)]
#[path = "route_compiler_tests.rs"]
mod tests;