camel_core/lifecycle/adapters/

route_controller.rs

//! Default implementation of RouteController.
//!
//! This module provides [`DefaultRouteController`], which manages route lifecycle
//! including starting, stopping, suspending, and resuming routes.

use std::collections::HashMap;
use std::sync::{Arc, Weak};
use std::time::Duration;

use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tower::{Layer, ServiceExt};
use tracing::{debug, info, warn};

use super::route_controller_trait::{
    BindExposureAcks, bind_key_from_uri, enforce_bind_exposure_gate,
};
use camel_api::error_handler::ErrorHandlerConfig;
use camel_api::metrics::MetricsCollector;
#[allow(unused_imports)]
use camel_api::{
    BoxProcessor, CamelError, Exchange, FunctionInvoker, IdentityProcessor, InFlightClaim,
    InFlightGauge, NoOpMetrics, NoopPlatformService, PlatformService, ProducerContext,
    RouteController, RuntimeHandle, StepLifecycle,
};
use camel_component_api::{Consumer, ConsumerContext, consumer::ExchangeEnvelope};
use camel_processor::aggregator::AggregatorService;
pub use camel_processor::aggregator::SharedLanguageRegistry;
use camel_processor::aggregator::{AggregateEmission, AggregationReceipt};

use crate::health_registry::HealthCheckRegistry;
use crate::intercept::InterceptRules;
use crate::lifecycle::CohortActivationGate;
use crate::lifecycle::adapters::controller_component_context::ControllerComponentContext;
use crate::lifecycle::adapters::route_compiler::TracerPipelineGating;
use crate::lifecycle::adapters::route_compiler_ext::{
    RouteCompilerExt, build_eh_config_pipeline, transport_from_uri,
};
use crate::lifecycle::adapters::route_helpers::{
    AggregateSplitInfo, CrashNotification, ManagedRoute, assert_no_mixed_top_level_splits,
    handle_is_running, inferred_lifecycle_label, is_pending, send_reply_or_b_prime,
};
#[cfg(test)]
pub(super) use crate::lifecycle::adapters::route_helpers::{
    emit_start_route_event, set_start_route_event_hook,
};
use crate::lifecycle::adapters::route_registry::RouteRegistry;
use crate::lifecycle::adapters::route_runtime_state;
use crate::lifecycle::adapters::step_compilers::CompiledStep;
use crate::lifecycle::application::route_definition::{BuilderStep, RouteDefinition};
pub(crate) use crate::lifecycle::domain::CompiledPipeline;
use crate::shared::components::domain::Registry;
use crate::shared::observability::domain::{DetailLevel, TracerConfig};
use camel_bean::BeanRegistry;

/// Default implementation of [`RouteController`].
///
/// Manages route lifecycle with support for:
/// - Starting/stopping individual routes
/// - Suspending and resuming routes
/// - Auto-startup with startup ordering
/// - Graceful shutdown
pub struct DefaultRouteController {
    /// Routes indexed by route ID.
    pub(super) routes: RouteRegistry,
    /// Reference to the component registry for resolving endpoints.
    pub(super) registry: Arc<std::sync::Mutex<Registry>>,
    /// Shared language registry for resolving declarative language expressions.
    pub(super) languages: SharedLanguageRegistry,
    /// Bean registry for bean method invocation.
    pub(super) beans: Arc<std::sync::Mutex<BeanRegistry>>,
    /// Runtime handle injected into ProducerContext for command/query operations.
    pub(super) runtime: Option<Weak<dyn RuntimeHandle>>,
    /// Optional global error handler applied to all routes without a per-route handler.
    pub(super) global_error_handler: Option<ErrorHandlerConfig>,
    /// Optional crash notifier for supervision.
    pub(super) crash_notifier: Option<mpsc::Sender<CrashNotification>>,
    /// Whether tracing is enabled for route pipelines.
    pub(super) tracer_gating: TracerPipelineGating,
    /// Detail level for tracing when enabled.
    pub(super) tracer_detail_level: DetailLevel,
    /// Metrics collector for tracing processor.
    pub(super) tracer_metrics: Option<Arc<dyn MetricsCollector>>,
    pub(super) platform_service: Arc<dyn PlatformService>,
    pub(super) function_invoker: Option<Arc<dyn FunctionInvoker>>,
    pub(super) health_registry: Option<Arc<HealthCheckRegistry>>,
    /// Shared idempotent repository registry. Defaults to an empty registry;
    /// the CamelContext builder installs a populated handle that includes the
    /// built-in `"memory"` repository.
    pub(super) idempotent_repositories: crate::SharedIdempotentRegistry,
    pub(super) claim_check_repositories: crate::SharedClaimCheckRegistry,
    pub(super) cache_repositories: crate::SharedCacheRegistry,
    /// F2 staging: prepared-but-not-inserted ManagedRoutes keyed by route_id.
    /// `prepare_*` writes here; `insert_prepared_route` drains via `remove()`.
    /// On insert-failure error paths, the caller (`reload_actions.rs`) must
    /// explicitly drain to avoid orphan CancellationToken/SharedPipeline leaks.
    pub(super) prepared_staging: HashMap<String, ManagedRoute>,
    /// Source endpoint URI to route_id index (one-to-many).
    pub(super) endpoint_index: super::endpoint_index::EndpointIndex,
    /// Operator acknowledgements for per-bind public exposure (ADR-0061).
    /// Empty by default → the gate fails closed on non-loopback binds.
    pub(super) bind_acks: super::route_controller_trait::BindExposureAcks,
    /// Route send-point interception rules captured by step compilation.
    pub(super) intercept: InterceptRules,
    /// Intercept-rules freeze. Trips on `add_route` success and on the
    /// `MarkStarted` actor command; never reset (stop/restart included),
    /// because compiled pipelines capture the rules at compile time.
    pub(super) frozen: bool,
    /// Startup-cohort activation barrier (rc-jxkj). Cloned into the
    /// `RouteControllerHandle` at spawn; reset/activate act on this shared
    /// gate directly, never through the actor.
    pub(super) cohort: Arc<CohortActivationGate>,
    /// Context-global accepted-not-completed gauge (drainclaim): the
    /// SAME `Arc` the owning `CamelContext` exposes through
    /// `total_in_flight()` (installed by the builder). Standalone
    /// controllers keep an isolated zero gauge — claims flow, but only
    /// the constructing scope can read them.
    pub(super) in_flight_total: Arc<InFlightGauge>,
}

impl DefaultRouteController {
    /// Open the startup-cohort activation barrier (rc-jxkj).
    ///
    /// The CamelContext lifecycle opens this automatically once the startup
    /// cohort completes. Consumers that drive a bare `DefaultRouteController`
    /// (outside a full context) must call this before dispatching
    /// (typically after starting routes), or pipeline dispatch parks
    /// every envelope until the caller's call timeout surfaces as a
    /// failure.
    pub fn activate_cohort(&self) {
        self.cohort.open();
    }

    pub(super) fn health_registry(&self) -> Arc<HealthCheckRegistry> {
        self.health_registry.clone().unwrap_or_else(|| {
            debug!("health_registry not configured — creating isolated fallback");
            Arc::new(HealthCheckRegistry::new(Duration::from_secs(5)))
        })
    }

    /// Create a new `DefaultRouteController` with the given registry.
    pub fn new(
        registry: Arc<std::sync::Mutex<Registry>>,
        platform_service: Arc<dyn PlatformService>,
    ) -> Self {
        Self::with_beans_and_platform_service(
            registry,
            Arc::new(std::sync::Mutex::new(BeanRegistry::new())),
            platform_service,
        )
    }

    /// Create a new `DefaultRouteController` with shared bean registry.
    pub fn with_beans(
        registry: Arc<std::sync::Mutex<Registry>>,
        beans: Arc<std::sync::Mutex<BeanRegistry>>,
    ) -> Self {
        Self::with_beans_and_platform_service(
            registry,
            beans,
            Arc::new(NoopPlatformService::default()),
        )
    }

    fn with_beans_and_platform_service(
        registry: Arc<std::sync::Mutex<Registry>>,
        beans: Arc<std::sync::Mutex<BeanRegistry>>,
        platform_service: Arc<dyn PlatformService>,
    ) -> Self {
        Self {
            routes: RouteRegistry::new(),
            registry,
            languages: Arc::new(std::sync::Mutex::new(HashMap::new())),
            beans,
            runtime: None,
            global_error_handler: None,
            crash_notifier: None,
            tracer_gating: TracerPipelineGating::off(),
            tracer_detail_level: DetailLevel::Minimal,
            tracer_metrics: None,
            platform_service,
            function_invoker: None,
            health_registry: None,
            idempotent_repositories: Arc::new(crate::IdempotentRegistry::new()),
            claim_check_repositories: Arc::new(crate::ClaimCheckRegistry::new()),
            cache_repositories: Arc::new(crate::CacheRegistry::new()),
            prepared_staging: HashMap::new(),
            endpoint_index: super::endpoint_index::EndpointIndex::new(),
            bind_acks: Default::default(),
            intercept: InterceptRules::default(),
            frozen: false,
            cohort: Arc::new(CohortActivationGate::new_closed()),
            in_flight_total: Arc::new(InFlightGauge::new()),
        }
    }

    /// Create a new `DefaultRouteController` with shared language registry.
    pub fn with_languages(
        registry: Arc<std::sync::Mutex<Registry>>,
        languages: SharedLanguageRegistry,
        platform_service: Arc<dyn PlatformService>,
    ) -> Self {
        Self {
            routes: RouteRegistry::new(),
            registry,
            languages,
            beans: Arc::new(std::sync::Mutex::new(BeanRegistry::new())),
            runtime: None,
            global_error_handler: None,
            crash_notifier: None,
            tracer_gating: TracerPipelineGating::off(),
            tracer_detail_level: DetailLevel::Minimal,
            tracer_metrics: None,
            platform_service,
            function_invoker: None,
            health_registry: None,
            idempotent_repositories: Arc::new(crate::IdempotentRegistry::new()),
            claim_check_repositories: Arc::new(crate::ClaimCheckRegistry::new()),
            cache_repositories: Arc::new(crate::CacheRegistry::new()),
            prepared_staging: HashMap::new(),
            endpoint_index: super::endpoint_index::EndpointIndex::new(),
            bind_acks: Default::default(),
            intercept: InterceptRules::default(),
            frozen: false,
            cohort: Arc::new(CohortActivationGate::new_closed()),
            in_flight_total: Arc::new(InFlightGauge::new()),
        }
    }

    pub fn with_languages_and_beans(
        registry: Arc<std::sync::Mutex<Registry>>,
        languages: SharedLanguageRegistry,
        platform_service: Arc<dyn PlatformService>,
        beans: Arc<std::sync::Mutex<BeanRegistry>>,
    ) -> Self {
        Self {
            routes: RouteRegistry::new(),
            registry,
            languages,
            beans,
            runtime: None,
            global_error_handler: None,
            crash_notifier: None,
            tracer_gating: TracerPipelineGating::off(),
            tracer_detail_level: DetailLevel::Minimal,
            tracer_metrics: None,
            platform_service,
            function_invoker: None,
            health_registry: None,
            idempotent_repositories: Arc::new(crate::IdempotentRegistry::new()),
            claim_check_repositories: Arc::new(crate::ClaimCheckRegistry::new()),
            cache_repositories: Arc::new(crate::CacheRegistry::new()),
            prepared_staging: HashMap::new(),
            endpoint_index: super::endpoint_index::EndpointIndex::new(),
            bind_acks: Default::default(),
            intercept: InterceptRules::default(),
            frozen: false,
            cohort: Arc::new(CohortActivationGate::new_closed()),
            in_flight_total: Arc::new(InFlightGauge::new()),
        }
    }

    pub fn with_function_invoker(mut self, function_invoker: Arc<dyn FunctionInvoker>) -> Self {
        self.function_invoker = Some(function_invoker);
        self
    }

    pub(crate) fn set_idempotent_repositories(
        &mut self,
        repositories: crate::SharedIdempotentRegistry,
    ) {
        self.idempotent_repositories = repositories;
    }

    pub(crate) fn set_claim_check_repositories(
        &mut self,
        repositories: crate::SharedClaimCheckRegistry,
    ) {
        self.claim_check_repositories = repositories;
    }

    pub(crate) fn set_cache_repositories(&mut self, repositories: crate::SharedCacheRegistry) {
        self.cache_repositories = repositories;
    }

    pub fn set_health_registry(&mut self, registry: Arc<HealthCheckRegistry>) {
        self.health_registry = Some(registry);
    }

    pub fn set_function_invoker(&mut self, invoker: Arc<dyn FunctionInvoker>) {
        self.function_invoker = Some(invoker);
    }

    /// Set runtime handle for ProducerContext creation.
    pub fn set_runtime_handle(&mut self, runtime: Arc<dyn RuntimeHandle>) {
        self.runtime = Some(Arc::downgrade(&runtime));
    }

    /// Set the crash notifier for supervision.
    ///
    /// When set, the controller will send a `CrashNotification` whenever
    /// a consumer crashes.
    pub fn set_crash_notifier(&mut self, tx: mpsc::Sender<CrashNotification>) {
        self.crash_notifier = Some(tx);
    }

    /// Set a global error handler applied to all routes without a per-route handler.
    /// Install operator acknowledgements for per-bind public exposure
    /// (ADR-0061). Built by the CLI from `CamelConfig.binds`.
    pub fn set_bind_exposure_acks(&mut self, acks: BindExposureAcks) {
        self.bind_acks = acks;
    }

    /// Install route send-point interception rules (pre-first-use only).
    ///
    /// Fails with `CamelError::Config` once the freeze has tripped: compiled
    /// pipelines capture the rules at compile time, so the rule set must not
    /// change after a route is added or the context is started.
    pub fn set_intercept_rules(&mut self, rules: InterceptRules) -> Result<(), CamelError> {
        if self.frozen {
            return Err(CamelError::Config(
                "intercept rules are frozen: a route was added or the context was started; \
                 rules cannot be changed after first use"
                    .into(),
            ));
        }
        self.intercept = rules;
        Ok(())
    }

    /// Builder-style build-time configuration on a fresh controller (which
    /// is never frozen); mirrors `with_function_invoker`.
    pub fn with_intercept_rules(mut self, rules: InterceptRules) -> Self {
        self.intercept = rules;
        self
    }

    /// Trip the intercept-rules freeze. Dispatched by the `MarkStarted`
    /// actor command so the freeze applies even with zero routes. Never
    /// unset.
    pub fn mark_started(&mut self) {
        self.frozen = true;
    }

    /// All compiled security plans whose routes bind the same listener
    /// address (bind key), for the per-bind exposure gate. Routes still
    /// staging (no plan yet) are skipped — classification failures already
    /// aborted their own staging (Task 1.8).
    pub(super) fn plans_for_bind(
        &self,
        bind_key: &str,
    ) -> Vec<(String, camel_api::security_policy::RouteSecurityPlan)> {
        self.routes
            .iter()
            .filter_map(|(route_id, managed)| {
                let plan = managed.compiled.security_plan.as_ref()?;
                let bind = bind_key_from_uri(&managed.from_uri)?;
                if bind.key == bind_key {
                    Some((route_id.clone(), plan.clone()))
                } else {
                    None
                }
            })
            .collect()
    }

    /// Return whether a listener for `bind_key` is already serving a route.
    ///
    /// This is deliberately based on the consumer handle, rather than on the
    /// presence of a compiled route: the exposure gate applies only to late
    /// registration against an already-running bind (Task 2.2). Startup and
    /// resume perform their own gate checks in the lifecycle implementation.
    fn bind_is_running(&self, bind_key: &str) -> bool {
        self.routes.iter().any(|(_, managed)| {
            bind_key_from_uri(&managed.from_uri).is_some_and(|bind| {
                bind.key == bind_key && handle_is_running(&managed.consumer_handle)
            })
        })
    }

    /// Gate a route before it is inserted into a listener that is already
    /// running. The candidate is included with all sibling plans so the gate
    /// has the same aggregation semantics as the start path.
    fn enforce_late_registration_gate(
        &self,
        from_uri: &str,
        route_id: &str,
        plan: Option<&camel_api::security_policy::RouteSecurityPlan>,
    ) -> Result<(), CamelError> {
        let Some(bind) = bind_key_from_uri(from_uri) else {
            return Ok(());
        };
        if !self.bind_is_running(&bind.key) {
            return Ok(());
        }

        let mut owned = self.plans_for_bind(&bind.key);
        if let Some(plan) = plan {
            owned.push((route_id.to_string(), plan.clone()));
        }
        let plans: Vec<(&str, &camel_api::security_policy::RouteSecurityPlan)> =
            owned.iter().map(|(id, plan)| (id.as_str(), plan)).collect();
        enforce_bind_exposure_gate(
            &bind.key,
            bind.loopback,
            &plans,
            self.bind_acks.acknowledged(&bind.key),
        )
        .map_err(|err| {
            CamelError::RouteError(format!(
                "late registration of route '{route_id}' rejected for bind '{}': {err}",
                bind.key
            ))
        })
    }

    pub fn set_error_handler(&mut self, config: ErrorHandlerConfig) {
        self.global_error_handler = Some(config);
    }

    /// Configure tracing for this route controller.
    pub fn set_tracer_config(&mut self, config: &TracerConfig) {
        // Pipeline wrapping follows tracing unless the effective assembly
        // raised it (exporter active); spans follow `enabled` alone.
        self.tracer_gating = TracerPipelineGating {
            pipeline_enabled: config.enabled || config.pipeline_enabled,
            spans_enabled: config.enabled,
            levers: config.metrics_levers.clone(),
        };
        self.tracer_detail_level = config.detail_level.clone();
    }

    /// Seed the tracer metrics collector — the shared late-bound
    /// `MetricsHandle` built once by `CamelContextBuilder::build()`.
    ///
    /// Replaces the deleted `TracerConfig.metrics_collector` snapshot
    /// injection: the collector is wired here, at construction, and late
    /// registrations flow through the handle without re-snapshotting.
    pub fn set_tracer_metrics(&mut self, metrics: Arc<dyn MetricsCollector>) {
        self.tracer_metrics = Some(metrics);
    }

    /// Install the context-global accepted-not-completed gauge
    /// (drainclaim) — the SAME `Arc<InFlightGauge>` the owning
    /// `CamelContext` reads through `total_in_flight()`. Called once by
    /// `CamelContextBuilder::build()`; contexts built without a builder
    /// (or standalone controllers) keep the isolated zero gauge from
    /// construction.
    pub fn set_in_flight_total(&mut self, counter: Arc<InFlightGauge>) {
        self.in_flight_total = counter;
    }

    fn build_producer_context(&self, route_id: &str) -> Result<ProducerContext, CamelError> {
        let mut producer_ctx = ProducerContext::new().with_route_id(route_id);
        if let Some(runtime) = self.runtime.as_ref().and_then(Weak::upgrade) {
            producer_ctx = producer_ctx.with_runtime(runtime);
        }
        Ok(producer_ctx)
    }

    /// Create a transient [`RouteCompilerExt`] from this controller's fields.
    fn route_compiler_ext(&self) -> RouteCompilerExt<'_> {
        RouteCompilerExt {
            registry: &self.registry,
            languages: &self.languages,
            beans: &self.beans,
            function_invoker: &self.function_invoker,
            tracer_gating: self.tracer_gating.clone(),
            tracer_detail_level: &self.tracer_detail_level,
            tracer_metrics: &self.tracer_metrics,
            platform_service: &self.platform_service,
            runtime: &self.runtime,
            global_error_handler: &self.global_error_handler,
            health_registry: &self.health_registry,
            route_registry: &self.routes,
            idempotent_repositories: Arc::clone(&self.idempotent_repositories),
            claim_check_repositories: Arc::clone(&self.claim_check_repositories),
            cache_repositories: Arc::clone(&self.cache_repositories),
            intercept: &self.intercept,
            in_flight_total: Arc::clone(&self.in_flight_total),
        }
    }

    /// Resolve BuilderSteps into BoxProcessors.
    #[allow(dead_code)] // used by tests and may be needed for future split paths
    pub(crate) fn resolve_steps(
        &self,
        steps: Vec<BuilderStep>,
        producer_ctx: &ProducerContext,
        registry: &Arc<std::sync::Mutex<Registry>>,
        route_id: Option<&str>,
        staging_mode: &super::step_resolution::FunctionStagingMode,
    ) -> Result<Vec<CompiledStep>, CamelError> {
        let component_ctx = Arc::new(ControllerComponentContext::new(
            Arc::clone(registry),
            Arc::clone(&self.languages),
            self.tracer_metrics
                .clone()
                .unwrap_or_else(|| Arc::new(NoOpMetrics)),
            Arc::clone(&self.platform_service),
            self.health_registry(),
            route_id.map(|s| s.to_string()),
            self.tracer_gating.levers.components_enabled(),
        ));
        let rt: Arc<dyn camel_component_api::RuntimeObservability> =
            Arc::clone(&component_ctx) as Arc<_>;

        super::step_resolution::resolve_steps(
            steps,
            producer_ctx,
            rt,
            registry,
            &self.languages,
            &self.beans,
            self.function_invoker.clone(),
            component_ctx,
            route_id,
            staging_mode,
            &self.idempotent_repositories,
            &self.claim_check_repositories,
            &self.cache_repositories,
            self.intercept.clone(),
        )
    }

    /// Add a route definition to the controller.
    ///
    /// Steps are resolved immediately using the registry.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - A route with the same ID already exists
    /// - Step resolution fails
    pub async fn add_route(&mut self, definition: RouteDefinition) -> Result<(), CamelError> {
        let route_id = definition.route_id().to_string();
        let from_uri = definition.from_uri().to_string();

        if self.routes.contains_key(&route_id) {
            return Err(CamelError::RouteError(format!(
                "duplicate route ID '{route_id}'"
            )));
        }

        debug!(route_id = %route_id, "Adding route to controller");

        let managed = match self.build_managed_route(
            definition,
            &super::step_resolution::FunctionStagingMode::DirectAdd,
        ) {
            Ok(managed) => managed,
            Err(err) => {
                self.discard_function_staging();
                return Err(err);
            }
        };

        // A running listener can observe a newly inserted route immediately.
        // Gate the complete sibling set before committing function staging or
        // inserting the route, so a rejected late registration is unreachable.
        if let Err(err) = self.enforce_late_registration_gate(
            &from_uri,
            &route_id,
            managed.compiled.security_plan.as_ref(),
        ) {
            self.discard_function_staging();
            return Err(err);
        }

        if let Some(invoker) = &self.function_invoker
            && let Err(err) = invoker.commit_staged().await
        {
            invoker.discard_staging(0);
            return Err(CamelError::Config(err.to_string()));
        }

        self.routes
            .insert(managed.definition.route_id().to_string(), managed);

        self.endpoint_index.insert(&from_uri, &route_id);
        // First successful route registration freezes the intercept rules:
        // this route's pipeline compiled against the current rule set.
        self.frozen = true;
        Ok(())
    }

    pub(super) fn build_managed_route(
        &self,
        definition: RouteDefinition,
        staging_mode: &super::step_resolution::FunctionStagingMode,
    ) -> Result<ManagedRoute, CamelError> {
        let route_id = definition.route_id().to_string();

        let definition_info = definition.to_info();

        // Security plan compilation (Task 1.8): consumer-backed routes get a
        // plan BEFORE any consumer starts; a declared route that fails
        // classification aborts staging (never a Public downgrade). Routes
        // without a provider registry compile against an empty view.
        let empty_providers;
        let providers = match &definition.provider_registry {
            Some(registry) => registry.as_ref(),
            None => {
                empty_providers = camel_auth::ProviderRegistry::new();
                &empty_providers
            }
        };
        let security_plan =
            super::route_compiler_ext::compile_route_security_plan(&definition, providers)?;

        let RouteDefinition {
            from_uri,
            steps,
            error_handler,
            circuit_breaker,
            circuit_breaker_fallback,
            security_policy,
            security_authenticator,
            provider_registry,
            unit_of_work,
            concurrency,
            ..
        } = definition;

        let producer_ctx = self.build_producer_context(&route_id)?;

        // N2: reject mixed Aggregate + Resequence top-level splits
        assert_no_mixed_top_level_splits(&steps)?;

        let (aggregate_split, processors_with_contracts) = self
            .route_compiler_ext()
            .detect_and_validate_route_split(steps, &producer_ctx, &route_id, staging_mode)?;
        let mut lifecycle = super::route_helpers::collect_lifecycle(&processors_with_contracts);

        // CB fallback (mirrors on_miss lifecycle packing): attach via the
        // shared helper, then merge the fallback lifecycle handles into the
        // route vec.
        let (circuit_breaker, fallback_lifecycle) = self.route_compiler_ext().attach_cb_fallback(
            circuit_breaker,
            circuit_breaker_fallback,
            &producer_ctx,
            &route_id,
            staging_mode,
        )?;
        lifecycle.extend(fallback_lifecycle);
        let route_id_for_tracing = route_id.clone();
        let eh_config = error_handler.or_else(|| self.global_error_handler.clone());
        let transport = transport_from_uri(&from_uri);

        let mut pipeline = build_eh_config_pipeline(
            eh_config.as_ref(),
            Arc::clone(&self.registry),
            Arc::clone(&self.languages),
            self.tracer_metrics.clone(),
            Arc::clone(&self.platform_service),
            self.health_registry(),
            &route_id_for_tracing,
            &producer_ctx,
            processors_with_contracts,
            self.tracer_gating.clone(),
            self.tracer_detail_level.clone(),
            security_policy.clone(),
            transport,
            circuit_breaker,
            Arc::clone(&self.in_flight_total),
        )?;

        let uow_counter = if let Some(uow_config) = &unit_of_work {
            let component_ctx = Arc::new(
                ControllerComponentContext::new(
                    Arc::clone(&self.registry),
                    Arc::clone(&self.languages),
                    self.tracer_metrics
                        .clone()
                        .unwrap_or_else(|| Arc::new(NoOpMetrics)),
                    Arc::clone(&self.platform_service),
                    self.health_registry(),
                    Some(route_id.clone()),
                    self.tracer_gating.levers.components_enabled(),
                )
                .with_in_flight(Arc::clone(&self.in_flight_total)),
            );
            let rt: Arc<dyn camel_component_api::RuntimeObservability> =
                Arc::clone(&component_ctx) as Arc<_>;
            let (uow_layer, counter) = super::route_compiler_ext::resolve_uow_layer(
                uow_config,
                &producer_ctx,
                rt,
                component_ctx.as_ref(),
                None,
            )?;
            pipeline = BoxProcessor::new(uow_layer.layer(pipeline));
            Some(counter)
        } else {
            None
        };

        Ok(ManagedRoute {
            definition: definition_info,
            from_uri,
            pipeline: super::pipeline_runtime::new_shared_pipeline_with_lifecycle(
                pipeline, lifecycle,
            ),
            concurrency,
            consumer_handle: None,
            pipeline_handle: None,
            consumer_cancel_token: CancellationToken::new(),
            pipeline_cancel_token: CancellationToken::new(),
            channel_sender: None,
            in_flight: uow_counter,
            drain_in_flight: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            aggregate_split,
            agg_service: None,
            stop_budget: None,
            compiled: route_runtime_state::CompiledRoute {
                security_policy,
                security_authenticator,
                provider_registry,
                security_plan,
            },
        })
    }

    pub async fn add_route_with_generation(
        &mut self,
        definition: RouteDefinition,
        generation: u64,
    ) -> Result<(), CamelError> {
        let route_id = definition.route_id().to_string();
        let from_uri = definition.from_uri().to_string();

        if self.routes.contains_key(&route_id) {
            return Err(CamelError::RouteError(format!(
                "duplicate route ID '{route_id}'"
            )));
        }

        debug!(route_id = %route_id, generation, "Adding route to controller with generation");

        let managed = self.build_managed_route(
            definition,
            &super::step_resolution::FunctionStagingMode::HotReload { generation },
        )?;

        // Symmetric with `add_route`: an already-running bind must never
        // observe an ungated insertion through the hot-reload path.
        if let Err(err) = self.enforce_late_registration_gate(
            &from_uri,
            &route_id,
            managed.compiled.security_plan.as_ref(),
        ) {
            self.discard_function_staging();
            return Err(err);
        }

        self.routes.insert(route_id.clone(), managed);

        self.endpoint_index.insert(&from_uri, &route_id);
        Ok(())
    }

    pub async fn remove_route_preserving_functions(
        &mut self,
        route_id: &str,
    ) -> Result<(), CamelError> {
        let managed = self.routes.get(route_id).ok_or_else(|| {
            CamelError::RouteError(format!("Route '{}' not found for removal", route_id))
        })?;
        if handle_is_running(&managed.consumer_handle)
            || handle_is_running(&managed.pipeline_handle)
        {
            return Err(CamelError::RouteError(format!(
                "Route '{}' must be stopped before removal (current execution lifecycle: {})",
                route_id,
                inferred_lifecycle_label(managed)
            )));
        }
        self.routes.remove(route_id);
        if let Some(reg) = &self.health_registry {
            reg.unregister_for_route(route_id);
        }
        self.endpoint_index.remove(route_id);
        debug!(route_id = %route_id, "Route removed from controller (functions preserved for reload finalize)");
        Ok(())
    }

    /// Compile a route definition into a processor pipeline, without adding it
    /// to the controller. Used for validation and testing.
    pub fn compile_route_definition(
        &self,
        def: RouteDefinition,
    ) -> Result<BoxProcessor, CamelError> {
        self.route_compiler_ext().compile_route_definition(def)
    }

    /// Compile a route definition with a specific generation (for hot-reload).
    pub fn compile_route_definition_with_generation(
        &self,
        def: RouteDefinition,
        generation: u64,
    ) -> Result<BoxProcessor, CamelError> {
        self.route_compiler_ext()
            .compile_route_definition_with_generation(def, generation)
    }

    /// Compile a route definition into a [`CompiledPipeline`] (processor +
    /// lifecycle handles). Used by the hot-reload Restart path so that
    /// lifecycle handles are threaded through
    /// [`swap_pipeline_raw`](Self::swap_pipeline_raw).
    pub(crate) fn compile_route_definition_pipeline(
        &self,
        def: RouteDefinition,
        generation: u64,
    ) -> Result<CompiledPipeline, CamelError> {
        self.route_compiler_ext()
            .compile_route_definition_pipeline(def, generation)
    }

    /// Compile without function generation, returning full [`CompiledPipeline`].
    ///
    /// Oracle Fix 1: used by the stateless hot-reload path so that
    /// lifecycle-bearing routes have their handles preserved.
    pub(crate) fn compile_route_definition_dry_pipeline(
        &self,
        def: RouteDefinition,
    ) -> Result<CompiledPipeline, CamelError> {
        self.route_compiler_ext()
            .compile_route_definition_dry_pipeline(def)
    }

    /// Remove a route from the controller map.
    ///
    /// The route **must** be stopped before removal (status `Stopped` or `Failed`).
    /// Returns an error if the route is still running or does not exist.
    /// Does not cancel any running tasks — call `stop_route` first.
    pub async fn remove_route(&mut self, route_id: &str) -> Result<(), CamelError> {
        let managed = self.routes.get(route_id).ok_or_else(|| {
            CamelError::RouteError(format!("Route '{}' not found for removal", route_id))
        })?;
        if handle_is_running(&managed.consumer_handle)
            || handle_is_running(&managed.pipeline_handle)
        {
            return Err(CamelError::RouteError(format!(
                "Route '{}' must be stopped before removal (current execution lifecycle: {})",
                route_id,
                inferred_lifecycle_label(managed)
            )));
        }
        if let Some(invoker) = &self.function_invoker {
            for (id, rid) in self.collect_function_refs(route_id) {
                if let Err(e) = invoker.unregister(&id, rid.as_deref()).await {
                    warn!(route_id = %route_id, error = %e, "Failed to unregister function during route removal");
                }
            }
        }
        self.routes.remove(route_id);
        if let Some(reg) = &self.health_registry {
            reg.unregister_for_route(route_id);
        }
        self.endpoint_index.remove(route_id);
        info!(route_id = %route_id, "Route removed from controller");
        Ok(())
    }

    fn collect_function_refs(
        &self,
        route_id: &str,
    ) -> Vec<(camel_api::FunctionId, Option<String>)> {
        self.function_invoker
            .as_ref()
            .map(|invoker| invoker.function_refs_for_route(route_id))
            .unwrap_or_default()
    }

    fn discard_function_staging(&self) {
        if let Some(invoker) = &self.function_invoker {
            invoker.discard_staging(0);
        }
    }

    /// Returns the number of routes in the controller.
    pub fn route_count(&self) -> usize {
        self.routes.route_count()
    }

    pub fn in_flight_count(&self, route_id: &str) -> Option<u64> {
        self.routes.in_flight_count(route_id)
    }

    /// Returns `true` if a route with the given ID exists.
    pub fn route_exists(&self, route_id: &str) -> bool {
        self.routes.route_exists(route_id)
    }

    /// Returns all route IDs.
    pub fn route_ids(&self) -> Vec<String> {
        self.routes.route_ids()
    }

    pub fn route_source_hash(&self, route_id: &str) -> Option<u64> {
        self.routes.route_source_hash(route_id)
    }

    /// Returns route IDs that should auto-start, sorted by startup order (ascending).
    pub fn auto_startup_route_ids(&self) -> Vec<String> {
        self.routes.auto_startup_route_ids()
    }

    /// Returns route IDs sorted by shutdown order (startup order descending).
    pub fn shutdown_route_ids(&self) -> Vec<String> {
        self.routes.shutdown_route_ids()
    }

    /// Atomically swap the pipeline of a route (zero-downtime).
    ///
    /// In-flight requests finish with the old pipeline (kept alive by Arc).
    /// New requests immediately use the new pipeline.
    ///
    /// ## Rejection policy
    ///
    /// Returns an error if the route has lifecycle-bearing steps or an active
    /// aggregate — these require the **Restart path** (stop → swap → start).
    ///
    /// The caller (e.g. `reload_actions::apply_swap`) MUST catch this rejection
    /// and fall back to:
    /// 1. `stop_route_reload` — drain lifecycle, stop consumer
    /// 2. `swap_pipeline_raw` — bypass the lifecycle check (route is stopped)
    /// 3. `start_route_reload` — re-create consumer with the new pipeline
    ///
    /// This is the "reject, don't defer" policy (oracle Fix 3): the swap is
    /// refused upfront rather than silently deferring or partially swapping.
    pub fn swap_pipeline(
        &self,
        route_id: &str,
        new_pipeline: BoxProcessor,
    ) -> Result<(), CamelError> {
        let managed = self
            .routes
            .get(route_id)
            .ok_or_else(|| CamelError::RouteError(format!("Route '{}' not found", route_id)))?;

        let assembly = managed.pipeline.load();
        let has_lifecycle = !assembly.lifecycle.is_empty();

        if has_lifecycle || managed.agg_service.is_some() {
            warn!(
                route_id = %route_id,
                "Hot-swap rejected — route has lifecycle/agg steps; use Restart path"
            );
            return Err(CamelError::RouteError(format!(
                "Route '{}' contains stateful steps (lifecycle-bearing). Hot-swap not supported — use restart.",
                route_id
            )));
        }

        drop(assembly);

        if managed.aggregate_split.is_some() {
            warn!(
                route_id = %route_id,
                "swap_pipeline: aggregate routes with timeout do not support hot-reload of pre/post segments"
            );
        }

        super::pipeline_runtime::swap_pipeline_raw(&managed.pipeline, new_pipeline, vec![]);
        debug!(route_id = %route_id, "Pipeline swapped atomically");
        Ok(())
    }

    /// Non-checking raw pipeline swap — bypasses lifecycle/aggregate rejection.
    ///
    /// Only for use after the route has been stopped (Restart path).
    /// Does NOT check for lifecycle handles or aggregate service — the caller
    /// is responsible for ensuring the route is safe to swap.
    ///
    /// Accepts `lifecycle` so that the new pipeline assembly records the
    /// lifecycle handles from the compiled steps.  When the route is
    /// subsequently stopped, these handles are drained.
    pub(crate) fn swap_pipeline_raw(
        &self,
        route_id: &str,
        new_pipeline: BoxProcessor,
        lifecycle: Vec<Arc<dyn StepLifecycle>>,
    ) -> Result<(), CamelError> {
        let managed = self
            .routes
            .get(route_id)
            .ok_or_else(|| CamelError::RouteError(format!("Route '{}' not found", route_id)))?;
        super::pipeline_runtime::swap_pipeline_raw(&managed.pipeline, new_pipeline, lifecycle);
        debug!(route_id = %route_id, "Pipeline swapped (raw — lifecycle bypass)");
        Ok(())
    }

    /// Returns the from_uri of a route, if it exists.
    pub fn route_from_uri(&self, route_id: &str) -> Option<String> {
        self.routes.route_from_uri(route_id)
    }

    /// Return all route_ids that consume from the given source endpoint URI.
    pub fn routes_for_endpoint(&self, uri: &str) -> Vec<String> {
        self.endpoint_index.routes_for(uri)
    }

    /// Return all registered source endpoint URIs.
    pub fn list_endpoint_uris(&self) -> Vec<String> {
        self.endpoint_index.list_uris()
    }

    /// Get a clone of the current pipeline for a route.
    ///
    /// This is useful for testing and introspection.
    /// Returns `None` if the route doesn't exist.
    pub fn get_pipeline(&self, route_id: &str) -> Option<BoxProcessor> {
        self.routes.get_pipeline(route_id)
    }

    /// Check whether the running route has lifecycle-bearing steps.
    ///
    /// Returns `false` when the route is missing.
    pub(crate) fn route_has_lifecycle(&self, route_id: &str) -> bool {
        self.routes
            .get(route_id)
            .map(|managed| !managed.pipeline.load().lifecycle.is_empty())
            .unwrap_or(false)
    }

    /// Internal stop implementation that can set custom status.
    pub(super) async fn stop_route_internal(&mut self, route_id: &str) -> Result<(), CamelError> {
        self.routes.stop_route(route_id).await
    }

    /// Internal stop with an explicit graceful drain budget.
    pub(super) async fn stop_route_with_budget_internal(
        &mut self,
        route_id: &str,
        budget: std::time::Duration,
    ) -> Result<(), CamelError> {
        self.routes.stop_route_with_budget(route_id, budget).await
    }

    pub async fn start_route_reload(&mut self, route_id: &str) -> Result<(), CamelError> {
        self.start_route(route_id).await
    }

    pub async fn stop_route_reload(&mut self, route_id: &str) -> Result<(), CamelError> {
        self.stop_route(route_id).await
    }
}

// ── Aggregator route helpers ──

impl DefaultRouteController {
    /// Start a route with an aggregate split (pre-pipeline → aggregator → post-pipeline).
    ///
    /// Spawns a biased-select forward loop that routes exchanges through the
    /// pre-pipeline, aggregator, and post-pipeline in sequence, with late-exchange
    /// handling and force-completion on stop.
    ///
    /// drainclaim claim propagation: the forward loop submits each
    /// envelope's [`InFlightClaim`] into the aggregator WITH its exchange
    /// ([`AggregatorService::submit_with_claim`]) — a pending stash parks
    /// the claim inside the bucket so the stashed exchange stays counted,
    /// and every emission path (sync completion, timeout `late_tx` fire,
    /// `force_complete_all` at stop) hands the bucket's claims back to
    /// this loop, which holds them across the post-pipeline
    /// continuation. Paths that drop a bucket without emitting (TTL
    /// eviction, unarmed-bucket release, discard-on-timeout, saturated
    /// late channel) release by dropping. Exactly one release per claim
    /// on every path. claimfamily (rc-e1a4f): every oneshot dispatch
    /// through the pre/post pipelines splits a sibling claim onto the
    /// exchange (in-band results take it back, stash emissions escape
    /// with theirs), so stash sites embedded in those pipelines stay
    /// counted.
    #[allow(clippy::too_many_arguments)]
    pub(super) async fn start_aggregate_route(
        &mut self,
        route_id: &str,
        split: AggregateSplitInfo,
        consumer: Box<dyn Consumer>,
        consumer_ctx: ConsumerContext,
        mut rx: mpsc::Receiver<ExchangeEnvelope>,
        crash_notifier: Option<mpsc::Sender<CrashNotification>>,
        runtime_for_consumer: Option<Weak<dyn RuntimeHandle>>,
        tx_for_storage: mpsc::Sender<ExchangeEnvelope>,
        // Pipeline cancellation — a child of the managed route's pipeline_cancel_token.
        pipeline_cancel: CancellationToken,
        drain_in_flight: Arc<std::sync::atomic::AtomicU64>,
    ) -> Result<(), CamelError> {
        // drainclaim claim propagation: the late channel carries each
        // emission's stashed claims alongside the aggregated exchange —
        // the loop holds them across the post-pipeline continuation.
        let (late_tx, late_rx) =
            mpsc::channel::<camel_processor::aggregator::AggregateEmission>(256);

        let route_cancel_clone = pipeline_cancel.clone();
        let mut svc = AggregatorService::new(
            split.agg_config.clone(),
            late_tx,
            Arc::clone(&self.languages),
            route_cancel_clone,
        );
        // Queue-depth visibility: the TTL-sweep pass reports the buffered
        // group count as camel_queue_depth{queue="aggregator:<route>"}.
        if let Some(metrics) = self.tracer_metrics.clone() {
            svc = svc.with_queue_metrics(metrics, format!("aggregator:{route_id}"));
        }
        let agg = Arc::new(svc);

        let pipeline_cancel_for_monitor = pipeline_cancel.clone();
        // rc-e2r9: capture route_id and metrics for b′ emission at reply-drop
        // sites inside the spawned task.
        let route_id_for_metrics = route_id.to_string();
        let metrics_for_reply_drop = self.tracer_metrics.clone();
        // rc-jxkj cohort gate: owned by the forward loop — the envelope arm
        // parks dispatch until the startup cohort opens the gate.
        let mut cohort_rx = self.cohort.subscribe();
        let agg_for_monitor = Arc::clone(&agg);

        {
            let managed = self
                .routes
                .get_mut(route_id)
                .expect("invariant: route must exist"); // allow-unwrap
            managed.agg_service = Some(Arc::clone(&agg));
        }

        let late_rx = Arc::new(tokio::sync::Mutex::new(late_rx));
        let pre_pipeline = split.pre_pipeline;
        let post_pipeline = split.post_pipeline;

        // Spawn biased select forward loop
        let pipeline_handle = tokio::spawn(async move {
            loop {
                tokio::select! {
                    biased;

                    // Ungated by design (D3, rc-jxkj): late exchanges exist
                    // only after a dispatched envelope traversed the
                    // aggregator — transitively post-activation. Gating here
                    // would be dead code and a self-deadlock risk.
                    late_ex = async {
                        let mut rx = late_rx.lock().await;
                        rx.recv().await
                    } => {
                        match late_ex {
                            Some(emission) => {
                                // drainclaim: hold the emission's stashed
                                // claims across the post-pipeline — drop
                                // at iteration end is the release.
                                let AggregateEmission {
                                    mut exchange,
                                    claims: _in_flight_claims,
                                } = emission;
                                // claimfamily (rc-e1a4f): split a sibling onto the emission exchange so
                                // residency inside stash sites embedded in the post-pipeline stays
                                // counted after this iteration. An in-band result drops at iteration end
                                // (its sibling releases with it); a stash emission escapes with its
                                // sibling; an Err drops it with the exchange.
                                exchange.in_flight_claim = _in_flight_claims
                                    .iter()
                                    .flatten()
                                    .next()
                                    .map(InFlightClaim::split);
                                let pipe = post_pipeline.load();
                                if let Err(e) =
                                    pipe.processor.clone_inner().oneshot(exchange).await
                                {
                                    tracing::warn!(error = %e, "late exchange post-pipeline failed");
                                }
                            }
                            None => return,
                        }
                    }

                    envelope_opt = rx.recv() => {
                        match envelope_opt {
                            Some(envelope) => {
                                // rc-jxkj cohort gate: park dispatch until
                                // the startup cohort completes (same guard
                                // as the non-aggregate drain loops).
                                // rc-z5qz: `biased` with the gate polled
                                // FIRST — with the cohort open AND the
                                // pipeline token cancelled, both branches
                                // are ready and an unbiased select picks
                                // randomly, letting the cancel arm drop a
                                // deliverable envelope (force_complete_all
                                // then sees no buckets). Gate-open must win
                                // deterministically; a genuinely closed
                                // gate still drops on cancel (rc-jxkj
                                // semantics preserved).
                                tokio::select! {
                                    biased;
                                    _ = cohort_rx.wait_for(|open| *open) => {}
                                    _ = pipeline_cancel.cancelled() => {
                                        // Drop the envelope; reply_tx (if
                                        // any) resolves to ChannelClosed for
                                        // the send_and_wait waiter.
                                        // `continue`, not `return`: the token
                                        // is already cancelled, so the next
                                        // loop iteration lands in the biased
                                        // outer select's cancel arm below,
                                        // which runs the force_complete_all +
                                        // late_rx cleanup. A `return` would
                                        // skip that cleanup and silently
                                        // drop a pending bucket across a
                                        // stop→restart gate re-arm.
                                        continue;
                                    }
                                }
                                let ExchangeEnvelope {
                                    exchange,
                                    reply_tx,
                                    in_flight_claim,
                                } = envelope;
                                let _drain_guard = super::route_helpers::DrainGuard::new(Arc::clone(&drain_in_flight));
                                // drainclaim claim PROPAGATION: the
                                // envelope's claim travels WITH the
                                // exchange into the aggregator — a
                                // pending stash parks it in the bucket
                                // (counted until the bucket completes),
                                // and a sync completion returns it (with
                                // the rest of the bucket's claims) to be
                                // held across the post-pipeline below.
                                // Rejection paths drop it inside the
                                // service — rejected = released.
                                let pre_pipe = pre_pipeline.load();
                                // claimfamily (rc-e1a4f): split a sibling claim onto the exchange so
                                // residency inside stash sites embedded in the pre-pipeline stays
                                // counted after this oneshot resolves. Taken back from an in-band Ok
                                // result below — the envelope's own claim still travels with the
                                // exchange into the aggregator; a stash emission escapes with its
                                // sibling; an Err drops it with the exchange.
                                let mut exchange = exchange;
                                exchange.in_flight_claim = in_flight_claim.as_ref().map(InFlightClaim::split);
                                let ex = match pre_pipe.processor.clone_inner().oneshot(exchange).await {
                                    // claimfamily: in-band completion — reclaim the sibling so release
                                    // stays at this loop iteration.
                                    Ok(mut ex) => {
                                        ex.in_flight_claim = None;
                                        ex
                                    }
                                    Err(e) => {
                                        // rc-e2r9: the real error rides with the
                                        // result so a dropped receiver still gets
                                        // it (ConsumerStopping suppressed).
                                        send_reply_or_b_prime(
                                            reply_tx,
                                            Err(e),
                                            &metrics_for_reply_drop,
                                            &route_id_for_metrics,
                                            "aggregate:pre-pipeline",
                                        );
                                        continue;
                                    }
                                };

                                let AggregationReceipt { reply, claims } =
                                    agg.submit_with_claim(ex, in_flight_claim).await;
                                // drainclaim: hold the completed bucket's
                                // claims across the post-pipeline
                                // continuation; dropped at iteration end or
                                // any `continue`/`return` (scope exit).
                                let _completed_bucket_claims = claims;

                                match reply {
                                    Ok(ex) => {
                                        if !is_pending(&ex) {
                                            let post_pipe = post_pipeline.load();
                                            // claimfamily (rc-e1a4f): split a sibling from the completed
                                            // bucket's claims onto the aggregated output so residency inside
                                            // stash sites embedded in the post-pipeline stays counted after
                                            // this iteration. Taken back from an in-band result before the
                                            // reply — release stays at iteration end; a stash emission
                                            // escapes with its sibling; an Err drops it with the exchange.
                                            let mut ex = ex;
                                            ex.in_flight_claim = _completed_bucket_claims
                                                .iter()
                                                .flatten()
                                                .next()
                                                .map(InFlightClaim::split);
                                            let mut out = post_pipe.processor.clone_inner().oneshot(ex).await;
                                            if let Ok(ref mut out_ex) = out {
                                                out_ex.in_flight_claim = None;
                                            }
                                            // rc-e2r9 review, Important 1: this site
                                            // previously `let _ = send`ed the
                                            // post-pipeline result — an Err with an
                                            // abandoned receiver vanished silently.
                                            // The helper closes that gap.
                                            send_reply_or_b_prime(
                                                reply_tx,
                                                out,
                                                &metrics_for_reply_drop,
                                                &route_id_for_metrics,
                                                "aggregate:post-pipeline",
                                            );
                                        } else {
                                            // Pending Ok: a dropped reply is silent —
                                            // b′ is an ERROR signal.
                                            send_reply_or_b_prime(
                                                reply_tx,
                                                Ok(ex),
                                                &metrics_for_reply_drop,
                                                &route_id_for_metrics,
                                                "aggregate:pending",
                                            );
                                        }
                                    }
                                    Err(e) => {
                                        send_reply_or_b_prime(
                                            reply_tx,
                                            Err(e),
                                            &metrics_for_reply_drop,
                                            &route_id_for_metrics,
                                            "aggregate:pipeline",
                                        );
                                    }
                                }
                            }
                            None => return,
                        }
                    }

                    _ = pipeline_cancel.cancelled() => {
                        agg.force_complete_all();
                        let mut rx_guard = late_rx.lock().await;
                        while let Ok(late_ex) = rx_guard.try_recv() {
                            // drainclaim: each forced emission's claims are
                            // held across its post-pipeline continuation —
                            // dropped after the oneshot completes (or
                            // immediately, which is also a release).
                            let AggregateEmission {
                                mut exchange,
                                claims: _in_flight_claims,
                            } = late_ex;
                            // claimfamily (rc-e1a4f): split a sibling onto the forced emission so
                            // residency inside stash sites embedded in the post-pipeline stays
                            // counted. The discarded result releases an in-band sibling with its
                            // drop; a stash emission escapes with its sibling.
                            exchange.in_flight_claim = _in_flight_claims
                                .iter()
                                .flatten()
                                .next()
                                .map(InFlightClaim::split);
                            let pipe = post_pipeline.load();
                            let _ = pipe.processor.clone_inner().oneshot(exchange).await;
                        }
                        break;
                    }
                }
            }
        });
        #[cfg(test)]
        emit_start_route_event("pipeline_spawned", route_id);

        // Start consumer after pipeline loop is spawned to avoid startup races
        // where consumers emit exchanges before the route pipeline begins polling.
        // rc-kh7c cleanup parity: capture the consumer's cancel token before
        // consumer_ctx moves into the task so the failure arm can stop child
        // tasks spawned by consumer.start().
        let consumer_cancel_for_cleanup = consumer_ctx.cancel_token();
        let (consumer_handle, startup_rx, watcher_inputs, outer_inputs, stop_budget_tx) =
            super::consumer_management::spawn_consumer_task(
                route_id.to_string(),
                consumer,
                consumer_ctx,
                crash_notifier,
                runtime_for_consumer,
                false,
            );

        // rc-w1u9: await consumer startup handshake for aggregate routes too
        // so bind failures surface as route-start errors.
        // For Immediate consumers the receiver is pre-resolved (rc-slvd).
        let startup_result =
            super::consumer_management::await_consumer_startup(startup_rx, "startup").await;
        // rc-kh7c: on failure, abort the orphaned consumer task and cancel the
        // pipeline so neither runs detached. The aggregate pipeline loop would
        // eventually self-clean via rx-drop + late_tx-drop, but cancelling
        // pipeline_cancel also triggers force_complete_all (aggregate cleanup).
        if let Err(e) = startup_result {
            consumer_handle.abort();
            pipeline_cancel_for_monitor.cancel();
            // Deliberate Explicit-failure cleanup parity with the trait start arm (rc-kh7c).
            consumer_cancel_for_cleanup.cancel();
            return Err(e);
        }

        // Detached failure watcher for Immediate consumers (rc-slvd).
        if let Some(inputs) = watcher_inputs {
            super::consumer_management::spawn_failure_watcher(inputs);
        }

        // Detached outer-task watcher for Explicit consumers (rc-a7rh):
        // spawned only after the handshake resolved Ok — rollback
        // terminations (abort-then-cancel above) happen before this point
        // and are never watched. Explicit consumers on aggregate routes
        // get identical coverage — no aggregate carve-out.
        if let Some(outer) = outer_inputs {
            super::consumer_management::spawn_outer_task_watcher(outer);
        }

        // Extend the stored consumer handle through aggregate force-completion.
        // While this monitor drains pending buckets, handle_is_running still reports
        // the Route as running because forced exchanges may still be in post-pipeline.
        //
        // bd rc-iioeq: a natural consumer exit (e.g. timer repeatCount
        // exhausted) must NOT destroy buckets whose inactivity timeout is
        // armed. With `force_completion_on_stop=false` (the default),
        // `force_complete_all` CANCELS the armed timeout task and silently
        // discards the bucket, so the inactivity emission never happens.
        // The forward loop stays up (the stored channel sender keeps the
        // input channel open), so an armed bucket still emits downstream
        // when its timeout fires. Buckets with no armed timeout task —
        // size/predicate-only, or timeout-configured but over the
        // `max_timeout_tasks` cap — can never complete after the consumer
        // exits; `release_unarmed_buckets` discards them eagerly so they
        // are not orphaned (the bucket_ttl sweep only runs inside the
        // pipeline's next exchange, which never arrives).
        let force_on_stop = agg_for_monitor.config().force_completion_on_stop;
        let consumer_handle = tokio::spawn(async move {
            let _ = consumer_handle.await;
            if !pipeline_cancel_for_monitor.is_cancelled() {
                if force_on_stop {
                    agg_for_monitor.force_complete_all();
                    pipeline_cancel_for_monitor.cancel();
                } else {
                    agg_for_monitor.release_unarmed_buckets();
                }
            }
        });
        #[cfg(test)]
        emit_start_route_event("consumer_spawned", route_id);

        {
            let managed = self
                .routes
                .get_mut(route_id)
                .expect("invariant: route must exist"); // allow-unwrap
            managed.consumer_handle = Some(consumer_handle);
            managed.pipeline_handle = Some(pipeline_handle);
            managed.channel_sender = Some(tx_for_storage);
            managed.stop_budget = Some(stop_budget_tx);
        }

        info!(route_id = %route_id, "Route started (aggregate with timeout)");
        Ok(())
    }

    /// Test-only: inject lifecycle handles into an existing route's pipeline
    /// assembly.  This makes the route lifecycle-bearing so that swap_pipeline
    /// rejects it, forcing callers (like reload_actions::apply_swap) to take
    /// the Restart path instead.
    #[cfg(test)]
    pub(crate) fn set_route_lifecycle_for_test(
        &mut self,
        route_id: &str,
        lifecycle: Vec<Arc<dyn StepLifecycle>>,
    ) -> Result<(), CamelError> {
        use super::pipeline_runtime::PipelineAssembly;
        use camel_api::SyncBoxProcessor;
        use std::sync::Arc;

        let managed = self
            .routes
            .get_mut(route_id)
            .ok_or_else(|| CamelError::RouteError(format!("Route '{}' not found", route_id)))?;
        let old_processor = managed.pipeline.load().processor.clone_inner();
        managed.pipeline.store(Arc::new(PipelineAssembly::new(
            SyncBoxProcessor::new(old_processor),
            lifecycle,
        )));
        Ok(())
    }
}

// ── rc-e2r9: b′ signal emission on reply-drop ──
// The shared `send_reply_or_b_prime` helper lives in
// `super::route_helpers` — it is used by both this module (aggregate drain
// loop) and `route_controller_trait` (Concurrent/Sequential pipelines).

#[cfg(test)]
impl crate::hot_reload::ports::ReloadIntrospectionPort for DefaultRouteController {
    fn route_ids(&self) -> Vec<String> {
        self.route_ids() // inherent pub fn
    }
    fn route_from_uri(&self, route_id: &str) -> Option<String> {
        self.route_from_uri(route_id) // inherent pub fn
    }
    fn route_source_hash(&self, route_id: &str) -> Option<u64> {
        self.route_source_hash(route_id) // inherent pub fn
    }
}

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

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